"""``muse verify`` — whole-repository integrity check. Walks every reachable commit from every branch ref and performs a five-tier integrity check: 1. Every branch ref points to an existing, well-formed commit. 2. Every commit's snapshot exists. 3. Every object referenced by every snapshot exists, and (unless ``--no-objects``) its SHA-256 is recomputed to detect silent corruption. 4. For every commit that carries an HMAC signature, the signature is verified against the agent key stored under ``.muse/keys/.key``. 5. Missing agent keys are reported separately as ``kind="key_missing"`` so agents can distinguish key-rotation events from genuine tamper detection (``kind="signature"``). This is Muse's equivalent of ``git fsck``. Run it periodically on long-lived agent repositories or after recovering from a storage failure. Usage:: muse verify # full check — re-hashes all objects muse verify --no-objects # existence check only (faster) muse verify --branch feat/x # check one branch only muse verify --fail-fast # stop on first failure (CI-friendly) muse verify --quiet # no output — exit code only muse verify --json # machine-readable report JSON output schema:: { "repo_id": "", "refs_checked": , "commits_checked": , "snapshots_checked": , "objects_checked": , "signatures_checked": , "all_ok": , "check_objects": , "branch": "", "fail_fast": , "failures": [ { "kind": "ref|commit|snapshot|object|signature|key_missing", "id": "", "error": "" } ] } Exit codes:: 0 — all checks passed 1 — one or more integrity failures detected 3 — I/O error reading repository files """ from __future__ import annotations import argparse import json import logging import sys from typing import TypedDict from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.validation import sanitize_display from muse.core.verify import VerifyFailure, VerifyResult, run_verify logger = logging.getLogger(__name__) class _VerifyJson(TypedDict): """JSON wire format for ``muse verify --json``.""" repo_id: str refs_checked: int commits_checked: int snapshots_checked: int objects_checked: int signatures_checked: int all_ok: bool check_objects: bool branch: str | None fail_fast: bool failures: list[VerifyFailure] def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the verify subcommand.""" parser = subparsers.add_parser( "verify", help="Check repository integrity — commits, snapshots, and objects.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--quiet", "-q", action="store_true", help="No output — exit code only.", ) parser.add_argument( "--no-objects", "-O", action="store_true", dest="no_objects", help="Existence check only — skip object re-hashing (faster).", ) parser.add_argument( "--branch", "-b", metavar="BRANCH", default=None, help="Verify only the named branch instead of all branches.", ) parser.add_argument( "--fail-fast", "-F", action="store_true", dest="fail_fast", help="Stop on the first failure (useful in CI pipelines).", ) parser.add_argument( "--json", action="store_true", dest="json_out", help="Emit a machine-readable JSON report on stdout.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Check repository integrity — commits, snapshots, and objects. Walks every reachable commit from every branch ref. For each commit, verifies that the snapshot exists. For each snapshot, verifies that every object file exists and (by default) re-hashes it to detect bit-rot. JSON output includes ``repo_id``, per-tier counters, and the full ``failures`` list with ``kind``, ``id``, and ``error`` for each failure. Failures with ``kind="key_missing"`` indicate that an agent key is absent (possible key rotation) and should not be treated as hard corruption. Exit code is 0 when all checks pass, 1 when any failure is found. Use ``--quiet`` in scripts that only care about the exit code. Use ``--json`` for agent pipelines that parse the failure list. Examples:: muse verify # full integrity check muse verify --no-objects # fast existence-only check muse verify --branch feat/x # check one branch only muse verify --fail-fast # abort on first failure muse verify --quiet && echo "healthy" muse verify --json | jq '.failures' """ quiet: bool = args.quiet no_objects: bool = args.no_objects json_out: bool = args.json_out branch: str | None = args.branch fail_fast: bool = args.fail_fast root = require_repo() try: result = run_verify( root, check_objects=not no_objects, branch=branch, fail_fast=fail_fast, ) except OSError as exc: if not quiet: print(f"❌ I/O error during verify: {exc}", file=sys.stderr) raise SystemExit(ExitCode.INTERNAL_ERROR) from exc if quiet: raise SystemExit(0 if result["all_ok"] else ExitCode.USER_ERROR) if json_out: repo_id = read_repo_id(root) or "" payload: _VerifyJson = { "repo_id": repo_id, "refs_checked": result["refs_checked"], "commits_checked": result["commits_checked"], "snapshots_checked": result["snapshots_checked"], "objects_checked": result["objects_checked"], "signatures_checked": result["signatures_checked"], "all_ok": result["all_ok"], "check_objects": not no_objects, "branch": branch, "fail_fast": fail_fast, "failures": result["failures"], } print(json.dumps(payload, indent=2)) else: _print_text(result, no_objects=no_objects, branch=branch) raise SystemExit(0 if result["all_ok"] else ExitCode.USER_ERROR) def _print_text( result: VerifyResult, *, no_objects: bool, branch: str | None, ) -> None: """Render a human-readable summary of the verify result.""" if branch: print(f"Scope: branch '{sanitize_display(branch)}'") print(f"Checking refs... {result['refs_checked']} ref(s)") print(f"Checking commits... {result['commits_checked']} commit(s)") print(f"Checking snapshots... {result['snapshots_checked']} snapshot(s)") action = "existence only" if no_objects else "re-hashed" print(f"Checking objects... {result['objects_checked']} object(s) [{action}]") print(f"Checking signatures... {result['signatures_checked']} signed commit(s)") if result["all_ok"]: print("✅ Repository is healthy.") else: failure_count = len(result["failures"]) print(f"\n❌ {failure_count} integrity failure(s):") for f in result["failures"]: kind = sanitize_display(f["kind"]) err = sanitize_display(f["error"]) print(f" {kind:<12} {f['id'][:24]} {err}")