aaronrene / overseer-kit public
next_extract.py python
319 lines 11.9 KB
Raw
sha256:0e9549ec7b463911bc08b7d586dc320b1ac9b1f5c943ee7e3865dcc6cb0f6f83 chore(governance): sync handover+roadmap to 84db8c8 (drift:… Human 1 day ago
1 """Parse PRIMARY / RELAY / PRODUCT RELAY / ARCHIVED / LANE TIP markers (§MR.6)."""
2
3 from __future__ import annotations
4
5 import hashlib
6 import re
7 from dataclasses import replace
8
9 from tools.workspace.types import NextBlock, NextRole
10
11 _MARKER_RE = re.compile(
12 r"<!--\s*overseer:next\s+"
13 r"role=(?P<role>primary|relay|product_relay|lane_tip|archived)"
14 r"(?:\s+lane=(?P<lane>[^\s>]+))?"
15 r"(?:\s+status=(?P<status>live|archived))?"
16 r"(?:\s+product_order=(?P<product_order>[^\s>]+))?"
17 r"(?:\s+tip_hash=sha256:(?P<tip_hash>[0-9a-fA-F]{64}))?"
18 r"\s*-->",
19 re.IGNORECASE,
20 )
21
22 _HEADING_NEXT = re.compile(r"^##\s+NEXT SESSION\s+[—\-]\s+(?P<title>.+?)\s*$")
23 _HEADING_PRODUCT_RELAY = re.compile(r"^##\s+PRODUCT RELAY\s+[—\-]\s+(?P<title>.+?)\s*$")
24 _HEADING_LANE_TIP = re.compile(r"^##\s+LANE TIP\s+[—\-]\s+(?P<title>.+?)\s*$")
25 _HEADING_ARCHIVED = re.compile(r"^##\s+ARCHIVED SESSION\s+[—\-]\s+(?P<title>.+?)\s*$")
26
27 _FENCE_RE = re.compile(r"```(?:[^\n]*)\n(.*?)```", re.DOTALL)
28 _STEP_RE = re.compile(r"(?im)^\s*Step:\s*(?P<v>.+?)\s*$")
29 _MODEL_RE = re.compile(r"(?im)^\s*Model:\s*(?P<v>.+?)\s*$")
30 _AUTHORITY_RE = re.compile(r"(?im)^\s*Authority:\s*(?P<v>.+?)\s*$")
31 _ID_ROW_RE = re.compile(
32 r"(?im)^\|\s*\*?\*?ID\*?\*?\s*\|\s*\*?\*?(?P<id>[^*|\s][^*|]*)\*?\*?\s*\|"
33 )
34
35
36 def lf_normalize(text: str) -> str:
37 """Normalize newlines to LF for stable hashing."""
38 return text.replace("\r\n", "\n").replace("\r", "\n")
39
40
41 def tip_hash_hex(fence_bytes_text: str) -> str:
42 """SHA-256 hex of LF-normalized UTF-8 paste-fence body (§MR.5.1)."""
43 normalized = lf_normalize(fence_bytes_text).encode("utf-8")
44 return hashlib.sha256(normalized).hexdigest()
45
46
47 def normalize_model_display(label: str) -> str | None:
48 """Map fence Model line to canonical display label from policy."""
49 cleaned = label.strip().strip("*").strip()
50 if not cleaned:
51 return None
52 mapping = {
53 "thinking": "Thinking",
54 "auto": "Auto",
55 "thinking → auto": "Thinking → Auto",
56 "thinking->auto": "Thinking → Auto",
57 "thinking_to_auto": "Thinking → Auto",
58 "operator + auto": "Operator + Auto",
59 "operator_plus_auto": "Operator + Auto",
60 "operator+auto": "Operator + Auto",
61 }
62 lowered = cleaned.lower().replace("—", "→").replace("–", "→")
63 lowered = re.sub(r"\s+", " ", lowered)
64 if cleaned in {"Thinking", "Auto", "Thinking → Auto", "Operator + Auto"}:
65 return cleaned
66 return mapping.get(lowered)
67
68
69 def _extract_fence_fields(fence: str | None) -> tuple[str | None, str | None, str | None]:
70 if not fence:
71 return None, None, None
72 step_m = _STEP_RE.search(fence)
73 model_m = _MODEL_RE.search(fence)
74 auth_m = _AUTHORITY_RE.search(fence)
75 step = step_m.group("v").strip() if step_m else None
76 model_raw = model_m.group("v").strip() if model_m else None
77 model = normalize_model_display(model_raw) if model_raw else None
78 authority = auth_m.group("v").strip().lower() if auth_m else None
79 return step, model, authority
80
81
82 def _step_from_body(body: str, fence_step: str | None) -> str | None:
83 if fence_step:
84 return fence_step
85 id_m = _ID_ROW_RE.search(body)
86 if id_m:
87 return id_m.group("id").strip()
88 return None
89
90
91 def _forbidden_archived_next_title(title: str) -> bool:
92 return "archived" in title.lower()
93
94
95 def _ambiguous_primary_phrase(title: str) -> bool:
96 lowered = title.lower()
97 return "primary relay" in lowered or "primary (relay)" in lowered
98
99
100 def extract_next_blocks(text: str) -> list[NextBlock]:
101 """Extract all marked (and legacy unmarked NEXT) blocks from handover text."""
102 lines = lf_normalize(text).split("\n")
103 blocks: list[NextBlock] = []
104 i = 0
105 while i < len(lines):
106 line = lines[i]
107 marker_m = _MARKER_RE.search(line.strip())
108 if marker_m and i + 1 < len(lines):
109 heading_line = i + 2 # 1-indexed heading
110 heading = lines[i + 1]
111 role = NextRole(marker_m.group("role").lower())
112 lane = marker_m.group("lane")
113 status = (marker_m.group("status") or ("archived" if role is NextRole.ARCHIVED else "live")).lower()
114 product_order = marker_m.group("product_order")
115 tip_hash = marker_m.group("tip_hash")
116 if tip_hash:
117 tip_hash = tip_hash.lower()
118
119 # Collect body until next marker or next major heading at ## level that starts a session block
120 j = i + 2
121 body_lines: list[str] = []
122 while j < len(lines):
123 peek = lines[j].strip()
124 if _MARKER_RE.search(peek):
125 break
126 if (
127 _HEADING_NEXT.match(lines[j])
128 or _HEADING_PRODUCT_RELAY.match(lines[j])
129 or _HEADING_LANE_TIP.match(lines[j])
130 or _HEADING_ARCHIVED.match(lines[j])
131 ):
132 # Unmarked heading — stop before it so legacy scanner can see it
133 break
134 body_lines.append(lines[j])
135 j += 1
136 body = "\n".join(body_lines)
137 fence_m = _FENCE_RE.search(body)
138 fence = fence_m.group(1) if fence_m else None
139 step, model, authority = _extract_fence_fields(fence)
140 step = _step_from_body(body, step)
141 blocks.append(
142 NextBlock(
143 role=role,
144 lane=lane,
145 status=status,
146 product_order=product_order,
147 tip_hash=tip_hash,
148 heading=heading.strip(),
149 heading_line=heading_line,
150 body=body,
151 fence=fence,
152 step_id=step,
153 model=model,
154 authority=authority,
155 unmarked=False,
156 )
157 )
158 i = j
159 continue
160
161 # Legacy / unmarked NEXT SESSION headings
162 next_m = _HEADING_NEXT.match(line)
163 if next_m:
164 # Skip if previous non-empty line was a marker (already consumed)
165 title = next_m.group("title")
166 j = i + 1
167 body_lines = []
168 while j < len(lines):
169 peek = lines[j].strip()
170 if _MARKER_RE.search(peek):
171 break
172 if (
173 _HEADING_NEXT.match(lines[j])
174 or _HEADING_PRODUCT_RELAY.match(lines[j])
175 or _HEADING_LANE_TIP.match(lines[j])
176 or _HEADING_ARCHIVED.match(lines[j])
177 ):
178 break
179 body_lines.append(lines[j])
180 j += 1
181 body = "\n".join(body_lines)
182 fence_m = _FENCE_RE.search(body)
183 fence = fence_m.group(1) if fence_m else None
184 step, model, authority = _extract_fence_fields(fence)
185 step = _step_from_body(body, step)
186 blocks.append(
187 NextBlock(
188 role=NextRole.PRIMARY,
189 lane=None,
190 status="live",
191 product_order=None,
192 tip_hash=None,
193 heading=line.strip(),
194 heading_line=i + 1,
195 body=body,
196 fence=fence,
197 step_id=step,
198 model=model,
199 authority=authority,
200 unmarked=True,
201 )
202 )
203 i = j
204 continue
205 i += 1
206 return blocks
207
208
209 def legacy_forbidden_archived_headings(text: str) -> list[tuple[int, str]]:
210 """Return (line, heading) for ``## NEXT SESSION — … archived …`` (forbidden)."""
211 out: list[tuple[int, str]] = []
212 for idx, line in enumerate(lf_normalize(text).split("\n"), start=1):
213 m = _HEADING_NEXT.match(line)
214 if m and _forbidden_archived_next_title(m.group("title")):
215 out.append((idx, line.strip()))
216 return out
217
218
219 def select_live_primary(blocks: list[NextBlock], *, lane: str = "product") -> NextBlock | None:
220 """Select LIVE PRIMARY for a lane (ignores archived)."""
221 live = [
222 b
223 for b in blocks
224 if b.role is NextRole.PRIMARY
225 and b.status == "live"
226 and not b.unmarked
227 and (b.lane is None or b.lane == lane)
228 ]
229 if len(live) == 1:
230 return live[0]
231 if len(live) > 1:
232 return None # ambiguous — caller treats as ambiguous_primary
233 return None
234
235
236 def select_product_tip(blocks: list[NextBlock], *, lane: str = "product") -> tuple[NextBlock | None, str | None]:
237 """Select relay XOR product_relay tip for product lane.
238
239 Returns ``(block, error_code)`` where error_code is ``ambiguous_primary`` when both present.
240 """
241 relays = [
242 b
243 for b in blocks
244 if b.role is NextRole.RELAY
245 and b.status == "live"
246 and (b.lane is None or b.lane == lane)
247 ]
248 product_relays = [
249 b
250 for b in blocks
251 if b.role is NextRole.PRODUCT_RELAY
252 and b.status == "live"
253 and (b.lane is None or b.lane == lane or b.lane == "product")
254 ]
255 if relays and product_relays:
256 return None, "ambiguous_primary"
257 if len(relays) > 1 or len(product_relays) > 1:
258 return None, "ambiguous_primary"
259 if len(relays) == 1:
260 return relays[0], None
261 if len(product_relays) == 1:
262 return product_relays[0], None
263 return None, None
264
265
266 def primary_paste_hash(block: NextBlock) -> str | None:
267 """Hash of PRIMARY paste fence bytes (required for tip freshness)."""
268 if not block.fence:
269 return None
270 return tip_hash_hex(block.fence)
271
272
273 def heading_role_mismatch(block: NextBlock) -> str | None:
274 """Return a short reason when heading pattern disagrees with marker role."""
275 heading = block.heading
276 if block.role is NextRole.PRIMARY:
277 if not heading.startswith("## NEXT SESSION"):
278 return "primary marker requires ## NEXT SESSION heading"
279 if "(PRIMARY)" not in heading:
280 return "PRIMARY heading must end with (PRIMARY)"
281 title = heading.split("—", 1)[-1] if "—" in heading else heading.split("-", 1)[-1]
282 if _forbidden_archived_next_title(title) or _ambiguous_primary_phrase(title):
283 return "forbidden PRIMARY title phrasing"
284 elif block.role is NextRole.RELAY:
285 if not heading.startswith("## NEXT SESSION"):
286 return "relay marker requires ## NEXT SESSION heading"
287 if "(RELAY →" not in heading and "(RELAY->" not in heading:
288 return "RELAY heading must contain (RELAY → …)"
289 title = heading.split("—", 1)[-1] if "—" in heading else heading
290 if _forbidden_archived_next_title(title):
291 return "forbidden RELAY title with archived"
292 elif block.role is NextRole.PRODUCT_RELAY:
293 if not heading.startswith("## PRODUCT RELAY"):
294 return "product_relay marker requires ## PRODUCT RELAY heading"
295 elif block.role is NextRole.LANE_TIP:
296 if not heading.startswith("## LANE TIP"):
297 return "lane_tip marker requires ## LANE TIP heading"
298 elif block.role is NextRole.ARCHIVED:
299 if heading.startswith("## NEXT SESSION"):
300 return "archived must not use ## NEXT SESSION heading"
301 if not heading.startswith("## ARCHIVED SESSION"):
302 return "archived marker requires ## ARCHIVED SESSION heading"
303 return None
304
305
306 def count_next_session_headings(text: str) -> int:
307 """Count ``## NEXT SESSION —`` headings (KH1 H2 / H13)."""
308 count = 0
309 for line in lf_normalize(text).split("\n"):
310 if _HEADING_NEXT.match(line):
311 count += 1
312 return count
313
314
315 def with_computed_hash(block: NextBlock) -> NextBlock:
316 """Return block with tip_hash filled from fence when role is primary."""
317 if block.role is NextRole.PRIMARY and block.fence and not block.tip_hash:
318 return replace(block, tip_hash=tip_hash_hex(block.fence))
319 return block
File History 1 commit
sha256:6abcf1fa82a7a621ccbc945f19acdba5bc0db54569599404a1452fb4a096a199 fix(ISR): default require_independent_second_reviewer to require Human minor ⚠ 1 day ago