aaronrene / overseer-kit public
base.py python
234 lines 8.0 KB
Raw
sha256:8461d44b77376fbf06fa7c3e085d309e3010fd8d5886d63c63e69ce118811ad4 docs: record AFF-b feature-tip SHAs after AFF-b-ISR commit. Human 3 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 def producer_identity(self) -> tuple[str, str]:
38 """Return (produced_by, provider_kind) per §FRV.4.2."""
39
40
41 @dataclass
42 class ChecklistEngine:
43 """Rule-based checklist evaluation shared by local and api providers.
44
45 Heuristic detectors emit findings only for concrete risk *surfaces* (missing
46 ground-truth/matrix evidence; absolute paths; secret-assignment patterns;
47 missing citation discipline). C5–C7 (irreversibility / real money / Tier-3
48 linkage) require judgment of whether the artifact *introduces* those risks —
49 normative discussion of the words is not a finding. Nuanced C5–C7 verdicts
50 come from scripted/model providers; this engine does not keyword-match
51 escalation vocabulary (§K5.5 / SPEC §6.3).
52 """
53
54 def evaluate(
55 self,
56 *,
57 artifact_text: str,
58 artifact_path: str,
59 checklist: list[ChecklistItem],
60 ) -> list[Finding]:
61 findings: list[Finding] = []
62 lines = artifact_text.splitlines()
63 check_ids = {item.id for item in checklist}
64
65 if "C1" in check_ids and not GROUND_TRUTH_RE.search(artifact_text):
66 findings.append(
67 Finding(
68 check="C1",
69 severity="MAJOR",
70 category="completeness",
71 path=artifact_path,
72 line=1,
73 message="Missing ground-truth edge declaration.",
74 ).with_citation()
75 )
76
77 if "C2" in check_ids and not TIER_MATRIX_RE.search(artifact_text):
78 findings.append(
79 Finding(
80 check="C2",
81 severity="BLOCKER",
82 category="completeness",
83 path=artifact_path,
84 line=1,
85 message="Missing seven-tier test matrix section.",
86 ).with_citation()
87 )
88
89 if "C4" in check_ids:
90 for line_no, line in enumerate(lines, start=1):
91 if ABSOLUTE_PATH_RE.search(line):
92 findings.append(
93 Finding(
94 check="C4",
95 severity="BLOCKER",
96 category="security",
97 path=artifact_path,
98 line=line_no,
99 message="Absolute machine path appears in artifact text.",
100 ).with_citation()
101 )
102 break
103 if SECRET_RE.search(line):
104 findings.append(
105 Finding(
106 check="C4",
107 severity="BLOCKER",
108 category="security",
109 path=artifact_path,
110 line=line_no,
111 message="Secret-like token pattern appears in artifact text.",
112 ).with_citation()
113 )
114 break
115
116 if "C8" in check_ids and "file+line" not in artifact_text.lower():
117 findings.append(
118 Finding(
119 check="C8",
120 severity="MINOR",
121 category="consistency",
122 path=artifact_path,
123 line=1,
124 message="Citation readiness discipline not evidenced in artifact.",
125 ).with_citation()
126 )
127
128 return findings
129
130
131 @dataclass
132 class LocalReviewProvider:
133 """Offline-capable local provider (§K5.8)."""
134
135 engine: ChecklistEngine = field(default_factory=ChecklistEngine)
136 force_unreachable: bool = False
137 unreachable_cause: str | None = None
138 scripted_findings: list[Finding] | None = None
139 review_calls: int = 0
140 reachable_calls: int = 0
141
142 def reachable(self) -> tuple[bool, str | None]:
143 self.reachable_calls += 1
144 if self.force_unreachable:
145 return False, self.unreachable_cause or "local runner unavailable"
146 return True, None
147
148 def review(
149 self,
150 *,
151 artifact_text: str,
152 artifact_path: str,
153 checklist: list[ChecklistItem],
154 reviewer: ReviewerSettings,
155 ) -> list[Finding]:
156 self.review_calls += 1
157 if self.scripted_findings is not None:
158 return list(self.scripted_findings)
159 return self.engine.evaluate(
160 artifact_text=artifact_text,
161 artifact_path=artifact_path,
162 checklist=checklist,
163 )
164
165 def producer_identity(self) -> tuple[str, str]:
166 """Return producer identity after review() (§FRV.4.2)."""
167 if self.scripted_findings is not None:
168 return ("scripted_provider", "rule_engine")
169 return ("checklist_engine", "rule_engine")
170
171
172 @dataclass
173 class ApiReviewProvider:
174 """Headless remote API provider (§K5.8 / K11)."""
175
176 kit_root: Path | None = None
177 client: ReviewApiClient | None = None
178 force_unreachable: bool = False
179 unreachable_cause: str | None = None
180 scripted_findings: list[Finding] | None = None
181 review_calls: int = 0
182 reachable_calls: int = 0
183
184 def _client(self) -> ReviewApiClient:
185 if self.client is not None:
186 return self.client
187 return ReviewApiClient(kit_root=self.kit_root)
188
189 def reachable(self) -> tuple[bool, str | None]:
190 self.reachable_calls += 1
191 if self.force_unreachable:
192 return False, self.unreachable_cause or "API provider unavailable"
193 return self._client().reachable()
194
195 def review(
196 self,
197 *,
198 artifact_text: str,
199 artifact_path: str,
200 checklist: list[ChecklistItem],
201 reviewer: ReviewerSettings,
202 ) -> list[Finding]:
203 self.review_calls += 1
204 if self.scripted_findings is not None:
205 return list(self.scripted_findings)
206 try:
207 return self._client().review(
208 artifact_text=artifact_text,
209 artifact_path=artifact_path,
210 checklist=checklist,
211 reviewer=reviewer,
212 )
213 except ProviderReviewError as exc:
214 raise ProviderReviewError(str(exc)) from exc
215
216 def producer_identity(self) -> tuple[str, str]:
217 """Return producer identity after review() (§FRV.4.2)."""
218 if self.scripted_findings is not None:
219 return ("scripted_provider", "rule_engine")
220 return ("api_model", "model_api")
221
222
223 def provider_for(
224 settings: ReviewerSettings,
225 provider: ReviewProvider | None = None,
226 *,
227 kit_root: Path | None = None,
228 ) -> ReviewProvider:
229 """Construct the effective provider unless a test double is injected."""
230 if provider is not None:
231 return provider
232 if settings.provider == "api":
233 return ApiReviewProvider(kit_root=kit_root)
234 return LocalReviewProvider()
File History 1 commit
sha256:8461d44b77376fbf06fa7c3e085d309e3010fd8d5886d63c63e69ce118811ad4 docs: record AFF-b feature-tip SHAs after AFF-b-ISR commit. Human 3 days ago