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