aaronrene / overseer-kit public
check.py python
320 lines 10.8 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 3 days ago
1 """Governance freshness probe — D1/D2 + marker; skips R4/gh (§GFG.4)."""
2
3 from __future__ import annotations
4
5 from dataclasses import dataclass
6 from pathlib import Path
7
8 from adapters.base import VcsAdapter
9 from adapters.config import OverseerConfig, resolve_lane_docs
10 from adapters.errors import ReadError
11 from adapters.factory import create_adapter
12 from adapters.runner import CommandRunner, SubprocessRunner
13 from cli.docs_paths import lane_living_doc_abs
14 from cli.version_lock import LockError, lock_path, read_version_lock
15 from tools.governance_hygiene.drift import detect_drift
16 from tools.governance_hygiene.types import VerifiedReads
17
18 GOVERNANCE_SYNC_MARKER = "last_governance_sync"
19 REMEDIATION_DRY_RUN = (
20 "ok governance-sync --dry-run then apply when the plan is correct "
21 "(ok governance-sync without dry-run / explicit apply path)"
22 )
23 REMEDIATION_RESTAMP = "ok governance-sync --dry-run"
24
25
26 @dataclass(frozen=True)
27 class SyncMarker:
28 """Parsed ``.overseer/last_governance_sync`` (§GFG.5.2)."""
29
30 timestamp: str
31 r1: str | None # None = absent key; "" = present empty
32 r3: str | None
33 legacy: bool # True when tip fields absent (timestamp-only)
34
35
36 @dataclass(frozen=True)
37 class GovernanceFreshnessReport:
38 """Result of a governance freshness probe (§GFG.4)."""
39
40 state: str # ok | drifted | stale_marker | unreadable | not_applicable
41 message: str
42 remediation: str | None
43 d1: str | None = None
44 d2: str | None = None
45 marker_present: bool = False
46 marker_r1: str | None = None
47 actual_r1: str | None = None
48
49 @property
50 def ok(self) -> bool:
51 return self.state in {"ok", "not_applicable"}
52
53
54 def parse_sync_marker(text: str) -> SyncMarker | None:
55 """Parse enriched or legacy marker body; empty input → None."""
56 lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
57 if not lines:
58 return None
59 timestamp = lines[0]
60 fields: dict[str, str] = {}
61 for line in lines[1:]:
62 if "=" not in line:
63 continue
64 key, _, value = line.partition("=")
65 fields[key.strip().lower()] = value.strip()
66 has_tip_key = "r1" in fields or "r3" in fields
67 return SyncMarker(
68 timestamp=timestamp,
69 r1=fields.get("r1") if "r1" in fields else None,
70 r3=fields.get("r3") if "r3" in fields else None,
71 legacy=not has_tip_key,
72 )
73
74
75 def check_governance_freshness(
76 config: OverseerConfig,
77 repo_root: Path,
78 *,
79 adapter: VcsAdapter | None = None,
80 runner: CommandRunner | None = None,
81 ) -> GovernanceFreshnessReport:
82 """Resolve freshness from D1/D2 + marker; never calls ``gh`` (§GFG.4.1)."""
83 try:
84 lock_file = lock_path(repo_root)
85 if not lock_file.is_file():
86 return GovernanceFreshnessReport(
87 state="not_applicable",
88 message="not initialized — no freshness obligation yet",
89 remediation=None,
90 )
91 read_version_lock(lock_file)
92 except LockError as exc:
93 return GovernanceFreshnessReport(
94 state="unreadable",
95 message=f"version.lock unreadable: {exc}",
96 remediation="ok sync",
97 )
98 except OSError as exc:
99 return GovernanceFreshnessReport(
100 state="unreadable",
101 message=f"could not load install state: {exc}",
102 remediation="ok sync",
103 )
104
105 active_runner = runner or SubprocessRunner()
106 active_adapter = adapter or create_adapter(config, repo_root, runner=active_runner)
107
108 reads_or_fail = _perform_freshness_reads(config, active_adapter)
109 if isinstance(reads_or_fail, str):
110 return GovernanceFreshnessReport(
111 state="unreadable",
112 message=reads_or_fail,
113 remediation=REMEDIATION_DRY_RUN,
114 )
115 reads = reads_or_fail
116
117 try:
118 handover_text, roadmap_text = _load_default_lane_docs(config, repo_root)
119 except OSError as exc:
120 return GovernanceFreshnessReport(
121 state="unreadable",
122 message=f"governance docs unreadable: {exc}",
123 remediation=REMEDIATION_DRY_RUN,
124 actual_r1=reads.r1_github_main_sha,
125 )
126
127 drift = detect_drift(reads, handover_text, roadmap_text)
128 d1 = drift.d1_handover_vs_git
129 d2 = drift.d2_anchor_vs_canonical
130
131 if d1 == "unreadable" or d2 == "unreadable":
132 return GovernanceFreshnessReport(
133 state="unreadable",
134 message="D1/D2 unreadable — failing closed",
135 remediation=REMEDIATION_DRY_RUN,
136 d1=d1,
137 d2=d2,
138 actual_r1=reads.r1_github_main_sha,
139 )
140
141 if d1 == "drifted" or d2 == "drifted":
142 which = "D1" if d1 == "drifted" else "D2"
143 if d1 == "drifted" and d2 == "drifted":
144 which = "D1/D2"
145 return GovernanceFreshnessReport(
146 state="drifted",
147 message=f"{which} drifted — handover/main freshness out of date",
148 remediation=REMEDIATION_DRY_RUN,
149 d1=d1,
150 d2=d2,
151 actual_r1=reads.r1_github_main_sha,
152 )
153
154 # D1 and D2 aligned — evaluate marker (§GFG.4.2 step 5). D3 ignored.
155 marker_path = repo_root / ".overseer" / GOVERNANCE_SYNC_MARKER
156 marker: SyncMarker | None = None
157 if marker_path.is_file():
158 try:
159 marker = parse_sync_marker(marker_path.read_text(encoding="utf-8"))
160 except OSError as exc:
161 return GovernanceFreshnessReport(
162 state="unreadable",
163 message=f"marker unreadable: {exc}",
164 remediation=REMEDIATION_RESTAMP,
165 d1=d1,
166 d2=d2,
167 actual_r1=reads.r1_github_main_sha,
168 )
169
170 tip_known, tip_value, tip_field = _regime_tip(config, reads)
171 marker_r1 = marker.r1 if marker is not None else None
172
173 if marker is None:
174 if tip_known:
175 return GovernanceFreshnessReport(
176 state="stale_marker",
177 message="last_governance_sync missing after main tip is known",
178 remediation=REMEDIATION_RESTAMP,
179 d1=d1,
180 d2=d2,
181 marker_present=False,
182 marker_r1=None,
183 actual_r1=reads.r1_github_main_sha,
184 )
185 return GovernanceFreshnessReport(
186 state="ok",
187 message="D1/D2 aligned; tip unknown — no marker obligation",
188 remediation=None,
189 d1=d1,
190 d2=d2,
191 marker_present=False,
192 actual_r1=reads.r1_github_main_sha,
193 )
194
195 if marker.legacy:
196 if tip_known:
197 return GovernanceFreshnessReport(
198 state="stale_marker",
199 message="legacy timestamp-only marker — re-stamp with enriched tip fields",
200 remediation=REMEDIATION_RESTAMP,
201 d1=d1,
202 d2=d2,
203 marker_present=True,
204 marker_r1=None,
205 actual_r1=reads.r1_github_main_sha,
206 )
207 return GovernanceFreshnessReport(
208 state="ok",
209 message="D1/D2 aligned; legacy marker present; tip unknown",
210 remediation=None,
211 d1=d1,
212 d2=d2,
213 marker_present=True,
214 actual_r1=reads.r1_github_main_sha,
215 )
216
217 stamped = marker.r1 if tip_field == "r1" else marker.r3
218 if tip_known and tip_value is not None:
219 stamped_norm = (stamped or "").lower()
220 if stamped_norm != tip_value.lower():
221 return GovernanceFreshnessReport(
222 state="stale_marker",
223 message=f"main advanced since last stamp ({tip_field})",
224 remediation=REMEDIATION_RESTAMP,
225 d1=d1,
226 d2=d2,
227 marker_present=True,
228 marker_r1=marker_r1,
229 actual_r1=reads.r1_github_main_sha,
230 )
231
232 return GovernanceFreshnessReport(
233 state="ok",
234 message="D1/D2 aligned and governance sync marker matches tip",
235 remediation=None,
236 d1=d1,
237 d2=d2,
238 marker_present=True,
239 marker_r1=marker_r1,
240 actual_r1=reads.r1_github_main_sha,
241 )
242
243
244 def _regime_tip(
245 config: OverseerConfig,
246 reads: VerifiedReads,
247 ) -> tuple[bool, str | None, str]:
248 """Return (tip_known, tip_sha, field_name) for marker comparison."""
249 regime = config.vcs.regime
250 if regime == "muse-only":
251 tip = reads.r3_canonical_main_sha
252 return tip is not None and bool(tip), tip, "r3"
253 tip = reads.r1_github_main_sha
254 return tip is not None and bool(tip), tip, "r1"
255
256
257 def _perform_freshness_reads(
258 config: OverseerConfig,
259 adapter: VcsAdapter,
260 ) -> VerifiedReads | str:
261 """R1/R2/R3/R5 only — empty R4; never invokes ``gh`` (§GFG.4.1)."""
262 regime = config.vcs.regime
263
264 status = adapter.status()
265 if isinstance(status, ReadError):
266 return f"{status.command}: {status}"
267
268 r1_sha: str | None = None
269 r1_cmd: str | None = None
270 if regime in {"git-only", "muse+git-mirror"}:
271 remote = config.vcs.git.remote
272 main = config.vcs.git.main_branch
273 r1_cmd = f"git rev-parse {remote}/{main}"
274 head = adapter.read_head(f"{remote}/{main}")
275 if isinstance(head, ReadError):
276 return f"{head.command}: {head}"
277 r1_sha = head.sha.lower()
278
279 anchor = adapter.read_canonical_anchor()
280 if isinstance(anchor, ReadError):
281 return f"{anchor.command}: {anchor}"
282
283 r3_sha: str | None = None
284 r3_cmd: str | None = None
285 if regime in {"muse+git-mirror", "muse-only"}:
286 muse_main = config.vcs.muse.main_branch
287 if not muse_main:
288 return "muse.main_branch not configured"
289 r3_cmd = f"muse rev-parse {muse_main}"
290 head = adapter.read_head(f"muse:{muse_main}")
291 if isinstance(head, ReadError):
292 return f"{head.command}: {head}"
293 r3_sha = head.sha.lower()
294 elif regime == "git-only":
295 r3_sha = r1_sha
296 r3_cmd = r1_cmd
297
298 return VerifiedReads(
299 regime=regime,
300 r1_github_main_sha=r1_sha,
301 r1_command=r1_cmd,
302 r2_anchor_sha=anchor.anchor_sha.lower(),
303 r2_source=anchor.source,
304 r3_canonical_main_sha=r3_sha,
305 r3_command=r3_cmd,
306 r4_merged_prs=(),
307 r5_branch=status.branch.strip(),
308 r5_dirty=status.dirty,
309 r5_regime=status.regime,
310 )
311
312
313 def _load_default_lane_docs(config: OverseerConfig, repo_root: Path) -> tuple[str, str]:
314 lane = config.docs.default_lane if config.docs.lanes is not None else None
315 lane_docs = resolve_lane_docs(config, lane)
316 handover_path = lane_living_doc_abs(repo_root, config, lane_docs, lane_docs.handover)
317 roadmap_path = lane_living_doc_abs(repo_root, config, lane_docs, lane_docs.roadmap)
318 handover = handover_path.read_text(encoding="utf-8") if handover_path.is_file() else ""
319 roadmap = roadmap_path.read_text(encoding="utf-8") if roadmap_path.is_file() else ""
320 return handover, roadmap
File History 1 commit
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 3 days ago