aaronrene / overseer-kit public
base.py python
219 lines 7.4 KB
Raw
sha256:0e9549ec7b463911bc08b7d586dc320b1ac9b1f5c943ee7e3865dcc6cb0f6f83 chore(governance): sync handover+roadmap to 84db8c8 (drift:… Human 2 days ago
1 """Review provider interface and implementations (§K5.8 / K11)."""
2
3 from __future__ import annotations
4
5 import re
6 from dataclasses import dataclass, field
7 from pathlib import Path
8 from typing import Protocol
9
10 from tools.freeze_reviewer.checklist import BUILTIN_CHECKLIST
11 from tools.freeze_reviewer.providers.api_client import ReviewApiClient
12 from tools.freeze_reviewer.providers.api_response import ProviderReviewError
13 from tools.freeze_reviewer.types import ChecklistItem, Finding, ReviewerSettings
14
15 ABSOLUTE_PATH_RE = re.compile(r"(?:^|[\s`'\"])(/[A-Za-z0-9._-]+){2,}")
16 SECRET_RE = re.compile(r"(?i)(api[_-]?key|secret|token|password)\s*[:=]\s*\S+")
17 TIER_MATRIX_RE = re.compile(r"seven[- ]tier|7[- ]tier", re.IGNORECASE)
18 GROUND_TRUTH_RE = re.compile(r"frozen:\s*true|ground truth|ground-truth", re.IGNORECASE)
19
20
21 class ReviewProvider(Protocol):
22 """Provider interface for freeze review."""
23
24 def reachable(self) -> tuple[bool, str | None]:
25 """Return reachability and optional non-secret cause."""
26
27 def review(
28 self,
29 *,
30 artifact_text: str,
31 artifact_path: str,
32 checklist: list[ChecklistItem],
33 reviewer: ReviewerSettings,
34 ) -> list[Finding]:
35 """Return pre-validation findings."""
36
37
38 @dataclass
39 class ChecklistEngine:
40 """Rule-based checklist evaluation shared by local and api providers.
41
42 Heuristic detectors emit findings only for concrete risk *surfaces* (missing
43 ground-truth/matrix evidence; absolute paths; secret-assignment patterns;
44 missing citation discipline). C5–C7 (irreversibility / real money / Tier-3
45 linkage) require judgment of whether the artifact *introduces* those risks —
46 normative discussion of the words is not a finding. Nuanced C5–C7 verdicts
47 come from scripted/model providers; this engine does not keyword-match
48 escalation vocabulary (§K5.5 / SPEC §6.3).
49 """
50
51 def evaluate(
52 self,
53 *,
54 artifact_text: str,
55 artifact_path: str,
56 checklist: list[ChecklistItem],
57 ) -> list[Finding]:
58 findings: list[Finding] = []
59 lines = artifact_text.splitlines()
60 check_ids = {item.id for item in checklist}
61
62 if "C1" in check_ids and not GROUND_TRUTH_RE.search(artifact_text):
63 findings.append(
64 Finding(
65 check="C1",
66 severity="MAJOR",
67 category="completeness",
68 path=artifact_path,
69 line=1,
70 message="Missing ground-truth edge declaration.",
71 ).with_citation()
72 )
73
74 if "C2" in check_ids and not TIER_MATRIX_RE.search(artifact_text):
75 findings.append(
76 Finding(
77 check="C2",
78 severity="BLOCKER",
79 category="completeness",
80 path=artifact_path,
81 line=1,
82 message="Missing seven-tier test matrix section.",
83 ).with_citation()
84 )
85
86 if "C4" in check_ids:
87 for line_no, line in enumerate(lines, start=1):
88 if ABSOLUTE_PATH_RE.search(line):
89 findings.append(
90 Finding(
91 check="C4",
92 severity="BLOCKER",
93 category="security",
94 path=artifact_path,
95 line=line_no,
96 message="Absolute machine path appears in artifact text.",
97 ).with_citation()
98 )
99 break
100 if SECRET_RE.search(line):
101 findings.append(
102 Finding(
103 check="C4",
104 severity="BLOCKER",
105 category="security",
106 path=artifact_path,
107 line=line_no,
108 message="Secret-like token pattern appears in artifact text.",
109 ).with_citation()
110 )
111 break
112
113 if "C8" in check_ids and "file+line" not in artifact_text.lower():
114 findings.append(
115 Finding(
116 check="C8",
117 severity="MINOR",
118 category="consistency",
119 path=artifact_path,
120 line=1,
121 message="Citation readiness discipline not evidenced in artifact.",
122 ).with_citation()
123 )
124
125 return findings
126
127
128 @dataclass
129 class LocalReviewProvider:
130 """Offline-capable local provider (§K5.8)."""
131
132 engine: ChecklistEngine = field(default_factory=ChecklistEngine)
133 force_unreachable: bool = False
134 unreachable_cause: str | None = None
135 scripted_findings: list[Finding] | None = None
136 review_calls: int = 0
137 reachable_calls: int = 0
138
139 def reachable(self) -> tuple[bool, str | None]:
140 self.reachable_calls += 1
141 if self.force_unreachable:
142 return False, self.unreachable_cause or "local runner unavailable"
143 return True, None
144
145 def review(
146 self,
147 *,
148 artifact_text: str,
149 artifact_path: str,
150 checklist: list[ChecklistItem],
151 reviewer: ReviewerSettings,
152 ) -> list[Finding]:
153 self.review_calls += 1
154 if self.scripted_findings is not None:
155 return list(self.scripted_findings)
156 return self.engine.evaluate(
157 artifact_text=artifact_text,
158 artifact_path=artifact_path,
159 checklist=checklist,
160 )
161
162
163 @dataclass
164 class ApiReviewProvider:
165 """Headless remote API provider (§K5.8 / K11)."""
166
167 kit_root: Path | None = None
168 client: ReviewApiClient | None = None
169 force_unreachable: bool = False
170 unreachable_cause: str | None = None
171 scripted_findings: list[Finding] | None = None
172 review_calls: int = 0
173 reachable_calls: int = 0
174
175 def _client(self) -> ReviewApiClient:
176 if self.client is not None:
177 return self.client
178 return ReviewApiClient(kit_root=self.kit_root)
179
180 def reachable(self) -> tuple[bool, str | None]:
181 self.reachable_calls += 1
182 if self.force_unreachable:
183 return False, self.unreachable_cause or "API provider unavailable"
184 return self._client().reachable()
185
186 def review(
187 self,
188 *,
189 artifact_text: str,
190 artifact_path: str,
191 checklist: list[ChecklistItem],
192 reviewer: ReviewerSettings,
193 ) -> list[Finding]:
194 self.review_calls += 1
195 if self.scripted_findings is not None:
196 return list(self.scripted_findings)
197 try:
198 return self._client().review(
199 artifact_text=artifact_text,
200 artifact_path=artifact_path,
201 checklist=checklist,
202 reviewer=reviewer,
203 )
204 except ProviderReviewError as exc:
205 raise ProviderReviewError(str(exc)) from exc
206
207
208 def provider_for(
209 settings: ReviewerSettings,
210 provider: ReviewProvider | None = None,
211 *,
212 kit_root: Path | None = None,
213 ) -> ReviewProvider:
214 """Construct the effective provider unless a test double is injected."""
215 if provider is not None:
216 return provider
217 if settings.provider == "api":
218 return ApiReviewProvider(kit_root=kit_root)
219 return LocalReviewProvider()
File History 1 commit
sha256:6abcf1fa82a7a621ccbc945f19acdba5bc0db54569599404a1452fb4a096a199 fix(ISR): default require_independent_second_reviewer to require Human minor ⚠ 2 days ago