aaronrene / overseer-kit public
status.py python
267 lines 8.9 KB
Raw
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439 feat: K1-P1 complete — agent provenance, build-verification… Sonnet 4.6 patch 53 days ago
1 """``overseer status`` command (§K4.4)."""
2
3 from __future__ import annotations
4
5 from argparse import Namespace
6 from pathlib import Path
7
8 from adapters.config import load_config
9 from adapters.errors import ConfigError, ReadError
10 from cli.context import CliContext
11 from cli.digest import sha256_hex
12 from cli.drift import compute_drift
13 from cli.footprint import resolve_footprint
14 from cli.kit_root import kit_version
15 from cli.output import CommandReport
16 from cli.paths import is_within_repo, resolve_config_path, resolve_repo_root
17 from cli.sanitize import format_config_error, sanitize_text
18 from cli.vcs_status import read_vcs_status, vcs_report
19 from cli.version_lock import LockError, lock_path, read_version_lock
20 from tools.governance_gates import scan_governance_gates
21 from tools.governance_gates.format import format_pending_gate_lines, pending_gates_payload
22 from tools.substrate_health import check_substrate
23
24
25 GOVERNANCE_SYNC_MARKER = "last_governance_sync"
26
27
28 def _read_governance_sync_marker(repo_root: Path) -> str | None:
29 marker = repo_root / ".overseer" / GOVERNANCE_SYNC_MARKER
30 if not marker.is_file():
31 return None
32 text = marker.read_text(encoding="utf-8").strip()
33 return text or None
34
35
36 def _compute_footprint_integrity(
37 repo_root: Path,
38 lock,
39 rendered,
40 ) -> tuple[str, list[str]]:
41 """Recompute kit-only digest; report preserved-living paths separately (§K6.4)."""
42 from cli.version_lock import ORIGIN_PRESERVED, compute_lock_digest, entry_origin
43
44 prior = {entry.path: entry for entry in lock.footprint}
45 preserved_living: list[str] = []
46 kit_entries = []
47 for item in rendered:
48 entry = prior.get(item.destination)
49 origin = entry_origin(entry) if entry is not None else "kit"
50 dest = repo_root / item.destination
51 if dest.is_file():
52 content = dest.read_bytes()
53 else:
54 content = b""
55 digest_hex = sha256_hex(content)
56 if origin == ORIGIN_PRESERVED:
57 preserved_living.append(item.destination)
58 continue
59 from cli.version_lock import FootprintEntry, ORIGIN_KIT
60
61 kit_entries.append(
62 FootprintEntry(
63 path=item.destination,
64 source=item.source,
65 sha256=digest_hex,
66 origin=ORIGIN_KIT,
67 )
68 )
69 computed = compute_lock_digest(kit_entries)
70 integrity = "ok" if computed == lock.footprint_digest else "mismatch"
71 return integrity, preserved_living
72
73
74 def _lock_summary(lock) -> dict:
75 return {
76 "lock_version": lock.lock_version,
77 "kit_version": lock.kit_version,
78 "config_version": lock.config_version,
79 "footprint_digest": lock.footprint_digest,
80 "installed_at": lock.installed_at,
81 "synced_at": lock.synced_at,
82 }
83
84
85 def _exit_code_from_conditions(
86 *,
87 config_error: bool,
88 integrity: str | None,
89 drift_status: str | None,
90 use_exit_code: bool,
91 substrate_ok: bool = True,
92 ) -> int:
93 """Apply frozen precedence: 2 > 6 > 3 > 0."""
94 if not use_exit_code:
95 return 0
96 if config_error or not substrate_ok:
97 return 2
98 if integrity == "mismatch":
99 return 6
100 if drift_status in {"behind", "ahead"}:
101 return 3
102 return 0
103
104
105 def run_status(args: Namespace, ctx: CliContext) -> int:
106 """Execute ``overseer status``."""
107 report = CommandReport()
108 repo_root = resolve_repo_root(cwd=ctx.cwd, repo_arg=args.repo, command="status")
109 overseer_dir = repo_root / ".overseer"
110
111 if not overseer_dir.is_dir():
112 payload = {
113 "initialized": False,
114 "warnings": [],
115 }
116 if ctx.output.json_mode:
117 ctx.output.emit_json(payload)
118 else:
119 ctx.output.emit("not initialized")
120 return 0
121
122 config_path = resolve_config_path(repo_root, args.config)
123 if not is_within_repo(repo_root, config_path):
124 ctx.output.error("refused: config path outside repo root")
125 return 4
126
127 config_error = False
128 lock_error = False
129
130 try:
131 config = load_config(config_path)
132 except ConfigError as exc:
133 config_error = True
134 ctx.output.error(format_config_error(exc, repo_root))
135 payload = {
136 "initialized": True,
137 "error": str(exc),
138 "warnings": report.warnings,
139 }
140 if ctx.output.json_mode:
141 ctx.output.emit_json(payload)
142 return 2
143
144 substrate = check_substrate(config, repo_root)
145 if not substrate.ok:
146 report.add_warning(f"substrate: {substrate.state} — {substrate.message}")
147 if substrate.remediation:
148 report.add_warning(f"substrate-remediation: {substrate.remediation}")
149
150 gate_scan = None
151 if config.governance_gates.remind and "status" in config.governance_gates.surfaces:
152 gate_scan = scan_governance_gates(config, repo_root)
153 if gate_scan.pending:
154 for line in format_pending_gate_lines(gate_scan):
155 report.add_warning(line)
156
157 lock_file = lock_path(repo_root)
158 lock = None
159 try:
160 lock = read_version_lock(lock_file)
161 except LockError as exc:
162 lock_error = True
163 report.add_warning(str(exc))
164
165 rendered: list = []
166 integrity: str | None = None
167 preserved_living: list[str] = []
168 if lock is not None:
169 try:
170 rendered = resolve_footprint(config, kit=ctx.kit)
171 except ConfigError as exc:
172 config_error = True
173 ctx.output.error(format_config_error(exc, repo_root))
174 return 2
175
176 if args.check_footprint:
177 integrity, preserved_living = _compute_footprint_integrity(repo_root, lock, rendered)
178 if integrity == "mismatch":
179 report.add_warning("footprint_integrity: mismatch")
180 for path in preserved_living:
181 report.add_warning(f"preserved-living: {path}")
182
183 drift = compute_drift(
184 cli_version=kit_version(),
185 lock=lock,
186 rendered=rendered,
187 repo_root=repo_root,
188 )
189 if drift["status"] in {"behind", "ahead"}:
190 report.add_warning(f"drift: {drift['status']}")
191
192 vcs_result = read_vcs_status(config, repo_root, ctx.runner)
193 if isinstance(vcs_result, ReadError):
194 ctx.output.error(sanitize_text(str(vcs_result), repo_root))
195 payload = {
196 "initialized": True,
197 "substrate": _substrate_payload(substrate),
198 "vcs": {
199 "error": sanitize_text(str(vcs_result), repo_root),
200 "command": sanitize_text(vcs_result.command, repo_root),
201 },
202 "warnings": report.warnings,
203 }
204 if ctx.output.json_mode:
205 ctx.output.emit_json(payload)
206 return 2
207
208 payload = {
209 "initialized": True,
210 "kit_version": kit_version(),
211 "substrate": _substrate_payload(substrate),
212 "lock": _lock_summary(lock) if lock else None,
213 "drift": drift,
214 "footprint_integrity": integrity,
215 "preserved_living": preserved_living if args.check_footprint else [],
216 "vcs": vcs_report(vcs_result, config),
217 "last_governance_sync": _read_governance_sync_marker(repo_root),
218 "governance_gates": pending_gates_payload(gate_scan)
219 if gate_scan is not None
220 else {"enabled": False, "suppressed": False, "active_phases": [], "pending": []},
221 "warnings": report.warnings,
222 }
223 if lock_error:
224 payload["lock_error"] = True
225
226 exit_code = _exit_code_from_conditions(
227 config_error=config_error,
228 integrity=integrity if args.check_footprint else None,
229 drift_status=drift["status"],
230 use_exit_code=args.exit_code,
231 substrate_ok=substrate.ok,
232 )
233 if lock_error and args.exit_code:
234 exit_code = 6 if exit_code == 0 else max(exit_code, 6)
235
236 if ctx.output.json_mode:
237 ctx.output.emit_json(payload)
238 else:
239 ctx.output.emit(f"kit_version: {payload['kit_version']}")
240 if not substrate.ok:
241 ctx.output.emit(f"substrate: {substrate.state} — {substrate.message}")
242 if substrate.remediation:
243 ctx.output.emit(f"substrate-remediation: {substrate.remediation}")
244 if lock:
245 ctx.output.emit(f"lock kit_version: {lock.kit_version}")
246 ctx.output.emit(f"drift: {drift['status']}")
247 if integrity:
248 ctx.output.emit(f"footprint_integrity: {integrity}")
249 if gate_scan is not None and gate_scan.pending:
250 ctx.output.emit("")
251 for line in format_pending_gate_lines(gate_scan):
252 ctx.output.emit(line)
253 ctx.output.emit(f"vcs.regime: {vcs_result.regime}")
254 ctx.output.emit(f"vcs.branch: {vcs_result.branch}")
255 ctx.output.emit(f"vcs.dirty: {vcs_result.dirty}")
256
257 return exit_code
258
259
260 def _substrate_payload(substrate) -> dict:
261 return {
262 "state": substrate.state,
263 "ok": substrate.ok,
264 "missing": list(substrate.missing),
265 "remediation": substrate.remediation,
266 "message": substrate.message,
267 }
File History 1 commit
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439 feat: K1-P1 complete — agent provenance, build-verification… Sonnet 4.6 patch 53 days ago