"""``muse.core.bridge.status`` — Bridge status reporting. Decomposed from ``muse.cli.commands.bridge`` (issue #14). Exports ------- run_git_status Entry point for ``muse bridge git-status``. _compute_drift Compute git/Muse commit drift since last sync. _count_muse_commits_since Walk Muse DAG to count commits since a ref. _print_bridge_status_text Human-readable status output. """ from __future__ import annotations import argparse import json import pathlib import sys from typing import Any from muse.core.bridge.git_primitives import _git, _validate_git_sha from muse.core.bridge.state import BridgeState, DriftInfo, read_bridge_state from muse.core.errors import ExitCode from muse.core.paths import head_path, ref_path def run_git_status(args: argparse.Namespace) -> None: """Entry point for ``muse bridge git-status``. Reads ``.muse/git-bridge.toml`` and reports the last import and export sync points, plus drift counts (commits in git since last import, commits in Muse since last export). Args: args: Parsed argparse namespace; see :func:`register` for flag definitions. Raises: SystemExit(ExitCode.USER_ERROR): No Muse repository found. """ from muse.core.repo import find_repo_root root = find_repo_root() json_out: bool = args.json_out if root is None: msg = "No Muse repository found. Run 'muse init' first." if json_out: print(json.dumps({"error": msg, "exit_code": ExitCode.USER_ERROR})) else: print(f"❌ {msg}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) state = read_bridge_state(root) git_dir_arg = getattr(args, "git_dir", None) drift: DriftInfo | None = None if git_dir_arg: git_dir = pathlib.Path(git_dir_arg).resolve() if (git_dir / ".git").exists(): drift = _compute_drift(root, git_dir, state) if json_out: payload = { "last_import": dict(state["last_import"]), "last_export": dict(state["last_export"]), } if drift is not None: payload["drift"] = drift print(json.dumps(payload)) else: _print_bridge_status_text(state, drift) def _compute_drift( root: pathlib.Path, git_dir: pathlib.Path, state: "BridgeState", ) -> DriftInfo: """Compute commit drift between git and Muse since the last bridge sync. Returns a dict with two keys: * ``git_commits_since_import``: Number of git commits on HEAD that come after ``last_import.git_sha``. ``None`` when no import state exists. * ``muse_commits_since_export``: Number of Muse commits on the current HEAD branch that come after ``last_export.muse_commit_id``. ``None`` when no export state exists. Args: root: Muse repository root. git_dir: Path to the git repository for git-side drift counting. state: Current :class:`BridgeState`. Returns: Dict with ``git_commits_since_import`` and ``muse_commits_since_export`` integer (or ``None``) values. """ result: DriftInfo = { "git_commits_since_import": None, "muse_commits_since_export": None, } # --- Git side: count commits since last import SHA --- li = state.get("last_import") or {} last_git_sha = li.get("git_sha", "") if last_git_sha and _validate_git_sha(last_git_sha): try: raw = _git( git_dir, "rev-list", f"{last_git_sha}..HEAD", "--count", check=True, ).strip() result["git_commits_since_import"] = int(raw) except Exception: # noqa: BLE001 result["git_commits_since_import"] = None # --- Muse side: walk commit chain since last export muse_commit_id --- le = state.get("last_export") or {} last_muse_cid = le.get("muse_commit_id", "") if last_muse_cid: result["muse_commits_since_export"] = _count_muse_commits_since( root, last_muse_cid ) return result def _count_muse_commits_since(root: pathlib.Path, since_commit_id: str) -> int | None: """Count Muse commits on HEAD since *since_commit_id*. Walks the Muse commit DAG backwards from HEAD through ``parent_commit_id`` until it reaches *since_commit_id* or a commit with no parent. Uses the store's ``read_commit`` so the path and JSON format are always correct. Args: root: Muse repository root. since_commit_id: ``sha256:``-prefixed commit ID to stop counting at. Returns: Number of commits after *since_commit_id*, or ``None`` on any error. """ from muse.core.commits import read_commit as _read_commit try: _head_file = head_path(root) if not _head_file.exists(): return None head_content = _head_file.read_text().strip() if head_content.startswith("ref: refs/heads/"): branch = head_content[len("ref: refs/heads/"):] _branch_ref = ref_path(root, branch) if not _branch_ref.exists(): return None current_id = _branch_ref.read_text().strip() else: current_id = head_content if not current_id: return None count = 0 visited: set[str] = set() while current_id and current_id != since_commit_id: if current_id in visited: break # cycle guard visited.add(current_id) record = _read_commit(root, current_id) if record is None: break count += 1 current_id = record.parent_commit_id or "" return count except Exception: # noqa: BLE001 return None def _print_bridge_status_text( state: "BridgeState", drift: "dict[str, Any] | None" = None, ) -> None: """Print a human-readable bridge status summary to stdout.""" SEP = "─" * 50 print("Muse Bridge Status") print(SEP) li = state["last_import"] git_behind = drift.get("git_commits_since_import") if drift else None if li: print("Last git-import:") print(f" git SHA: {li.get('git_sha', '(none)')[:16]}…") print(f" git ref: {li.get('git_ref', '(none)')}") print(f" muse commit: {li.get('muse_commit_id', '(none)')[:23]}…") print(f" imported: {li.get('imported_at', '(none)')}") if git_behind is not None: print(f" commits behind: {git_behind}") else: print("Last git-import: (none)") print() le = state["last_export"] muse_ahead = drift.get("muse_commits_since_export") if drift else None if le: print("Last git-export:") print(f" muse commit: {le.get('muse_commit_id', '(none)')[:23]}…") print(f" git ref: {le.get('git_ref', '(none)')}") print(f" exported: {le.get('exported_at', '(none)')}") if muse_ahead is not None: print(f" commits ahead: {muse_ahead}") else: print("Last git-export: (none)") if drift is not None: print() print("Drift:") m_ahead = drift.get("muse_commits_since_export") g_behind = drift.get("git_commits_since_import") if m_ahead is not None: suffix = "commit" if m_ahead == 1 else "commits" print(f" Muse→Git: {m_ahead} {suffix} to export") else: print(" Muse→Git: (unknown)") if g_behind is not None: suffix = "commit" if g_behind == 1 else "commits" print(f" Git→Muse: {g_behind} {suffix} to import") else: print(" Git→Muse: (unknown)") def _register_git_status_parser( subs: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the ``git-status`` subcommand onto *subs*.""" import argparse as _ap p = subs.add_parser( "git-status", help="Show bridge state: last import/export and drift counts.", formatter_class=_ap.RawDescriptionHelpFormatter, description=( "Display the current Muse bridge state — what was last imported " "and exported, and how far ahead/behind each repo is." ), ) p.add_argument( "--git-dir", default=None, metavar="PATH", help="Path to the git repository for drift calculation.", ) p.add_argument( "--json", "-j", dest="json_out", action="store_true", default=False, help="Emit machine-readable JSON.", ) p.set_defaults(func=run_git_status)