base.py
python
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1
docs: MuseHub-first before ISR #74 — staging solidify NEXT
Human
1 day 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
2 commits
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1
docs: MuseHub-first before ISR #74 — staging solidify NEXT
Human
1 day ago
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439
feat: K1-P1 complete — agent provenance, build-verification…
Sonnet 4.6
patch
53 days ago