aaronrene / overseer-kit public
patch.py python
328 lines 11.7 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago
1 """Templated section replacement for handover + roadmap (§4 / §GSP)."""
2
3 from __future__ import annotations
4
5 import re
6 from datetime import date
7 from pathlib import Path
8
9 from adapters.config import OverseerConfig
10 from tools.governance_hygiene.anchors import replace_anchor_block
11 from tools.governance_hygiene.drift import merged_prs_missing_from_done
12 from tools.governance_hygiene.next_regen import (
13 LAND_PHASE_A,
14 LAND_PHASE_B,
15 LAND_PHASE_UNREADABLE,
16 NextRegenDecision,
17 REASON_LAND_A_WAIT,
18 REASON_LAND_PHASE_UNREADABLE,
19 ensure_primary_next_marker,
20 extract_paste_pr_number,
21 format_change_log_fragment,
22 format_next_regen_token,
23 plan_land_b,
24 plan_next_regen,
25 render_land_b_next_session,
26 render_land_b_paste,
27 render_next_session,
28 render_paste_ready,
29 resolve_land_phase,
30 set_marker_land_phase,
31 )
32 from tools.governance_hygiene.parse import normalize_status, parse_queue_rows, phase_tokens, pr_matches_row
33 from tools.governance_hygiene.types import QueueRow
34 from tools.governance_hygiene.types import DriftReport, MergedPullRequest, VerifiedReads
35
36
37 def build_handover_patches(
38 handover_text: str,
39 reads: VerifiedReads,
40 drift: DriftReport,
41 *,
42 realign_summary: str | None,
43 sync_date: str | None = None,
44 config: OverseerConfig,
45 roadmap_text: str,
46 repo_root: Path,
47 ) -> tuple[str, tuple[str, ...], str]:
48 """Return patched handover text, touched sections, and ``next_regen`` token.
49
50 ``roadmap_text`` must be the D3-reconciled roadmap (§GSP.6.1).
51 """
52 today = sync_date or date.today().isoformat()
53 sections: list[str] = []
54 text = handover_text
55
56 vcs_body = _render_vcs_table(reads, today)
57 text = replace_anchor_block(text, "vcs-table", vcs_body)
58 sections.append("vcs-table")
59
60 missing_prs = merged_prs_missing_from_done(text, reads.r4_merged_prs)
61 if missing_prs:
62 done_body = _render_done_recently(text, missing_prs)
63 text = replace_anchor_block(text, "done-recently", done_body)
64 sections.append("done-recently")
65
66 snapshot_body = _render_verified_snapshot(reads, drift)
67 text = replace_anchor_block(text, "verified-snapshot", snapshot_body)
68 sections.append("verified-snapshot")
69
70 # §PMHF.3.4: land-a NEXT + post-merge-incomplete signals → emit land-b for the
71 # same slice; never re-emit land-a and never clobber a mid-wait land-a paste.
72 main_branch = config.vcs.git.main_branch or "main"
73 merged_pr_numbers = {pr.number for pr in reads.r4_merged_prs}
74 paste_pr = extract_paste_pr_number(text)
75 land_b = plan_land_b(
76 text,
77 d1=drift.d1_handover_vs_git,
78 merged_pr_signal=paste_pr is not None and paste_pr in merged_pr_numbers,
79 main_branch=main_branch,
80 )
81 current_land_phase = resolve_land_phase(text)
82
83 if land_b is not None:
84 text = ensure_primary_next_marker(text, config)
85 next_body = render_land_b_next_session(land_b, config=config, sync_date=today)
86 paste_body = render_land_b_paste(land_b, config=config)
87 text = replace_anchor_block(text, "next-session", next_body)
88 text = replace_anchor_block(text, "paste-ready-prompt", paste_body)
89 text = set_marker_land_phase(text, LAND_PHASE_B)
90 sections.append("next-session")
91 sections.append("paste-ready-prompt")
92 next_regen_token = "next_regen: regenerated (land-b)"
93 next_regen_fragment = "next_regen=regenerated:land-b"
94 elif current_land_phase == LAND_PHASE_A:
95 # Mid-land wait: preserve the land-a NEXT/paste; fail closed on regen.
96 decision = NextRegenDecision(None, REASON_LAND_A_WAIT, None, None, False)
97 next_regen_token = format_next_regen_token(decision)
98 next_regen_fragment = format_change_log_fragment(decision)
99 elif current_land_phase == LAND_PHASE_UNREADABLE:
100 decision = NextRegenDecision(None, REASON_LAND_PHASE_UNREADABLE, None, None, False)
101 next_regen_token = format_next_regen_token(decision)
102 next_regen_fragment = format_change_log_fragment(decision)
103 else:
104 decision = plan_next_regen(
105 roadmap_text=roadmap_text,
106 handover_text=text,
107 config=config,
108 repo_root=repo_root,
109 )
110 if decision.row is not None and decision.reason is None:
111 text = ensure_primary_next_marker(text, config)
112 next_body = render_next_session(
113 decision=decision,
114 roadmap_text=roadmap_text,
115 config=config,
116 sync_date=today,
117 )
118 paste_body = render_paste_ready(decision=decision, config=config)
119 text = replace_anchor_block(text, "next-session", next_body)
120 text = replace_anchor_block(text, "paste-ready-prompt", paste_body)
121 # §PMHF.3.4 rule 2: a regenerated non-land NEXT clears land-phase.
122 text = set_marker_land_phase(text, None)
123 sections.append("next-session")
124 sections.append("paste-ready-prompt")
125 next_regen_token = format_next_regen_token(decision)
126 next_regen_fragment = format_change_log_fragment(decision)
127 change_line = _render_change_log_line(
128 drift,
129 reads,
130 realign_summary,
131 today,
132 next_regen_fragment=next_regen_fragment,
133 )
134 text = _append_change_log(text, change_line)
135 sections.append("change-log")
136
137 return text, tuple(sections), next_regen_token
138
139
140 def build_roadmap_patches(
141 roadmap_text: str,
142 reads: VerifiedReads,
143 drift: DriftReport,
144 ) -> tuple[str, tuple[str, ...]]:
145 """Return patched roadmap text and touched section names."""
146 sections: list[str] = []
147 text = roadmap_text
148
149 if drift.d3_queue_vs_merged == "drifted":
150 text = _patch_queue_rows(text, reads.r4_merged_prs)
151 sections.append("build-queue")
152
153 glance = _render_next_step_glance(text)
154 text = replace_anchor_block(text, "next-step-glance", glance)
155 sections.append("next-step-glance")
156
157 return text, tuple(sections)
158
159
160 def _render_vcs_table(reads: VerifiedReads, today: str) -> str:
161 dirty = "yes" if reads.r5_dirty else "no"
162 lines = [
163 f"## VCS (verified {today})",
164 "",
165 "| Item | Value |",
166 "| --- | --- |",
167 f"| Branch | `{reads.r5_branch}` |",
168 ]
169 if reads.r1_github_main_sha:
170 lines.append(f"| GitHub `main` | `{reads.r1_github_main_sha}` |")
171 lines.append(f"| Canonical anchor | `{reads.r2_anchor_sha}` ({reads.r2_source}) |")
172 if reads.r3_canonical_main_sha and reads.regime != "git-only":
173 lines.append(f"| Muse `main` | `{reads.r3_canonical_main_sha}` |")
174 lines.append(f"| Dirty | {dirty} |")
175 return "\n".join(lines)
176
177
178 def _render_done_recently(handover_text: str, new_prs: list[MergedPullRequest]) -> str:
179 existing_rows: list[str] = []
180 capture = False
181 for line in handover_text.splitlines():
182 if line.strip().startswith("### What just landed"):
183 capture = True
184 continue
185 if capture and line.startswith("### "):
186 break
187 if capture and line.startswith("|") and "---" not in line and "Slice" not in line:
188 existing_rows.append(line)
189
190 new_rows = [
191 f"| PR #{pr.number} | {pr.title} (merged {pr.merged_at[:10] if pr.merged_at else 'unknown'}) |"
192 for pr in new_prs
193 ]
194 rows = new_rows + existing_rows
195 body = [
196 "### What just landed",
197 "",
198 "| Slice | Deliverable |",
199 "| --- | --- |",
200 *rows,
201 ]
202 return "\n".join(body)
203
204
205 def _render_verified_snapshot(reads: VerifiedReads, drift: DriftReport) -> str:
206 lines = [
207 "## Verified snapshot",
208 "",
209 "| Area | State |",
210 "| --- | --- |",
211 f"| **VCS regime** | `{reads.r5_regime}` |",
212 ]
213 if reads.r1_github_main_sha:
214 lines.append(f"| **GitHub main** | `{reads.r1_github_main_sha}` |")
215 lines.append(f"| **Canonical anchor** | `{reads.r2_anchor_sha}` |")
216 if reads.r3_canonical_main_sha:
217 lines.append(f"| **Canonical main** | `{reads.r3_canonical_main_sha}` |")
218 lines.append(f"| **Branch** | `{reads.r5_branch}` |")
219 lines.append(f"| **Dirty** | `{'yes' if reads.r5_dirty else 'no'}` |")
220 lines.append(
221 f"| **Drift** | D1={drift.d1_handover_vs_git}, D2={drift.d2_anchor_vs_canonical}, "
222 f"D3={drift.d3_queue_vs_merged} |"
223 )
224 return "\n".join(lines)
225
226
227 def _render_change_log_line(
228 drift: DriftReport,
229 reads: VerifiedReads,
230 realign_summary: str | None,
231 today: str,
232 *,
233 next_regen_fragment: str | None = None,
234 ) -> str:
235 parts = [
236 f"D1={drift.d1_handover_vs_git}",
237 f"D2={drift.d2_anchor_vs_canonical}",
238 f"D3={drift.d3_queue_vs_merged}",
239 ]
240 summary = f"governance-sync: drift ({', '.join(parts)})"
241 if reads.r1_github_main_sha:
242 summary += f" @ `{reads.r1_github_main_sha[:7]}`"
243 if realign_summary:
244 summary += f"; realign: {realign_summary}"
245 if next_regen_fragment:
246 summary += f"; {next_regen_fragment}"
247 return f"- **{today}** — {summary}"
248
249
250 def _append_change_log(text: str, line: str) -> str:
251 if line in text:
252 return text
253 marker = "## Change log"
254 if marker not in text:
255 return text.rstrip() + f"\n\n{marker}\n\n{line}\n"
256 return text.replace(marker, f"{marker}\n\n{line}", 1)
257
258
259 def _patch_queue_rows(roadmap_text: str, merged_prs: tuple[MergedPullRequest, ...]) -> str:
260 lines = roadmap_text.splitlines()
261 out: list[str] = []
262 in_queue = False
263 for line in lines:
264 if line.strip().startswith("## Build queue"):
265 in_queue = True
266 out.append(line)
267 continue
268 if in_queue and line.startswith("## "):
269 in_queue = False
270 if in_queue and line.startswith("|") and "---" not in line and "Phase |" not in line:
271 row = _maybe_merge_row(line, merged_prs)
272 out.append(row)
273 continue
274 out.append(line)
275 return "\n".join(out) + ("\n" if roadmap_text.endswith("\n") else "")
276
277
278 def _maybe_merge_row(line: str, merged_prs: tuple[MergedPullRequest, ...]) -> str:
279 cells = [cell.strip() for cell in line.strip("|").split("|")]
280 if len(cells) < 4:
281 return line
282 phase_label = cells[0]
283 status = normalize_status(cells[2])
284 if status in {"DONE", "MERGED"}:
285 return line
286 probe = QueueRow(
287 phase_label=phase_label,
288 model="",
289 status="",
290 deliverable="",
291 raw_line=line,
292 )
293 for pr in merged_prs:
294 if pr_matches_row(pr.title, probe):
295 cells[2] = f"**DONE** (PR #{pr.number}, `{pr.merge_commit_sha[:7]}`)"
296 return "| " + " | ".join(cells) + " |"
297 return line
298
299
300 def _render_next_step_glance(roadmap_text: str) -> str:
301 """Fail-closed glance: only emit a NEXT when open-row count is exactly one (§GSP.4.2)."""
302 rows = parse_queue_rows(roadmap_text)
303 next_rows = [row for row in rows if normalize_status(row.status) in {"TODO", "NEXT", "WIP"}]
304 if len(next_rows) != 1:
305 return "\n".join(
306 [
307 "## Next step at a glance",
308 "",
309 "_No unambiguous NEXT row — operator authorship required._",
310 ]
311 )
312 row = next_rows[0]
313 tokens = phase_tokens(row.phase_label)
314 phase_id = tokens[0] if tokens else row.phase_label
315 return "\n".join(
316 [
317 "## Next step at a glance",
318 "",
319 f"**Next:** {row.phase_label} — **Model:** {row.model} — **Status:** {row.status}",
320 f"**Phase ID:** {phase_id}",
321 ]
322 )
323
324
325 def extract_paste_ready_block(handover_text: str) -> str | None:
326 """Return the fenced paste-ready prompt block if present."""
327 match = re.search(r"### Paste-ready prompt[\s\S]*?```[\s\S]*?```", handover_text)
328 return match.group(0) if match else None
File History 2 commits
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439 feat: K1-P1 complete — agent provenance, build-verification… Sonnet 4.6 patch 53 days ago