status.py
python
| 1 | """muse status — show working-tree drift against HEAD. |
| 2 | |
| 3 | Output modes |
| 4 | ------------ |
| 5 | |
| 6 | Default (color when stdout is a TTY):: |
| 7 | |
| 8 | On branch main |
| 9 | |
| 10 | Changes since last commit: |
| 11 | (use "muse commit -m <msg>" to record changes) |
| 12 | |
| 13 | modified: tracks/drums.mid |
| 14 | new file: tracks/lead.mp3 |
| 15 | deleted: tracks/scratch.mid |
| 16 | |
| 17 | --short (color letter prefix when stdout is a TTY):: |
| 18 | |
| 19 | M tracks/drums.mid |
| 20 | A tracks/lead.mp3 |
| 21 | D tracks/scratch.mid |
| 22 | |
| 23 | --porcelain (machine-readable, stable for scripting — no color ever):: |
| 24 | |
| 25 | ## main |
| 26 | M tracks/drums.mid |
| 27 | A tracks/lead.mp3 |
| 28 | D tracks/scratch.mid |
| 29 | |
| 30 | Color convention |
| 31 | ---------------- |
| 32 | - yellow modified — file exists in both old and new snapshot, content changed |
| 33 | - green new file — file is new, not present in last commit |
| 34 | - red deleted — file was removed since last commit |
| 35 | """ |
| 36 | |
| 37 | from __future__ import annotations |
| 38 | |
| 39 | import json |
| 40 | import logging |
| 41 | import pathlib |
| 42 | import sys |
| 43 | |
| 44 | import typer |
| 45 | |
| 46 | from muse.core.errors import ExitCode |
| 47 | from muse.core.repo import require_repo |
| 48 | from muse.core.store import get_head_snapshot_manifest, read_current_branch |
| 49 | from muse.domain import SnapshotManifest |
| 50 | from muse.plugins.registry import resolve_plugin_by_domain |
| 51 | |
| 52 | logger = logging.getLogger(__name__) |
| 53 | |
| 54 | app = typer.Typer() |
| 55 | |
| 56 | # Change-type colors. Applied only when stdout is a TTY so piped output stays |
| 57 | # clean without needing --porcelain. |
| 58 | _YELLOW = typer.colors.YELLOW |
| 59 | _GREEN = typer.colors.GREEN |
| 60 | _RED = typer.colors.RED |
| 61 | |
| 62 | |
| 63 | def _read_repo_meta(root: pathlib.Path) -> tuple[str, str]: |
| 64 | """Read ``.muse/repo.json`` once and return ``(repo_id, domain)``. |
| 65 | |
| 66 | Returns sensible defaults on any read or parse failure rather than |
| 67 | propagating an unhandled exception to the user. The caller never needs |
| 68 | to guard against a missing or corrupt ``repo.json`` — status degrades |
| 69 | gracefully to an empty diff in the worst case. |
| 70 | """ |
| 71 | repo_json = root / ".muse" / "repo.json" |
| 72 | try: |
| 73 | data = json.loads(repo_json.read_text(encoding="utf-8")) |
| 74 | repo_id_raw = data.get("repo_id", "") |
| 75 | repo_id = str(repo_id_raw) if isinstance(repo_id_raw, str) and repo_id_raw else "" |
| 76 | domain_raw = data.get("domain", "") |
| 77 | domain = str(domain_raw) if isinstance(domain_raw, str) and domain_raw else "midi" |
| 78 | return repo_id, domain |
| 79 | except (OSError, json.JSONDecodeError): |
| 80 | return "", "midi" |
| 81 | |
| 82 | |
| 83 | @app.callback(invoke_without_command=True) |
| 84 | def status( |
| 85 | ctx: typer.Context, |
| 86 | short: bool = typer.Option(False, "--short", "-s", help="Condensed output."), |
| 87 | porcelain: bool = typer.Option(False, "--porcelain", help="Machine-readable output (no color)."), |
| 88 | branch_only: bool = typer.Option(False, "--branch", "-b", help="Show branch info only."), |
| 89 | fmt: str = typer.Option("text", "--format", "-f", help="Output format: text or json."), |
| 90 | ) -> None: |
| 91 | """Show working-tree drift against HEAD. |
| 92 | |
| 93 | Agents should pass ``--format json`` to receive structured output with |
| 94 | ``branch``, ``clean`` (bool), and ``added``, ``modified``, ``deleted`` |
| 95 | file lists. |
| 96 | """ |
| 97 | if fmt not in ("text", "json"): |
| 98 | from muse.core.validation import sanitize_display as _sd |
| 99 | typer.echo(f"❌ Unknown --format '{_sd(fmt)}'. Choose text or json.", err=True) |
| 100 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 101 | |
| 102 | root = require_repo() |
| 103 | try: |
| 104 | branch = read_current_branch(root) |
| 105 | except ValueError as exc: |
| 106 | typer.echo(f"fatal: {exc}", err=True) |
| 107 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 108 | |
| 109 | # Read repo.json exactly once — repo_id and domain both come from here. |
| 110 | # resolve_plugin_by_domain() uses the pre-read domain string, eliminating |
| 111 | # the two additional repo.json reads that resolve_plugin() and read_domain() |
| 112 | # would otherwise each trigger independently. |
| 113 | repo_id, domain = _read_repo_meta(root) |
| 114 | |
| 115 | # JSON mode prints everything at once at the end — skip all partial prints. |
| 116 | if fmt != "json": |
| 117 | if porcelain: |
| 118 | typer.echo(f"## {branch}") |
| 119 | elif not short: |
| 120 | typer.echo(f"On branch {branch}") |
| 121 | |
| 122 | # --branch: print only the branch header then exit, regardless of mode. |
| 123 | if branch_only: |
| 124 | if fmt == "json": |
| 125 | typer.echo(json.dumps({"branch": branch})) |
| 126 | return |
| 127 | |
| 128 | # Compute isatty once; it is a syscall and must not be repeated per line. |
| 129 | # Porcelain output is never colored, even on a TTY. |
| 130 | is_tty = sys.stdout.isatty() and not porcelain and fmt != "json" |
| 131 | |
| 132 | def _color(text: str, color: str) -> str: |
| 133 | return typer.style(text, fg=color, bold=True) if is_tty else text |
| 134 | |
| 135 | head_manifest = get_head_snapshot_manifest(root, repo_id, branch) or {} |
| 136 | plugin = resolve_plugin_by_domain(domain) |
| 137 | committed_snap = SnapshotManifest(files=head_manifest, domain=domain) |
| 138 | report = plugin.drift(committed_snap, root) |
| 139 | delta = report.delta |
| 140 | |
| 141 | added: set[str] = {op["address"] for op in delta["ops"] if op["op"] == "insert"} |
| 142 | modified: set[str] = {op["address"] for op in delta["ops"] if op["op"] in ("replace", "patch")} |
| 143 | deleted: set[str] = {op["address"] for op in delta["ops"] if op["op"] == "delete"} |
| 144 | |
| 145 | clean = not (added or modified or deleted) |
| 146 | |
| 147 | # --format json: always wins, no color, fully structured. |
| 148 | if fmt == "json": |
| 149 | typer.echo(json.dumps({ |
| 150 | "branch": branch, |
| 151 | "clean": clean, |
| 152 | "added": sorted(added), |
| 153 | "modified": sorted(modified), |
| 154 | "deleted": sorted(deleted), |
| 155 | })) |
| 156 | return |
| 157 | |
| 158 | if clean: |
| 159 | if not short and not porcelain: |
| 160 | typer.echo("\nNothing to commit, working tree clean") |
| 161 | return |
| 162 | |
| 163 | # --porcelain: stable machine-readable output, no color, ever. |
| 164 | if porcelain: |
| 165 | for p in sorted(modified): |
| 166 | typer.echo(f" M {p}") |
| 167 | for p in sorted(added): |
| 168 | typer.echo(f" A {p}") |
| 169 | for p in sorted(deleted): |
| 170 | typer.echo(f" D {p}") |
| 171 | return |
| 172 | |
| 173 | # --short: compact one-line-per-file, colored letter prefix. |
| 174 | if short: |
| 175 | for p in sorted(modified): |
| 176 | typer.echo(f" {_color('M', _YELLOW)} {p}") |
| 177 | for p in sorted(added): |
| 178 | typer.echo(f" {_color('A', _GREEN)} {p}") |
| 179 | for p in sorted(deleted): |
| 180 | typer.echo(f" {_color('D', _RED)} {p}") |
| 181 | return |
| 182 | |
| 183 | # Default: human-readable, colored label. |
| 184 | typer.echo("\nChanges since last commit:") |
| 185 | typer.echo(' (use "muse commit -m <msg>" to record changes)\n') |
| 186 | for p in sorted(modified): |
| 187 | typer.echo(f"\t{_color(' modified:', _YELLOW)} {p}") |
| 188 | for p in sorted(added): |
| 189 | typer.echo(f"\t{_color(' new file:', _GREEN)} {p}") |
| 190 | for p in sorted(deleted): |
| 191 | typer.echo(f"\t{_color(' deleted:', _RED)} {p}") |