aaronrene / overseer-kit public
adapter.py python
206 lines 6.6 KB
Raw
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439 feat: K1-P1 complete — agent provenance, build-verification… Sonnet 4.6 patch 54 days ago
1 """Muse+git-mirror VCS adapter backend (§4, SD-14)."""
2
3 from __future__ import annotations
4
5 from adapters.base import BaseAdapter, read_bridge_git_sha
6 from adapters.errors import ReadError, WriteError
7 from adapters.types import (
8 AnchorResult,
9 CommitResult,
10 HeadResult,
11 MirrorResult,
12 RealignResult,
13 StatusResult,
14 )
15
16
17 class MuseGitMirrorAdapter(BaseAdapter):
18 """Scooling/Knowtation backend — Muse canonical with GitHub mirror."""
19
20 def status(self) -> StatusResult | ReadError:
21 muse_branch = self._muse("rev-parse", "--abbrev-ref", "HEAD")
22 if isinstance(muse_branch, ReadError):
23 return muse_branch
24 muse_dirty = self._muse_dirty()
25 if isinstance(muse_dirty, ReadError):
26 return muse_dirty
27 git_branch = self._git("rev-parse", "--abbrev-ref", "HEAD")
28 if isinstance(git_branch, ReadError):
29 return git_branch
30 git_dirty = self._git("status", "--porcelain")
31 if isinstance(git_dirty, ReadError):
32 return git_dirty
33
34 notes = [
35 "canonical=muse",
36 f"git-branch={git_branch.stdout}",
37 "sd-14: never git push origin main",
38 ]
39 dirty = muse_dirty or bool(git_dirty.stdout.strip())
40 return StatusResult(
41 regime=self.regime,
42 dirty=dirty,
43 branch=muse_branch.stdout,
44 notes=notes,
45 )
46
47 def read_head(self, ref: str) -> HeadResult | ReadError:
48 if ref.startswith("muse:") or ref.startswith("sha256:"):
49 muse_ref = ref.removeprefix("muse:")
50 return self._muse_rev_parse_sha(muse_ref)
51
52 result = self._git("rev-parse", ref)
53 if isinstance(result, ReadError):
54 return result
55 sha = result.stdout.strip()
56 if not sha:
57 return ReadError(f"git rev-parse {ref}", "empty sha")
58 return HeadResult(sha=sha, kind="git")
59
60 def read_canonical_anchor(self) -> AnchorResult | ReadError:
61 export_sha = read_bridge_git_sha(self.repo_root, "last_export")
62 if export_sha:
63 return AnchorResult(anchor_sha=export_sha, source=".muse/git-bridge.toml:last_export")
64
65 mirror = self.config.vcs.git.mirror_branch
66 remote = self.config.vcs.git.remote
67 if mirror:
68 ref = f"{remote}/{mirror}"
69 head = self.read_head(ref)
70 if not isinstance(head, ReadError):
71 return AnchorResult(anchor_sha=head.sha, source=ref)
72
73 return ReadError(
74 "read_canonical_anchor",
75 "no bridge anchor in .muse/git-bridge.toml or mirror ref",
76 )
77
78 def realign(self, *, dry_run: bool, max_commits: int) -> RealignResult | ReadError:
79 from_ref = read_bridge_git_sha(self.repo_root, "last_import")
80 if not from_ref:
81 from_ref = read_bridge_git_sha(self.repo_root, "last_export")
82 if not from_ref:
83 return ReadError(
84 "realign",
85 "cannot determine from_ref (missing .muse/git-bridge.toml anchor)",
86 )
87
88 remote = self.config.vcs.git.remote
89 main = self.config.vcs.git.main_branch
90 to_ref = f"{remote}/{main}"
91 to_head = self.read_head(to_ref)
92 if isinstance(to_head, ReadError):
93 return to_head
94
95 count_cmd = self._git(
96 "rev-list",
97 "--count",
98 f"{from_ref}..{to_head.sha}",
99 )
100 if isinstance(count_cmd, ReadError):
101 return count_cmd
102 try:
103 would_import = int(count_cmd.stdout.strip())
104 except ValueError:
105 return ReadError(count_cmd.stdout, "invalid rev-list count")
106
107 if would_import > max_commits:
108 return RealignResult(
109 would_import=would_import,
110 applied=False,
111 from_ref=from_ref,
112 to_ref=to_head.sha,
113 reason=f"exceeds max_commits ({max_commits})",
114 )
115
116 if dry_run:
117 return RealignResult(
118 would_import=would_import,
119 applied=False,
120 from_ref=from_ref,
121 to_ref=to_head.sha,
122 reason="dry-run",
123 )
124
125 muse_main = self.config.vcs.muse.main_branch or "main"
126 import_cmd = self._muse(
127 "bridge",
128 "git-import",
129 ".",
130 "--branch",
131 muse_main,
132 "--from-ref",
133 from_ref,
134 "--incremental",
135 "--preserve-merge-commits",
136 )
137 if isinstance(import_cmd, ReadError):
138 return import_cmd
139
140 return RealignResult(
141 would_import=would_import,
142 applied=True,
143 from_ref=from_ref,
144 to_ref=to_head.sha,
145 )
146
147 def commit_feature(
148 self,
149 *,
150 branch: str,
151 message: str,
152 paths: list[str],
153 ) -> CommitResult | ReadError | WriteError:
154 unsafe = self._validate_paths(paths)
155 if unsafe:
156 return unsafe
157 if self._is_protected_branch(branch):
158 return WriteError(
159 "commit_feature",
160 f"refused protected branch {branch!r}",
161 )
162
163 checkout = self._muse("checkout", branch)
164 if isinstance(checkout, ReadError):
165 return checkout
166
167 current = self._muse("rev-parse", "--abbrev-ref", "HEAD")
168 if isinstance(current, ReadError):
169 return current
170 if current.stdout.strip() != branch:
171 return WriteError(
172 "commit_feature",
173 f"branch mismatch after checkout: {current.stdout!r}",
174 )
175
176 if paths:
177 for path in paths:
178 add = self._muse("add", path)
179 if isinstance(add, ReadError):
180 return add
181
182 commit = self._muse("commit", "-m", message)
183 if isinstance(commit, ReadError):
184 return commit
185
186 head = self._muse_rev_parse_sha("HEAD")
187 if isinstance(head, ReadError):
188 return head
189 return CommitResult(committed=True, sha=head.sha)
190
191 def mirror(self, *, dry_run: bool) -> MirrorResult | ReadError:
192 status = self._muse("bridge", "git-status")
193 if isinstance(status, ReadError):
194 return status
195 diff_summary = status.stdout or status.stderr
196 if dry_run:
197 return MirrorResult(
198 diff_summary=diff_summary,
199 pushed=False,
200 reason="dry-run",
201 )
202 return MirrorResult(
203 diff_summary=diff_summary,
204 pushed=False,
205 reason="operator-authorization-required",
206 )
File History 1 commit
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439 feat: K1-P1 complete — agent provenance, build-verification… Sonnet 4.6 patch 54 days ago