aaronrene / overseer-kit public
drift.py python
99 lines 3.3 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago
1 """Drift detection D1–D3 (§3)."""
2
3 from __future__ import annotations
4
5 from tools.governance_hygiene.parse import normalize_status, parse_handover_github_main_sha, parse_queue_rows, pr_matches_row
6 from tools.governance_hygiene.types import DriftReport, DriftState, MergedPullRequest, VerifiedReads
7
8
9 def detect_drift(
10 reads: VerifiedReads,
11 handover_text: str,
12 roadmap_text: str,
13 ) -> DriftReport:
14 """Compare verified reads to doc claims; never infer VCS state from docs alone."""
15 d1 = _detect_d1(reads, handover_text)
16 d2 = _detect_d2(reads)
17 d3 = _detect_d3(reads, roadmap_text)
18 details: dict[str, str] = {}
19 if d1 == "drifted":
20 claimed = parse_handover_github_main_sha(handover_text)
21 details["d1"] = f"claimed={claimed} actual={reads.r1_github_main_sha}"
22 if d2 == "drifted":
23 details["d2"] = f"anchor={reads.r2_anchor_sha} canonical={reads.r3_canonical_main_sha}"
24 if d3 == "drifted":
25 details["d3"] = "queue status mismatch vs merged PRs"
26 return DriftReport(
27 d1_handover_vs_git=d1,
28 d2_anchor_vs_canonical=d2,
29 d3_queue_vs_merged=d3,
30 details=details,
31 )
32
33
34 def _detect_d1(reads: VerifiedReads, handover_text: str) -> DriftState:
35 if reads.regime == "muse-only":
36 return "aligned"
37 if reads.r1_github_main_sha is None:
38 return "unreadable"
39 claimed = parse_handover_github_main_sha(handover_text)
40 if claimed is None:
41 return "drifted"
42 return "aligned" if claimed == reads.r1_github_main_sha.lower() else "drifted"
43
44
45 def _detect_d2(reads: VerifiedReads) -> DriftState:
46 if reads.regime == "git-only":
47 if reads.r3_canonical_main_sha is None:
48 reads_r3 = reads.r1_github_main_sha
49 else:
50 reads_r3 = reads.r3_canonical_main_sha
51 if reads_r3 is None:
52 return "unreadable"
53 return "aligned" if reads.r2_anchor_sha.lower() == reads_r3.lower() else "drifted"
54
55 if reads.r3_canonical_main_sha is None:
56 return "unreadable"
57 return (
58 "aligned"
59 if reads.r2_anchor_sha.lower() == reads.r3_canonical_main_sha.lower()
60 else "drifted"
61 )
62
63
64 def _detect_d3(reads: VerifiedReads, roadmap_text: str) -> DriftState:
65 if reads.regime == "muse-only":
66 return "aligned"
67
68 rows = parse_queue_rows(roadmap_text)
69 merged = list(reads.r4_merged_prs)
70
71 for pr in merged:
72 matching = [row for row in rows if pr_matches_row(pr.title, row)]
73 if not matching:
74 continue
75 for row in matching:
76 status = normalize_status(row.status)
77 if status not in {"DONE", "MERGED"}:
78 return "drifted"
79
80 for row in rows:
81 status = normalize_status(row.status)
82 if status != "MERGED":
83 continue
84 if not any(pr_matches_row(pr.title, row) for pr in merged):
85 return "drifted"
86
87 return "aligned"
88
89
90 def merged_prs_missing_from_done(handover_text: str, merged: tuple[MergedPullRequest, ...]) -> list[MergedPullRequest]:
91 """Return merged PRs not yet mentioned in the handover done-recently block."""
92 missing: list[MergedPullRequest] = []
93 for pr in merged:
94 if f"PR #{pr.number}" in handover_text or pr.merge_commit_sha[:7] in handover_text:
95 continue
96 if pr.title in handover_text:
97 continue
98 missing.append(pr)
99 return missing
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