"""``overseer status`` command (§K4.4).""" from __future__ import annotations from argparse import Namespace from pathlib import Path from adapters.config import load_config from adapters.errors import ConfigError, ReadError from cli.context import CliContext from cli.digest import sha256_hex from cli.drift import compute_drift from cli.footprint import resolve_footprint from cli.kit_root import kit_version from cli.output import CommandReport from cli.paths import is_within_repo, resolve_config_path, resolve_repo_root from cli.sanitize import format_config_error, sanitize_text from cli.vcs_status import read_vcs_status, vcs_report from cli.version_lock import LockError, lock_path, read_version_lock from cli.commands.route import routing_policy_status from tools.cost_awareness.format import format_cost_awareness_lines from tools.cost_awareness.surface import build_cost_awareness_report, cost_awareness_payload from adapters.factory import create_adapter from tools.footprint_coverage import check_footprint_coverage from tools.footprint_integrity import check_footprint_integrity from tools.governance_freshness import check_governance_freshness from tools.governance_gates import scan_governance_gates from tools.governance_gates.format import format_pending_gate_lines, pending_gates_payload from tools.land_closeout import check_land_closeout, land_closeout_payload from tools.muse_sync import check_muse_sync from tools.substrate_health import check_substrate from tools.verification_evidence_gate import ( build_verification_evidence_gate, format_verification_evidence_gate_line, verification_evidence_gate_payload, ) from tools.independent_second_reviewer import ( build_independent_second_reviewer_gate, format_independent_second_reviewer_gate_line, independent_second_reviewer_gate_payload, ) from tools.optional_feature_tips import ( build_optional_feature_tips, optional_feature_tips_payload, ) from tools.workspace import build_status_report from tools.workspace.board_names import status_board_name_advisory GOVERNANCE_SYNC_MARKER = "last_governance_sync" IDE_WORKSPACE_HINT = ( "ide: open the repo root (folder containing .overseer) so .cursor/rules load" ) def _read_governance_sync_marker(repo_root: Path) -> str | None: """Return timestamp line only (backward-compatible; tip fields live under governance_freshness).""" marker = repo_root / ".overseer" / GOVERNANCE_SYNC_MARKER if not marker.is_file(): return None text = marker.read_text(encoding="utf-8") line = text.splitlines()[0].strip() if text.strip() else "" return line or None def _governance_freshness_payload(report) -> dict: return { "state": report.state, "ok": report.ok, "message": report.message, "remediation": report.remediation, "d1": report.d1, "d2": report.d2, "marker_present": report.marker_present, "marker_r1": report.marker_r1, "actual_r1": report.actual_r1, } def _compute_footprint_integrity( repo_root: Path, lock, rendered, ) -> tuple[str, list[str]]: """Recompute kit-only digest; report preserved-living paths separately (§K6.4).""" from cli.version_lock import ORIGIN_PRESERVED, compute_lock_digest, entry_origin prior = {entry.path: entry for entry in lock.footprint} preserved_living: list[str] = [] kit_entries = [] for item in rendered: entry = prior.get(item.destination) origin = entry_origin(entry) if entry is not None else "kit" dest = repo_root / item.destination if dest.is_file(): content = dest.read_bytes() else: content = b"" digest_hex = sha256_hex(content) if origin == ORIGIN_PRESERVED: preserved_living.append(item.destination) continue from cli.version_lock import FootprintEntry, ORIGIN_KIT kit_entries.append( FootprintEntry( path=item.destination, source=item.source, sha256=digest_hex, origin=ORIGIN_KIT, ) ) computed = compute_lock_digest(kit_entries) integrity = "ok" if computed == lock.footprint_digest else "mismatch" return integrity, preserved_living def _lock_summary(lock) -> dict: return { "lock_version": lock.lock_version, "kit_version": lock.kit_version, "config_version": lock.config_version, "footprint_digest": lock.footprint_digest, "installed_at": lock.installed_at, "synced_at": lock.synced_at, } def _exit_code_from_conditions( *, config_error: bool, integrity: str | None, drift_status: str | None, use_exit_code: bool, substrate_ok: bool = True, muse_sync_ok: bool = True, footprint_self_integrity_ok: bool = True, footprint_coverage_ok: bool = True, verification_evidence_gate_ok: bool = True, independent_second_reviewer_gate_ok: bool = True, governance_freshness_ok: bool = True, land_closeout_ok: bool = True, workspace_ok: bool = True, ) -> int: """Apply frozen precedence: 2 > 6 > 35 > 3 > 0 (§MR.7.2). §KH2.5: ``muse_sync_ok`` folds into the 2 tier. §KH3.5: ``footprint_self_integrity_ok`` (declared-but-absent kit-owned files) also folds into the same 2 tier, distinct from and independent of the opt-in ``--check-footprint`` content-digest ``integrity`` tier (6). §GFG.6: ``governance_freshness_ok`` (D1/D2 drift or stale marker) folds into the same 2 tier. §PMHF.6.1: ``land_closeout_ok`` folds into the same 2 tier; ``land_a_in_progress`` has ``ok=True`` so waiting for merge never false-fails status. §MR.7.2: workspace relay failure is ``35`` and overrides drift ``3`` / clean ``0`` without overriding config/substrate/muse_sync/footprint-self-integrity/governance-freshness (``2``) or lock integrity (``6``). """ if not use_exit_code: return 0 if ( config_error or not substrate_ok or not muse_sync_ok or not footprint_self_integrity_ok or not footprint_coverage_ok or not verification_evidence_gate_ok or not independent_second_reviewer_gate_ok or not governance_freshness_ok or not land_closeout_ok ): return 2 if integrity == "mismatch": return 6 if not workspace_ok: return 35 if drift_status in {"behind", "ahead"}: return 3 return 0 def run_status(args: Namespace, ctx: CliContext) -> int: """Execute ``overseer status``.""" report = CommandReport() repo_root = resolve_repo_root(cwd=ctx.cwd, repo_arg=args.repo, command="status") overseer_dir = repo_root / ".overseer" if not overseer_dir.is_dir(): payload = { "initialized": False, "warnings": [], } if ctx.output.json_mode: ctx.output.emit_json(payload) else: ctx.output.emit("not initialized") return 0 config_path = resolve_config_path(repo_root, args.config) if not is_within_repo(repo_root, config_path): ctx.output.error("refused: config path outside repo root") return 4 config_error = False lock_error = False try: config = load_config(config_path) except ConfigError as exc: config_error = True ctx.output.error(format_config_error(exc, repo_root)) payload = { "initialized": True, "error": str(exc), "warnings": report.warnings, } if ctx.output.json_mode: ctx.output.emit_json(payload) return 2 substrate = check_substrate(config, repo_root) if not substrate.ok: report.add_warning(f"substrate: {substrate.state} — {substrate.message}") if substrate.remediation: report.add_warning(f"substrate-remediation: {substrate.remediation}") gate_scan = None if config.governance_gates.remind and "status" in config.governance_gates.surfaces: gate_scan = scan_governance_gates(config, repo_root) if gate_scan.pending: for line in format_pending_gate_lines(gate_scan): report.add_warning(line) lock_file = lock_path(repo_root) lock = None try: lock = read_version_lock(lock_file) except LockError as exc: lock_error = True report.add_warning(str(exc)) rendered: list = [] integrity: str | None = None preserved_living: list[str] = [] if lock is not None: try: rendered = resolve_footprint(config, kit=ctx.kit) except ConfigError as exc: config_error = True ctx.output.error(format_config_error(exc, repo_root)) return 2 if args.check_footprint: integrity, preserved_living = _compute_footprint_integrity(repo_root, lock, rendered) if integrity == "mismatch": report.add_warning("footprint_integrity: mismatch") for path in preserved_living: report.add_warning(f"preserved-living: {path}") footprint_self_integrity = check_footprint_integrity(repo_root, lock=lock) if not footprint_self_integrity.ok: report.add_warning( f"footprint_self_integrity: {footprint_self_integrity.state} — " f"{footprint_self_integrity.message}" ) if footprint_self_integrity.remediation: report.add_warning( f"footprint_self_integrity-remediation: {footprint_self_integrity.remediation}" ) footprint_coverage = check_footprint_coverage( repo_root, config, lock=lock, rendered=rendered if rendered else None, kit=ctx.kit, ) if not footprint_coverage.ok: report.add_warning( f"footprint_coverage: {footprint_coverage.state} — {footprint_coverage.message}" ) if footprint_coverage.remediation: report.add_warning( f"footprint_coverage-remediation: {footprint_coverage.remediation}" ) drift = compute_drift( cli_version=kit_version(), lock=lock, rendered=rendered, repo_root=repo_root, ) if drift["status"] in {"behind", "ahead"}: report.add_warning(f"drift: {drift['status']}") vcs_result = read_vcs_status(config, repo_root, ctx.runner) if isinstance(vcs_result, ReadError): ctx.output.error(sanitize_text(str(vcs_result), repo_root)) payload = { "initialized": True, "substrate": _substrate_payload(substrate), "vcs": { "error": sanitize_text(str(vcs_result), repo_root), "command": sanitize_text(vcs_result.command, repo_root), }, "warnings": report.warnings, } if ctx.output.json_mode: ctx.output.emit_json(payload) return 2 muse_sync = check_muse_sync(config, vcs_result) if not muse_sync.ok: report.add_warning(f"muse_sync: {muse_sync.state} — {muse_sync.message}") if muse_sync.remediation: report.add_warning(f"muse_sync-remediation: {muse_sync.remediation}") governance_freshness = check_governance_freshness( config, repo_root, adapter=create_adapter(config, repo_root, runner=ctx.runner), runner=ctx.runner, ) if not governance_freshness.ok: report.add_warning( f"governance_freshness: {governance_freshness.state} — {governance_freshness.message}" ) if governance_freshness.remediation: report.add_warning( f"governance_freshness-remediation: {governance_freshness.remediation}" ) # §PMHF.6.1: always-on land closeout probe; never invokes gh from status. land_closeout = check_land_closeout( config, repo_root, runner=ctx.runner, probe_merged_pr=False, freshness=governance_freshness, ) if not land_closeout.ok: report.add_warning(f"land_closeout: {land_closeout.state} — {land_closeout.message}") if land_closeout.remediation: report.add_warning(f"land_closeout-remediation: {land_closeout.remediation}") model_routing_status = routing_policy_status(config, repo_root, kit_root=ctx.kit) if model_routing_status.get("enabled") and not model_routing_status.get("valid"): violation = model_routing_status.get("violation") or "invalid" report.add_warning(f"model_routing: invalid — {violation}") cost_report = None if config.cost_awareness.enabled and "status" in config.cost_awareness.surfaces: cost_report = build_cost_awareness_report(config, repo_root, kit_root=ctx.kit) if cost_report.invalid: violation = cost_report.violation or "invalid" report.add_warning(f"cost_awareness: invalid — {violation}") elif cost_report.exit_code == 31: ctx.output.error(cost_report.violation or "routing policy file missing or unreadable") payload = { "initialized": True, "cost_awareness": cost_awareness_payload(cost_report), "warnings": report.warnings, } if ctx.output.json_mode: ctx.output.emit_json(payload) return 31 else: for line in format_cost_awareness_lines(cost_report): report.add_warning(line) handover_path = repo_root / config.repo.root_relative_docs / config.docs.handover roadmap_path = repo_root / config.repo.root_relative_docs / config.docs.roadmap handover_text = handover_path.read_text(encoding="utf-8") if handover_path.is_file() else None roadmap_text = roadmap_path.read_text(encoding="utf-8") if roadmap_path.is_file() else None verification_evidence_gate = build_verification_evidence_gate( config, repo_root, handover_text=handover_text, roadmap_text=roadmap_text, ) ve_line = format_verification_evidence_gate_line(verification_evidence_gate) if ve_line: report.add_warning(ve_line) independent_second_reviewer_gate = build_independent_second_reviewer_gate( config, repo_root, handover_text=handover_text, roadmap_text=roadmap_text, ) isr_line = format_independent_second_reviewer_gate_line(independent_second_reviewer_gate) if isr_line: report.add_warning(isr_line) optional_feature_tips = build_optional_feature_tips(config) # §NXP.6 — warn only; must not fold into --exit-code. board_name_advisory = status_board_name_advisory(config) if board_name_advisory is not None: report.add_warning(board_name_advisory) payload = { "initialized": True, "kit_version": kit_version(), "substrate": _substrate_payload(substrate), "muse_sync": _muse_sync_payload(muse_sync), "footprint_self_integrity": _footprint_self_integrity_payload(footprint_self_integrity), "footprint_coverage": _footprint_coverage_payload(footprint_coverage), "ide_workspace_hint": IDE_WORKSPACE_HINT, "optional_feature_tips": optional_feature_tips_payload(optional_feature_tips), "governance_freshness": _governance_freshness_payload(governance_freshness), "land_closeout": land_closeout_payload(land_closeout), "lock": _lock_summary(lock) if lock else None, "drift": drift, "footprint_integrity": integrity, "preserved_living": preserved_living if args.check_footprint else [], "vcs": vcs_report(vcs_result, config), "last_governance_sync": _read_governance_sync_marker(repo_root), "governance_gates": pending_gates_payload(gate_scan) if gate_scan is not None else {"enabled": False, "suppressed": False, "active_phases": [], "pending": []}, "model_routing": model_routing_status, "warnings": report.warnings, "board_name_advisory": board_name_advisory, } if cost_report is not None: payload["cost_awareness"] = cost_awareness_payload(cost_report) ve_payload = verification_evidence_gate_payload(verification_evidence_gate) if ve_payload is not None: payload["verification_evidence_gate"] = ve_payload isr_payload = independent_second_reviewer_gate_payload(independent_second_reviewer_gate) if isr_payload is not None: payload["independent_second_reviewer_gate"] = isr_payload if lock_error: payload["lock_error"] = True workspace_report = None workspace_ok = True if getattr(args, "workspace", False): workspace_report = build_status_report(config, repo_root) payload["workspace"] = workspace_report.to_json() # S9: single-repo green must not imply workspace.ok; workspace failure # contributes exit 35 only when --exit-code is set. if workspace_report.configured and not workspace_report.ok: workspace_ok = False report.add_warning(f"workspace: {workspace_report.state}") exit_code = _exit_code_from_conditions( config_error=config_error, integrity=integrity if args.check_footprint else None, drift_status=drift["status"], use_exit_code=args.exit_code, substrate_ok=substrate.ok, muse_sync_ok=muse_sync.ok, footprint_self_integrity_ok=footprint_self_integrity.ok, footprint_coverage_ok=footprint_coverage.ok, verification_evidence_gate_ok=verification_evidence_gate.ok, independent_second_reviewer_gate_ok=independent_second_reviewer_gate.ok, governance_freshness_ok=governance_freshness.ok, land_closeout_ok=land_closeout.ok, workspace_ok=workspace_ok, ) if lock_error and args.exit_code: exit_code = 6 if exit_code in {0, 3, 35} else max(exit_code, 6) if ctx.output.json_mode: ctx.output.emit_json(payload) else: ctx.output.emit(f"kit_version: {payload['kit_version']}") if not substrate.ok: ctx.output.emit(f"substrate: {substrate.state} — {substrate.message}") if substrate.remediation: ctx.output.emit(f"substrate-remediation: {substrate.remediation}") if not muse_sync.ok: ctx.output.emit(f"muse_sync: {muse_sync.state} — {muse_sync.message}") if muse_sync.remediation: ctx.output.emit(f"muse_sync-remediation: {muse_sync.remediation}") if not governance_freshness.ok: ctx.output.emit( f"governance_freshness: {governance_freshness.state} — " f"{governance_freshness.message}" ) if governance_freshness.remediation: ctx.output.emit( f"governance_freshness-remediation: {governance_freshness.remediation}" ) if not land_closeout.ok: ctx.output.emit( f"land_closeout: {land_closeout.state} — {land_closeout.message}" ) if land_closeout.remediation: ctx.output.emit(f"land_closeout-remediation: {land_closeout.remediation}") if lock: ctx.output.emit(f"lock kit_version: {lock.kit_version}") ctx.output.emit(f"drift: {drift['status']}") if integrity: ctx.output.emit(f"footprint_integrity: {integrity}") if model_routing_status.get("enabled"): if model_routing_status.get("valid"): ctx.output.emit("model_routing: valid") else: violation = model_routing_status.get("violation") or "invalid" ctx.output.emit(f"model_routing: invalid — {violation}") if gate_scan is not None and gate_scan.pending: ctx.output.emit("") for line in format_pending_gate_lines(gate_scan): ctx.output.emit(line) if cost_report is not None and config.cost_awareness.enabled: if cost_report.invalid: violation = cost_report.violation or "invalid" ctx.output.emit(f"cost_awareness: invalid — {violation}") else: for line in format_cost_awareness_lines(cost_report): ctx.output.emit(line) if not footprint_coverage.ok: ctx.output.emit( f"footprint_coverage: {footprint_coverage.state} — {footprint_coverage.message}" ) if footprint_coverage.remediation: ctx.output.emit( f"footprint_coverage-remediation: {footprint_coverage.remediation}" ) ctx.output.emit(IDE_WORKSPACE_HINT) for line in optional_feature_tips: ctx.output.emit(line) if board_name_advisory is not None: ctx.output.emit(board_name_advisory) if ve_line: ctx.output.emit(ve_line) ctx.output.emit(f"vcs.regime: {vcs_result.regime}") ctx.output.emit(f"vcs.branch: {vcs_result.branch}") ctx.output.emit(f"vcs.dirty: {vcs_result.dirty}") if workspace_report is not None: ctx.output.emit(f"workspace.configured: {str(workspace_report.configured).lower()}") ctx.output.emit(f"workspace.ok: {str(workspace_report.ok).lower()}") ctx.output.emit(f"workspace.state: {workspace_report.state}") if workspace_report.authoritative_handover: ctx.output.emit( f"authoritative_handover: {workspace_report.authoritative_handover}" ) for member in workspace_report.members: base = member.get("handover_basename") or "?" ctx.output.emit(f"workspace.member {member['id']}: handover={base}") return exit_code def _substrate_payload(substrate) -> dict: return { "state": substrate.state, "ok": substrate.ok, "missing": list(substrate.missing), "remediation": substrate.remediation, "message": substrate.message, } def _muse_sync_payload(muse_sync) -> dict: return { "state": muse_sync.state, "ok": muse_sync.ok, "remediation": muse_sync.remediation, "message": muse_sync.message, } def _footprint_self_integrity_payload(report) -> dict: return { "state": report.state, "ok": report.ok, "missing": list(report.missing), "remediation": report.remediation, "message": report.message, } def _footprint_coverage_payload(report) -> dict: return { "state": report.state, "ok": report.ok, "missing": list(report.missing), "remediation": report.remediation, "message": report.message, }