aaronrene / overseer-kit public
review.py python
227 lines 7.9 KB
Raw
sha256:8461d44b77376fbf06fa7c3e085d309e3010fd8d5886d63c63e69ce118811ad4 docs: record AFF-b feature-tip SHAs after AFF-b-ISR commit. Human 5 days ago
1 """``overseer review --freeze`` command (§K5.2)."""
2
3 from __future__ import annotations
4
5 import sys
6 from argparse import Namespace
7 from pathlib import Path
8
9 from adapters.config import load_config
10 from adapters.errors import ConfigError, ReadError
11 from adapters.factory import create_adapter
12 from cli.context import CliContext
13 from cli.kit_root import kit_version
14 from cli.paths import PathEscapeError, confine_path, repo_relative, resolve_config_path, resolve_repo_root
15 from cli.sanitize import format_config_error
16 from tools.freeze_reviewer.checklist import builtin_checklist, load_checklist_file
17 from tools.freeze_reviewer.engine import ReviewOptions, resolve_exit_code, resolve_reviewer_settings, run_freeze_review
18 from tools.freeze_reviewer.labels import validate_reviewer_model
19 from tools.footprint_coverage import check_footprint_coverage
20 from tools.footprint_integrity import check_footprint_integrity
21 from tools.freeze_reviewer.report import build_report, render_human_report
22 from tools.muse_sync import check_muse_sync
23 from tools.substrate_health import check_substrate
24
25 DISALLOWED_FLAGS = frozenset(
26 {
27 "--write-vcs",
28 "--commit",
29 "--push",
30 "--escalate-force-pass",
31 }
32 )
33
34
35 def _validate_raw_argv(argv: list[str]) -> int | None:
36 """Reject disallowed flags with USAGE exit 1."""
37 for token in argv:
38 if token in DISALLOWED_FLAGS:
39 return 1
40 if token.startswith("--model") and "=" in token:
41 value = token.split("=", 1)[1]
42 if _looks_like_vendor_slug(value):
43 return 1
44 if token == "--model":
45 return None
46 return None
47
48
49 def _looks_like_vendor_slug(value: str) -> bool:
50 lowered = value.lower()
51 return any(marker in lowered for marker in ("gpt-", "claude-", "composer-"))
52
53
54 def _resolve_effective_checklist(args: Namespace, repo_root: Path) -> tuple[list | None, int | None]:
55 if args.checklist is None:
56 return builtin_checklist(), None
57 try:
58 checklist_path = confine_path(repo_root, args.checklist)
59 except PathEscapeError:
60 return None, 4
61 if not checklist_path.is_file():
62 return None, 4
63 try:
64 return load_checklist_file(checklist_path), None
65 except ConfigError:
66 return None, 2
67
68
69 def _resolve_artifact(args: Namespace, repo_root: Path) -> tuple[Path | None, str | None, int | None]:
70 try:
71 artifact_path = confine_path(repo_root, args.freeze_path)
72 except PathEscapeError:
73 return None, None, 4
74 if not artifact_path.is_file():
75 return None, None, 4
76 rel = repo_relative(repo_root, artifact_path)
77 return artifact_path, rel, None
78
79
80 def run_review(args: Namespace, ctx: CliContext, *, raw_argv: list[str] | None = None) -> int:
81 """Execute ``overseer review --freeze``."""
82 argv = raw_argv or []
83 disallowed = _validate_raw_argv(argv)
84 if disallowed is not None:
85 ctx.output.error("usage: invalid or disallowed flag")
86 return disallowed
87
88 repo_root = resolve_repo_root(cwd=ctx.cwd, repo_arg=args.repo, command="review")
89 overseer_dir = repo_root / ".overseer"
90 if not overseer_dir.is_dir():
91 ctx.output.error("not initialized — run ok init first")
92 return 2
93
94 config_path = resolve_config_path(repo_root, args.config)
95 try:
96 config = load_config(config_path)
97 except ConfigError as exc:
98 ctx.output.error(format_config_error(exc, repo_root))
99 return 2
100
101 checklist, checklist_code = _resolve_effective_checklist(args, repo_root)
102 if checklist_code is not None:
103 if checklist_code == 2:
104 ctx.output.error("invalid checklist file")
105 else:
106 ctx.output.error("refused: checklist path")
107 return checklist_code
108
109 artifact_path, rel_path, artifact_code = _resolve_artifact(args, repo_root)
110 if artifact_code is not None or artifact_path is None or rel_path is None:
111 ctx.output.error("refused: artifact path")
112 return artifact_code or 4
113
114 # Validate model label when agent mode effective
115 effective_mode = args.mode or config.freeze_contract.reviewer.mode
116 effective_model = args.model or config.freeze_contract.reviewer.model
117 if effective_mode != "human":
118 if args.model and _looks_like_vendor_slug(args.model):
119 ctx.output.error("reviewer.model must be a label, not a vendor slug")
120 return 1
121 try:
122 validate_reviewer_model(effective_model, ctx.kit)
123 except ConfigError as exc:
124 ctx.output.error(format_config_error(exc, repo_root))
125 return 2
126 if args.provider and args.provider not in {"local", "api"}:
127 ctx.output.error("invalid --provider value")
128 return 1
129 if args.mode and args.mode not in {"agent", "human"}:
130 ctx.output.error("invalid --mode value")
131 return 1
132
133 adapter = create_adapter(config, repo_root, runner=ctx.runner)
134 substrate = check_substrate(config, repo_root)
135 if not substrate.ok:
136 ctx.output.error(f"substrate: {substrate.state} — {substrate.message}")
137 if substrate.remediation:
138 ctx.output.error(f"remediation: {substrate.remediation}")
139 return 2
140
141 status = adapter.status()
142 if isinstance(status, ReadError):
143 ctx.output.error(str(status))
144 return 2
145
146 muse_sync = check_muse_sync(config, status)
147 if not muse_sync.ok:
148 ctx.output.error(f"muse_sync: {muse_sync.state} — {muse_sync.message}")
149 if muse_sync.remediation:
150 ctx.output.error(f"remediation: {muse_sync.remediation}")
151 return 2
152
153 footprint_self_integrity = check_footprint_integrity(repo_root)
154 if not footprint_self_integrity.ok:
155 ctx.output.error(
156 f"footprint_self_integrity: {footprint_self_integrity.state} — "
157 f"{footprint_self_integrity.message}"
158 )
159 if footprint_self_integrity.remediation:
160 ctx.output.error(f"remediation: {footprint_self_integrity.remediation}")
161 return 2
162
163 footprint_coverage = check_footprint_coverage(
164 repo_root,
165 config,
166 kit=ctx.kit,
167 )
168 if not footprint_coverage.ok:
169 ctx.output.error(
170 f"footprint_coverage: {footprint_coverage.state} — {footprint_coverage.message}"
171 )
172 if footprint_coverage.remediation:
173 ctx.output.error(f"remediation: {footprint_coverage.remediation}")
174 return 2
175
176 injected_provider = None
177 if ctx.review_provider_factory is not None:
178 injected_provider = ctx.review_provider_factory(config.freeze_contract.reviewer.provider)
179
180 options = ReviewOptions(
181 dry_run=args.dry_run,
182 no_stamp=args.no_stamp,
183 mode=args.mode,
184 provider=args.provider,
185 model=args.model,
186 checklist=checklist,
187 kit_version=kit_version(),
188 kit_root=ctx.kit,
189 injected_provider=injected_provider,
190 override_non_pass_stamp=bool(getattr(args, "override_non_pass_stamp", False)),
191 checklist_source=("operator_file" if getattr(args, "checklist", None) else "builtin"),
192 )
193
194 try:
195 result = run_freeze_review(
196 artifact_path=artifact_path,
197 rel_path=rel_path,
198 config=config,
199 options=options,
200 )
201 except ValueError as exc:
202 if str(exc) == "not-utf8":
203 ctx.output.error("refused: artifact not utf-8")
204 return 4
205 raise
206
207 reviewer = resolve_reviewer_settings(config.freeze_contract, options)
208 report = build_report(
209 freeze_path=rel_path,
210 result=result,
211 reviewer=reviewer,
212 config=config.freeze_contract,
213 enabled=config.freeze_contract.enabled,
214 )
215 exit_code = resolve_exit_code(
216 result,
217 config_error=False,
218 refused=result.refused,
219 )
220 report["exit_code"] = exit_code
221
222 if ctx.output.json_mode:
223 ctx.output.emit_json(report)
224 else:
225 ctx.output.emit(render_human_report(freeze_path=rel_path, result=result))
226
227 return exit_code
File History 1 commit
sha256:8461d44b77376fbf06fa7c3e085d309e3010fd8d5886d63c63e69ce118811ad4 docs: record AFF-b feature-tip SHAs after AFF-b-ISR commit. Human 5 days ago