aaronrene / overseer-kit public
patch.py python
233 lines 7.8 KB
Raw
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439 feat: K1-P1 complete — agent provenance, build-verification… Sonnet 4.6 patch 56 days ago
1 """Templated section replacement for handover + roadmap (§4)."""
2
3 from __future__ import annotations
4
5 import re
6 from datetime import date
7
8 from adapters.config import OverseerConfig
9 from tools.governance_hygiene.anchors import replace_anchor_block
10 from tools.governance_hygiene.drift import merged_prs_missing_from_done
11 from tools.governance_hygiene.parse import normalize_status, parse_queue_rows, phase_tokens, pr_matches_row
12 from tools.governance_hygiene.types import QueueRow
13 from tools.governance_hygiene.types import DriftReport, MergedPullRequest, VerifiedReads
14
15
16 def build_handover_patches(
17 handover_text: str,
18 reads: VerifiedReads,
19 drift: DriftReport,
20 *,
21 realign_summary: str | None,
22 sync_date: str | None = None,
23 ) -> tuple[str, tuple[str, ...]]:
24 """Return patched handover text and the list of touched section names."""
25 today = sync_date or date.today().isoformat()
26 sections: list[str] = []
27 text = handover_text
28
29 vcs_body = _render_vcs_table(reads, today)
30 text = replace_anchor_block(text, "vcs-table", vcs_body)
31 sections.append("vcs-table")
32
33 missing_prs = merged_prs_missing_from_done(text, reads.r4_merged_prs)
34 if missing_prs:
35 done_body = _render_done_recently(text, missing_prs)
36 text = replace_anchor_block(text, "done-recently", done_body)
37 sections.append("done-recently")
38
39 snapshot_body = _render_verified_snapshot(reads, drift)
40 text = replace_anchor_block(text, "verified-snapshot", snapshot_body)
41 sections.append("verified-snapshot")
42
43 change_line = _render_change_log_line(drift, reads, realign_summary, today)
44 text = _append_change_log(text, change_line)
45 sections.append("change-log")
46
47 return text, tuple(sections)
48
49
50 def build_roadmap_patches(
51 roadmap_text: str,
52 reads: VerifiedReads,
53 drift: DriftReport,
54 ) -> tuple[str, tuple[str, ...]]:
55 """Return patched roadmap text and touched section names."""
56 sections: list[str] = []
57 text = roadmap_text
58
59 if drift.d3_queue_vs_merged == "drifted":
60 text = _patch_queue_rows(text, reads.r4_merged_prs)
61 sections.append("build-queue")
62
63 glance = _render_next_step_glance(text)
64 text = replace_anchor_block(text, "next-step-glance", glance)
65 sections.append("next-step-glance")
66
67 return text, tuple(sections)
68
69
70 def _render_vcs_table(reads: VerifiedReads, today: str) -> str:
71 dirty = "yes" if reads.r5_dirty else "no"
72 lines = [
73 f"## VCS (verified {today})",
74 "",
75 "| Item | Value |",
76 "| --- | --- |",
77 f"| Branch | `{reads.r5_branch}` |",
78 ]
79 if reads.r1_github_main_sha:
80 lines.append(f"| GitHub `main` | `{reads.r1_github_main_sha}` |")
81 lines.append(f"| Canonical anchor | `{reads.r2_anchor_sha}` ({reads.r2_source}) |")
82 if reads.r3_canonical_main_sha and reads.regime != "git-only":
83 lines.append(f"| Muse `main` | `{reads.r3_canonical_main_sha}` |")
84 lines.append(f"| Dirty | {dirty} |")
85 return "\n".join(lines)
86
87
88 def _render_done_recently(handover_text: str, new_prs: list[MergedPullRequest]) -> str:
89 existing_rows: list[str] = []
90 capture = False
91 for line in handover_text.splitlines():
92 if line.strip().startswith("### What just landed"):
93 capture = True
94 continue
95 if capture and line.startswith("### "):
96 break
97 if capture and line.startswith("|") and "---" not in line and "Slice" not in line:
98 existing_rows.append(line)
99
100 new_rows = [
101 f"| PR #{pr.number} | {pr.title} (merged {pr.merged_at[:10] if pr.merged_at else 'unknown'}) |"
102 for pr in new_prs
103 ]
104 rows = new_rows + existing_rows
105 body = [
106 "### What just landed",
107 "",
108 "| Slice | Deliverable |",
109 "| --- | --- |",
110 *rows,
111 ]
112 return "\n".join(body)
113
114
115 def _render_verified_snapshot(reads: VerifiedReads, drift: DriftReport) -> str:
116 lines = [
117 "## Verified snapshot",
118 "",
119 "| Area | State |",
120 "| --- | --- |",
121 f"| **VCS regime** | `{reads.r5_regime}` |",
122 ]
123 if reads.r1_github_main_sha:
124 lines.append(f"| **GitHub main** | `{reads.r1_github_main_sha}` |")
125 lines.append(f"| **Canonical anchor** | `{reads.r2_anchor_sha}` |")
126 if reads.r3_canonical_main_sha:
127 lines.append(f"| **Canonical main** | `{reads.r3_canonical_main_sha}` |")
128 lines.append(f"| **Branch** | `{reads.r5_branch}` |")
129 lines.append(f"| **Dirty** | `{'yes' if reads.r5_dirty else 'no'}` |")
130 lines.append(
131 f"| **Drift** | D1={drift.d1_handover_vs_git}, D2={drift.d2_anchor_vs_canonical}, "
132 f"D3={drift.d3_queue_vs_merged} |"
133 )
134 return "\n".join(lines)
135
136
137 def _render_change_log_line(
138 drift: DriftReport,
139 reads: VerifiedReads,
140 realign_summary: str | None,
141 today: str,
142 ) -> str:
143 parts = [
144 f"D1={drift.d1_handover_vs_git}",
145 f"D2={drift.d2_anchor_vs_canonical}",
146 f"D3={drift.d3_queue_vs_merged}",
147 ]
148 summary = f"governance-sync: drift ({', '.join(parts)})"
149 if reads.r1_github_main_sha:
150 summary += f" @ `{reads.r1_github_main_sha[:7]}`"
151 if realign_summary:
152 summary += f"; realign: {realign_summary}"
153 return f"- **{today}** — {summary}"
154
155
156 def _append_change_log(text: str, line: str) -> str:
157 if line in text:
158 return text
159 marker = "## Change log"
160 if marker not in text:
161 return text.rstrip() + f"\n\n{marker}\n\n{line}\n"
162 return text.replace(marker, f"{marker}\n\n{line}", 1)
163
164
165 def _patch_queue_rows(roadmap_text: str, merged_prs: tuple[MergedPullRequest, ...]) -> str:
166 lines = roadmap_text.splitlines()
167 out: list[str] = []
168 in_queue = False
169 for line in lines:
170 if line.strip().startswith("## Build queue"):
171 in_queue = True
172 out.append(line)
173 continue
174 if in_queue and line.startswith("## "):
175 in_queue = False
176 if in_queue and line.startswith("|") and "---" not in line and "Phase |" not in line:
177 row = _maybe_merge_row(line, merged_prs)
178 out.append(row)
179 continue
180 out.append(line)
181 return "\n".join(out) + ("\n" if roadmap_text.endswith("\n") else "")
182
183
184 def _maybe_merge_row(line: str, merged_prs: tuple[MergedPullRequest, ...]) -> str:
185 cells = [cell.strip() for cell in line.strip("|").split("|")]
186 if len(cells) < 4:
187 return line
188 phase_label = cells[0]
189 status = normalize_status(cells[2])
190 if status in {"DONE", "MERGED"}:
191 return line
192 probe = QueueRow(
193 phase_label=phase_label,
194 model="",
195 status="",
196 deliverable="",
197 raw_line=line,
198 )
199 for pr in merged_prs:
200 if pr_matches_row(pr.title, probe):
201 cells[2] = f"**DONE** (PR #{pr.number}, `{pr.merge_commit_sha[:7]}`)"
202 return "| " + " | ".join(cells) + " |"
203 return line
204
205
206 def _render_next_step_glance(roadmap_text: str) -> str:
207 rows = parse_queue_rows(roadmap_text)
208 next_rows = [row for row in rows if normalize_status(row.status) in {"TODO", "NEXT", "WIP"}]
209 if not next_rows:
210 return "\n".join(
211 [
212 "## Next step at a glance",
213 "",
214 "_No unambiguous NEXT row — operator authorship required._",
215 ]
216 )
217 row = next_rows[0]
218 tokens = phase_tokens(row.phase_label)
219 phase_id = tokens[0] if tokens else row.phase_label
220 return "\n".join(
221 [
222 "## Next step at a glance",
223 "",
224 f"**Next:** {row.phase_label} — **Model:** {row.model} — **Status:** {row.status}",
225 f"**Phase ID:** {phase_id}",
226 ]
227 )
228
229
230 def extract_paste_ready_block(handover_text: str) -> str | None:
231 """Return the fenced paste-ready prompt block if present."""
232 match = re.search(r"### Paste-ready prompt[\s\S]*?```[\s\S]*?```", handover_text)
233 return match.group(0) if match else None
File History 1 commit
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439 feat: K1-P1 complete — agent provenance, build-verification… Sonnet 4.6 patch 56 days ago