aaronrene / overseer-kit public
base.py python
217 lines 7.2 KB
Raw
sha256:0e9549ec7b463911bc08b7d586dc320b1ac9b1f5c943ee7e3865dcc6cb0f6f83 chore(governance): sync handover+roadmap to 84db8c8 (drift:… Human 20 hours ago
1 """Shared adapter helpers and base protocol."""
2
3 from __future__ import annotations
4
5 import json
6 import re
7 from pathlib import Path
8 from typing import Protocol, Union
9
10 from adapters.config import OverseerConfig
11 from adapters.errors import ReadError, WriteError
12 from adapters.runner import CommandResult, CommandRunner, SubprocessRunner, quote_arg
13 from adapters.types import (
14 AnchorResult,
15 CommitResult,
16 HeadResult,
17 MirrorResult,
18 RealignResult,
19 StatusResult,
20 )
21
22 AdapterResult = Union[
23 StatusResult,
24 HeadResult,
25 AnchorResult,
26 RealignResult,
27 CommitResult,
28 MirrorResult,
29 ReadError,
30 WriteError,
31 ]
32
33
34 class VcsAdapter(Protocol):
35 """Frozen §4 adapter interface."""
36
37 def status(self) -> StatusResult | ReadError: ...
38
39 def read_head(self, ref: str) -> HeadResult | ReadError: ...
40
41 def read_canonical_anchor(self) -> AnchorResult | ReadError: ...
42
43 def realign(self, *, dry_run: bool, max_commits: int) -> RealignResult | ReadError: ...
44
45 def commit_feature(
46 self,
47 *,
48 branch: str,
49 message: str,
50 paths: list[str],
51 ) -> CommitResult | ReadError | WriteError: ...
52
53 def mirror(self, *, dry_run: bool) -> MirrorResult | ReadError: ...
54
55
56 class BaseAdapter:
57 """Common helpers for all backends."""
58
59 def __init__(
60 self,
61 config: OverseerConfig,
62 repo_root: Path,
63 runner: CommandRunner | None = None,
64 ) -> None:
65 self.config = config
66 self.repo_root = repo_root.resolve()
67 self.runner = runner or SubprocessRunner()
68 self._muse_cwd = self._resolve_muse_cwd()
69
70 def _resolve_muse_cwd(self) -> Path:
71 """Install root, or ``install_root / vcs.muse.working_dir`` (§K6.5.1)."""
72 from cli.docs_paths import validate_muse_working_dir
73
74 working = validate_muse_working_dir(self.repo_root, self.config.vcs.muse.working_dir)
75 return working if working is not None else self.repo_root
76
77 @property
78 def regime(self) -> str:
79 return self.config.vcs.regime
80
81 @property
82 def muse_cwd(self) -> Path:
83 """Absolute directory passed to ``muse -C``."""
84 return self._muse_cwd
85
86 def _git(self, *args: str) -> CommandResult | ReadError:
87 cmd = "git " + " ".join(quote_arg(a) for a in args)
88 result = self.runner.run(cmd, cwd=str(self.repo_root))
89 if not result.ok:
90 return ReadError(cmd, result.stderr or result.stdout, result.exit_code)
91 return result
92
93 def _muse(self, *args: str) -> CommandResult | ReadError:
94 cmd = (
95 "muse -C "
96 + quote_arg(str(self._muse_cwd))
97 + " "
98 + " ".join(quote_arg(a) for a in args)
99 )
100 result = self.runner.run(cmd, cwd=str(self.repo_root))
101 if not result.ok:
102 return ReadError(cmd, result.stderr or result.stdout, result.exit_code)
103 return result
104
105 def _muse_rev_parse_sha(self, ref: str) -> "HeadResult | ReadError":
106 """Resolve a muse ref to its commit id via ``muse rev-parse`` (Muse 0.2+).
107
108 Muse 0.2.x prints the commit id as a bare string on stdout (exit 0).
109 On failure (unknown ref / empty repo) it exits non-zero with JSON on
110 stdout — ``_muse`` already converts non-zero exits to ``ReadError``,
111 so by the time we read ``result.stdout`` the value is always a plain SHA.
112 """
113 result = self._muse("rev-parse", ref)
114 if isinstance(result, ReadError):
115 return result
116 sha = result.stdout.strip()
117 if not sha:
118 return ReadError(f"muse rev-parse {ref}", "empty sha")
119 return HeadResult(sha=sha, kind="muse")
120
121 def _muse_dirty(self) -> bool | ReadError:
122 """Return Muse working-tree dirty flag (Muse 0.2+ ``status --json``; legacy ``--porcelain``)."""
123 json_result = self._muse("status", "--json")
124 if not isinstance(json_result, ReadError):
125 try:
126 payload = json.loads(json_result.stdout)
127 except json.JSONDecodeError:
128 return ReadError(
129 "muse status --json",
130 "invalid JSON from muse status",
131 )
132 if isinstance(payload, dict):
133 if "dirty" in payload:
134 return bool(payload["dirty"])
135 total = payload.get("total_changes")
136 if isinstance(total, int):
137 return total > 0
138 return ReadError("muse status --json", "missing dirty/total_changes field")
139
140 porcelain = self._muse("status", "--porcelain")
141 if isinstance(porcelain, ReadError):
142 return porcelain
143 return bool(porcelain.stdout.strip())
144
145 def _is_protected_branch(self, branch: str) -> bool:
146 git_main = self.config.vcs.git.main_branch
147 muse_main = self.config.vcs.muse.main_branch
148 normalized = branch.removeprefix("muse:").removeprefix("refs/heads/")
149 protected = {git_main}
150 if muse_main:
151 protected.add(muse_main)
152 return normalized in protected
153
154 def _validate_paths(self, paths: list[str]) -> ReadError | None:
155 for path in paths:
156 if not path or path.startswith("-") or ".." in Path(path).parts:
157 return ReadError(
158 "commit_feature",
159 f"refused unsafe path: {path!r}",
160 )
161 return None
162
163
164 BRIDGE_SHA_RE = re.compile(
165 r'^\s*git_sha\s*=\s*"([0-9a-fA-F]+)"\s*$',
166 re.MULTILINE,
167 )
168
169 BRIDGE_MUSE_COMMIT_RE = re.compile(
170 r'^\s*muse_commit_id\s*=\s*"([^"]+)"\s*$',
171 re.MULTILINE,
172 )
173
174
175 def _bridge_section_body(repo_root: Path, section: str) -> str | None:
176 """Return the body of ``[section]`` from ``.muse/git-bridge.toml``, or ``None``."""
177 bridge_path = repo_root / ".muse" / "git-bridge.toml"
178 if not bridge_path.is_file():
179 return None
180 text = bridge_path.read_text(encoding="utf-8")
181 marker = f"[{section}]"
182 if marker not in text:
183 return None
184 section_text = text.split(marker, 1)[1]
185 if "\n[" in section_text:
186 section_text = section_text.split("\n[", 1)[0]
187 return section_text
188
189
190 def bridge_section_present(repo_root: Path, section: str) -> bool:
191 """True when ``.muse/git-bridge.toml`` contains a ``[section]`` header."""
192 return _bridge_section_body(repo_root, section) is not None
193
194
195 def read_bridge_git_sha(repo_root: Path, section: str) -> str | None:
196 """Read ``git_sha`` from ``.muse/git-bridge.toml`` for ``last_export`` or ``last_import``."""
197 section_text = _bridge_section_body(repo_root, section)
198 if section_text is None:
199 return None
200 match = BRIDGE_SHA_RE.search(section_text)
201 return match.group(1) if match else None
202
203
204 def read_bridge_muse_commit_id(repo_root: Path, section: str) -> str | None:
205 """Read ``muse_commit_id`` from ``.muse/git-bridge.toml`` (Muse ID space).
206
207 Used for D2 / canonical-anchor equality under Muse 0.2.x content-hash tips.
208 ``git_sha`` remains for realign ``from_ref`` / Git ancestry only.
209 """
210 section_text = _bridge_section_body(repo_root, section)
211 if section_text is None:
212 return None
213 match = BRIDGE_MUSE_COMMIT_RE.search(section_text)
214 if not match:
215 return None
216 value = match.group(1).strip()
217 return value or None
File History 1 commit
sha256:6abcf1fa82a7a621ccbc945f19acdba5bc0db54569599404a1452fb4a096a199 fix(ISR): default require_independent_second_reviewer to require Human minor ⚠ 20 hours ago