branch.py
python
sha256:e237dc0e8122609f5131d11c9dda9bba480395a5a4355cda0c9fa7e634fddd29
fix(branch): guard -d --dry-run against destructive writes;…
Sonnet 4.6
patch
101 days ago
| 1 | """``muse branch`` — list, create, rename, copy, and delete branches. |
| 2 | |
| 3 | Git-idiomatic flags:: |
| 4 | |
| 5 | muse branch # list all local branches |
| 6 | muse branch <name> # create branch at HEAD |
| 7 | muse branch <name> <start-point> # create at commit SHA, SHA prefix, or branch |
| 8 | muse branch -d <name> # safe delete (must be merged) |
| 9 | muse branch -D <name> # force delete |
| 10 | muse branch -dr <remote>/<branch> # delete local remote-tracking ref (no server call) |
| 11 | muse branch -Dr <remote>/<branch> # same, force (no merge check) |
| 12 | muse branch -m [<old>] <new> # rename (safe) |
| 13 | muse branch -M [<old>] <new> # rename (force) |
| 14 | muse branch -c [<src>] <dest> # copy (safe) |
| 15 | muse branch -C [<src>] <dest> # copy (force) |
| 16 | muse branch -v # list with last commit SHA + subject |
| 17 | muse branch -vv # also show upstream tracking ref |
| 18 | muse branch -r # list remote-tracking branches |
| 19 | muse branch -a # list local + remote-tracking branches |
| 20 | muse branch --merged [<commit>] # only branches merged into commit |
| 21 | muse branch --no-merged [<commit>] # only branches NOT merged into commit |
| 22 | muse branch --contains <commit> # only branches that contain commit |
| 23 | muse branch --sort name # sort by name (default) |
| 24 | muse branch --sort committeddate # sort by date of most recent commit |
| 25 | |
| 26 | To delete a branch on the remote **and** prune the local tracking ref in one |
| 27 | step, use ``muse push``:: |
| 28 | |
| 29 | muse push <remote> --delete <branch> |
| 30 | |
| 31 | Agents should pass ``--format json`` (or ``--json``) for machine-readable |
| 32 | output on all operations. The listing JSON schema is:: |
| 33 | |
| 34 | [ |
| 35 | { |
| 36 | "name": "feat/my-thing", |
| 37 | "current": false, |
| 38 | "commit_id": "<sha256> | null", |
| 39 | "committed_at": "2026-03-21T12:00:00+00:00 | null", |
| 40 | "last_message": "Add feature X", |
| 41 | "upstream": "origin/feat/my-thing" |
| 42 | }, |
| 43 | ... |
| 44 | ] |
| 45 | |
| 46 | Exit codes:: |
| 47 | |
| 48 | 0 — success |
| 49 | 1 — invalid branch name, branch not found, attempting to delete checked-out branch |
| 50 | """ |
| 51 | |
| 52 | import argparse |
| 53 | import json |
| 54 | import logging |
| 55 | import pathlib |
| 56 | import re |
| 57 | import sys |
| 58 | import tomllib |
| 59 | from typing import TypedDict |
| 60 | |
| 61 | from muse.cli.config import delete_branch_meta, get_protected_branches, get_remote_head, is_branch_protected, read_branch_meta, write_branch_meta |
| 62 | from muse.core.reflog import append_reflog |
| 63 | from muse.core.types import MsgpackDict, short_id |
| 64 | from muse.core.paths import ref_path as _ref_path, heads_dir as _heads_dir, remotes_dir as _remotes_dir, config_toml_path as _config_toml_path, reflog_branch_path as _reflog_branch_path, reflog_heads_dir as _reflog_heads_dir |
| 65 | from muse.core.envelope import EnvelopeJson, make_envelope |
| 66 | from muse.core.timing import start_timer |
| 67 | from muse.core.errors import ExitCode |
| 68 | from muse.core.repo import require_repo |
| 69 | from muse.core.refs import read_ref |
| 70 | from muse.core.io import write_text_atomic |
| 71 | from muse.core.refs import ( |
| 72 | get_head_commit_id, |
| 73 | read_current_branch, |
| 74 | write_branch_ref, |
| 75 | write_head_branch, |
| 76 | ) |
| 77 | from muse.core.commits import ( |
| 78 | read_commit, |
| 79 | resolve_commit_ref, |
| 80 | ) |
| 81 | from muse.core.validation import clamp_int, sanitize_display, validate_branch_name |
| 82 | |
| 83 | type _Payload = dict[str, str | None] |
| 84 | logger = logging.getLogger(__name__) |
| 85 | |
| 86 | class _BranchCreateJson(EnvelopeJson): |
| 87 | """JSON output for ``muse branch -b <name> --json``.""" |
| 88 | |
| 89 | action: str |
| 90 | branch: str |
| 91 | commit_id: str | None |
| 92 | intent: str | None |
| 93 | resumable: bool |
| 94 | |
| 95 | class _BranchEntryJson(TypedDict): |
| 96 | name: str |
| 97 | current: bool |
| 98 | commit_id: str | None |
| 99 | committed_at: str | None |
| 100 | last_message: str | None |
| 101 | upstream: str | None |
| 102 | intent: str | None |
| 103 | resumable: bool |
| 104 | created_by: str | None |
| 105 | |
| 106 | class _BranchListJson(EnvelopeJson): |
| 107 | """JSON output for ``muse branch --json``.""" |
| 108 | |
| 109 | branches: list[_BranchEntryJson] |
| 110 | |
| 111 | |
| 112 | class _PruneConfigJson(TypedDict): |
| 113 | """JSON output for ``muse branch --prune-config``.""" |
| 114 | |
| 115 | action: str |
| 116 | pruned: int |
| 117 | kept: int |
| 118 | dry_run: bool |
| 119 | pruned_branches: list[str] |
| 120 | |
| 121 | # --------------------------------------------------------------------------- |
| 122 | # ANSI helpers — emitted only when stdout is a TTY. |
| 123 | # --------------------------------------------------------------------------- |
| 124 | |
| 125 | _RESET = "\033[0m" |
| 126 | _BOLD = "\033[1m" |
| 127 | _DIM = "\033[2m" |
| 128 | _GREEN = "\033[32m" |
| 129 | _RED = "\033[31m" |
| 130 | _YELLOW = "\033[33m" |
| 131 | _CYAN = "\033[36m" |
| 132 | |
| 133 | def _c(text: str, *codes: str, tty: bool) -> str: |
| 134 | """Wrap *text* in ANSI escape *codes* only when writing to a TTY.""" |
| 135 | if not tty: |
| 136 | return text |
| 137 | return "".join(codes) + text + _RESET |
| 138 | |
| 139 | # --------------------------------------------------------------------------- |
| 140 | # Internal helpers |
| 141 | # --------------------------------------------------------------------------- |
| 142 | |
| 143 | def _ref_file(root: pathlib.Path, branch: str) -> pathlib.Path: |
| 144 | """Return the ref-file path for a local branch.""" |
| 145 | return _ref_path(root, branch) |
| 146 | |
| 147 | def _list_local_branches(root: pathlib.Path) -> list[str]: |
| 148 | """Return a sorted list of all local branch names. |
| 149 | |
| 150 | Only plain files are considered; directories, symlinks and any file not |
| 151 | directly under ``refs/heads/`` (e.g. lock files) are silently skipped. |
| 152 | """ |
| 153 | heads_dir = _heads_dir(root) |
| 154 | if not heads_dir.exists(): |
| 155 | return [] |
| 156 | return sorted( |
| 157 | p.relative_to(heads_dir).as_posix() |
| 158 | for p in heads_dir.rglob("*") |
| 159 | if p.is_file() and not p.name.startswith(".") |
| 160 | ) |
| 161 | |
| 162 | def _list_remotes(root: pathlib.Path) -> list[str]: |
| 163 | """Return sorted remote-tracking branch names as ``remote/branch``. |
| 164 | |
| 165 | Only plain files are visited; symlinks, hidden files, and directories |
| 166 | are skipped to avoid leaking internal artefacts into the listing. |
| 167 | """ |
| 168 | remotes_dir = _remotes_dir(root) |
| 169 | if not remotes_dir.exists(): |
| 170 | return [] |
| 171 | results: list[str] = [] |
| 172 | for remote_dir in sorted(remotes_dir.iterdir()): |
| 173 | if not remote_dir.is_dir(): |
| 174 | continue |
| 175 | remote = remote_dir.name |
| 176 | for ref_file in sorted(remote_dir.rglob("*")): |
| 177 | if ref_file.is_file() and not ref_file.name.startswith("."): |
| 178 | branch_rel = ref_file.relative_to(remote_dir).as_posix() |
| 179 | results.append(f"{remote}/{branch_rel}") |
| 180 | return results |
| 181 | |
| 182 | def _resolve_commit_id(root: pathlib.Path, b: str) -> str: |
| 183 | """Return the current commit ID for a branch listing entry. |
| 184 | |
| 185 | *b* is the display name (e.g. ``"main"`` or ``"remotes/origin/dev"``). |
| 186 | Remote entries are read from the remote tracking file under |
| 187 | ``.muse/remotes/``; local entries use the standard head ref. |
| 188 | """ |
| 189 | if b.startswith("remotes/"): |
| 190 | rest = b.removeprefix("remotes/") |
| 191 | remote, _, branch_name = rest.partition("/") |
| 192 | if branch_name: |
| 193 | return get_remote_head(remote, branch_name, root) or "" |
| 194 | return get_head_commit_id(root, b) or "" |
| 195 | |
| 196 | def _upstream_for(root: pathlib.Path, branch: str) -> str | None: |
| 197 | """Return the upstream tracking ref for *branch*, or ``None`` if unset.""" |
| 198 | config_path = _config_toml_path(root) |
| 199 | if not config_path.exists(): |
| 200 | return None |
| 201 | try: |
| 202 | with config_path.open("rb") as f: |
| 203 | config = tomllib.load(f) |
| 204 | section = config.get("branch", {}).get(branch, {}) |
| 205 | remote: str | None = section.get("remote") |
| 206 | merge_ref: str | None = section.get("merge") |
| 207 | if remote and merge_ref: |
| 208 | short = merge_ref.removeprefix("refs/heads/") |
| 209 | return f"{remote}/{short}" |
| 210 | except Exception: |
| 211 | pass |
| 212 | return None |
| 213 | |
| 214 | def _commit_ancestors(root: pathlib.Path, commit_id: str) -> set[str]: |
| 215 | """Return the set of all commit IDs reachable from *commit_id* (inclusive).""" |
| 216 | from muse.core.graph import ancestor_ids |
| 217 | return ancestor_ids(root, commit_id) |
| 218 | |
| 219 | def _is_merged(root: pathlib.Path, branch: str, into: str) -> bool: |
| 220 | """Return ``True`` if the tip of *branch* is an ancestor of the tip of *into*.""" |
| 221 | branch_tip = get_head_commit_id(root, branch) |
| 222 | into_tip = get_head_commit_id(root, into) |
| 223 | if branch_tip is None or into_tip is None: |
| 224 | return False |
| 225 | return branch_tip in _commit_ancestors(root, into_tip) |
| 226 | |
| 227 | def _contains_commit(root: pathlib.Path, branch: str, commit_id: str) -> bool: |
| 228 | """Return ``True`` if *commit_id* is reachable from the tip of *branch*.""" |
| 229 | tip = get_head_commit_id(root, branch) |
| 230 | if tip is None: |
| 231 | return False |
| 232 | return commit_id in _commit_ancestors(root, tip) |
| 233 | |
| 234 | def _cleanup_empty_dirs(ref_file: pathlib.Path, heads_dir: pathlib.Path) -> None: |
| 235 | """Remove any empty parent directories left behind after unlinking *ref_file*.""" |
| 236 | for parent in ref_file.parents: |
| 237 | if parent == heads_dir: |
| 238 | break |
| 239 | try: |
| 240 | parent.rmdir() |
| 241 | except OSError: |
| 242 | break |
| 243 | |
| 244 | def _resolve_start_point(root: pathlib.Path, current: str, start_point: str) -> str: |
| 245 | """Resolve *start_point* to a full commit ID. |
| 246 | |
| 247 | Accepts branch names, full SHA-256 commit IDs, and abbreviated SHA |
| 248 | prefixes (any unambiguous prefix works). Returns the raw *start_point* |
| 249 | string unchanged if resolution fails — the caller is responsible for |
| 250 | surfacing a meaningful error in that case. |
| 251 | """ |
| 252 | # Try as branch name first — skip if it contains characters forbidden in |
| 253 | # branch names (e.g. ':' in sha256:-prefixed IDs) to avoid ValueError. |
| 254 | try: |
| 255 | branch_tip = get_head_commit_id(root, start_point) |
| 256 | if branch_tip is not None: |
| 257 | return branch_tip |
| 258 | except ValueError: |
| 259 | pass # Not a valid branch name — fall through to SHA resolution. |
| 260 | # Fall back to SHA / SHA-prefix resolution. |
| 261 | # resolve_commit_ref handles both bare hex and sha256:-prefixed IDs. |
| 262 | rec = resolve_commit_ref(root, current, start_point) |
| 263 | if rec is not None: |
| 264 | return rec.commit_id |
| 265 | # Return as-is; the caller's write_branch_ref will expose the invalid ID. |
| 266 | return start_point |
| 267 | |
| 268 | # --------------------------------------------------------------------------- |
| 269 | # CLI registration |
| 270 | # --------------------------------------------------------------------------- |
| 271 | |
| 272 | def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: |
| 273 | """Register the ``muse branch`` subcommand and all its flags.""" |
| 274 | parser = subparsers.add_parser( |
| 275 | "branch", |
| 276 | help="List, create, rename, copy, or delete branches.", |
| 277 | description=__doc__, |
| 278 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 279 | ) |
| 280 | parser.add_argument("args", nargs="*", help="Branch name(s) — context-sensitive.") |
| 281 | |
| 282 | # Mutually exclusive operation flags (mirrors git branch). |
| 283 | ops = parser.add_mutually_exclusive_group() |
| 284 | ops.add_argument( |
| 285 | "-d", "--delete", dest="op", action="store_const", const="delete", |
| 286 | help="Delete a branch (safe — must be fully merged).", |
| 287 | ) |
| 288 | ops.add_argument( |
| 289 | "-D", dest="op", action="store_const", const="force_delete", |
| 290 | help="Force-delete a branch regardless of merge status.", |
| 291 | ) |
| 292 | ops.add_argument( |
| 293 | "-m", "--move", dest="op", action="store_const", const="rename", |
| 294 | help="Rename a branch (safe).", |
| 295 | ) |
| 296 | ops.add_argument( |
| 297 | "-M", dest="op", action="store_const", const="force_rename", |
| 298 | help="Force-rename a branch.", |
| 299 | ) |
| 300 | ops.add_argument( |
| 301 | "-c", "--copy", dest="op", action="store_const", const="copy", |
| 302 | help="Copy a branch (safe).", |
| 303 | ) |
| 304 | ops.add_argument( |
| 305 | "-C", dest="op", action="store_const", const="force_copy", |
| 306 | help="Force-copy a branch.", |
| 307 | ) |
| 308 | |
| 309 | # Listing modifiers. |
| 310 | parser.add_argument( |
| 311 | "-v", action="count", default=0, dest="verbose", |
| 312 | help="Show last commit SHA + subject. Repeat (-vv) to also show upstream.", |
| 313 | ) |
| 314 | parser.add_argument( |
| 315 | "-r", "--remotes", action="store_true", |
| 316 | help="List remote-tracking branches.", |
| 317 | ) |
| 318 | parser.add_argument( |
| 319 | "-a", "--all", action="store_true", dest="all_branches", |
| 320 | help="List both local and remote-tracking branches.", |
| 321 | ) |
| 322 | parser.add_argument( |
| 323 | "--merged", metavar="COMMIT", nargs="?", const="HEAD", |
| 324 | help="Only list branches merged into COMMIT (default HEAD).", |
| 325 | ) |
| 326 | parser.add_argument( |
| 327 | "--no-merged", metavar="COMMIT", nargs="?", const="HEAD", |
| 328 | help="Only list branches NOT merged into COMMIT (default HEAD).", |
| 329 | ) |
| 330 | parser.add_argument( |
| 331 | "--contains", metavar="COMMIT", |
| 332 | help="Only list branches that contain COMMIT.", |
| 333 | ) |
| 334 | parser.add_argument( |
| 335 | "--sort", default="name", metavar="KEY", |
| 336 | choices=["name", "committeddate"], |
| 337 | help="Sort branches by 'name' (default) or 'committeddate'.", |
| 338 | ) |
| 339 | parser.add_argument( |
| 340 | "--intent", default=None, metavar="TEXT", |
| 341 | help="Short description of what this branch is for (stored in config).", |
| 342 | ) |
| 343 | parser.add_argument( |
| 344 | "--resumable", action="store_true", default=False, |
| 345 | help=( |
| 346 | "On create: mark this branch as a resumable agent checkpoint. " |
| 347 | "On list (no name): filter to resumable branches only." |
| 348 | ), |
| 349 | ) |
| 350 | parser.add_argument( |
| 351 | "--prune-config", action="store_true", dest="prune_config", |
| 352 | help=( |
| 353 | "Remove stale [branch.*] entries from .muse/config.toml — " |
| 354 | "entries for branches whose ref no longer exists. " |
| 355 | "Use --dry-run to preview without writing." |
| 356 | ), |
| 357 | ) |
| 358 | parser.add_argument( |
| 359 | "--dry-run", action="store_true", dest="dry_run", |
| 360 | help="With --prune-config: report what would be removed without writing.", |
| 361 | ) |
| 362 | parser.add_argument( |
| 363 | "--json", "-j", action="store_true", dest="json_out", |
| 364 | help="Emit machine-readable JSON.", |
| 365 | ) |
| 366 | parser.set_defaults(func=run, op=None, prune_config=False, dry_run=False) |
| 367 | |
| 368 | # --------------------------------------------------------------------------- |
| 369 | # Command handler |
| 370 | # --------------------------------------------------------------------------- |
| 371 | |
| 372 | _SHA256_FULL_RE = re.compile(r"^sha256:[0-9a-f]{64}$") |
| 373 | |
| 374 | |
| 375 | def _resolve_into_tip(root: pathlib.Path, into: str, current: str) -> str | None: |
| 376 | """Resolve the ``--merged``/``--no-merged`` argument to a commit ID. |
| 377 | |
| 378 | *into* may be ``"HEAD"``, a branch name, or an already-resolved |
| 379 | ``sha256:``-prefixed commit ID (e.g. the tip of a remote-tracking ref |
| 380 | a caller looked up separately). A well-formed commit ID is used |
| 381 | directly, skipping the branch-name lookup and its ``validate_branch_name`` |
| 382 | check entirely — that check rejects the ``:`` character a commit ID |
| 383 | always contains. |
| 384 | """ |
| 385 | resolved = current if into == "HEAD" else into |
| 386 | if _SHA256_FULL_RE.match(resolved): |
| 387 | return resolved |
| 388 | return get_head_commit_id(root, resolved) |
| 389 | |
| 390 | |
| 391 | def run(args: argparse.Namespace) -> None: |
| 392 | """List, create, rename, copy, or delete branches. |
| 393 | |
| 394 | Without a subcommand flag, lists all local branches. With ``--format json`` |
| 395 | the output is a stable JSON array; mutation ops (create, rename, copy, |
| 396 | delete) emit a single result object with an ``"action"`` key. |
| 397 | |
| 398 | Agent quickstart |
| 399 | ---------------- |
| 400 | :: |
| 401 | |
| 402 | muse branch --json # list all branches |
| 403 | muse branch --json --resumable # list resumable branches only |
| 404 | muse branch -b feat/thing --json # create branch |
| 405 | muse branch -d feat/thing --json # delete branch |
| 406 | |
| 407 | JSON fields (list mode — top-level is a bare array) |
| 408 | ---------------------------------------------------- |
| 409 | name Branch name. |
| 410 | current ``true`` for the currently checked-out branch. |
| 411 | commit_id Full ``sha256:…`` commit ID at the tip. |
| 412 | last_message Commit message at the tip. |
| 413 | upstream Upstream tracking ref; ``null`` if none. |
| 414 | intent Branch intent annotation (``--intent`` flag). |
| 415 | resumable ``true`` if the branch was created with ``--resumable``. |
| 416 | |
| 417 | JSON fields (mutation mode) |
| 418 | --------------------------- |
| 419 | action What was done: ``"created"``, ``"deleted"``, ``"renamed"``, etc. |
| 420 | name Branch name acted upon. |
| 421 | |
| 422 | Exit codes |
| 423 | ---------- |
| 424 | 0 Success. |
| 425 | 1 Invalid arguments, branch not found, or operation conflicts. |
| 426 | 2 Not inside a Muse repository. |
| 427 | """ |
| 428 | elapsed = start_timer() |
| 429 | positional: list[str] = args.args |
| 430 | op: str | None = args.op |
| 431 | verbose: int = clamp_int(args.verbose, 0, 4, 'verbose') |
| 432 | remotes_only: bool = args.remotes |
| 433 | all_branches: bool = args.all_branches |
| 434 | merged_into: str | None = args.merged |
| 435 | not_merged_into: str | None = args.no_merged |
| 436 | contains_commit: str | None = args.contains |
| 437 | sort_key: str = args.sort |
| 438 | intent: str | None = args.intent |
| 439 | resumable: bool = args.resumable |
| 440 | json_out: bool = args.json_out |
| 441 | tty: bool = sys.stdout.isatty() |
| 442 | |
| 443 | root = require_repo() |
| 444 | current = read_current_branch(root) |
| 445 | heads_dir = _heads_dir(root) |
| 446 | |
| 447 | # ------------------------------------------------------------------ |
| 448 | # PRUNE-CONFIG — remove stale [branch.*] entries from config.toml |
| 449 | # ------------------------------------------------------------------ |
| 450 | if args.prune_config: |
| 451 | config_path = _config_toml_path(root) |
| 452 | config: MsgpackDict = {} |
| 453 | if config_path.exists(): |
| 454 | import tomllib as _tomllib |
| 455 | config = _tomllib.loads(config_path.read_text()) |
| 456 | branch_sections: MsgpackDict = dict(config.get("branch") or {}) |
| 457 | pruned: list[str] = [] |
| 458 | kept: list[str] = [] |
| 459 | for bname in sorted(branch_sections): |
| 460 | ref_file = _heads_dir(root) / bname |
| 461 | if ref_file.exists(): |
| 462 | kept.append(bname) |
| 463 | else: |
| 464 | pruned.append(bname) |
| 465 | if not args.dry_run: |
| 466 | delete_branch_meta(root, bname) |
| 467 | result: _PruneConfigJson = { |
| 468 | "action": "prune_config", |
| 469 | "pruned": len(pruned), |
| 470 | "kept": len(kept), |
| 471 | "dry_run": args.dry_run, |
| 472 | "pruned_branches": pruned, |
| 473 | } |
| 474 | if json_out: |
| 475 | print(json.dumps(result)) |
| 476 | else: |
| 477 | prefix = "[dry-run] " if args.dry_run else "" |
| 478 | print(f"{prefix}Pruned {len(pruned)} stale config entries, kept {len(kept)} live entries.") |
| 479 | for b in pruned: |
| 480 | print(f" - {b}") |
| 481 | return |
| 482 | |
| 483 | # ------------------------------------------------------------------ |
| 484 | # DELETE / FORCE-DELETE |
| 485 | # Supports two modes: |
| 486 | # muse branch -d|-D <local-branch> — delete a local branch |
| 487 | # muse branch -d|-D -r <remote>/<branch> — prune a remote-tracking ref |
| 488 | # ------------------------------------------------------------------ |
| 489 | if op in ("delete", "force_delete"): |
| 490 | if not positional: |
| 491 | if json_out: |
| 492 | print(json.dumps({"error": "usage", "message": "muse branch -d|-D [-r] <branch> …"})) |
| 493 | print("❌ Usage: muse branch -d|-D [-r] <branch> …", file=sys.stderr) |
| 494 | raise SystemExit(ExitCode.USER_ERROR) |
| 495 | |
| 496 | # -r flag: delete local remote-tracking refs (no server call). |
| 497 | if remotes_only: |
| 498 | from muse.cli.config import delete_remote_head |
| 499 | for spec in positional: |
| 500 | # Accept both "remote/branch" and "remotes/remote/branch" spellings. |
| 501 | clean = spec.removeprefix("remotes/") |
| 502 | slash = clean.find("/") |
| 503 | if slash == -1: |
| 504 | if json_out: |
| 505 | print(json.dumps({"error": "invalid_ref", "ref": spec, "message": "remote-tracking ref must be '<remote>/<branch>'"})) |
| 506 | print( |
| 507 | f"❌ Remote-tracking ref must be '<remote>/<branch>', got " |
| 508 | f"'{sanitize_display(spec)}'.", |
| 509 | file=sys.stderr, |
| 510 | ) |
| 511 | raise SystemExit(ExitCode.USER_ERROR) |
| 512 | remote_name = clean[:slash] |
| 513 | branch_name = clean[slash + 1:] |
| 514 | removed = delete_remote_head(remote_name, branch_name, root) |
| 515 | if not removed: |
| 516 | if json_out: |
| 517 | print(json.dumps({"error": "not_found", "ref": clean, "message": f"remote-tracking ref '{clean}' not found"})) |
| 518 | print( |
| 519 | f"❌ Remote-tracking ref '{sanitize_display(clean)}' not found.", |
| 520 | file=sys.stderr, |
| 521 | ) |
| 522 | raise SystemExit(ExitCode.USER_ERROR) |
| 523 | if json_out: |
| 524 | print(json.dumps({ |
| 525 | "action": "deleted_remote_tracking", |
| 526 | "remote": remote_name, |
| 527 | "branch": branch_name, |
| 528 | })) |
| 529 | else: |
| 530 | print( |
| 531 | f"Deleted remote-tracking ref " |
| 532 | f"{_c(sanitize_display(clean), _RED, tty=tty)}." |
| 533 | ) |
| 534 | return |
| 535 | |
| 536 | force = op == "force_delete" |
| 537 | for branch_name in positional: |
| 538 | try: |
| 539 | validate_branch_name(branch_name) |
| 540 | except ValueError as exc: |
| 541 | if json_out: |
| 542 | print(json.dumps({"error": "invalid_branch_name", "branch": branch_name, "message": str(exc)})) |
| 543 | print(f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr) |
| 544 | raise SystemExit(ExitCode.USER_ERROR) |
| 545 | if branch_name == current: |
| 546 | if json_out: |
| 547 | print(json.dumps({"error": "current_branch", "branch": branch_name, "message": f"cannot delete the currently checked-out branch '{branch_name}'"})) |
| 548 | print( |
| 549 | f"❌ Cannot delete the currently checked-out branch " |
| 550 | f"'{sanitize_display(branch_name)}'.", |
| 551 | file=sys.stderr, |
| 552 | ) |
| 553 | raise SystemExit(ExitCode.USER_ERROR) |
| 554 | rf = _ref_file(root, branch_name) |
| 555 | if not rf.is_file(): |
| 556 | if json_out: |
| 557 | print(json.dumps({"error": "not_found", "branch": branch_name, "message": f"branch '{branch_name}' not found"})) |
| 558 | print(f"❌ Branch '{sanitize_display(branch_name)}' not found.", file=sys.stderr) |
| 559 | raise SystemExit(ExitCode.USER_ERROR) |
| 560 | if not force and not _is_merged(root, branch_name, current): |
| 561 | if json_out: |
| 562 | print(json.dumps({"error": "not_merged", "branch": branch_name, "message": f"branch '{branch_name}' is not fully merged", "hint": "use -D to force-delete"})) |
| 563 | print( |
| 564 | f"❌ Branch '{sanitize_display(branch_name)}' is not fully merged.\n" |
| 565 | f" Use -D to force-delete.", |
| 566 | file=sys.stderr, |
| 567 | ) |
| 568 | raise SystemExit(ExitCode.USER_ERROR) |
| 569 | protected = get_protected_branches(root) |
| 570 | if is_branch_protected(branch_name, protected): |
| 571 | if json_out: |
| 572 | print(json.dumps({"error": "protected", "branch": branch_name, "message": f"branch '{branch_name}' is protected and cannot be deleted"})) |
| 573 | else: |
| 574 | print( |
| 575 | f"❌ Branch '{sanitize_display(branch_name)}' is protected and cannot be deleted.", |
| 576 | file=sys.stderr, |
| 577 | ) |
| 578 | raise SystemExit(ExitCode.USER_ERROR) |
| 579 | tip = read_ref(rf) or "" |
| 580 | if not args.dry_run: |
| 581 | rf.unlink() |
| 582 | _cleanup_empty_dirs(rf, heads_dir) |
| 583 | reflog_file = _reflog_branch_path(root, branch_name) |
| 584 | reflog_file.unlink(missing_ok=True) |
| 585 | _cleanup_empty_dirs(reflog_file, _reflog_heads_dir(root)) |
| 586 | delete_branch_meta(root, branch_name) |
| 587 | if json_out: |
| 588 | print(json.dumps({"action": "deleted", "branch": branch_name, "was": tip, "dry_run": args.dry_run})) |
| 589 | else: |
| 590 | prefix = "[dry-run] " if args.dry_run else "" |
| 591 | print( |
| 592 | f"{prefix}Deleted branch {_c(sanitize_display(branch_name), _RED, tty=tty)} " |
| 593 | f"({_c('was ' + (tip or 'unknown'), _DIM, tty=tty)})." |
| 594 | ) |
| 595 | return |
| 596 | |
| 597 | # ------------------------------------------------------------------ |
| 598 | # RENAME / FORCE-RENAME |
| 599 | # ------------------------------------------------------------------ |
| 600 | if op in ("rename", "force_rename"): |
| 601 | force = op == "force_rename" |
| 602 | if len(positional) == 1: |
| 603 | old_name, new_name = current, positional[0] |
| 604 | elif len(positional) == 2: |
| 605 | old_name, new_name = positional[0], positional[1] |
| 606 | else: |
| 607 | if json_out: |
| 608 | print(json.dumps({"error": "usage", "message": "muse branch -m|-M [<old>] <new>"})) |
| 609 | print("❌ Usage: muse branch -m|-M [<old>] <new>", file=sys.stderr) |
| 610 | raise SystemExit(ExitCode.USER_ERROR) |
| 611 | for n in (old_name, new_name): |
| 612 | try: |
| 613 | validate_branch_name(n) |
| 614 | except ValueError as exc: |
| 615 | if json_out: |
| 616 | print(json.dumps({"error": "invalid_branch_name", "branch": n, "message": str(exc)})) |
| 617 | print(f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr) |
| 618 | raise SystemExit(ExitCode.USER_ERROR) |
| 619 | src = _ref_file(root, old_name) |
| 620 | dst = _ref_file(root, new_name) |
| 621 | if not src.is_file(): |
| 622 | if json_out: |
| 623 | print(json.dumps({"error": "not_found", "branch": old_name, "message": f"branch '{old_name}' not found"})) |
| 624 | print(f"❌ Branch '{sanitize_display(old_name)}' not found.", file=sys.stderr) |
| 625 | raise SystemExit(ExitCode.USER_ERROR) |
| 626 | if dst.is_file() and not force: |
| 627 | if json_out: |
| 628 | print(json.dumps({"error": "already_exists", "branch": new_name, "message": f"branch '{new_name}' already exists", "hint": "use -M to force"})) |
| 629 | print( |
| 630 | f"❌ Branch '{sanitize_display(new_name)}' already exists. Use -M to force.", |
| 631 | file=sys.stderr, |
| 632 | ) |
| 633 | raise SystemExit(ExitCode.USER_ERROR) |
| 634 | tip = read_ref(src) or "" |
| 635 | if tip: |
| 636 | write_branch_ref(root, new_name, tip) |
| 637 | else: |
| 638 | write_text_atomic(dst, "") |
| 639 | src.unlink() |
| 640 | _cleanup_empty_dirs(src, heads_dir) |
| 641 | if old_name == current: |
| 642 | write_head_branch(root, new_name) |
| 643 | if json_out: |
| 644 | print(json.dumps({"action": "renamed", "from": old_name, "to": new_name})) |
| 645 | else: |
| 646 | print( |
| 647 | f"Renamed branch " |
| 648 | f"{_c(sanitize_display(old_name), _YELLOW, tty=tty)} → " |
| 649 | f"{_c(sanitize_display(new_name), _GREEN, tty=tty)}." |
| 650 | ) |
| 651 | return |
| 652 | |
| 653 | # ------------------------------------------------------------------ |
| 654 | # COPY / FORCE-COPY |
| 655 | # ------------------------------------------------------------------ |
| 656 | if op in ("copy", "force_copy"): |
| 657 | force = op == "force_copy" |
| 658 | if len(positional) == 1: |
| 659 | src_name, dst_name = current, positional[0] |
| 660 | elif len(positional) == 2: |
| 661 | src_name, dst_name = positional[0], positional[1] |
| 662 | else: |
| 663 | if json_out: |
| 664 | print(json.dumps({"error": "usage", "message": "muse branch -c|-C [<src>] <dest>"})) |
| 665 | print("❌ Usage: muse branch -c|-C [<src>] <dest>", file=sys.stderr) |
| 666 | raise SystemExit(ExitCode.USER_ERROR) |
| 667 | for n in (src_name, dst_name): |
| 668 | try: |
| 669 | validate_branch_name(n) |
| 670 | except ValueError as exc: |
| 671 | if json_out: |
| 672 | print(json.dumps({"error": "invalid_branch_name", "branch": n, "message": str(exc)})) |
| 673 | print(f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr) |
| 674 | raise SystemExit(ExitCode.USER_ERROR) |
| 675 | src = _ref_file(root, src_name) |
| 676 | dst = _ref_file(root, dst_name) |
| 677 | if not src.is_file(): |
| 678 | if json_out: |
| 679 | print(json.dumps({"error": "not_found", "branch": src_name, "message": f"branch '{src_name}' not found"})) |
| 680 | print(f"❌ Branch '{sanitize_display(src_name)}' not found.", file=sys.stderr) |
| 681 | raise SystemExit(ExitCode.USER_ERROR) |
| 682 | if dst.is_file() and not force: |
| 683 | if json_out: |
| 684 | print(json.dumps({"error": "already_exists", "branch": dst_name, "message": f"branch '{dst_name}' already exists", "hint": "use -C to force"})) |
| 685 | print( |
| 686 | f"❌ Branch '{sanitize_display(dst_name)}' already exists. Use -C to force.", |
| 687 | file=sys.stderr, |
| 688 | ) |
| 689 | raise SystemExit(ExitCode.USER_ERROR) |
| 690 | tip = read_ref(src) or "" |
| 691 | if tip: |
| 692 | write_branch_ref(root, dst_name, tip) |
| 693 | else: |
| 694 | write_text_atomic(dst, "") |
| 695 | if json_out: |
| 696 | print(json.dumps({"action": "copied", "from": src_name, "to": dst_name})) |
| 697 | else: |
| 698 | print( |
| 699 | f"Copied branch " |
| 700 | f"{_c(sanitize_display(src_name), _YELLOW, tty=tty)} → " |
| 701 | f"{_c(sanitize_display(dst_name), _GREEN, tty=tty)}." |
| 702 | ) |
| 703 | return |
| 704 | |
| 705 | # ------------------------------------------------------------------ |
| 706 | # CREATE |
| 707 | # ------------------------------------------------------------------ |
| 708 | if op is None and positional: |
| 709 | new_name = positional[0] |
| 710 | start_point: str | None = positional[1] if len(positional) > 1 else None |
| 711 | try: |
| 712 | validate_branch_name(new_name) |
| 713 | except ValueError as exc: |
| 714 | if json_out: |
| 715 | print(json.dumps({"error": "invalid_branch_name", "branch": new_name, "message": str(exc)})) |
| 716 | print(f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr) |
| 717 | raise SystemExit(ExitCode.USER_ERROR) |
| 718 | rf = _ref_file(root, new_name) |
| 719 | if rf.is_file(): |
| 720 | # Branch exists. If --intent or --resumable given (no start_point), |
| 721 | # treat as a metadata update rather than a failed create. |
| 722 | if (intent is not None or resumable) and start_point is None: |
| 723 | write_branch_meta( |
| 724 | root, |
| 725 | new_name, |
| 726 | intent=intent, |
| 727 | resumable=resumable if resumable else None, |
| 728 | ) |
| 729 | meta = read_branch_meta(root, new_name) |
| 730 | if json_out: |
| 731 | print(json.dumps({ |
| 732 | "action": "updated", |
| 733 | "branch": new_name, |
| 734 | "intent": meta.get("intent"), |
| 735 | "resumable": bool(meta.get("resumable", False)), |
| 736 | })) |
| 737 | else: |
| 738 | parts: list[str] = [] |
| 739 | if intent is not None: |
| 740 | parts.append(f"intent={sanitize_display(intent)!r}") |
| 741 | if resumable: |
| 742 | parts.append("resumable=true") |
| 743 | print( |
| 744 | f"Updated branch {_c(sanitize_display(new_name), _YELLOW, tty=tty)}" |
| 745 | f"{' (' + ', '.join(parts) + ')' if parts else ''}." |
| 746 | ) |
| 747 | return |
| 748 | if json_out: |
| 749 | print(json.dumps({"error": "already_exists", "branch": new_name, "message": f"branch '{new_name}' already exists"})) |
| 750 | print(f"❌ Branch '{sanitize_display(new_name)}' already exists.", file=sys.stderr) |
| 751 | raise SystemExit(ExitCode.USER_ERROR) |
| 752 | |
| 753 | if start_point is not None: |
| 754 | # Resolve branch names, full SHAs, and abbreviated SHA prefixes. |
| 755 | sp_tip: str = _resolve_start_point(root, current, start_point) |
| 756 | else: |
| 757 | sp_tip = get_head_commit_id(root, current) or "" |
| 758 | |
| 759 | if sp_tip: |
| 760 | write_branch_ref(root, new_name, sp_tip) |
| 761 | source_label = start_point or current or "HEAD" |
| 762 | append_reflog(root, new_name, None, sp_tip, "", f"branch: Created from {source_label}") |
| 763 | else: |
| 764 | write_text_atomic(rf, "") |
| 765 | |
| 766 | # Persist intent / resumable if supplied. |
| 767 | if intent is not None or resumable: |
| 768 | write_branch_meta( |
| 769 | root, |
| 770 | new_name, |
| 771 | intent=intent, |
| 772 | resumable=resumable if resumable else None, |
| 773 | ) |
| 774 | |
| 775 | if json_out: |
| 776 | print(json.dumps({ |
| 777 | **make_envelope(elapsed), |
| 778 | **_BranchCreateJson( |
| 779 | action="created", |
| 780 | branch=new_name, |
| 781 | commit_id=sp_tip or None, |
| 782 | intent=intent, |
| 783 | resumable=resumable, |
| 784 | ), |
| 785 | "from": start_point, |
| 786 | })) |
| 787 | else: |
| 788 | print(f"Created branch {_c(sanitize_display(new_name), _GREEN, tty=tty)}.") |
| 789 | return |
| 790 | |
| 791 | # ------------------------------------------------------------------ |
| 792 | # LIST |
| 793 | # ------------------------------------------------------------------ |
| 794 | local_branches = _list_local_branches(root) |
| 795 | if remotes_only: |
| 796 | display_branches = [f"remotes/{b}" for b in _list_remotes(root)] |
| 797 | elif all_branches: |
| 798 | display_branches = local_branches + [f"remotes/{b}" for b in _list_remotes(root)] |
| 799 | else: |
| 800 | display_branches = list(local_branches) |
| 801 | |
| 802 | # --resumable filter: only show branches marked resumable in config. |
| 803 | if resumable and not positional: |
| 804 | filtered_resumable: list[str] = [] |
| 805 | for b in display_branches: |
| 806 | local_b = b.removeprefix("remotes/") |
| 807 | meta = read_branch_meta(root, local_b) |
| 808 | if meta.get("resumable") is True: |
| 809 | filtered_resumable.append(b) |
| 810 | display_branches = filtered_resumable |
| 811 | |
| 812 | # --merged / --no-merged / --contains filters |
| 813 | if merged_into or not_merged_into or contains_commit: |
| 814 | resolved_current = current |
| 815 | |
| 816 | # Pre-compute ancestor sets once — not once per branch. |
| 817 | # _commit_ancestors walks the full commit DAG; recomputing it for every |
| 818 | # branch being checked is O(branches × commits) instead of O(commits). |
| 819 | _merged_ancestors: set[str] | None = None |
| 820 | if merged_into: |
| 821 | _into_tip = _resolve_into_tip(root, merged_into, resolved_current) |
| 822 | _merged_ancestors = _commit_ancestors(root, _into_tip) if _into_tip else set() |
| 823 | |
| 824 | _not_merged_ancestors: set[str] | None = None |
| 825 | if not_merged_into: |
| 826 | _into_tip = _resolve_into_tip(root, not_merged_into, resolved_current) |
| 827 | _not_merged_ancestors = _commit_ancestors(root, _into_tip) if _into_tip else set() |
| 828 | |
| 829 | def _passes(b: str) -> bool: |
| 830 | local_b = b.removeprefix("remotes/") |
| 831 | if _merged_ancestors is not None: |
| 832 | tip = get_head_commit_id(root, local_b) |
| 833 | if tip is None or tip not in _merged_ancestors: |
| 834 | return False |
| 835 | if _not_merged_ancestors is not None: |
| 836 | tip = get_head_commit_id(root, local_b) |
| 837 | if tip is not None and tip in _not_merged_ancestors: |
| 838 | return False |
| 839 | if contains_commit: |
| 840 | if not _contains_commit(root, local_b, contains_commit): |
| 841 | return False |
| 842 | return True |
| 843 | |
| 844 | display_branches = [b for b in display_branches if _passes(b)] |
| 845 | |
| 846 | # --sort: sort by committed date if requested. |
| 847 | # Name sort is the default (already applied by _list_local_branches). |
| 848 | if sort_key == "committeddate": |
| 849 | def _committed_ts(b: str) -> str: |
| 850 | cid = _resolve_commit_id(root, b) |
| 851 | if not cid: |
| 852 | return "" |
| 853 | rec = read_commit(root, cid) |
| 854 | return rec.committed_at.isoformat() if rec else "" |
| 855 | |
| 856 | display_branches = sorted(display_branches, key=_committed_ts, reverse=True) |
| 857 | |
| 858 | if json_out: |
| 859 | result: list[_BranchEntryJson] = [] |
| 860 | for b in display_branches: |
| 861 | local_b = b.removeprefix("remotes/") |
| 862 | commit_id = _resolve_commit_id(root, b) |
| 863 | rec = read_commit(root, commit_id) if commit_id else None |
| 864 | last_message: str | None = ( |
| 865 | sanitize_display(rec.message.splitlines()[0][:72]) if rec and rec.message else None |
| 866 | ) |
| 867 | upstream: str | None = _upstream_for(root, local_b) |
| 868 | meta = read_branch_meta(root, local_b) |
| 869 | branch_intent: str | None = meta.get("intent") or None # type: ignore[assignment] |
| 870 | branch_intent = sanitize_display(branch_intent) if branch_intent else None |
| 871 | branch_resumable: bool = bool(meta.get("resumable", False)) |
| 872 | created_by: str | None = (rec.agent_id if rec and rec.agent_id else None) |
| 873 | result.append({ |
| 874 | "name": b, |
| 875 | "current": local_b == current, |
| 876 | "commit_id": commit_id or None, |
| 877 | "committed_at": rec.committed_at.isoformat() if rec else None, |
| 878 | "last_message": last_message, |
| 879 | "upstream": upstream, |
| 880 | "intent": branch_intent, |
| 881 | "resumable": branch_resumable, |
| 882 | "created_by": created_by, |
| 883 | }) |
| 884 | print(json.dumps(result)) |
| 885 | return |
| 886 | |
| 887 | for b in display_branches: |
| 888 | is_remote_entry = b.startswith("remotes/") |
| 889 | local_b = b.removeprefix("remotes/") |
| 890 | is_current = (local_b == current) and not is_remote_entry |
| 891 | marker = _c("* ", _GREEN, tty=tty) if is_current else " " |
| 892 | # Build the display name once; apply sanitization before any coloring |
| 893 | # so that ANSI codes from _c() are not accidentally re-sanitized. |
| 894 | safe_name = sanitize_display(b) |
| 895 | name_str = _c(safe_name, _GREEN, tty=tty) if is_current else safe_name |
| 896 | if verbose >= 1: |
| 897 | commit_id = _resolve_commit_id(root, b) |
| 898 | short = short_id(commit_id) if commit_id else "(empty)" |
| 899 | rec = read_commit(root, commit_id) if commit_id else None |
| 900 | msg = sanitize_display(rec.message.splitlines()[0][:48]) if rec and rec.message else "" |
| 901 | short_str = _c(short, _YELLOW, tty=tty) |
| 902 | if verbose >= 2: |
| 903 | upstream = _upstream_for(root, local_b) |
| 904 | up_str = ( |
| 905 | f" [{_c(sanitize_display(upstream), _CYAN, tty=tty)}]" |
| 906 | if upstream else "" |
| 907 | ) |
| 908 | print(f"{marker}{name_str} {short_str}{up_str} {msg}") |
| 909 | else: |
| 910 | print(f"{marker}{name_str} {short_str} {msg}") |
| 911 | else: |
| 912 | print(f"{marker}{name_str}") |
File History
4 commits
sha256:e237dc0e8122609f5131d11c9dda9bba480395a5a4355cda0c9fa7e634fddd29
fix(branch): guard -d --dry-run against destructive writes;…
Sonnet 4.6
patch
101 days ago
sha256:42d0a10d093980afe543a88e9ed75c5ad0ac339026e419ee3b07b8a57c73ed5b
fix(branch): dry-run flag ignored on branch delete (closes #25)
Sonnet 4.6
patch
101 days ago
sha256:8860dea10c653956b613a814cc752a6d34cb3986cdf16749a49172affdabf045
fix tests
Human
minor
⚠
107 days ago
sha256:99f8eb388d9a9c353e68b9a4e5bebe1b4240a8f511e6f0928e58c0e95153e103
feat: branch --prune-config, fix hub repo delete docstrings…
Sonnet 4.6
minor
⚠
108 days ago