aaronrene / overseer-kit public
scan.py python
241 lines 8.0 KB
Raw
sha256:0e9549ec7b463911bc08b7d586dc320b1ac9b1f5c943ee7e3865dcc6cb0f6f83 chore(governance): sync handover+roadmap to 84db8c8 (drift:… Human 2 days ago
1 """Read-only governance gate scan against roadmap + handover (§KH1.9)."""
2
3 from __future__ import annotations
4
5 import re
6 from pathlib import Path
7
8 from adapters.config import GovernanceGatesConfig, OverseerConfig
9 from tools.freeze_reviewer.artifact import extract_existing_stamp, parse_artifact
10 from tools.governance_gates.checklist import (
11 BUILD_VERIFICATION_INVOKE,
12 FREEZE_REVIEW_INVOKE,
13 )
14 from tools.governance_gates.types import GateScanResult, PendingGate
15
16 ROADMAP_ROW_RE = re.compile(
17 r"^\|\s*\*\*(?P<phase>[^|*]+)\*\*\s*\|\s*(?P<model>[^|]+)\|\s*\*\*(?P<status>[^|*]+)\*\*\s*\|\s*(?P<deliverable>[^|]+)",
18 re.MULTILINE,
19 )
20 PHASE_DOC_RE = re.compile(r"docs/archive/phases/PHASE-[A-Za-z0-9_.-]+\.md")
21 NEXT_ID_RE = re.compile(r"\|\s*\*\*ID\*\*\s*\|\s*\*\*([^*]+)\*\*", re.MULTILINE)
22 BUILD_VERIFICATION_PASS_RE = re.compile(
23 r"build[- ]verification(?:[- ]review)?[^\n]{0,80}\bpass\b",
24 re.IGNORECASE,
25 )
26 HANDOVER_PASTE_MARKER = "Governance gates"
27
28
29 def scan_governance_gates(
30 config: OverseerConfig,
31 repo_root: Path,
32 *,
33 handover_text: str | None = None,
34 roadmap_text: str | None = None,
35 ) -> GateScanResult:
36 """Scan active roadmap slice for pending freeze review / build verification gates."""
37 gates = config.governance_gates
38 if not gates.remind:
39 return GateScanResult(
40 enabled=True,
41 suppressed=True,
42 pending=(),
43 active_phases=(),
44 )
45
46 handover_path = repo_root / config.repo.root_relative_docs / config.docs.handover
47 roadmap_path = repo_root / config.repo.root_relative_docs / config.docs.roadmap
48 handover = handover_text if handover_text is not None else _read_text(handover_path)
49 roadmap = roadmap_text if roadmap_text is not None else _read_text(roadmap_path)
50
51 active = _active_phases(handover, roadmap)
52 pending: list[PendingGate] = []
53
54 if gates.freeze_review_required:
55 pending.extend(_scan_freeze_review(repo_root, config, roadmap, active))
56
57 if gates.build_verification_required:
58 pending.extend(_scan_build_verification(handover, roadmap, active))
59
60 if "handover-paste" in gates.surfaces and handover is not None:
61 if HANDOVER_PASTE_MARKER not in handover:
62 pending.append(
63 PendingGate(
64 gate_id="handover_paste",
65 phase_id=_next_phase_id(handover) or "unknown",
66 artifact=config.docs.handover,
67 message="handover paste-ready prompt missing Governance gates checklist",
68 invoke="include §KH1.9 checklist in paste-ready prompt fence",
69 )
70 )
71
72 return GateScanResult(
73 enabled=True,
74 suppressed=False,
75 pending=tuple(pending),
76 active_phases=active,
77 )
78
79
80 def _read_text(path: Path) -> str | None:
81 if not path.is_file():
82 return None
83 return path.read_text(encoding="utf-8")
84
85
86 def _active_phases(handover: str | None, roadmap: str | None) -> tuple[str, ...]:
87 phases: set[str] = set()
88 if handover:
89 next_id = _next_phase_id(handover)
90 if next_id:
91 phases.add(_normalize_phase_id(next_id))
92 if roadmap:
93 for match in ROADMAP_ROW_RE.finditer(roadmap):
94 status = match.group("status").strip().upper()
95 if status in {"WIP", "TODO", "BLOCKED"}:
96 phases.add(_normalize_phase_id(match.group("phase")))
97 return tuple(sorted(phases))
98
99
100 def _next_phase_id(handover: str) -> str | None:
101 match = NEXT_ID_RE.search(handover)
102 if not match:
103 return None
104 return match.group(1).strip()
105
106
107 def _normalize_phase_id(text: str) -> str:
108 cleaned = text.strip()
109 cleaned = re.sub(r"\s+", " ", cleaned)
110 return cleaned
111
112
113 def _scan_freeze_review(
114 repo_root: Path,
115 config: OverseerConfig,
116 roadmap: str | None,
117 active: tuple[str, ...],
118 ) -> list[PendingGate]:
119 pending: list[PendingGate] = []
120 docs_root = repo_root / config.repo.root_relative_docs
121 for phase_id in active:
122 contract = _contract_path_for_phase(repo_root, docs_root, roadmap, phase_id)
123 if contract is None or not contract.is_file():
124 continue
125 rel = contract.relative_to(repo_root).as_posix()
126 try:
127 parsed = parse_artifact(contract, rel_path=rel)
128 except (ValueError, OSError):
129 continue
130 if parsed.declaration != "present":
131 continue
132 stamp = extract_existing_stamp(parsed)
133 if stamp and stamp.get("verdict") == "pass":
134 continue
135 if _narrative_freeze_pass(contract.read_text(encoding="utf-8")):
136 continue
137 pending.append(
138 PendingGate(
139 gate_id="freeze_review",
140 phase_id=phase_id,
141 artifact=rel,
142 message=f"frozen artifact lacks reviewed → pass ({rel})",
143 invoke=FREEZE_REVIEW_INVOKE,
144 )
145 )
146 return pending
147
148
149 def _scan_build_verification(
150 handover: str | None,
151 roadmap: str | None,
152 active: tuple[str, ...],
153 ) -> list[PendingGate]:
154 if roadmap is None:
155 return []
156 pending: list[PendingGate] = []
157 corpus = (handover or "") + "\n" + roadmap
158 for match in ROADMAP_ROW_RE.finditer(roadmap):
159 phase_label = _normalize_phase_id(match.group("phase"))
160 if phase_label not in active:
161 continue
162 model = match.group("model")
163 status = match.group("status").strip().upper()
164 if not _is_auto_model(model):
165 continue
166 if status not in {"WIP", "DONE"}:
167 continue
168 if _build_verification_recorded(corpus, phase_label):
169 continue
170 pending.append(
171 PendingGate(
172 gate_id="build_verification",
173 phase_id=phase_label,
174 artifact=None,
175 message=(
176 f"Auto phase {status} without recorded build-verification pass "
177 f"({phase_label})"
178 ),
179 invoke=BUILD_VERIFICATION_INVOKE,
180 )
181 )
182 return pending
183
184
185 def _is_auto_model(model: str) -> bool:
186 lowered = model.lower()
187 if "auto" not in lowered:
188 return False
189 if lowered.strip() == "thinking":
190 return False
191 return True
192
193
194 def _build_verification_recorded(corpus: str, phase_id: str) -> bool:
195 if not BUILD_VERIFICATION_PASS_RE.search(corpus):
196 return False
197 phase_tokens = [token for token in re.split(r"[\s/]+", phase_id.lower()) if token]
198 if not phase_tokens:
199 return False
200 window = 400
201 for match in BUILD_VERIFICATION_PASS_RE.finditer(corpus):
202 start = max(0, match.start() - window)
203 end = min(len(corpus), match.end() + window)
204 snippet = corpus[start:end].lower()
205 if any(token in snippet for token in phase_tokens):
206 return True
207 return False
208
209
210 def _contract_path_for_phase(
211 repo_root: Path,
212 docs_root: Path,
213 roadmap: str | None,
214 phase_id: str,
215 ) -> Path | None:
216 if roadmap:
217 for row in ROADMAP_ROW_RE.finditer(roadmap):
218 if _normalize_phase_id(row.group("phase")) != phase_id:
219 continue
220 deliverable = row.group("deliverable")
221 doc_match = PHASE_DOC_RE.search(deliverable)
222 if doc_match:
223 candidate = (repo_root / doc_match.group(0)).resolve()
224 if not str(candidate).startswith(str(repo_root.resolve())):
225 return None
226 return candidate
227 slug = re.sub(r"[^A-Za-z0-9]+", "-", phase_id).strip("-").upper()
228 candidates = sorted(docs_root.glob(f"PHASE-{slug}*.md"))
229 if candidates:
230 return candidates[0]
231 token = phase_id.split()[0].upper()
232 candidates = sorted(docs_root.glob(f"PHASE-{token}*.md"))
233 return candidates[0] if candidates else None
234
235
236 def _narrative_freeze_pass(text: str) -> bool:
237 return bool(
238 re.search(r"reviewed\s*→\s*`pass`", text, re.IGNORECASE)
239 or re.search(r"Freeze status:.*\bpass\b", text, re.IGNORECASE)
240 or re.search(r"→\s*`pass`.*Cleared", text, re.IGNORECASE)
241 )
File History 1 commit
sha256:6abcf1fa82a7a621ccbc945f19acdba5bc0db54569599404a1452fb4a096a199 fix(ISR): default require_independent_second_reviewer to require Human minor ⚠ 2 days ago