"""muse status — show working-tree drift against HEAD. Output modes ------------ Default (color when stdout is a TTY):: On branch main Your branch is up to date with 'origin/main'. Changes since last commit: (use "muse commit -m " to record changes) modified: tracks/drums.mid new file: tracks/lead.mp3 deleted: tracks/scratch.mid renamed: tracks/old.mid → tracks/new.mid --short (color letter prefix when stdout is a TTY):: M tracks/drums.mid A tracks/lead.mp3 D tracks/scratch.mid R tracks/old.mid → tracks/new.mid --porcelain (machine-readable, stable for scripting — no color ever):: ## main M tracks/drums.mid A tracks/lead.mp3 D tracks/scratch.mid R tracks/old.mid → tracks/new.mid --json (agent-native, always stable):: { "branch": "main", "head_commit": "abc123…", "clean": true, "dirty": false, "upstream": null, "ahead": null, "behind": null, "total_changes": 0, "added": [], "modified": [], "deleted": [], "renamed": {}, "conflict_paths": [], "merge_in_progress": false, "merge_from": null, "conflict_count": 0, "checkout_interrupted": false, "checkout_target": null } --exit-code Exits 0 when the working tree is clean, 1 when dirty. Combine with ``--json`` for structured output plus a testable exit code. Color convention ---------------- yellow modified — file exists in both old and new snapshot, content changed green new file — file is new, not present in last commit red deleted — file was removed since last commit cyan renamed — file was moved or renamed since last commit """ from __future__ import annotations import argparse import json import logging import pathlib import sys from typing import TypedDict from muse.cli.commands.checkout import read_checkout_head from muse.cli.config import get_remote_head, get_upstream from muse.core._types import Manifest, Metadata from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.store import ( get_head_commit_id, get_head_snapshot_manifest, read_current_branch, walk_commits_between, ) from muse.core.validation import sanitize_display from muse.core.snapshot import directories_from_manifest from muse.domain import SnapshotManifest, StagePlugin from muse.plugins.registry import resolve_plugin logger = logging.getLogger(__name__) type _StageEntryMap = dict[str, Manifest] # Default domain when repo.json is absent or corrupt. Must match # the default used by ``muse init`` (currently "code"). _DEFAULT_DOMAIN = "code" class _UpstreamInfo(TypedDict): """Computed ahead/behind counts for the current branch vs its upstream.""" tracking_ref: str ahead: int | None behind: int | None line: str class _BranchOnlyJson(TypedDict, total=False): """JSON payload for ``--branch-only`` output.""" branch: str head_commit: str | None upstream: str | None ahead: int | None behind: int | None merge_in_progress: bool merge_from: str | None conflict_count: int class _StatusJson(TypedDict): """JSON payload for full ``muse status --json`` output. All keys are always present so agents can read them without ``dict.get`` guards. Schema ------ branch Current branch name. head_commit SHA-256 of the current HEAD commit; null on empty repo. upstream Remote name if a tracking upstream is configured, else null. clean True when working tree exactly matches HEAD. dirty Alias for ``not clean`` — conventional for CI checks. ahead Commits local is ahead of remote; null when no upstream. behind Commits local is behind remote; null when no upstream. total_changes Total count of added + modified + deleted + renamed items. added Paths/symbols added since HEAD. modified Paths/symbols modified since HEAD. deleted Paths/symbols deleted since HEAD. renamed Mapping of old → new path for renamed items. conflict_paths Symbol addresses with unresolved conflicts ([] when clean). merge_in_progress True when a merge is in progress. merge_from Branch being merged in; null when no merge in progress. conflict_count Number of unresolved conflicts (0 when clean). checkout_interrupted True when a previous checkout was killed mid-flight. checkout_target Branch or snapshot that the interrupted checkout was targeting. """ branch: str head_commit: str | None upstream: str | None clean: bool dirty: bool ahead: int | None behind: int | None total_changes: int added: list[str] modified: list[str] deleted: list[str] renamed: Manifest conflict_paths: list[str] merge_in_progress: bool merge_from: str | None conflict_count: int checkout_interrupted: bool checkout_target: str | None class _StagedStatusJson(TypedDict, total=False): """JSON payload for staged-status (code-domain) output.""" branch: str head_commit: str | None clean: bool dirty: bool staged: _StageEntryMap unstaged: Manifest untracked: list[str] conflict_paths: list[str] merge_in_progress: bool merge_from: str | None conflict_count: int checkout_interrupted: bool checkout_target: str | None _YELLOW = "\033[33m" _GREEN = "\033[32m" _RED = "\033[31m" _CYAN = "\033[36m" _BOLD = "\033[1m" _RESET = "\033[0m" def _color(text: str, ansi: str, is_tty: bool) -> str: """Wrap *text* in ANSI color codes only when writing to a TTY.""" return f"{_BOLD}{ansi}{text}{_RESET}" if is_tty else text def _compute_upstream_info( root: pathlib.Path, branch: str, upstream: str, ) -> _UpstreamInfo: """Compute ahead/behind counts and the human-readable tracking line once. Centralises all ``walk_commits_between`` calls so they are executed exactly once per ``muse status`` invocation regardless of output format. Previously the text path called ``_tracking_line`` (two BFS walks) and the JSON path re-implemented the same logic inline (two more BFS walks) — four total BFS traversals per status call. This helper performs at most two walks and returns a typed result consumed by both paths. Args: root: Repository root. branch: Current branch name. upstream: Upstream remote name (e.g. ``"origin"``). Returns: :class:`_UpstreamInfo` with ``tracking_ref``, ``ahead``, ``behind``, and a pre-formatted ``line`` for the text output. """ tracking_ref = f"{upstream}/{branch}" remote_head = get_remote_head(upstream, branch, root) if not remote_head: return _UpstreamInfo( tracking_ref=tracking_ref, ahead=None, behind=None, line=f"Tracking: {tracking_ref} (not yet pushed)", ) local_head = get_head_commit_id(root, branch) if not local_head: return _UpstreamInfo( tracking_ref=tracking_ref, ahead=None, behind=None, line=f"Tracking: {tracking_ref}", ) if local_head == remote_head: return _UpstreamInfo( tracking_ref=tracking_ref, ahead=0, behind=0, line=f"Your branch is up to date with '{tracking_ref}'.", ) # Both walks are necessary only for the diverged case; the common case # (up-to-date) returns early above without any BFS at all. ahead = len(walk_commits_between(root, local_head, remote_head)) behind = len(walk_commits_between(root, remote_head, local_head)) if ahead and behind: line = ( f"Your branch and '{tracking_ref}' have diverged, " f"and have {ahead} and {behind} different commits each." ) elif ahead: suffix = "commit" if ahead == 1 else "commits" line = f"Your branch is ahead of '{tracking_ref}' by {ahead} {suffix}." elif behind: suffix = "commit" if behind == 1 else "commits" line = f"Your branch is behind '{tracking_ref}' by {behind} {suffix}." else: line = f"Your branch is up to date with '{tracking_ref}'." return _UpstreamInfo( tracking_ref=tracking_ref, ahead=ahead, behind=behind, line=line, ) def _read_repo_meta(root: pathlib.Path) -> tuple[str, str]: """Read ``.muse/repo.json`` once and return ``(repo_id, domain)``. Returns sensible defaults on any read or parse failure rather than propagating an unhandled exception to the user. Status degrades gracefully to an empty diff in the worst case. The domain default is ``"code"`` — matching ``muse init``'s default — so that a corrupt or absent ``repo.json`` produces sensible ignore rules rather than silently switching to the ``midi`` domain. """ repo_json = root / ".muse" / "repo.json" try: data = json.loads(repo_json.read_text(encoding="utf-8")) repo_id_raw = data.get("repo_id", "") repo_id = str(repo_id_raw) if isinstance(repo_id_raw, str) and repo_id_raw else "" domain_raw = data.get("domain", "") domain = str(domain_raw) if isinstance(domain_raw, str) and domain_raw else _DEFAULT_DOMAIN return repo_id, domain except (OSError, json.JSONDecodeError): return "", _DEFAULT_DOMAIN def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse status`` subcommand and its flags.""" parser = subparsers.add_parser( "status", help="Show working-tree drift against HEAD.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--short", "-s", action="store_true", help="Condensed one-letter-per-file output.", ) parser.add_argument( "--porcelain", action="store_true", help="Machine-readable output (no color, stable format for scripts).", ) parser.add_argument( "--branch", "-b", action="store_true", dest="branch_only", help="Show branch/upstream info only — skip the file diff.", ) parser.add_argument( "--format", "-f", dest="fmt", default="text", metavar="FORMAT", help="Output format: text (default) or json.", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json. Implies structured, colorless output.", ) parser.add_argument( "--exit-code", action="store_true", dest="exit_code", help=( "Exit 0 when the working tree is clean, 1 when dirty. " "Combines with --json for structured output plus a testable exit code." ), ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Show working-tree drift against HEAD. Covers four scenarios in a single command: 1. **Clean tree** — confirms the working tree matches HEAD exactly. 2. **Dirty tree** — lists added / modified / deleted / renamed files. 3. **Merge in progress** — reports conflict count, the branch being merged, and the exact next steps needed to complete or abort the merge. This section appears before the normal diff so it is never overlooked. 4. **Staged index active** (code plugin) — shows the three-bucket staged / unstaged / untracked view identical to ``git status``. JSON output schema (all keys always present):: { "branch": "feat/foo", "head_commit": "abc123…", "clean": false, "dirty": true, "upstream": "origin", "ahead": 2, "behind": 0, "total_changes": 3, "added": ["new.py"], "modified": ["src/main.py"], "deleted": [], "renamed": {}, "conflict_paths": [], "merge_in_progress": false, "merge_from": null, "conflict_count": 0 } Exit codes: 0 — success (or clean tree when ``--exit-code`` is set). 1 — dirty working tree (only when ``--exit-code`` is given). 2 — usage error (invalid ``--format`` value). 3 — internal error (e.g. repository not found). """ from muse.core.merge_engine import read_merge_state fmt: str = args.fmt short: bool = args.short porcelain: bool = args.porcelain branch_only: bool = args.branch_only exit_code_flag: bool = args.exit_code if fmt not in ("text", "json"): print( f"❌ Unknown --format '{sanitize_display(fmt)}'. Choose text or json.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() try: branch = read_current_branch(root) except ValueError as exc: print(f"fatal: {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) repo_id, domain = _read_repo_meta(root) upstream = get_upstream(branch, root) # ── Checkout-interrupted state ──────────────────────────────────────────── # .muse/CHECKOUT_HEAD exists only when a previous checkout was killed # mid-flight. The working tree may be partially mutated; warn loudly so # the user knows to retry the checkout rather than treating missing files # as uncommitted deletions. checkout_target: str | None = read_checkout_head(root) checkout_interrupted: bool = checkout_target is not None # ── Merge-in-progress state ─────────────────────────────────────────────── merge_state = read_merge_state(root) merge_in_progress = merge_state is not None conflict_paths: list[str] = merge_state.conflict_paths if merge_state else [] conflict_count = len(conflict_paths) merge_from: str | None = merge_state.other_branch if merge_state else None # ── HEAD commit id ──────────────────────────────────────────────────────── head_commit: str | None = get_head_commit_id(root, branch) # ── Upstream ahead/behind (computed once, shared by all output formats) ── upstream_info: _UpstreamInfo | None = None if upstream: upstream_info = _compute_upstream_info(root, branch, upstream) # ── Text/porcelain: checkout-interrupted banner ─────────────────────────── if fmt != "json" and not porcelain and checkout_interrupted: safe_target = sanitize_display(checkout_target) if checkout_target else "" print( f"\n🚨 CHECKOUT INTERRUPTED — the previous checkout to " f"'{safe_target}' did not complete.", file=sys.stderr, ) print( " The working tree may be partially mutated.\n" " Files shown as 'deleted' below may be missing because of the\n" " interrupted checkout, not because you deleted them.\n" " Next steps:", file=sys.stderr, ) print( f" muse checkout {safe_target} # retry the checkout\n" " muse checkout # switch to a different branch", file=sys.stderr, ) # ── Text/porcelain: merge banner and branch line ────────────────────────── if fmt != "json" and not porcelain: if not short: print(f"On branch {sanitize_display(branch)}") if upstream_info: print(upstream_info["line"]) if merge_in_progress: safe_merge_from = sanitize_display(merge_from) if merge_from else "" label = f" merging '{safe_merge_from}'" if safe_merge_from else "" print(f"\n⚠️ You have an unresolved merge in progress{label}.") if conflict_count: print(f" {conflict_count} unresolved conflict(s).") print(" Next steps:") print(" muse conflicts # see all conflicts") print(" muse checkout --ours # accept your version") print(" muse checkout --theirs # accept their version") print(" muse checkout --ours --all # resolve all — keep ours") print(" muse checkout --theirs --all # resolve all — keep theirs") print(" muse commit # once all resolved") print(" muse merge --abort # cancel the merge") else: print(" All conflicts resolved — run `muse commit` to complete the merge.") # ── Branch-only mode ────────────────────────────────────────────────────── if branch_only: if fmt == "json": out = _BranchOnlyJson( branch=branch, head_commit=head_commit, upstream=upstream, ahead=upstream_info["ahead"] if upstream_info else None, behind=upstream_info["behind"] if upstream_info else None, ) if merge_in_progress: out["merge_in_progress"] = True out["merge_from"] = merge_from out["conflict_count"] = conflict_count print(json.dumps(out)) return is_tty = sys.stdout.isatty() and not porcelain and fmt != "json" plugin = resolve_plugin(root) # ── Staged-index path (code domain with active stage) ──────────────────── if isinstance(plugin, StagePlugin) and plugin.stage_index_path(root).exists(): _render_staged_status( root, plugin, branch, head_commit, fmt, short, porcelain, is_tty, merge_in_progress=merge_in_progress, conflict_paths=conflict_paths, merge_from=merge_from, exit_code_flag=exit_code_flag, checkout_interrupted=checkout_interrupted, checkout_target=checkout_target, ) return # ── Drift computation ───────────────────────────────────────────────────── head_manifest = get_head_snapshot_manifest(root, repo_id, branch) or {} committed_snap = SnapshotManifest(files=head_manifest, domain=domain, directories=directories_from_manifest(head_manifest)) report = plugin.drift(committed_snap, root) delta = report.delta added: set[str] = set() modified: set[str] = set() deleted: set[str] = set() renamed: Manifest = {} for op in delta["ops"]: op_type = op["op"] addr = op["address"] if op_type == "insert": added.add(addr) elif op_type == "delete": deleted.add(addr) elif op_type == "replace": modified.add(addr) elif op_type == "patch": from_addr = op.get("from_address") if from_addr: renamed[str(from_addr)] = addr else: modified.add(addr) elif op_type == "directory_rename": from_addr = op.get("from_address") if from_addr: renamed[str(from_addr)] = addr clean = not (added or modified or deleted or renamed) dirty = not clean # ── JSON output ─────────────────────────────────────────────────────────── if fmt == "json": out_json = _StatusJson( branch=branch, head_commit=head_commit, upstream=upstream, clean=clean, dirty=dirty, ahead=upstream_info["ahead"] if upstream_info else None, behind=upstream_info["behind"] if upstream_info else None, total_changes=len(added) + len(modified) + len(deleted) + len(renamed), added=sorted(added), modified=sorted(modified), deleted=sorted(deleted), renamed=renamed, conflict_paths=conflict_paths, merge_in_progress=merge_in_progress, merge_from=merge_from, conflict_count=conflict_count, checkout_interrupted=checkout_interrupted, checkout_target=checkout_target, ) print(json.dumps(out_json)) if exit_code_flag and dirty: raise SystemExit(1) return # ── Porcelain output ────────────────────────────────────────────────────── if porcelain: suffix = " (MERGING)" if merge_in_progress else "" print(f"## {sanitize_display(branch)}{suffix}") for p in sorted(modified): print(f" M {p}") for p in sorted(added): print(f" A {p}") for p in sorted(deleted): print(f" D {p}") for old, new in sorted(renamed.items()): print(f" R {old} → {new}") if exit_code_flag and dirty: raise SystemExit(1) return # ── Short output ────────────────────────────────────────────────────────── if short: for p in sorted(modified): print(f" {_color('M', _YELLOW, is_tty)} {p}") for p in sorted(added): print(f" {_color('A', _GREEN, is_tty)} {p}") for p in sorted(deleted): print(f" {_color('D', _RED, is_tty)} {p}") for old, new in sorted(renamed.items()): print(f" {_color('R', _CYAN, is_tty)} {old} → {new}") if exit_code_flag and dirty: raise SystemExit(1) return # ── Long text output ────────────────────────────────────────────────────── if clean and not merge_in_progress: print("\nNothing to commit, working tree clean") if exit_code_flag: raise SystemExit(0) return if not clean: print("\nChanges since last commit:") print(' (use "muse commit -m " to record changes)\n') for p in sorted(modified): print(f"\t{_color(' modified:', _YELLOW, is_tty)} {sanitize_display(p)}") for p in sorted(added): print(f"\t{_color(' new file:', _GREEN, is_tty)} {sanitize_display(p)}") for p in sorted(deleted): print(f"\t{_color(' deleted:', _RED, is_tty)} {sanitize_display(p)}") for old, new in sorted(renamed.items()): print( f"\t{_color(' renamed:', _CYAN, is_tty)} " f"{sanitize_display(old)} → {sanitize_display(new)}" ) if exit_code_flag and dirty: raise SystemExit(1) def _render_staged_status( root: pathlib.Path, plugin: StagePlugin, branch: str, head_commit: str | None, fmt: str, short: bool, porcelain: bool, is_tty: bool, *, merge_in_progress: bool = False, conflict_paths: list[str] | None = None, merge_from: str | None = None, exit_code_flag: bool = False, checkout_interrupted: bool = False, checkout_target: str | None = None, ) -> None: """Render the three-bucket staged / unstaged / untracked view. Displayed when the active plugin implements :class:`~muse.domain.StagePlugin` and a stage index is present. Mirrors ``git status`` long-form output. Args: root: Repository root. plugin: Active plugin (must implement :class:`StagePlugin`). branch: Current branch name. head_commit: SHA-256 of HEAD commit (null on empty repo). fmt: Output format — ``"text"`` or ``"json"``. short: Render condensed one-letter-per-file output. porcelain: Render machine-readable no-color output. is_tty: True when stdout is a terminal (enables color). merge_in_progress: True when a merge is in progress. conflict_paths: Paths with unresolved merge conflicts. merge_from: Branch being merged in. exit_code_flag: Exit 1 when dirty. """ status = plugin.stage_status(root) staged = status["staged"] unstaged = status["unstaged"] untracked = status["untracked"] clean = not staged and not unstaged and not untracked dirty = not clean _conflict_paths: list[str] = conflict_paths or [] _MODE_LABEL: Metadata = { "A": "new file", "M": "modified", "D": "deleted", } if fmt == "json": out = _StagedStatusJson( branch=branch, head_commit=head_commit, clean=clean, dirty=dirty, staged={ p: {"mode": e["mode"], "object_id": e["object_id"]} for p, e in staged.items() }, unstaged=unstaged, untracked=untracked, conflict_paths=_conflict_paths, merge_in_progress=merge_in_progress, merge_from=merge_from, conflict_count=len(_conflict_paths), checkout_interrupted=checkout_interrupted, checkout_target=checkout_target, ) print(json.dumps(out)) if exit_code_flag and dirty: raise SystemExit(1) return if porcelain: suffix = " (MERGING)" if merge_in_progress else "" print(f"## {sanitize_display(branch)}{suffix}") for p, entry in sorted(staged.items()): print(f"{entry['mode']} {p}") for p, label in sorted(unstaged.items()): pu_letter = "M" if label == "modified" else "D" print(f" {pu_letter} {p}") for p in untracked: print(f"?? {p}") if exit_code_flag and dirty: raise SystemExit(1) return if short: for p, entry in sorted(staged.items()): s_mode = entry["mode"] color = _GREEN if s_mode == "A" else _YELLOW if s_mode == "M" else _RED print(f"{_color(s_mode, color, is_tty)} {p}") for p, label in sorted(unstaged.items()): u_letter = "M" if label == "modified" else "D" u_color = _YELLOW if label == "modified" else _RED print(f" {_color(u_letter, u_color, is_tty)} {p}") for p in untracked: print(f"?? {p}") if exit_code_flag and dirty: raise SystemExit(1) return # Long form — mirrors git status exactly. if staged: print("\nChanges staged for commit:") print(' (use "muse code reset HEAD " to unstage)\n') for p, entry in sorted(staged.items()): label = _MODE_LABEL.get(entry["mode"], entry["mode"]) color = _GREEN if entry["mode"] == "A" else _YELLOW if entry["mode"] == "M" else _RED pad = max(0, 10 - len(label)) print(f"\t{_color(label + ':', color, is_tty)}{' ' * pad} {p}") if unstaged: print("\nChanges not staged for commit:") print(' (use "muse code add " to update what will be committed)\n') for p, label in sorted(unstaged.items()): color = _YELLOW if label == "modified" else _RED pad = max(0, 10 - len(label)) print(f"\t{_color(label + ':', color, is_tty)}{' ' * pad} {p}") if untracked: print("\nUntracked files:") print(' (use "muse code add " to include in what will be committed)\n') for p in untracked: print(f"\t{p}") if clean and not merge_in_progress: print("\nNothing to commit, working tree clean") if staged: print() # trailing newline after last section if exit_code_flag and dirty: raise SystemExit(1)