aaronrene / overseer-kit public
reads.py python
175 lines 5.6 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 2 days ago
1 """Verified reads R1–R5 via kit adapter + gh (§2)."""
2
3 from __future__ import annotations
4
5 import json
6 from dataclasses import dataclass
7 from pathlib import Path
8
9 from adapters.base import VcsAdapter
10 from adapters.config import OverseerConfig
11 from adapters.errors import ReadError
12 from adapters.runner import CommandRunner
13 from tools.footprint_coverage import check_footprint_coverage
14 from tools.footprint_integrity import check_footprint_integrity
15 from tools.governance_hygiene.types import MergedPullRequest, VerifiedReads
16 from tools.muse_sync import check_muse_sync
17 from tools.substrate_health import check_substrate
18
19 GH_MERGED_LIMIT = 5
20
21
22 @dataclass(frozen=True)
23 class ReadFailure:
24 """Fail-closed read error with the exact failing command."""
25
26 command: str
27 message: str
28 regime: str
29
30
31 def perform_verified_reads(
32 config: OverseerConfig,
33 adapter: VcsAdapter,
34 runner: CommandRunner,
35 *,
36 repo_root: Path | None = None,
37 ) -> VerifiedReads | ReadFailure:
38 """Execute R1–R5; stop at the first failure."""
39 regime = config.vcs.regime
40
41 if repo_root is not None:
42 substrate = check_substrate(config, repo_root)
43 if not substrate.ok:
44 return ReadFailure(
45 "substrate-health",
46 substrate.message,
47 regime,
48 )
49
50 status = adapter.status()
51 if isinstance(status, ReadError):
52 return ReadFailure(status.command, str(status), regime)
53
54 muse_sync = check_muse_sync(config, status)
55 if not muse_sync.ok:
56 return ReadFailure("muse-sync", muse_sync.message, regime)
57
58 if repo_root is not None:
59 footprint_self_integrity = check_footprint_integrity(repo_root)
60 if not footprint_self_integrity.ok:
61 return ReadFailure(
62 "footprint-self-integrity",
63 footprint_self_integrity.message,
64 regime,
65 )
66
67 footprint_coverage = check_footprint_coverage(repo_root, config)
68 if not footprint_coverage.ok:
69 return ReadFailure(
70 "footprint-coverage",
71 footprint_coverage.message,
72 regime,
73 )
74
75 r1_sha: str | None = None
76 r1_cmd: str | None = None
77 if regime in {"git-only", "muse+git-mirror"}:
78 remote = config.vcs.git.remote
79 main = config.vcs.git.main_branch
80 r1_cmd = f"git rev-parse {remote}/{main}"
81 head = adapter.read_head(f"{remote}/{main}")
82 if isinstance(head, ReadError):
83 return ReadFailure(head.command, str(head), regime)
84 r1_sha = head.sha.lower()
85
86 anchor = adapter.read_canonical_anchor()
87 if isinstance(anchor, ReadError):
88 return ReadFailure(anchor.command, str(anchor), regime)
89
90 r3_sha: str | None = None
91 r3_cmd: str | None = None
92 if regime in {"muse+git-mirror", "muse-only"}:
93 muse_main = config.vcs.muse.main_branch
94 if not muse_main:
95 return ReadFailure(
96 "read_head",
97 "muse.main_branch not configured",
98 regime,
99 )
100 r3_cmd = f"muse log -1 --format=%H {muse_main}"
101 head = adapter.read_head(f"muse:{muse_main}")
102 if isinstance(head, ReadError):
103 return ReadFailure(head.command, str(head), regime)
104 r3_sha = head.sha.lower()
105 elif regime == "git-only":
106 r3_sha = r1_sha
107 r3_cmd = r1_cmd
108
109 merged: tuple[MergedPullRequest, ...] = ()
110 if regime in {"git-only", "muse+git-mirror"}:
111 gh_result = fetch_merged_prs(runner)
112 if isinstance(gh_result, ReadFailure):
113 return gh_result
114 merged = gh_result
115
116 return VerifiedReads(
117 regime=regime,
118 r1_github_main_sha=r1_sha,
119 r1_command=r1_cmd,
120 r2_anchor_sha=anchor.anchor_sha.lower(),
121 r2_source=anchor.source,
122 r3_canonical_main_sha=r3_sha,
123 r3_command=r3_cmd,
124 r4_merged_prs=merged,
125 r5_branch=status.branch.strip(),
126 r5_dirty=status.dirty,
127 r5_regime=status.regime,
128 )
129
130
131 def fetch_merged_prs(runner: CommandRunner) -> tuple[MergedPullRequest, ...] | ReadFailure:
132 """R4: recent merged PRs via gh (never parse docs)."""
133 command = (
134 "gh pr list --state merged --limit "
135 f"{GH_MERGED_LIMIT} --json number,title,mergeCommit,mergedAt"
136 )
137 result = runner.run(command)
138 if not result.ok:
139 message = result.stderr or result.stdout or "gh command failed"
140 return ReadFailure(command, message, "gh")
141 if not result.stdout.strip():
142 return ()
143
144 try:
145 payload = json.loads(result.stdout)
146 except json.JSONDecodeError as exc:
147 return ReadFailure(command, f"invalid json: {exc}", "gh")
148
149 if not isinstance(payload, list):
150 return ReadFailure(command, "expected JSON array", "gh")
151
152 merged: list[MergedPullRequest] = []
153 for item in payload:
154 if not isinstance(item, dict):
155 continue
156 number = item.get("number")
157 title = item.get("title")
158 merge_commit = item.get("mergeCommit") or {}
159 sha = merge_commit.get("oid") if isinstance(merge_commit, dict) else None
160 merged_at = item.get("mergedAt")
161 if not isinstance(number, int) or not isinstance(title, str):
162 continue
163 if not isinstance(sha, str) or not sha:
164 continue
165 if not isinstance(merged_at, str):
166 merged_at = ""
167 merged.append(
168 MergedPullRequest(
169 number=number,
170 title=title,
171 merge_commit_sha=sha.lower(),
172 merged_at=merged_at,
173 )
174 )
175 return tuple(merged)
File History 2 commits
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 2 days ago
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439 feat: K1-P1 complete — agent provenance, build-verification… Sonnet 4.6 patch 53 days ago