branch.py
python
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8
docs(KD-STAGING): sync governance after KD-6b DONE
Human
13 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>", |
| 39 | "last_message": "Add feature X", |
| 40 | "upstream": "origin/feat/my-thing" |
| 41 | }, |
| 42 | ... |
| 43 | ] |
| 44 | |
| 45 | Exit codes:: |
| 46 | |
| 47 | 0 — success |
| 48 | 1 — invalid branch name, branch not found, attempting to delete checked-out branch |
| 49 | """ |
| 50 | |
| 51 | from __future__ import annotations |
| 52 | |
| 53 | import argparse |
| 54 | import json |
| 55 | import logging |
| 56 | import pathlib |
| 57 | import sys |
| 58 | import tomllib |
| 59 | |
| 60 | from muse.cli.config import get_remote_head |
| 61 | from muse.core.errors import ExitCode |
| 62 | from muse.core.repo import read_repo_id, require_repo |
| 63 | from muse.core.store import ( |
| 64 | get_head_commit_id, |
| 65 | read_commit, |
| 66 | read_current_branch, |
| 67 | resolve_commit_ref, |
| 68 | write_branch_ref, |
| 69 | write_head_branch, |
| 70 | write_text_atomic, |
| 71 | ) |
| 72 | from muse.core.validation import clamp_int, sanitize_display, validate_branch_name |
| 73 | |
| 74 | |
| 75 | type _Payload = dict[str, str | None] |
| 76 | logger = logging.getLogger(__name__) |
| 77 | |
| 78 | # --------------------------------------------------------------------------- |
| 79 | # ANSI helpers — emitted only when stdout is a TTY. |
| 80 | # --------------------------------------------------------------------------- |
| 81 | |
| 82 | _RESET = "\033[0m" |
| 83 | _BOLD = "\033[1m" |
| 84 | _DIM = "\033[2m" |
| 85 | _GREEN = "\033[32m" |
| 86 | _RED = "\033[31m" |
| 87 | _YELLOW = "\033[33m" |
| 88 | _CYAN = "\033[36m" |
| 89 | |
| 90 | |
| 91 | def _c(text: str, *codes: str, tty: bool) -> str: |
| 92 | """Wrap *text* in ANSI escape *codes* only when writing to a TTY.""" |
| 93 | if not tty: |
| 94 | return text |
| 95 | return "".join(codes) + text + _RESET |
| 96 | |
| 97 | |
| 98 | # --------------------------------------------------------------------------- |
| 99 | # Internal helpers |
| 100 | # --------------------------------------------------------------------------- |
| 101 | |
| 102 | |
| 103 | def _ref_file(root: pathlib.Path, branch: str) -> pathlib.Path: |
| 104 | """Return the ref-file path for a local branch.""" |
| 105 | return root / ".muse" / "refs" / "heads" / branch |
| 106 | |
| 107 | |
| 108 | def _list_local_branches(root: pathlib.Path) -> list[str]: |
| 109 | """Return a sorted list of all local branch names. |
| 110 | |
| 111 | Only plain files are considered; directories, symlinks and any file not |
| 112 | directly under ``refs/heads/`` (e.g. lock files) are silently skipped. |
| 113 | """ |
| 114 | heads_dir = root / ".muse" / "refs" / "heads" |
| 115 | if not heads_dir.exists(): |
| 116 | return [] |
| 117 | return sorted( |
| 118 | p.relative_to(heads_dir).as_posix() |
| 119 | for p in heads_dir.rglob("*") |
| 120 | if p.is_file() and not p.name.startswith(".") |
| 121 | ) |
| 122 | |
| 123 | |
| 124 | def _list_remotes(root: pathlib.Path) -> list[str]: |
| 125 | """Return sorted remote-tracking branch names as ``remote/branch``. |
| 126 | |
| 127 | Only plain files are visited; symlinks, hidden files, and directories |
| 128 | are skipped to avoid leaking internal artefacts into the listing. |
| 129 | """ |
| 130 | remotes_dir = root / ".muse" / "remotes" |
| 131 | if not remotes_dir.exists(): |
| 132 | return [] |
| 133 | results: list[str] = [] |
| 134 | for remote_dir in sorted(remotes_dir.iterdir()): |
| 135 | if not remote_dir.is_dir(): |
| 136 | continue |
| 137 | remote = remote_dir.name |
| 138 | for ref_file in sorted(remote_dir.rglob("*")): |
| 139 | if ref_file.is_file() and not ref_file.name.startswith("."): |
| 140 | branch_rel = ref_file.relative_to(remote_dir).as_posix() |
| 141 | results.append(f"{remote}/{branch_rel}") |
| 142 | return results |
| 143 | |
| 144 | |
| 145 | def _resolve_commit_id(root: pathlib.Path, b: str) -> str: |
| 146 | """Return the current commit ID for a branch listing entry. |
| 147 | |
| 148 | *b* is the display name (e.g. ``"main"`` or ``"remotes/origin/dev"``). |
| 149 | Remote entries are read from the remote tracking file under |
| 150 | ``.muse/remotes/``; local entries use the standard head ref. |
| 151 | """ |
| 152 | if b.startswith("remotes/"): |
| 153 | rest = b.removeprefix("remotes/") |
| 154 | remote, _, branch_name = rest.partition("/") |
| 155 | if branch_name: |
| 156 | return get_remote_head(remote, branch_name, root) or "" |
| 157 | return get_head_commit_id(root, b) or "" |
| 158 | |
| 159 | |
| 160 | def _upstream_for(root: pathlib.Path, branch: str) -> str | None: |
| 161 | """Return the upstream tracking ref for *branch*, or ``None`` if unset.""" |
| 162 | config_path = root / ".muse" / "config.toml" |
| 163 | if not config_path.exists(): |
| 164 | return None |
| 165 | try: |
| 166 | with config_path.open("rb") as f: |
| 167 | config = tomllib.load(f) |
| 168 | section = config.get("branch", {}).get(branch, {}) |
| 169 | remote: str | None = section.get("remote") |
| 170 | merge_ref: str | None = section.get("merge") |
| 171 | if remote and merge_ref: |
| 172 | short = merge_ref.removeprefix("refs/heads/") |
| 173 | return f"{remote}/{short}" |
| 174 | except Exception: |
| 175 | pass |
| 176 | return None |
| 177 | |
| 178 | |
| 179 | def _commit_ancestors(root: pathlib.Path, commit_id: str) -> set[str]: |
| 180 | """Return the set of all commit IDs reachable from *commit_id* (inclusive).""" |
| 181 | seen: set[str] = set() |
| 182 | queue: list[str] = [commit_id] |
| 183 | while queue: |
| 184 | cid = queue.pop() |
| 185 | if cid in seen: |
| 186 | continue |
| 187 | seen.add(cid) |
| 188 | rec = read_commit(root, cid) |
| 189 | if rec is None: |
| 190 | continue |
| 191 | if rec.parent_commit_id: |
| 192 | queue.append(rec.parent_commit_id) |
| 193 | if rec.parent2_commit_id: |
| 194 | queue.append(rec.parent2_commit_id) |
| 195 | return seen |
| 196 | |
| 197 | |
| 198 | def _is_merged(root: pathlib.Path, branch: str, into: str) -> bool: |
| 199 | """Return ``True`` if the tip of *branch* is an ancestor of the tip of *into*.""" |
| 200 | branch_tip = get_head_commit_id(root, branch) |
| 201 | into_tip = get_head_commit_id(root, into) |
| 202 | if branch_tip is None or into_tip is None: |
| 203 | return False |
| 204 | return branch_tip in _commit_ancestors(root, into_tip) |
| 205 | |
| 206 | |
| 207 | def _contains_commit(root: pathlib.Path, branch: str, commit_id: str) -> bool: |
| 208 | """Return ``True`` if *commit_id* is reachable from the tip of *branch*.""" |
| 209 | tip = get_head_commit_id(root, branch) |
| 210 | if tip is None: |
| 211 | return False |
| 212 | return commit_id in _commit_ancestors(root, tip) |
| 213 | |
| 214 | |
| 215 | def _cleanup_empty_dirs(ref_file: pathlib.Path, heads_dir: pathlib.Path) -> None: |
| 216 | """Remove any empty parent directories left behind after unlinking *ref_file*.""" |
| 217 | for parent in ref_file.parents: |
| 218 | if parent == heads_dir: |
| 219 | break |
| 220 | try: |
| 221 | parent.rmdir() |
| 222 | except OSError: |
| 223 | break |
| 224 | |
| 225 | |
| 226 | def _resolve_start_point(root: pathlib.Path, repo_id: str, current: str, start_point: str) -> str: |
| 227 | """Resolve *start_point* to a full commit ID. |
| 228 | |
| 229 | Accepts branch names, full SHA-256 commit IDs, and abbreviated SHA |
| 230 | prefixes (any unambiguous prefix works). Returns the raw *start_point* |
| 231 | string unchanged if resolution fails — the caller is responsible for |
| 232 | surfacing a meaningful error in that case. |
| 233 | """ |
| 234 | # Try as branch name first. |
| 235 | branch_tip = get_head_commit_id(root, start_point) |
| 236 | if branch_tip is not None: |
| 237 | return branch_tip |
| 238 | # Fall back to SHA / SHA-prefix resolution. |
| 239 | rec = resolve_commit_ref(root, repo_id, current, start_point) |
| 240 | if rec is not None: |
| 241 | return rec.commit_id |
| 242 | # Return as-is; the caller's write_branch_ref will expose the invalid ID. |
| 243 | return start_point |
| 244 | |
| 245 | |
| 246 | # --------------------------------------------------------------------------- |
| 247 | # CLI registration |
| 248 | # --------------------------------------------------------------------------- |
| 249 | |
| 250 | |
| 251 | def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: |
| 252 | """Register the ``muse branch`` subcommand and all its flags.""" |
| 253 | parser = subparsers.add_parser( |
| 254 | "branch", |
| 255 | help="List, create, rename, copy, or delete branches.", |
| 256 | description=__doc__, |
| 257 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 258 | ) |
| 259 | parser.add_argument("args", nargs="*", help="Branch name(s) — context-sensitive.") |
| 260 | |
| 261 | # Mutually exclusive operation flags (mirrors git branch). |
| 262 | ops = parser.add_mutually_exclusive_group() |
| 263 | ops.add_argument( |
| 264 | "-d", "--delete", dest="op", action="store_const", const="delete", |
| 265 | help="Delete a branch (safe — must be fully merged).", |
| 266 | ) |
| 267 | ops.add_argument( |
| 268 | "-D", dest="op", action="store_const", const="force_delete", |
| 269 | help="Force-delete a branch regardless of merge status.", |
| 270 | ) |
| 271 | ops.add_argument( |
| 272 | "-m", "--move", dest="op", action="store_const", const="rename", |
| 273 | help="Rename a branch (safe).", |
| 274 | ) |
| 275 | ops.add_argument( |
| 276 | "-M", dest="op", action="store_const", const="force_rename", |
| 277 | help="Force-rename a branch.", |
| 278 | ) |
| 279 | ops.add_argument( |
| 280 | "-c", "--copy", dest="op", action="store_const", const="copy", |
| 281 | help="Copy a branch (safe).", |
| 282 | ) |
| 283 | ops.add_argument( |
| 284 | "-C", dest="op", action="store_const", const="force_copy", |
| 285 | help="Force-copy a branch.", |
| 286 | ) |
| 287 | |
| 288 | # Listing modifiers. |
| 289 | parser.add_argument( |
| 290 | "-v", action="count", default=0, dest="verbose", |
| 291 | help="Show last commit SHA + subject. Repeat (-vv) to also show upstream.", |
| 292 | ) |
| 293 | parser.add_argument( |
| 294 | "-r", "--remotes", action="store_true", |
| 295 | help="List remote-tracking branches.", |
| 296 | ) |
| 297 | parser.add_argument( |
| 298 | "-a", "--all", action="store_true", dest="all_branches", |
| 299 | help="List both local and remote-tracking branches.", |
| 300 | ) |
| 301 | parser.add_argument( |
| 302 | "--merged", metavar="COMMIT", nargs="?", const="HEAD", |
| 303 | help="Only list branches merged into COMMIT (default HEAD).", |
| 304 | ) |
| 305 | parser.add_argument( |
| 306 | "--no-merged", metavar="COMMIT", nargs="?", const="HEAD", |
| 307 | help="Only list branches NOT merged into COMMIT (default HEAD).", |
| 308 | ) |
| 309 | parser.add_argument( |
| 310 | "--contains", metavar="COMMIT", |
| 311 | help="Only list branches that contain COMMIT.", |
| 312 | ) |
| 313 | parser.add_argument( |
| 314 | "--sort", default="name", metavar="KEY", |
| 315 | choices=["name", "committeddate"], |
| 316 | help="Sort branches by 'name' (default) or 'committeddate'.", |
| 317 | ) |
| 318 | parser.add_argument( |
| 319 | "--format", "-f", default="text", dest="fmt", |
| 320 | help="Output format: text or json.", |
| 321 | ) |
| 322 | parser.add_argument( |
| 323 | "--json", action="store_const", const="json", dest="fmt", |
| 324 | help="Shorthand for --format json.", |
| 325 | ) |
| 326 | parser.set_defaults(func=run, op=None) |
| 327 | |
| 328 | |
| 329 | # --------------------------------------------------------------------------- |
| 330 | # Command handler |
| 331 | # --------------------------------------------------------------------------- |
| 332 | |
| 333 | |
| 334 | def run(args: argparse.Namespace) -> None: |
| 335 | """List, create, rename, copy, or delete branches. |
| 336 | |
| 337 | Agents should pass ``--format json`` when listing to receive a stable JSON |
| 338 | array:: |
| 339 | |
| 340 | [ |
| 341 | { |
| 342 | "name": "feat/my-thing", |
| 343 | "current": false, |
| 344 | "commit_id": "<sha256>", |
| 345 | "last_message": "Add feature X", |
| 346 | "upstream": null |
| 347 | } |
| 348 | ] |
| 349 | |
| 350 | Mutation operations (create, rename, copy, delete) emit a single result |
| 351 | object with an ``"action"`` key describing what happened. |
| 352 | """ |
| 353 | positional: list[str] = args.args |
| 354 | op: str | None = args.op |
| 355 | verbose: int = clamp_int(args.verbose, 0, 4, 'verbose') |
| 356 | remotes_only: bool = args.remotes |
| 357 | all_branches: bool = args.all_branches |
| 358 | merged_into: str | None = args.merged |
| 359 | not_merged_into: str | None = args.no_merged |
| 360 | contains_commit: str | None = args.contains |
| 361 | sort_key: str = args.sort |
| 362 | fmt: str = args.fmt |
| 363 | tty: bool = sys.stdout.isatty() |
| 364 | |
| 365 | if fmt not in ("text", "json"): |
| 366 | print(f"❌ Unknown --format '{sanitize_display(fmt)}'. Choose text or json.", file=sys.stderr) |
| 367 | raise SystemExit(ExitCode.USER_ERROR) |
| 368 | |
| 369 | root = require_repo() |
| 370 | repo_id = read_repo_id(root) |
| 371 | current = read_current_branch(root) |
| 372 | heads_dir = root / ".muse" / "refs" / "heads" |
| 373 | |
| 374 | # ------------------------------------------------------------------ |
| 375 | # DELETE / FORCE-DELETE |
| 376 | # Supports two modes: |
| 377 | # muse branch -d|-D <local-branch> — delete a local branch |
| 378 | # muse branch -d|-D -r <remote>/<branch> — prune a remote-tracking ref |
| 379 | # ------------------------------------------------------------------ |
| 380 | if op in ("delete", "force_delete"): |
| 381 | if not positional: |
| 382 | print("❌ Usage: muse branch -d|-D [-r] <branch> …", file=sys.stderr) |
| 383 | raise SystemExit(ExitCode.USER_ERROR) |
| 384 | |
| 385 | # -r flag: delete local remote-tracking refs (no server call). |
| 386 | if remotes_only: |
| 387 | from muse.cli.config import delete_remote_head |
| 388 | for spec in positional: |
| 389 | # Accept both "remote/branch" and "remotes/remote/branch" spellings. |
| 390 | clean = spec.removeprefix("remotes/") |
| 391 | slash = clean.find("/") |
| 392 | if slash == -1: |
| 393 | print( |
| 394 | f"❌ Remote-tracking ref must be '<remote>/<branch>', got " |
| 395 | f"'{sanitize_display(spec)}'.", |
| 396 | file=sys.stderr, |
| 397 | ) |
| 398 | raise SystemExit(ExitCode.USER_ERROR) |
| 399 | remote_name = clean[:slash] |
| 400 | branch_name = clean[slash + 1:] |
| 401 | removed = delete_remote_head(remote_name, branch_name, root) |
| 402 | if not removed: |
| 403 | print( |
| 404 | f"❌ Remote-tracking ref '{sanitize_display(clean)}' not found.", |
| 405 | file=sys.stderr, |
| 406 | ) |
| 407 | raise SystemExit(ExitCode.USER_ERROR) |
| 408 | if fmt == "json": |
| 409 | print(json.dumps({ |
| 410 | "action": "deleted_remote_tracking", |
| 411 | "remote": remote_name, |
| 412 | "branch": branch_name, |
| 413 | })) |
| 414 | else: |
| 415 | print( |
| 416 | f"Deleted remote-tracking ref " |
| 417 | f"{_c(sanitize_display(clean), _RED, tty=tty)}." |
| 418 | ) |
| 419 | return |
| 420 | |
| 421 | force = op == "force_delete" |
| 422 | for branch_name in positional: |
| 423 | try: |
| 424 | validate_branch_name(branch_name) |
| 425 | except ValueError as exc: |
| 426 | print(f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr) |
| 427 | raise SystemExit(ExitCode.USER_ERROR) |
| 428 | if branch_name == current: |
| 429 | print( |
| 430 | f"❌ Cannot delete the currently checked-out branch " |
| 431 | f"'{sanitize_display(branch_name)}'.", |
| 432 | file=sys.stderr, |
| 433 | ) |
| 434 | raise SystemExit(ExitCode.USER_ERROR) |
| 435 | rf = _ref_file(root, branch_name) |
| 436 | if not rf.is_file(): |
| 437 | print(f"❌ Branch '{sanitize_display(branch_name)}' not found.", file=sys.stderr) |
| 438 | raise SystemExit(ExitCode.USER_ERROR) |
| 439 | if not force and not _is_merged(root, branch_name, current): |
| 440 | print( |
| 441 | f"❌ Branch '{sanitize_display(branch_name)}' is not fully merged.\n" |
| 442 | f" Use -D to force-delete.", |
| 443 | file=sys.stderr, |
| 444 | ) |
| 445 | raise SystemExit(ExitCode.USER_ERROR) |
| 446 | tip = rf.read_text().strip() |
| 447 | rf.unlink() |
| 448 | _cleanup_empty_dirs(rf, heads_dir) |
| 449 | if fmt == "json": |
| 450 | print(json.dumps({"action": "deleted", "branch": branch_name, "was": tip})) |
| 451 | else: |
| 452 | short = tip[:8] if tip else "unknown" |
| 453 | print( |
| 454 | f"Deleted branch {_c(sanitize_display(branch_name), _RED, tty=tty)} " |
| 455 | f"({_c('was ' + short, _DIM, tty=tty)})." |
| 456 | ) |
| 457 | return |
| 458 | |
| 459 | # ------------------------------------------------------------------ |
| 460 | # RENAME / FORCE-RENAME |
| 461 | # ------------------------------------------------------------------ |
| 462 | if op in ("rename", "force_rename"): |
| 463 | force = op == "force_rename" |
| 464 | if len(positional) == 1: |
| 465 | old_name, new_name = current, positional[0] |
| 466 | elif len(positional) == 2: |
| 467 | old_name, new_name = positional[0], positional[1] |
| 468 | else: |
| 469 | print("❌ Usage: muse branch -m|-M [<old>] <new>", file=sys.stderr) |
| 470 | raise SystemExit(ExitCode.USER_ERROR) |
| 471 | for n in (old_name, new_name): |
| 472 | try: |
| 473 | validate_branch_name(n) |
| 474 | except ValueError as exc: |
| 475 | print(f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr) |
| 476 | raise SystemExit(ExitCode.USER_ERROR) |
| 477 | src = _ref_file(root, old_name) |
| 478 | dst = _ref_file(root, new_name) |
| 479 | if not src.is_file(): |
| 480 | print(f"❌ Branch '{sanitize_display(old_name)}' not found.", file=sys.stderr) |
| 481 | raise SystemExit(ExitCode.USER_ERROR) |
| 482 | if dst.is_file() and not force: |
| 483 | print( |
| 484 | f"❌ Branch '{sanitize_display(new_name)}' already exists. Use -M to force.", |
| 485 | file=sys.stderr, |
| 486 | ) |
| 487 | raise SystemExit(ExitCode.USER_ERROR) |
| 488 | tip = src.read_text().strip() |
| 489 | if tip: |
| 490 | write_branch_ref(root, new_name, tip) |
| 491 | else: |
| 492 | write_text_atomic(dst, "") |
| 493 | src.unlink() |
| 494 | _cleanup_empty_dirs(src, heads_dir) |
| 495 | if old_name == current: |
| 496 | write_head_branch(root, new_name) |
| 497 | if fmt == "json": |
| 498 | print(json.dumps({"action": "renamed", "from": old_name, "to": new_name})) |
| 499 | else: |
| 500 | print( |
| 501 | f"Renamed branch " |
| 502 | f"{_c(sanitize_display(old_name), _YELLOW, tty=tty)} → " |
| 503 | f"{_c(sanitize_display(new_name), _GREEN, tty=tty)}." |
| 504 | ) |
| 505 | return |
| 506 | |
| 507 | # ------------------------------------------------------------------ |
| 508 | # COPY / FORCE-COPY |
| 509 | # ------------------------------------------------------------------ |
| 510 | if op in ("copy", "force_copy"): |
| 511 | force = op == "force_copy" |
| 512 | if len(positional) == 1: |
| 513 | src_name, dst_name = current, positional[0] |
| 514 | elif len(positional) == 2: |
| 515 | src_name, dst_name = positional[0], positional[1] |
| 516 | else: |
| 517 | print("❌ Usage: muse branch -c|-C [<src>] <dest>", file=sys.stderr) |
| 518 | raise SystemExit(ExitCode.USER_ERROR) |
| 519 | for n in (src_name, dst_name): |
| 520 | try: |
| 521 | validate_branch_name(n) |
| 522 | except ValueError as exc: |
| 523 | print(f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr) |
| 524 | raise SystemExit(ExitCode.USER_ERROR) |
| 525 | src = _ref_file(root, src_name) |
| 526 | dst = _ref_file(root, dst_name) |
| 527 | if not src.is_file(): |
| 528 | print(f"❌ Branch '{sanitize_display(src_name)}' not found.", file=sys.stderr) |
| 529 | raise SystemExit(ExitCode.USER_ERROR) |
| 530 | if dst.is_file() and not force: |
| 531 | print( |
| 532 | f"❌ Branch '{sanitize_display(dst_name)}' already exists. Use -C to force.", |
| 533 | file=sys.stderr, |
| 534 | ) |
| 535 | raise SystemExit(ExitCode.USER_ERROR) |
| 536 | tip = src.read_text().strip() |
| 537 | if tip: |
| 538 | write_branch_ref(root, dst_name, tip) |
| 539 | else: |
| 540 | write_text_atomic(dst, "") |
| 541 | if fmt == "json": |
| 542 | print(json.dumps({"action": "copied", "from": src_name, "to": dst_name})) |
| 543 | else: |
| 544 | print( |
| 545 | f"Copied branch " |
| 546 | f"{_c(sanitize_display(src_name), _YELLOW, tty=tty)} → " |
| 547 | f"{_c(sanitize_display(dst_name), _GREEN, tty=tty)}." |
| 548 | ) |
| 549 | return |
| 550 | |
| 551 | # ------------------------------------------------------------------ |
| 552 | # CREATE |
| 553 | # ------------------------------------------------------------------ |
| 554 | if op is None and positional: |
| 555 | new_name = positional[0] |
| 556 | start_point: str | None = positional[1] if len(positional) > 1 else None |
| 557 | try: |
| 558 | validate_branch_name(new_name) |
| 559 | except ValueError as exc: |
| 560 | print(f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr) |
| 561 | raise SystemExit(ExitCode.USER_ERROR) |
| 562 | rf = _ref_file(root, new_name) |
| 563 | if rf.is_file(): |
| 564 | print(f"❌ Branch '{sanitize_display(new_name)}' already exists.", file=sys.stderr) |
| 565 | raise SystemExit(ExitCode.USER_ERROR) |
| 566 | |
| 567 | if start_point is not None: |
| 568 | # Resolve branch names, full SHAs, and abbreviated SHA prefixes. |
| 569 | sp_tip: str = _resolve_start_point(root, repo_id, current, start_point) |
| 570 | else: |
| 571 | sp_tip = get_head_commit_id(root, current) or "" |
| 572 | |
| 573 | if sp_tip: |
| 574 | write_branch_ref(root, new_name, sp_tip) |
| 575 | else: |
| 576 | write_text_atomic(rf, "") |
| 577 | |
| 578 | if fmt == "json": |
| 579 | payload: _Payload = { |
| 580 | "action": "created", |
| 581 | "branch": new_name, |
| 582 | "commit_id": sp_tip or None, |
| 583 | "from": start_point, |
| 584 | } |
| 585 | print(json.dumps(payload)) |
| 586 | else: |
| 587 | print(f"Created branch {_c(sanitize_display(new_name), _GREEN, tty=tty)}.") |
| 588 | return |
| 589 | |
| 590 | # ------------------------------------------------------------------ |
| 591 | # LIST |
| 592 | # ------------------------------------------------------------------ |
| 593 | local_branches = _list_local_branches(root) |
| 594 | if remotes_only: |
| 595 | display_branches = [f"remotes/{b}" for b in _list_remotes(root)] |
| 596 | elif all_branches: |
| 597 | display_branches = local_branches + [f"remotes/{b}" for b in _list_remotes(root)] |
| 598 | else: |
| 599 | display_branches = list(local_branches) |
| 600 | |
| 601 | # --merged / --no-merged / --contains filters |
| 602 | if merged_into or not_merged_into or contains_commit: |
| 603 | resolved_current = current |
| 604 | |
| 605 | def _passes(b: str) -> bool: |
| 606 | local_b = b.removeprefix("remotes/") |
| 607 | if merged_into: |
| 608 | into = resolved_current if merged_into == "HEAD" else merged_into |
| 609 | if not _is_merged(root, local_b, into): |
| 610 | return False |
| 611 | if not_merged_into: |
| 612 | into = resolved_current if not_merged_into == "HEAD" else not_merged_into |
| 613 | if _is_merged(root, local_b, into): |
| 614 | return False |
| 615 | if contains_commit: |
| 616 | if not _contains_commit(root, local_b, contains_commit): |
| 617 | return False |
| 618 | return True |
| 619 | |
| 620 | display_branches = [b for b in display_branches if _passes(b)] |
| 621 | |
| 622 | # --sort: sort by committed date if requested. |
| 623 | # Name sort is the default (already applied by _list_local_branches). |
| 624 | if sort_key == "committeddate": |
| 625 | def _committed_ts(b: str) -> str: |
| 626 | cid = _resolve_commit_id(root, b) |
| 627 | if not cid: |
| 628 | return "" |
| 629 | rec = read_commit(root, cid) |
| 630 | return rec.committed_at.isoformat() if rec else "" |
| 631 | |
| 632 | display_branches = sorted(display_branches, key=_committed_ts, reverse=True) |
| 633 | |
| 634 | if fmt == "json": |
| 635 | result: list[dict[str, str | bool | None]] = [] |
| 636 | for b in display_branches: |
| 637 | local_b = b.removeprefix("remotes/") |
| 638 | commit_id = _resolve_commit_id(root, b) |
| 639 | rec = read_commit(root, commit_id) if commit_id else None |
| 640 | last_message: str | None = ( |
| 641 | sanitize_display(rec.message.splitlines()[0][:72]) if rec and rec.message else None |
| 642 | ) |
| 643 | upstream: str | None = _upstream_for(root, local_b) |
| 644 | result.append({ |
| 645 | "name": b, |
| 646 | "current": local_b == current, |
| 647 | "commit_id": commit_id, |
| 648 | "last_message": last_message, |
| 649 | "upstream": upstream, |
| 650 | }) |
| 651 | print(json.dumps(result)) |
| 652 | return |
| 653 | |
| 654 | for b in display_branches: |
| 655 | is_remote_entry = b.startswith("remotes/") |
| 656 | local_b = b.removeprefix("remotes/") |
| 657 | is_current = (local_b == current) and not is_remote_entry |
| 658 | marker = _c("* ", _GREEN, tty=tty) if is_current else " " |
| 659 | # Build the display name once; apply sanitization before any coloring |
| 660 | # so that ANSI codes from _c() are not accidentally re-sanitized. |
| 661 | safe_name = sanitize_display(b) |
| 662 | name_str = _c(safe_name, _GREEN, tty=tty) if is_current else safe_name |
| 663 | if verbose >= 1: |
| 664 | commit_id = _resolve_commit_id(root, b) |
| 665 | short = commit_id[:8] if commit_id else "(empty)" |
| 666 | rec = read_commit(root, commit_id) if commit_id else None |
| 667 | msg = sanitize_display(rec.message.splitlines()[0][:48]) if rec and rec.message else "" |
| 668 | short_str = _c(short, _YELLOW, tty=tty) |
| 669 | if verbose >= 2: |
| 670 | upstream = _upstream_for(root, local_b) |
| 671 | up_str = ( |
| 672 | f" [{_c(sanitize_display(upstream), _CYAN, tty=tty)}]" |
| 673 | if upstream else "" |
| 674 | ) |
| 675 | print(f"{marker}{name_str} {short_str}{up_str} {msg}") |
| 676 | else: |
| 677 | print(f"{marker}{name_str} {short_str} {msg}") |
| 678 | else: |
| 679 | print(f"{marker}{name_str}") |
File History
1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8
docs(KD-STAGING): sync governance after KD-6b DONE
Human
13 days ago