muse/cli/commands/branch.py · gabriel/muse

print(json.dumps({"error": "not_merged", "branch": branch_name, "message": f"branch '{branch_name}' is not fully merged", "hint": "use -D to force-delete"}))

518

print(

519

f"❌ Branch '{sanitize_display(branch_name)}' is not fully merged.\n"

520

f" Use -D to force-delete.",

521

file=sys.stderr,

522

)

523

raise SystemExit(ExitCode.USER_ERROR)

524

tip = read_ref(rf) or ""

525

rf.unlink()

526

_cleanup_empty_dirs(rf, heads_dir)

527

if json_out:

528

print(json.dumps({"action": "deleted", "branch": branch_name, "was": tip}))

529

else:

530

short = short_id(tip) if tip else "unknown"

531

print(

532

f"Deleted branch {_c(sanitize_display(branch_name), _RED, tty=tty)} "

533

f"({_c('was ' + short, _DIM, tty=tty)})."

)

return

# ------------------------------------------------------------------

538

# RENAME / FORCE-RENAME

539

# ------------------------------------------------------------------

540

if op in ("rename", "force_rename"):

541

force = op == "force_rename"

542

if len(positional) == 1:

543

old_name, new_name = current, positional[0]

544

elif len(positional) == 2:

545

old_name, new_name = positional[0], positional[1]

546

else:

547

if json_out:

548

print(json.dumps({"error": "usage", "message": "muse branch -m|-M [<old>] <new>"}))

549

print("❌ Usage: muse branch -m|-M [<old>] <new>", file=sys.stderr)

550

raise SystemExit(ExitCode.USER_ERROR)

551

for n in (old_name, new_name):

552

try:

553

validate_branch_name(n)

554

except ValueError as exc:

555

if json_out:

556

print(json.dumps({"error": "invalid_branch_name", "branch": n, "message": str(exc)}))

557

print(f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr)

558

raise SystemExit(ExitCode.USER_ERROR)

559

src = _ref_file(root, old_name)

560

dst = _ref_file(root, new_name)

561

if not src.is_file():

562

if json_out:

563

print(json.dumps({"error": "not_found", "branch": old_name, "message": f"branch '{old_name}' not found"}))

564

print(f"❌ Branch '{sanitize_display(old_name)}' not found.", file=sys.stderr)

565

raise SystemExit(ExitCode.USER_ERROR)

566

if dst.is_file() and not force:

567

if json_out:

568

print(json.dumps({"error": "already_exists", "branch": new_name, "message": f"branch '{new_name}' already exists", "hint": "use -M to force"}))

569

print(

570

f"❌ Branch '{sanitize_display(new_name)}' already exists. Use -M to force.",

571

file=sys.stderr,

572

)

573

raise SystemExit(ExitCode.USER_ERROR)

574

tip = read_ref(src) or ""

575

if tip:

576

write_branch_ref(root, new_name, tip)

577

else:

578

write_text_atomic(dst, "")

579

src.unlink()

580

_cleanup_empty_dirs(src, heads_dir)

581

if old_name == current:

582

write_head_branch(root, new_name)

583

if json_out:

584

print(json.dumps({"action": "renamed", "from": old_name, "to": new_name}))

else:

print(

f"Renamed branch "

f"{_c(sanitize_display(old_name), _YELLOW, tty=tty)} → "

589

f"{_c(sanitize_display(new_name), _GREEN, tty=tty)}."

)

return

# ------------------------------------------------------------------

594

# COPY / FORCE-COPY

595

# ------------------------------------------------------------------

596

if op in ("copy", "force_copy"):

597

force = op == "force_copy"

598

if len(positional) == 1:

599

src_name, dst_name = current, positional[0]

600

elif len(positional) == 2:

601

src_name, dst_name = positional[0], positional[1]

602

else:

603

if json_out:

604

print(json.dumps({"error": "usage", "message": "muse branch -c|-C [<src>] <dest>"}))

605

print("❌ Usage: muse branch -c|-C [<src>] <dest>", file=sys.stderr)

606

raise SystemExit(ExitCode.USER_ERROR)

607

for n in (src_name, dst_name):

608

try:

609

validate_branch_name(n)

610

except ValueError as exc:

611

if json_out:

612

print(json.dumps({"error": "invalid_branch_name", "branch": n, "message": str(exc)}))

613

print(f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr)

614

raise SystemExit(ExitCode.USER_ERROR)

615

src = _ref_file(root, src_name)

616

dst = _ref_file(root, dst_name)

617

if not src.is_file():

618

if json_out:

619

print(json.dumps({"error": "not_found", "branch": src_name, "message": f"branch '{src_name}' not found"}))

620

print(f"❌ Branch '{sanitize_display(src_name)}' not found.", file=sys.stderr)

621

raise SystemExit(ExitCode.USER_ERROR)

622

if dst.is_file() and not force:

623

if json_out:

624

print(json.dumps({"error": "already_exists", "branch": dst_name, "message": f"branch '{dst_name}' already exists", "hint": "use -C to force"}))

625

print(

626

f"❌ Branch '{sanitize_display(dst_name)}' already exists. Use -C to force.",

627

file=sys.stderr,

628

)

629

raise SystemExit(ExitCode.USER_ERROR)

630

tip = read_ref(src) or ""

631

if tip:

632

write_branch_ref(root, dst_name, tip)

633

else:

634

write_text_atomic(dst, "")

635

if json_out:

636

print(json.dumps({"action": "copied", "from": src_name, "to": dst_name}))

else:

print(

f"Copied branch "

f"{_c(sanitize_display(src_name), _YELLOW, tty=tty)} → "

641

f"{_c(sanitize_display(dst_name), _GREEN, tty=tty)}."

)

return

# ------------------------------------------------------------------

646

# CREATE

647

# ------------------------------------------------------------------

648

if op is None and positional:

649

new_name = positional[0]

650

start_point: str | None = positional[1] if len(positional) > 1 else None

651

try:

652

validate_branch_name(new_name)

653

except ValueError as exc:

654

if json_out:

655

print(json.dumps({"error": "invalid_branch_name", "branch": new_name, "message": str(exc)}))

656

print(f"❌ Invalid branch name: {sanitize_display(str(exc))}", file=sys.stderr)

657

raise SystemExit(ExitCode.USER_ERROR)

658

rf = _ref_file(root, new_name)

659

if rf.is_file():

660

# Branch exists. If --intent or --resumable given (no start_point),

661

# treat as a metadata update rather than a failed create.

662

if (intent is not None or resumable) and start_point is None:

write_branch_meta(

root,

new_name,

intent=intent,

resumable=resumable if resumable else None,

668

)

669

meta = read_branch_meta(root, new_name)

if json_out:

print(json.dumps({

"action": "updated",

"branch": new_name,

"intent": meta.get("intent"),

675

"resumable": bool(meta.get("resumable", False)),

676

}))

677

else:

678

parts: list[str] = []

679

if intent is not None:

680

parts.append(f"intent={sanitize_display(intent)!r}")

681

if resumable:

682

parts.append("resumable=true")

683

print(

684

f"Updated branch {_c(sanitize_display(new_name), _YELLOW, tty=tty)}"

685

f"{' (' + ', '.join(parts) + ')' if parts else ''}."

)

return

if json_out:

print(json.dumps({"error": "already_exists", "branch": new_name, "message": f"branch '{new_name}' already exists"}))

690

print(f"❌ Branch '{sanitize_display(new_name)}' already exists.", file=sys.stderr)

691

raise SystemExit(ExitCode.USER_ERROR)

692

693

if start_point is not None:

694

# Resolve branch names, full SHAs, and abbreviated SHA prefixes.

695

sp_tip: str = _resolve_start_point(root, repo_id, current, start_point)

696

else:

697

sp_tip = get_head_commit_id(root, current) or ""

698

699

if sp_tip:

700

write_branch_ref(root, new_name, sp_tip)

701

else:

702

write_text_atomic(rf, "")

703

704

# Persist intent / resumable if supplied.

705

if intent is not None or resumable:

write_branch_meta(

root,

new_name,

intent=intent,

resumable=resumable if resumable else None,

)

if json_out:

print(json.dumps({

**make_envelope(elapsed),

**_BranchCreateJson(

action="created",

branch=new_name,

commit_id=sp_tip or None,

intent=intent,

resumable=resumable,

),

"from": start_point,

}))

else:

print(f"Created branch {_c(sanitize_display(new_name), _GREEN, tty=tty)}.")

727

return

728

729

# ------------------------------------------------------------------

730

# LIST

731

# ------------------------------------------------------------------

732

local_branches = _list_local_branches(root)

733

if remotes_only:

734

display_branches = [f"remotes/{b}" for b in _list_remotes(root)]

735

elif all_branches:

736

display_branches = local_branches + [f"remotes/{b}" for b in _list_remotes(root)]

737

else:

738

display_branches = list(local_branches)

739

740

# --resumable filter: only show branches marked resumable in config.

741

if resumable and not positional:

742

filtered_resumable: list[str] = []

743

for b in display_branches:

744

local_b = b.removeprefix("remotes/")

745

meta = read_branch_meta(root, local_b)

746

if meta.get("resumable") is True:

747

filtered_resumable.append(b)

748

display_branches = filtered_resumable

749

750

# --merged / --no-merged / --contains filters

751

if merged_into or not_merged_into or contains_commit:

752

resolved_current = current

753

754

# Pre-compute ancestor sets once — not once per branch.

755

# _commit_ancestors walks the full commit DAG; recomputing it for every

756

# branch being checked is O(branches × commits) instead of O(commits).

757

_merged_ancestors: set[str] | None = None

758

if merged_into:

759

_into = resolved_current if merged_into == "HEAD" else merged_into

760

_into_tip = get_head_commit_id(root, _into)

761

_merged_ancestors = _commit_ancestors(root, _into_tip) if _into_tip else set()

762

763

_not_merged_ancestors: set[str] | None = None

764

if not_merged_into:

765

_into = resolved_current if not_merged_into == "HEAD" else not_merged_into

766

_into_tip = get_head_commit_id(root, _into)

767

_not_merged_ancestors = _commit_ancestors(root, _into_tip) if _into_tip else set()

768

769

def _passes(b: str) -> bool:

770

local_b = b.removeprefix("remotes/")

771

if _merged_ancestors is not None:

772

tip = get_head_commit_id(root, local_b)

773

if tip is None or tip not in _merged_ancestors:

774

return False

775

if _not_merged_ancestors is not None:

776

tip = get_head_commit_id(root, local_b)

777

if tip is not None and tip in _not_merged_ancestors:

778

return False

779

if contains_commit:

780

if not _contains_commit(root, local_b, contains_commit):

return False

return True

display_branches = [b for b in display_branches if _passes(b)]

785

786

# --sort: sort by committed date if requested.

787

# Name sort is the default (already applied by _list_local_branches).

788

if sort_key == "committeddate":

789

def _committed_ts(b: str) -> str:

790

cid = _resolve_commit_id(root, b)

791

if not cid:

792

return ""

793

rec = read_commit(root, cid)

794

return rec.committed_at.isoformat() if rec else ""

795

796

display_branches = sorted(display_branches, key=_committed_ts, reverse=True)

797

798

if json_out:

799

result: list[_BranchEntryJson] = []

800

for b in display_branches:

801

local_b = b.removeprefix("remotes/")

802

commit_id = _resolve_commit_id(root, b)

803

rec = read_commit(root, commit_id) if commit_id else None

804

last_message: str | None = (

805

sanitize_display(rec.message.splitlines()[0][:72]) if rec and rec.message else None

806

)

807

upstream: str | None = _upstream_for(root, local_b)

808

meta = read_branch_meta(root, local_b)

809

branch_intent: str | None = meta.get("intent") or None # type: ignore[assignment]

810

branch_intent = sanitize_display(branch_intent) if branch_intent else None

811

branch_resumable: bool = bool(meta.get("resumable", False))

812

created_by: str | None = (rec.agent_id if rec and rec.agent_id else None)

813

result.append({

814

"name": b,

815

"current": local_b == current,

816

"commit_id": commit_id or None,

817

"committed_at": rec.committed_at.isoformat() if rec else None,

818

"last_message": last_message,

819

"upstream": upstream,

820

"intent": branch_intent,

821

"resumable": branch_resumable,

822

"created_by": created_by,

823

})

824

print(json.dumps(result))

825

return

826

827

for b in display_branches:

828

is_remote_entry = b.startswith("remotes/")

829

local_b = b.removeprefix("remotes/")

830

is_current = (local_b == current) and not is_remote_entry

831

marker = _c("* ", _GREEN, tty=tty) if is_current else " "

832

# Build the display name once; apply sanitization before any coloring

833

# so that ANSI codes from _c() are not accidentally re-sanitized.

834

safe_name = sanitize_display(b)

835

name_str = _c(safe_name, _GREEN, tty=tty) if is_current else safe_name

836

if verbose >= 1:

837

commit_id = _resolve_commit_id(root, b)

838

short = short_id(commit_id) if commit_id else "(empty)"

839

rec = read_commit(root, commit_id) if commit_id else None

840

msg = sanitize_display(rec.message.splitlines()[0][:48]) if rec and rec.message else ""

841

short_str = _c(short, _YELLOW, tty=tty)

842

if verbose >= 2:

843

upstream = _upstream_for(root, local_b)

844

up_str = (

845

f" [{_c(sanitize_display(upstream), _CYAN, tty=tty)}]"

846

if upstream else ""

847

)

848

print(f"{marker}{name_str} {short_str}{up_str} {msg}")

849

else:

850

print(f"{marker}{name_str} {short_str} {msg}")

851

else:

852

print(f"{marker}{name_str}")