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