aaronrene / overseer-kit public
surface.py python
211 lines 6.3 KB
Raw
sha256:8461d44b77376fbf06fa7c3e085d309e3010fd8d5886d63c63e69ce118811ad4 docs: record AFF-b feature-tip SHAs after AFF-b-ISR commit. Human 6 days ago
1 """Active-slice adversarial-freeze status / governance-sync surface (§AFF.9)."""
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 tools.adversarial_freeze.authorize import (
11 adversarial_authorization_state,
12 aff_hold_bypassed,
13 aggregate_adversarial_authorization_states,
14 frv_authorizing_freeze_paths,
15 )
16 from tools.governance_gates.scan import (
17 ROADMAP_ROW_RE,
18 _is_auto_model,
19 _normalize_phase_id,
20 scan_governance_gates,
21 )
22 from tools.governance_hygiene.parse import compact_step_id
23
24 # discover_freeze_candidates is imported lazily inside build_adversarial_freeze_gate
25 # to avoid a circular import:
26 # next_regen → adversarial_freeze.authorize → package __init__ → surface → next_regen
27
28 AUTO_B_STEP_RE = re.compile(r"\bb\b$", re.IGNORECASE)
29
30 WARN_PENDING = (
31 "warning: no authorizing adversarial_freeze entry for active Auto slice"
32 )
33 WARN_ABSENT = (
34 "warning: adversarial_freeze unreadable (broken ledger or artifact) "
35 "for active Auto slice"
36 )
37
38
39 @dataclass(frozen=True)
40 class AdversarialFreezeGateReport:
41 """Result of the active-slice adversarial-freeze probe (§AFF.9)."""
42
43 skipped: bool
44 ok: bool
45 mode: str | None = None
46 state: str | None = None
47 matched: bool | None = None
48 message: str | None = None
49 token: str | None = None
50
51
52 def build_adversarial_freeze_gate(
53 config: OverseerConfig,
54 repo_root: Path,
55 *,
56 handover_text: str | None = None,
57 roadmap_text: str | None = None,
58 ) -> AdversarialFreezeGateReport:
59 """Run active-slice AFF probe when honesty suggest/require is enabled."""
60 if aff_hold_bypassed(config):
61 return AdversarialFreezeGateReport(skipped=True, ok=True)
62
63 mode = config.honesty.adversarial_freeze
64 if mode not in {"suggest", "require"}:
65 return AdversarialFreezeGateReport(skipped=True, ok=True)
66
67 docs_root = config.repo.root_relative_docs
68 handover_path = repo_root / docs_root / config.docs.handover
69 roadmap_path = repo_root / docs_root / config.docs.roadmap
70 handover = handover_text if handover_text is not None else _read_text(handover_path)
71 roadmap = roadmap_text if roadmap_text is not None else _read_text(roadmap_path)
72
73 gate_scan = scan_governance_gates(
74 config,
75 repo_root,
76 handover_text=handover,
77 roadmap_text=roadmap,
78 )
79 if not gate_scan.active_phases:
80 return AdversarialFreezeGateReport(skipped=True, ok=True)
81
82 phase_id = _select_active_auto_phase(roadmap, gate_scan.active_phases)
83 if phase_id is None:
84 return AdversarialFreezeGateReport(skipped=True, ok=True)
85
86 row = _roadmap_row(roadmap or "", phase_id)
87 if row is None:
88 return AdversarialFreezeGateReport(skipped=True, ok=True)
89
90 step_id = compact_step_id(row.group("phase"))
91 from tools.governance_hygiene.next_regen import discover_freeze_candidates
92
93 candidates = discover_freeze_candidates(
94 repo_root, step_id, row.group("deliverable")
95 )
96 if not candidates:
97 return AdversarialFreezeGateReport(skipped=True, ok=True)
98
99 authorizing = frv_authorizing_freeze_paths(
100 repo_root, candidates, phase_id=step_id, config=config
101 )
102 if not authorizing:
103 return AdversarialFreezeGateReport(skipped=True, ok=True)
104
105 states: list[str] = []
106 for path in authorizing:
107 auth = adversarial_authorization_state(repo_root, path, config=config)
108 if auth.state == "off":
109 continue
110 states.append(auth.state)
111
112 if not states:
113 return AdversarialFreezeGateReport(skipped=True, ok=True)
114
115 aggregated = aggregate_adversarial_authorization_states(states)
116 matched = aggregated in {"pass", "skipped"}
117
118 if matched:
119 return AdversarialFreezeGateReport(
120 skipped=False,
121 ok=True,
122 mode=mode,
123 state=aggregated,
124 matched=True,
125 )
126
127 if mode == "require":
128 return AdversarialFreezeGateReport(
129 skipped=False,
130 ok=False,
131 mode=mode,
132 state=aggregated,
133 matched=False,
134 message="missing adversarial_freeze ledger entry for active Auto slice",
135 token="missing_adversarial_freeze",
136 )
137
138 message = WARN_ABSENT if aggregated == "absent" else WARN_PENDING
139 return AdversarialFreezeGateReport(
140 skipped=False,
141 ok=True,
142 mode=mode,
143 state=aggregated,
144 matched=False,
145 message=message,
146 )
147
148
149 def adversarial_freeze_gate_payload(
150 report: AdversarialFreezeGateReport,
151 ) -> dict | None:
152 if report.skipped:
153 return None
154 payload: dict = {
155 "ok": report.ok,
156 "mode": report.mode,
157 "state": report.state,
158 "matched": report.matched,
159 }
160 if report.token:
161 payload["token"] = report.token
162 return payload
163
164
165 def format_adversarial_freeze_gate_line(
166 report: AdversarialFreezeGateReport,
167 ) -> str | None:
168 if report.skipped or report.message is None:
169 return None
170 return report.message
171
172
173 def _read_text(path: Path) -> str | None:
174 if not path.is_file():
175 return None
176 return path.read_text(encoding="utf-8")
177
178
179 def _select_active_auto_phase(roadmap: str | None, active: tuple[str, ...]) -> str | None:
180 if roadmap is None:
181 return None
182 for phase_id in active:
183 row = _roadmap_row(roadmap, phase_id)
184 if row is None:
185 continue
186 model = row.group("model")
187 if _is_active_auto_model(model, phase_id):
188 return phase_id
189 return None
190
191
192 def _is_active_auto_model(model: str, phase_id: str) -> bool:
193 if not _is_auto_model(model):
194 return False
195 lowered = model.lower()
196 if "thinking" in lowered and "auto" in lowered:
197 return bool(AUTO_B_STEP_RE.search(phase_id.replace(" ", "")))
198 return True
199
200
201 def _roadmap_row(roadmap: str, phase_id: str):
202 normalized = _normalize_phase_id(phase_id)
203 tokens = [token for token in re.split(r"[\s/]+", normalized.lower()) if token]
204 for match in ROADMAP_ROW_RE.finditer(roadmap):
205 phase_label = _normalize_phase_id(match.group("phase"))
206 if phase_label == normalized:
207 return match
208 phase_lower = phase_label.lower()
209 if tokens and all(token in phase_lower for token in tokens):
210 return match
211 return None
File History 1 commit
sha256:8461d44b77376fbf06fa7c3e085d309e3010fd8d5886d63c63e69ce118811ad4 docs: record AFF-b feature-tip SHAs after AFF-b-ISR commit. Human 6 days ago