surface.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Active-slice Mode D independent second reviewer surface (§ISR.6).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import re |
| 6 | from dataclasses import dataclass |
| 7 | from pathlib import Path |
| 8 | |
| 9 | from adapters.config import OverseerConfig |
| 10 | from cli.paths import confine_path |
| 11 | from tools.governance_gates.scan import ( |
| 12 | BUILD_VERIFICATION_PASS_RE, |
| 13 | PHASE_DOC_RE, |
| 14 | ROADMAP_ROW_RE, |
| 15 | _is_auto_model, |
| 16 | _normalize_phase_id, |
| 17 | scan_governance_gates, |
| 18 | ) |
| 19 | from tools.honesty.ledger import read_ledger_entries |
| 20 | from tools.honesty.validate import find_matching_independent_second_review |
| 21 | |
| 22 | AUTO_B_STEP_RE = re.compile(r"\bb\b$", re.IGNORECASE) |
| 23 | |
| 24 | REQUIRE_REMEDIATION = ( |
| 25 | "second chat: append independent_second_review then " |
| 26 | "ok honesty-status --independent-second-review PHASE" |
| 27 | ) |
| 28 | |
| 29 | |
| 30 | @dataclass(frozen=True) |
| 31 | class IndependentSecondReviewerGateReport: |
| 32 | """Result of the active-slice independent second reviewer probe.""" |
| 33 | |
| 34 | skipped: bool |
| 35 | ok: bool |
| 36 | mode: str | None = None |
| 37 | matched: bool | None = None |
| 38 | phase_id: str | None = None |
| 39 | message: str | None = None |
| 40 | token: str | None = None |
| 41 | remediation: str | None = None |
| 42 | |
| 43 | |
| 44 | def build_independent_second_reviewer_gate( |
| 45 | config: OverseerConfig, |
| 46 | repo_root: Path, |
| 47 | *, |
| 48 | handover_text: str | None = None, |
| 49 | roadmap_text: str | None = None, |
| 50 | ) -> IndependentSecondReviewerGateReport: |
| 51 | """Run active-slice Mode D when honesty warn/require is enabled.""" |
| 52 | require = config.honesty.require_independent_second_reviewer |
| 53 | if not config.honesty.enabled or require not in {"warn", "require"}: |
| 54 | return IndependentSecondReviewerGateReport(skipped=True, ok=True) |
| 55 | |
| 56 | docs_root = config.repo.root_relative_docs |
| 57 | handover_path = repo_root / docs_root / config.docs.handover |
| 58 | roadmap_path = repo_root / docs_root / config.docs.roadmap |
| 59 | handover = handover_text if handover_text is not None else _read_text(handover_path) |
| 60 | roadmap = roadmap_text if roadmap_text is not None else _read_text(roadmap_path) |
| 61 | |
| 62 | gate_scan = scan_governance_gates( |
| 63 | config, |
| 64 | repo_root, |
| 65 | handover_text=handover, |
| 66 | roadmap_text=roadmap, |
| 67 | ) |
| 68 | if not gate_scan.active_phases: |
| 69 | return IndependentSecondReviewerGateReport(skipped=True, ok=True) |
| 70 | |
| 71 | phase_id = _select_active_auto_phase(roadmap, gate_scan.active_phases) |
| 72 | if phase_id is None: |
| 73 | return IndependentSecondReviewerGateReport(skipped=True, ok=True) |
| 74 | |
| 75 | if not _should_run_isr_gate(phase_id, handover, roadmap): |
| 76 | return IndependentSecondReviewerGateReport(skipped=True, ok=True) |
| 77 | |
| 78 | frozen_spec = _frozen_spec_for_phase(roadmap, phase_id) |
| 79 | matched = _ledger_match(config, repo_root, phase_id, frozen_spec) |
| 80 | |
| 81 | if matched: |
| 82 | return IndependentSecondReviewerGateReport( |
| 83 | skipped=False, |
| 84 | ok=True, |
| 85 | mode=require, |
| 86 | matched=True, |
| 87 | phase_id=phase_id, |
| 88 | ) |
| 89 | |
| 90 | if require == "require": |
| 91 | return IndependentSecondReviewerGateReport( |
| 92 | skipped=False, |
| 93 | ok=False, |
| 94 | mode=require, |
| 95 | matched=False, |
| 96 | phase_id=phase_id, |
| 97 | message="missing independent_second_review ledger entry for active Auto slice", |
| 98 | token="missing_independent_second_review", |
| 99 | remediation=REQUIRE_REMEDIATION, |
| 100 | ) |
| 101 | |
| 102 | return IndependentSecondReviewerGateReport( |
| 103 | skipped=False, |
| 104 | ok=True, |
| 105 | mode="warn", |
| 106 | matched=False, |
| 107 | phase_id=phase_id, |
| 108 | message="warning: no matching independent_second_review entry for active Auto slice", |
| 109 | ) |
| 110 | |
| 111 | |
| 112 | def independent_second_reviewer_gate_payload( |
| 113 | report: IndependentSecondReviewerGateReport, |
| 114 | ) -> dict | None: |
| 115 | if report.skipped: |
| 116 | return None |
| 117 | payload: dict = { |
| 118 | "ok": report.ok, |
| 119 | "mode": report.mode, |
| 120 | "matched": report.matched, |
| 121 | } |
| 122 | if report.phase_id: |
| 123 | payload["phase_id"] = report.phase_id |
| 124 | if report.token: |
| 125 | payload["token"] = report.token |
| 126 | if report.remediation: |
| 127 | payload["remediation"] = report.remediation |
| 128 | return payload |
| 129 | |
| 130 | |
| 131 | def format_independent_second_reviewer_gate_line( |
| 132 | report: IndependentSecondReviewerGateReport, |
| 133 | ) -> str | None: |
| 134 | if report.skipped or report.message is None: |
| 135 | return None |
| 136 | return f"independent_second_reviewer_gate: {report.message}" |
| 137 | |
| 138 | |
| 139 | def _read_text(path: Path) -> str | None: |
| 140 | if not path.is_file(): |
| 141 | return None |
| 142 | return path.read_text(encoding="utf-8") |
| 143 | |
| 144 | |
| 145 | def _select_active_auto_phase(roadmap: str | None, active: tuple[str, ...]) -> str | None: |
| 146 | if roadmap is None: |
| 147 | return None |
| 148 | for phase_id in active: |
| 149 | row = _roadmap_row(roadmap, phase_id) |
| 150 | if row is None: |
| 151 | continue |
| 152 | model = row.group("model") |
| 153 | if _is_active_auto_model(model, phase_id): |
| 154 | return phase_id |
| 155 | return None |
| 156 | |
| 157 | |
| 158 | def _is_active_auto_model(model: str, phase_id: str) -> bool: |
| 159 | if not _is_auto_model(model): |
| 160 | return False |
| 161 | lowered = model.lower() |
| 162 | if "thinking" in lowered and "auto" in lowered: |
| 163 | return bool(AUTO_B_STEP_RE.search(phase_id.replace(" ", ""))) |
| 164 | return True |
| 165 | |
| 166 | |
| 167 | def _roadmap_row(roadmap: str, phase_id: str): |
| 168 | normalized = _normalize_phase_id(phase_id) |
| 169 | tokens = [token for token in re.split(r"[\s/]+", normalized.lower()) if token] |
| 170 | for match in ROADMAP_ROW_RE.finditer(roadmap): |
| 171 | phase_label = _normalize_phase_id(match.group("phase")) |
| 172 | if phase_label == normalized: |
| 173 | return match |
| 174 | phase_lower = phase_label.lower() |
| 175 | if tokens and all(token in phase_lower for token in tokens): |
| 176 | return match |
| 177 | return None |
| 178 | |
| 179 | |
| 180 | def _roadmap_status(roadmap: str, phase_id: str) -> str | None: |
| 181 | row = _roadmap_row(roadmap, phase_id) |
| 182 | if row is None: |
| 183 | return None |
| 184 | return row.group("status").strip().upper() |
| 185 | |
| 186 | |
| 187 | def _claims_done_or_bv_pass(corpus: str, phase_id: str) -> bool: |
| 188 | if _roadmap_status(corpus, phase_id) == "DONE": |
| 189 | return True |
| 190 | if not BUILD_VERIFICATION_PASS_RE.search(corpus): |
| 191 | return False |
| 192 | phase_tokens = [token for token in re.split(r"[\s/]+", phase_id.lower()) if token] |
| 193 | if not phase_tokens: |
| 194 | return False |
| 195 | window = 400 |
| 196 | for match in BUILD_VERIFICATION_PASS_RE.finditer(corpus): |
| 197 | start = max(0, match.start() - window) |
| 198 | end = min(len(corpus), match.end() + window) |
| 199 | snippet = corpus[start:end].lower() |
| 200 | if any(token in snippet for token in phase_tokens): |
| 201 | return True |
| 202 | return False |
| 203 | |
| 204 | |
| 205 | def _should_run_isr_gate(phase_id: str, handover: str | None, roadmap: str | None) -> bool: |
| 206 | corpus = (handover or "") + "\n" + (roadmap or "") |
| 207 | status = _roadmap_status(roadmap or "", phase_id) |
| 208 | if status in {"TODO", "WIP"} and not _claims_done_or_bv_pass(corpus, phase_id): |
| 209 | return False |
| 210 | return _claims_done_or_bv_pass(corpus, phase_id) |
| 211 | |
| 212 | |
| 213 | def _frozen_spec_for_phase(roadmap: str | None, phase_id: str) -> str | None: |
| 214 | if roadmap is None: |
| 215 | return None |
| 216 | row = _roadmap_row(roadmap, phase_id) |
| 217 | if row is None: |
| 218 | return None |
| 219 | matches = PHASE_DOC_RE.findall(row.group("deliverable")) |
| 220 | if len(matches) != 1: |
| 221 | return None |
| 222 | return matches[0] |
| 223 | |
| 224 | |
| 225 | def _ledger_match( |
| 226 | config: OverseerConfig, |
| 227 | repo_root: Path, |
| 228 | phase_id: str, |
| 229 | frozen_spec: str | None, |
| 230 | ) -> bool: |
| 231 | ledger_rel = config.honesty.ledger |
| 232 | if ledger_rel is None or not ledger_rel.strip(): |
| 233 | return False |
| 234 | try: |
| 235 | ledger_path = confine_path(repo_root, ledger_rel) |
| 236 | except Exception: |
| 237 | return False |
| 238 | if not ledger_path.is_file() or ledger_path.stat().st_size == 0: |
| 239 | return False |
| 240 | try: |
| 241 | entries = read_ledger_entries(ledger_path) |
| 242 | except (ValueError, OSError): |
| 243 | return False |
| 244 | return ( |
| 245 | find_matching_independent_second_review( |
| 246 | entries, |
| 247 | phase_id=phase_id, |
| 248 | frozen_spec=frozen_spec, |
| 249 | producer_session=None, |
| 250 | ) |
| 251 | is not None |
| 252 | ) |