aaronrene / overseer-kit public
support.py python
1,214 lines 44.4 KB
Raw
sha256:0e9549ec7b463911bc08b7d586dc320b1ac9b1f5c943ee7e3865dcc6cb0f6f83 chore(governance): sync handover+roadmap to 84db8c8 (drift:… Human 1 day ago
1 """Test helpers (not pytest fixtures)."""
2
3 from __future__ import annotations
4
5 import json
6 import os
7 import subprocess
8 from dataclasses import dataclass
9 from pathlib import Path
10
11 from adapters.config import OverseerConfig, load_config
12 from adapters.factory import create_adapter
13 from adapters.runner import CommandResult, RecordingRunner
14
15 FIXTURES = Path(__file__).resolve().parent / "fixtures"
16 PILOT = FIXTURES / "pilot"
17 CHECKPOINTS = FIXTURES / "checkpoints"
18 HONESTY = FIXTURES / "honesty"
19 KIT_ROOT = Path(__file__).resolve().parent.parent
20 OVERSEER_DEPRECATION_LINE = "warning: 'overseer' is deprecated; use 'ok' (same commands).\n"
21
22
23 @dataclass(frozen=True)
24 class ShimResult:
25 """Captured POSIX shim subprocess output."""
26
27 exit_code: int
28 stdout: str
29 stderr: str
30
31
32 def seed_git_repo(repo_root: Path) -> None:
33 """Initialize a minimal git repo for subprocess shim tests."""
34 subprocess.run(["git", "init", "-b", "main"], cwd=repo_root, check=True, capture_output=True)
35 subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=repo_root, check=True, capture_output=True)
36 subprocess.run(["git", "config", "user.name", "test"], cwd=repo_root, check=True, capture_output=True)
37 subprocess.run(
38 ["git", "commit", "--allow-empty", "-m", "seed"],
39 cwd=repo_root,
40 check=True,
41 capture_output=True,
42 )
43
44
45 def run_shim(
46 shim: str,
47 argv: list[str],
48 *,
49 cwd: Path,
50 env: dict[str, str] | None = None,
51 ) -> ShimResult:
52 """Invoke ``cli/<shim>`` as a subprocess (§Q2A.10 integration/e2e helper)."""
53 script = KIT_ROOT / "cli" / shim
54 proc_env = os.environ.copy()
55 if env:
56 proc_env.update(env)
57 completed = subprocess.run(
58 [str(script), *argv],
59 cwd=cwd,
60 env=proc_env,
61 capture_output=True,
62 text=True,
63 check=False,
64 )
65 return ShimResult(
66 exit_code=completed.returncode,
67 stdout=completed.stdout,
68 stderr=completed.stderr,
69 )
70
71
72 def write_config(repo_root: Path, name: str) -> Path:
73 src = FIXTURES / name
74 dest = repo_root / ".overseer" / "config.yaml"
75 dest.parent.mkdir(parents=True, exist_ok=True)
76 dest.write_text(src.read_text(encoding="utf-8"), encoding="utf-8")
77 return dest
78
79
80 def load_fixture_config(repo_root: Path, name: str) -> OverseerConfig:
81 return load_config(write_config(repo_root, name))
82
83
84 def pls_config(
85 repo_root: Path,
86 fixture: str = "config-git-only.yaml",
87 *,
88 enabled: bool = True,
89 ) -> OverseerConfig:
90 """Fixture config with ``close_ritual.post_land_sync.enabled`` overridden (§PLS tests)."""
91 from dataclasses import replace
92
93 from adapters.config import PostLandSyncConfig
94
95 config = load_fixture_config(repo_root, fixture)
96 return replace(
97 config,
98 close_ritual=replace(
99 config.close_ritual,
100 post_land_sync=PostLandSyncConfig(enabled=enabled),
101 ),
102 )
103
104
105 def gh_merged_runner(*, pr_state: str = "OPEN", check_state: str = "pass"):
106 """Fake ``gh`` runner driving ``run_pr_land`` to a merge outcome (§PLS tests)."""
107 import json as _json
108 import subprocess as _sp
109
110 def _runner(cmd: list[str]) -> "_sp.CompletedProcess[str]":
111 if cmd[:3] == ["gh", "pr", "view"]:
112 return _sp.CompletedProcess(
113 cmd, 0, stdout=_json.dumps({"number": 1, "state": pr_state}), stderr=""
114 )
115 if cmd[:3] == ["gh", "pr", "checks"]:
116 rc = 0 if check_state == "pass" else 1
117 return _sp.CompletedProcess(
118 cmd, rc, stdout=f"verify\t{check_state}\t0s\thttps://example.test/v\n", stderr=""
119 )
120 if cmd[:3] == ["gh", "pr", "merge"]:
121 return _sp.CompletedProcess(cmd, 0, stdout="", stderr="")
122 raise AssertionError(f"unexpected gh argv: {cmd}")
123
124 return _runner
125
126
127 class FakeGitRunner:
128 """Recording git fake for §PLS post-land sync argv assertions.
129
130 ``fail`` is a set of git subcommands ({"fetch", "status", "rev-parse",
131 "checkout", "pull"}) that return exit 1.
132 """
133
134 def __init__(
135 self,
136 *,
137 porcelain: str = "",
138 branch: str = "main",
139 fail: set[str] | tuple[str, ...] = (),
140 ) -> None:
141 self.porcelain = porcelain
142 self.branch = branch
143 self.fail = set(fail)
144 self.calls: list[list[str]] = []
145
146 def __call__(self, cmd: list[str]):
147 import subprocess as _sp
148
149 self.calls.append(list(cmd))
150 op = cmd[1] if len(cmd) > 1 else ""
151 rc = 1 if op in self.fail else 0
152 stdout = ""
153 if rc == 0:
154 if op == "status":
155 stdout = self.porcelain
156 elif op == "rev-parse":
157 stdout = self.branch + "\n"
158 elif op == "checkout":
159 self.branch = cmd[2]
160 return _sp.CompletedProcess(
161 list(cmd), rc, stdout=stdout, stderr="induced failure" if rc else ""
162 )
163
164
165 def ok(stdout: str = "") -> CommandResult:
166 return CommandResult(stdout=stdout, stderr="", exit_code=0)
167
168
169 def fail(stderr: str = "error", code: int = 1) -> CommandResult:
170 return CommandResult(stdout="", stderr=stderr, exit_code=code)
171
172
173 def make_runner(responses: dict[str, CommandResult]) -> RecordingRunner:
174 return RecordingRunner(responses=responses, calls=[])
175
176
177 class BranchStateRunner:
178 """Stateful fake VCS host for §GSW write-path and §GSB reconcile tests.
179
180 Tracks the current branch of both histories, refuses a bare Muse checkout
181 of an existing branch while the Muse tree is dirty (Muse 0.2.x live
182 behavior — the GSW incident), honors ``--autoshelf`` dirty-carry, and
183 records every command for order assertions. ``--force`` always fails so
184 any forbidden use surfaces in tests (§GSW.8).
185
186 §GSB additions — per-branch tips (``git_tips`` / ``muse_tips``, Git SHAs
187 and Muse ``sha256:`` ids in separate spaces), ancestry maps
188 (``git_ancestors`` / ``muse_ancestors``: descendant → set of ancestors),
189 and a shared-worktree content token. ``content_map`` maps a tip id to a
190 content token (default: everything is ``content:base``); a Muse checkout
191 of a branch rewrites ``worktree`` to that branch tip's content token —
192 modeling the live defect where a stale Muse checkout dirties the shared
193 tree so a Git checkout of the same name refuses.
194 """
195
196 def __init__(
197 self,
198 root: str,
199 *,
200 git_branch: str = "main",
201 muse_branch: str = "main",
202 git_dirty: bool = False,
203 muse_dirty: bool = False,
204 origin_main_tip: str = "cafebabe",
205 muse_main_tip: str = "sha256:musetip",
206 merged_prs_json: str = "[]",
207 muse_rev_parse_main_values: list[str] | None = None,
208 git_commit_fails: bool = False,
209 muse_commit_fails: bool = False,
210 existing_git_branches: set[str] | None = None,
211 existing_muse_branches: set[str] | None = None,
212 git_tips: dict[str, str] | None = None,
213 muse_tips: dict[str, str] | None = None,
214 git_ancestors: dict[str, set[str]] | None = None,
215 muse_ancestors: dict[str, set[str]] | None = None,
216 content_map: dict[str, str] | None = None,
217 worktree: str | None = None,
218 ) -> None:
219 self.root = root
220 self.git_branch = git_branch
221 self.muse_branch = muse_branch
222 self.git_dirty = git_dirty
223 self.muse_dirty = muse_dirty
224 self.origin_main_tip = origin_main_tip
225 self.muse_main_tip = muse_main_tip
226 self.merged_prs_json = merged_prs_json
227 self.muse_rev_parse_main_values = list(muse_rev_parse_main_values or [])
228 self.git_commit_fails = git_commit_fails
229 self.muse_commit_fails = muse_commit_fails
230 self.git_branches = {git_branch} | (existing_git_branches or set())
231 self.muse_branches = {muse_branch} | (existing_muse_branches or set())
232 self.git_tips: dict[str, str] = dict(git_tips or {})
233 for name in self.git_branches:
234 self.git_tips.setdefault(name, "feedface")
235 self.muse_tips: dict[str, str] = dict(muse_tips or {})
236 for name in self.muse_branches:
237 self.muse_tips.setdefault(name, muse_main_tip)
238 self.git_ancestors: dict[str, set[str]] = {
239 sha: set(parents) for sha, parents in (git_ancestors or {}).items()
240 }
241 self.muse_ancestors: dict[str, set[str]] = {
242 sha: set(parents) for sha, parents in (muse_ancestors or {}).items()
243 }
244 self.content_map: dict[str, str] = dict(content_map or {})
245 self.worktree = (
246 worktree
247 if worktree is not None
248 else self._content(self.git_tips[self.git_branch])
249 )
250 self.git_commit_count = 0
251 self.muse_commit_count = 0
252 self.calls: list[tuple[str, str | None]] = []
253
254 def _content(self, tip: str | None) -> str:
255 return self.content_map.get(tip or "", "content:base")
256
257 def _git_worktree_dirty(self) -> bool:
258 return self.git_dirty or self.worktree != self._content(
259 self.git_tips.get(self.git_branch)
260 )
261
262 def _muse_worktree_dirty(self) -> bool:
263 return self.muse_dirty or self.worktree != self._content(
264 self.muse_tips.get(self.muse_branch)
265 )
266
267 def _git_known_commits(self) -> set[str]:
268 known = set(self.git_tips.values()) | set(self.git_ancestors)
269 for parents in self.git_ancestors.values():
270 known |= parents
271 return known
272
273 def run(self, command: str, *, cwd: str | None = None) -> CommandResult:
274 import shlex
275
276 self.calls.append((command, cwd))
277 tokens = shlex.split(command)
278 if tokens[0] == "git":
279 return self._git(tokens[1:])
280 if tokens[0] == "muse":
281 # muse -C <root> <args...>
282 return self._muse(tokens[3:])
283 if tokens[0] == "gh":
284 return CommandResult(stdout=self.merged_prs_json, stderr="", exit_code=0)
285 return CommandResult(stdout="", stderr="unmocked command", exit_code=127)
286
287 def _git(self, args: list[str]) -> CommandResult:
288 if args[:3] == ["rev-parse", "--abbrev-ref", "HEAD"]:
289 return _ok_result(self.git_branch)
290 if args == ["rev-parse", "origin/main"]:
291 return _ok_result(self.origin_main_tip)
292 if args == ["rev-parse", "HEAD"]:
293 return _ok_result(self.git_tips.get(self.git_branch, "feedface"))
294 if args[:2] == ["rev-parse", "--verify"] and len(args) == 3:
295 name = args[2].removeprefix("refs/heads/")
296 if name in self.git_branches:
297 return _ok_result(self.git_tips[name])
298 return _fail_result(f"unknown ref {name!r}")
299 if args[0] == "rev-parse" and len(args) == 2:
300 name = args[1]
301 if name in self.git_branches:
302 return _ok_result(self.git_tips[name])
303 return _fail_result(f"unknown ref {name!r}")
304 if args[:2] == ["status", "--porcelain"]:
305 return _ok_result(" M tracked.md" if self._git_worktree_dirty() else "")
306 if args[:2] == ["checkout", "-b"]:
307 branch = args[2]
308 if branch in self.git_branches:
309 return _fail_result(f"branch {branch!r} already exists")
310 self.git_tips[branch] = self.git_tips.get(self.git_branch, "feedface")
311 self.git_branches.add(branch)
312 self.git_branch = branch
313 return _ok_result("")
314 if args[0] == "checkout":
315 if "--force" in args:
316 return _fail_result("--force forbidden in GSW tests")
317 branch = args[-1]
318 if branch not in self.git_branches:
319 return _fail_result(f"unknown branch {branch!r}")
320 target_content = self._content(self.git_tips.get(branch))
321 current_content = self._content(self.git_tips.get(self.git_branch))
322 if (
323 not self.git_dirty
324 and self.worktree != current_content
325 and self.worktree != target_content
326 ):
327 # Live git two-tree rule: foreign worktree bytes that match
328 # neither HEAD nor the target refuse the switch (the GSB
329 # incident after a stale Muse checkout).
330 return _fail_result(
331 "error: Your local changes to the following files would be "
332 "overwritten by checkout"
333 )
334 self.git_branch = branch # git carries dirty changes across checkout
335 self.worktree = target_content
336 return _ok_result("")
337 if args[:2] == ["branch", "-f"] and len(args) == 4:
338 name, tip = args[2], args[3]
339 if name == self.git_branch:
340 return _fail_result(
341 f"cannot force update the currently checked out branch {name!r}"
342 )
343 self.git_branches.add(name)
344 self.git_tips[name] = tip
345 return _ok_result("")
346 if args[0] == "update-ref" and len(args) == 3:
347 name = args[1].removeprefix("refs/heads/")
348 self.git_branches.add(name)
349 self.git_tips[name] = args[2]
350 return _ok_result("")
351 if args[:2] == ["reset", "--hard"] and len(args) == 3:
352 tip = args[2]
353 self.git_tips[self.git_branch] = tip
354 self.worktree = self._content(tip)
355 self.git_dirty = False
356 return _ok_result("")
357 if args[0] == "add":
358 return _ok_result("")
359 if args[0] == "commit":
360 if self.git_commit_fails:
361 return _fail_result("induced git commit failure")
362 self.git_commit_count += 1
363 parent = self.git_tips.get(self.git_branch, "feedface")
364 new_sha = f"gitc{self.git_commit_count:04d}"
365 self.git_ancestors[new_sha] = {parent} | self.git_ancestors.get(parent, set())
366 self.git_tips[self.git_branch] = new_sha
367 self.content_map[new_sha] = self.worktree
368 self.git_dirty = False
369 return _ok_result("")
370 if args[0] == "push":
371 return _ok_result("")
372 if args[:2] == ["remote", "get-url"]:
373 return _ok_result("[email protected]:owner/repo.git")
374 if args[:2] == ["merge-base", "--is-ancestor"] and len(args) == 4:
375 ancestor, descendant = args[2], args[3]
376 if ancestor == descendant or ancestor in self.git_ancestors.get(
377 descendant, set()
378 ):
379 return _ok_result("")
380 known = self._git_known_commits()
381 if ancestor in known and descendant in known:
382 return CommandResult(stdout="", stderr="", exit_code=1)
383 # Legacy permissive default for ids outside the modeled graph
384 # (e.g. realign superset probes against bridge anchors).
385 return _ok_result("")
386 if args[0] == "merge-base":
387 return _ok_result("")
388 if args[0] == "rev-list":
389 return _ok_result("0")
390 return CommandResult(stdout="", stderr="unmocked git command", exit_code=127)
391
392 def _muse(self, args: list[str]) -> CommandResult:
393 if args[:3] == ["rev-parse", "--abbrev-ref", "HEAD"]:
394 return _ok_result(self.muse_branch)
395 if args == ["rev-parse", "main"]:
396 if self.muse_rev_parse_main_values:
397 return _ok_result(self.muse_rev_parse_main_values.pop(0))
398 return _ok_result(self.muse_tips.get("main", self.muse_main_tip))
399 if args == ["rev-parse", "HEAD"]:
400 return _ok_result(self.muse_tips.get(self.muse_branch, self.muse_main_tip))
401 if args[0] == "rev-parse" and len(args) == 2:
402 name = args[1]
403 if name in self.muse_branches:
404 return _ok_result(self.muse_tips[name])
405 return _fail_result(f"'{name}' not found")
406 if args[:2] == ["status", "--json"]:
407 dirty = self._muse_worktree_dirty()
408 return _ok_result(
409 json.dumps({"dirty": dirty, "total_changes": 1 if dirty else 0})
410 )
411 if args[:2] == ["status", "--porcelain"]:
412 return _ok_result(" M tracked.md" if self._muse_worktree_dirty() else "")
413 if args[:2] == ["branch", "--show-current"]:
414 return _ok_result(self.muse_branch)
415 if args[:2] == ["checkout", "-b"]:
416 branch = args[2]
417 if branch in self.muse_branches:
418 return _fail_result(f"branch {branch!r} already exists")
419 self.muse_tips[branch] = self.muse_tips.get(
420 self.muse_branch, self.muse_main_tip
421 )
422 self.muse_branches.add(branch)
423 self.muse_branch = branch
424 return _ok_result("")
425 if args[:2] == ["checkout", "--autoshelf"]:
426 branch = args[2]
427 if branch not in self.muse_branches:
428 return _fail_result(f"unknown branch {branch!r}")
429 self.muse_branch = branch # dirty changes shelved + reapplied
430 self.worktree = self._content(self.muse_tips.get(branch))
431 return _ok_result("")
432 if args[0] == "checkout":
433 if "--force" in args:
434 return _fail_result("--force forbidden in GSW tests")
435 branch = args[-1]
436 if branch not in self.muse_branches:
437 return _fail_result(f"unknown branch {branch!r}")
438 if self._muse_worktree_dirty():
439 # Muse 0.2.x live behavior: refuse dirty tracked checkout.
440 return _fail_result("dirty tracked files present; use --autoshelf or --merge")
441 self.muse_branch = branch
442 # Muse rewrites the shared worktree to the branch tip's content —
443 # the live §GSB defect when that tip is stale.
444 self.worktree = self._content(self.muse_tips.get(branch))
445 return _ok_result("")
446 if args[0] == "update-ref" and len(args) == 3:
447 name = args[1]
448 self.muse_branches.add(name)
449 self.muse_tips[name] = args[2]
450 return _ok_result("")
451 if args[0] == "reset" and "--hard" in args:
452 if "--force" in args:
453 return _fail_result("--force forbidden in GSW tests")
454 target = next(a for a in args[1:] if not a.startswith("-"))
455 if self._muse_worktree_dirty():
456 # Live Muse 0.2.x refuses reset --hard on tracked changes.
457 return _fail_result(
458 "error: Your local changes would be overwritten by reset --hard"
459 )
460 self.muse_tips[self.muse_branch] = target
461 self.worktree = self._content(target)
462 return _ok_result("")
463 if args[0] == "merge-base":
464 operands = [a for a in args[1:] if not a.startswith("-")]
465 if len(operands) != 2:
466 return _fail_result("merge-base needs two commits")
467 commit_a, commit_b = operands
468 if commit_a == commit_b or commit_a in self.muse_ancestors.get(
469 commit_b, set()
470 ):
471 base: str | None = commit_a
472 elif commit_b in self.muse_ancestors.get(commit_a, set()):
473 base = commit_b
474 else:
475 base = None
476 return _ok_result(
477 json.dumps(
478 {
479 "commit_a": commit_a,
480 "commit_b": commit_b,
481 "merge_base": base,
482 "exit_code": 0,
483 }
484 )
485 )
486 if args[:2] == ["code", "add"]:
487 return _ok_result("")
488 if args[0] == "add":
489 # Muse 0.2.x live behavior: no top-level `add` subcommand.
490 return CommandResult(
491 stdout="",
492 stderr="muse: error: argument COMMAND: invalid choice: 'add'",
493 exit_code=2,
494 )
495 if args[0] == "commit":
496 if self.muse_commit_fails:
497 return _fail_result("induced muse commit failure")
498 self.muse_commit_count += 1
499 parent = self.muse_tips.get(self.muse_branch, self.muse_main_tip)
500 new_sha = f"sha256:musec{self.muse_commit_count:04d}"
501 self.muse_ancestors[new_sha] = {parent} | self.muse_ancestors.get(
502 parent, set()
503 )
504 self.muse_tips[self.muse_branch] = new_sha
505 self.content_map[new_sha] = self.worktree
506 self.muse_dirty = False
507 return _ok_result("")
508 if args[0] == "bridge":
509 return _ok_result("")
510 return CommandResult(stdout="", stderr="unmocked muse command", exit_code=127)
511
512
513 def _ok_result(stdout: str) -> CommandResult:
514 return CommandResult(stdout=stdout, stderr="", exit_code=0)
515
516
517 def _fail_result(stderr: str) -> CommandResult:
518 return CommandResult(stdout="", stderr=stderr, exit_code=1)
519
520
521 GSW_CONFIG_BY_REGIME = {
522 "git-only": "config-git-only.yaml",
523 "muse-only": "config-muse-only.yaml",
524 "muse+git-mirror": "config-muse-git-mirror.yaml",
525 }
526
527 GSW_DOC_NAMES = {
528 "git-only": ("OVERSEER-HANDOVER.md", "ROADMAP.md"),
529 "muse-only": ("MUSEHUB-OVERSEER-HANDOVER.md", "MUSEHUB-ROADMAP.md"),
530 "muse+git-mirror": ("OVERSEER-HANDOVER.md", "ROADMAP.md"),
531 }
532
533 _GSW_MUSE_ONLY_HANDOVER = "# Handover — muse-only\n\n## Change log\n\n- **2026-07-01** — initial\n"
534 _GSW_MUSE_ONLY_ROADMAP = (
535 "# Roadmap — muse-only\n\n## Build queue\n\n"
536 "| Phase | Model | Status | Deliverable |\n| --- | --- | --- | --- |\n"
537 "| **X1** | Auto | **TODO** | thing |\n"
538 )
539
540
541 def seed_gsw_repo(
542 repo_root: Path,
543 regime: str,
544 *,
545 handover_text: str | None = None,
546 roadmap_text: str | None = None,
547 muse_main_tip: str = "sha256:musetip",
548 ) -> tuple[Path, Path]:
549 """Seed config, docs, and Muse substrate/bridge for §GSW write-path tests.
550
551 Returns ``(handover_path, roadmap_path)``. Default docs produce D1 drift
552 for git regimes (claim ``deadbeef`` vs actual ``cafebabe``); muse-only
553 docs are minimal (its drift is driven by sequenced R2/R3 reads).
554 """
555 write_config(repo_root, GSW_CONFIG_BY_REGIME[regime])
556 if regime != "git-only":
557 seed_muse_substrate(repo_root)
558 if regime == "muse+git-mirror":
559 (repo_root / ".muse" / "git-bridge.toml").write_text(
560 f'[last_export]\nmuse_commit_id = "{muse_main_tip}"\ngit_sha = "{"1" * 40}"\n',
561 encoding="utf-8",
562 )
563 docs = repo_root / "docs"
564 docs.mkdir(parents=True, exist_ok=True)
565 handover_name, roadmap_name = GSW_DOC_NAMES[regime]
566 if handover_text is None:
567 if regime == "muse-only":
568 handover_text = _GSW_MUSE_ONLY_HANDOVER
569 else:
570 handover_text = (FIXTURES / "governance-handover-drift.md").read_text(encoding="utf-8")
571 if roadmap_text is None:
572 if regime == "muse-only":
573 roadmap_text = _GSW_MUSE_ONLY_ROADMAP
574 else:
575 roadmap_text = (FIXTURES / "governance-roadmap-drift.md").read_text(encoding="utf-8")
576 handover_path = docs / handover_name
577 roadmap_path = docs / roadmap_name
578 handover_path.write_text(handover_text, encoding="utf-8")
579 roadmap_path.write_text(roadmap_text, encoding="utf-8")
580 return handover_path, roadmap_path
581
582
583 def gsw_runner(repo_root: Path, regime: str, **kwargs) -> BranchStateRunner:
584 """``BranchStateRunner`` pre-configured per regime for §GSW tests.
585
586 muse-only drives its only possible drift dimension (D2) via sequenced
587 ``muse rev-parse main`` values (R2 anchor read, then R3 canonical read).
588 """
589 if regime == "muse-only" and "muse_rev_parse_main_values" not in kwargs:
590 kwargs["muse_rev_parse_main_values"] = ["sha256:anchor", "sha256:moved"]
591 return BranchStateRunner(str(repo_root.resolve()), **kwargs)
592
593
594 def adapter_for(config: OverseerConfig, repo_root: Path, runner: RecordingRunner):
595 return create_adapter(config, repo_root, runner=runner)
596
597
598 def git_status_runner(
599 branch: str = "main",
600 dirty: bool = False,
601 tip: str = "cafebabe",
602 ) -> RecordingRunner:
603 """Recording runner with git-only ``status()`` + origin/main tip responses."""
604 dirty_out = " M file" if dirty else ""
605 return make_runner(
606 {
607 "git rev-parse --abbrev-ref HEAD": ok(branch),
608 "git status --porcelain": ok(dirty_out),
609 "git rev-parse origin/main": ok(tip),
610 }
611 )
612
613
614 def muse_status_runner(
615 repo_root: Path,
616 branch: str = "main",
617 dirty: bool = False,
618 tip: str = "cafebabe",
619 ) -> RecordingRunner:
620 """Recording runner with muse-only ``status()`` responses."""
621 root = str(repo_root.resolve())
622 dirty_out = " M file" if dirty else ""
623 return make_runner(
624 {
625 f"muse -C {root} branch --show-current": ok(branch),
626 f"muse -C {root} status --porcelain": ok(dirty_out),
627 f"muse -C {root} rev-parse {branch}": ok(tip),
628 f"muse -C {root} rev-parse main": ok(tip),
629 }
630 )
631
632
633 def muse_mirror_status_runner(
634 repo_root: Path,
635 branch: str = "main",
636 dirty: bool = False,
637 tip: str = "cafebabe",
638 ) -> RecordingRunner:
639 """Recording runner with muse+git-mirror ``status()`` responses."""
640 root = str(repo_root.resolve())
641 dirty_out = " M file" if dirty else ""
642 return make_runner(
643 {
644 f"muse -C {root} rev-parse --abbrev-ref HEAD": ok(branch),
645 f"muse -C {root} status --porcelain": ok(dirty_out),
646 f"muse -C {root} status --json": ok(
647 '{"dirty": true}' if dirty else '{"dirty": false}'
648 ),
649 "git rev-parse --abbrev-ref HEAD": ok(branch),
650 "git status --porcelain": ok(dirty_out),
651 "git rev-parse origin/main": ok(tip),
652 f"muse -C {root} rev-parse main": ok(tip),
653 f"muse -C {root} rev-parse {branch}": ok(tip),
654 }
655 )
656
657
658 def seed_governance_freshness(
659 repo_root: Path,
660 tip: str = "cafebabe",
661 *,
662 handover_path: Path | None = None,
663 ) -> None:
664 """Align default-lane handover GitHub-main claim + write enriched marker (§GFG).
665
666 Call after ``ok init`` so ``status --exit-code`` is not fail-closed solely on freshness.
667 """
668 from datetime import datetime, timezone
669
670 from adapters.config import load_config, resolve_lane_docs
671 from cli.docs_paths import lane_living_doc_abs
672 from cli.paths import resolve_config_path
673
674 config_path = resolve_config_path(repo_root, None)
675 config = load_config(config_path)
676 lane = config.docs.default_lane if config.docs.lanes is not None else None
677 lane_docs = resolve_lane_docs(config, lane)
678 path = handover_path or lane_living_doc_abs(
679 repo_root, config, lane_docs, lane_docs.handover
680 )
681 if path.is_file():
682 text = path.read_text(encoding="utf-8")
683 claim = f"| GitHub `main` | `{tip}` |"
684 if f"`{tip}`" not in text or "GitHub `main`" not in text:
685 text = text.rstrip() + f"\n\n| Item | Value |\n| --- | --- |\n{claim}\n"
686 path.write_text(text, encoding="utf-8")
687
688 stamp = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
689 marker = repo_root / ".overseer" / "last_governance_sync"
690 marker.parent.mkdir(parents=True, exist_ok=True)
691 r1 = "" if config.vcs.regime == "muse-only" else tip
692 r3 = tip
693 marker.write_text(f"{stamp}\nr1={r1}\nr3={r3}\n", encoding="utf-8")
694
695 if config.vcs.regime == "muse+git-mirror":
696 muse_dir = repo_root / ".muse"
697 muse_dir.mkdir(parents=True, exist_ok=True)
698 bridge = muse_dir / "git-bridge.toml"
699 # muse_commit_id = tip (R2 Muse space); git_sha uses a distinct 40-hex so
700 # fixtures prove git_sha≠tip alone is not D2 drift (§D2F.3).
701 git_export = "1" * 40 if tip.startswith("sha256:") else tip
702 bridge.write_text(
703 f'[last_export]\n'
704 f'muse_commit_id = "{tip}"\n'
705 f'git_sha = "{git_export}"\n',
706 encoding="utf-8",
707 )
708
709
710 LAND_A_MARKER = (
711 "<!-- overseer:next role=primary lane=product status=live land-phase=land-a -->"
712 )
713 LAND_B_MARKER = (
714 "<!-- overseer:next role=primary lane=product status=live land-phase=land-b -->"
715 )
716
717
718 def land_a_fence_body(slice_id: str = "PMHF", *, paste_extra: str = "") -> str:
719 """Frozen land-a paste body (§PMHF.3.1) for fixtures."""
720 return (
721 "Model: Operator + Auto\n"
722 f"ID: {slice_id} → main (land-a)\n"
723 "land-phase: land-a\n"
724 "\n"
725 "Deliver:\n"
726 "1. Open/update PR (or SD-21 authorized land path)\n"
727 "2. Stop for Tier 3 merge authorization when required\n"
728 "3. Do NOT claim land complete\n"
729 "4. Do NOT regenerate post-merge NEXT in this paste\n"
730 f"{paste_extra}"
731 "\n"
732 "After merge is confirmed on main: paste land-b (same slice). "
733 "Land is incomplete until land-b.\n"
734 )
735
736
737 def land_b_fence_body(slice_id: str = "PMHF") -> str:
738 """Frozen land-b paste body (§PMHF.3.2) for fixtures."""
739 return (
740 "Model: Auto\n"
741 f"ID: {slice_id} land-b (post-merge sync)\n"
742 "land-phase: land-b\n"
743 "\n"
744 "Deliver:\n"
745 "1. Fetch/pull latest main (regime-appropriate)\n"
746 "2. ok governance-sync --dry-run then apply when the plan is correct\n"
747 "3. Regenerate NEXT + paste so they no longer say wait-for-merge / land-a\n"
748 "4. Feature-branch commit bundling ROADMAP + HANDOVER (SD-17)\n"
749 "5. ok status --exit-code → 0 and ok land-closeout → 0 before claiming land complete\n"
750 )
751
752
753 def land_handover_text(
754 claim: str = "cafebabe",
755 *,
756 marker: str | None = LAND_A_MARKER,
757 slice_id: str = "PMHF",
758 heading: str | None = None,
759 fence_body: str | None = None,
760 ) -> str:
761 """Handover fixture with NEXT marker, paste fence, and GitHub-main claim (§PMHF)."""
762 if fence_body is None:
763 fence_body = land_a_fence_body(slice_id)
764 if heading is None:
765 heading = f"{slice_id} → main (land-a)"
766 marker_line = f"{marker}\n" if marker else ""
767 return (
768 "# Overseer Handover — fixture\n"
769 "\n"
770 f"{marker_line}## NEXT SESSION — {heading}\n"
771 "\n"
772 "**Model:** Operator + Auto\n"
773 "\n"
774 "### What just landed\n"
775 "\n"
776 "| Slice | Deliverable |\n"
777 "| --- | --- |\n"
778 "| prior | prior slice |\n"
779 "\n"
780 "### THE ONE NEXT STEP — **Model: Operator + Auto**\n"
781 "\n"
782 "| | |\n"
783 "| --- | --- |\n"
784 f"| **ID** | **{heading}** |\n"
785 "\n"
786 f"### Paste-ready prompt — {slice_id}\n"
787 "\n"
788 "```text\n"
789 f"{fence_body}"
790 "```\n"
791 "\n"
792 "---\n"
793 "\n"
794 "## Verified snapshot\n"
795 "\n"
796 "| Area | State |\n"
797 "| --- | --- |\n"
798 "| **VCS regime** | `git-only` |\n"
799 "\n"
800 "<!-- overseer:anchor:vcs-table -->\n"
801 "## VCS (verified 2026-07-30)\n"
802 "\n"
803 "| Item | Value |\n"
804 "| --- | --- |\n"
805 "| Branch | `main` |\n"
806 f"| GitHub `main` | `{claim}` |\n"
807 "<!-- /overseer:anchor:vcs-table -->\n"
808 "\n"
809 "## Change log\n"
810 "\n"
811 "- **2026-07-30** — land-a opened\n"
812 )
813
814
815 def land_roadmap_text(*rows: str) -> str:
816 """Roadmap fixture with a build queue; pass full ``| … |`` rows."""
817 default_rows = (
818 "| **PMHF-b Build** | Auto | **DONE** | land closeout build |",
819 "| **PMHF → main** | Operator + Auto | **TODO** | Land PMHF (Tier 3 merge) |",
820 )
821 body = "\n".join(rows or default_rows)
822 return (
823 "# Roadmap — fixture\n"
824 "\n"
825 "## Build queue\n"
826 "\n"
827 "| Phase | Model | Status | Deliverable |\n"
828 "| --- | --- | --- | --- |\n"
829 f"{body}\n"
830 "\n"
831 "## Definition of Done (every phase)\n"
832 "\n"
833 "- Tests green\n"
834 )
835
836
837 def seed_land_repo(
838 repo_root: Path,
839 *,
840 claim: str = "cafebabe",
841 handover_text: str | None = None,
842 roadmap_text: str | None = None,
843 marker_tip: str | None = "cafebabe",
844 ) -> None:
845 """Seed config, lock, docs, and enriched sync marker for land-closeout tests."""
846 write_config(repo_root, "config-git-only.yaml")
847 (repo_root / ".overseer" / "version.lock").write_text(
848 "lock_version: 1\nkit_version: 0.1.0\nconfig_version: 1\n"
849 "footprint_digest: sha256:" + ("0" * 64) + "\n"
850 'installed_at: "2026-01-01T00:00:00Z"\nsynced_at: "2026-01-01T00:00:00Z"\n'
851 "footprint: []\n",
852 encoding="utf-8",
853 )
854 docs = repo_root / "docs"
855 docs.mkdir(parents=True, exist_ok=True)
856 (docs / "OVERSEER-HANDOVER.md").write_text(
857 handover_text if handover_text is not None else land_handover_text(claim),
858 encoding="utf-8",
859 )
860 (docs / "ROADMAP.md").write_text(
861 roadmap_text if roadmap_text is not None else land_roadmap_text(),
862 encoding="utf-8",
863 )
864 if marker_tip is not None:
865 (repo_root / ".overseer" / "last_governance_sync").write_text(
866 f"2026-07-30T00:00:00Z\nr1={marker_tip}\nr3={marker_tip}\n",
867 encoding="utf-8",
868 )
869
870
871 def run_cli(
872 argv: list[str],
873 *,
874 cwd: Path,
875 runner: RecordingRunner | None = None,
876 kit: Path | None = None,
877 review_provider_factory=None,
878 script_executor=None,
879 json_mode: bool = False,
880 ) -> int:
881 """Invoke ``cli.main`` with an injected runner and working directory."""
882 from cli.context import CliContext
883 from cli.main import main
884 from cli.output import OutputContext
885
886 old_cwd = Path.cwd()
887 os.chdir(cwd)
888 try:
889 ctx = CliContext.create(
890 runner=runner or make_runner({}),
891 cwd=cwd,
892 kit=kit,
893 output=OutputContext(json_mode=json_mode),
894 review_provider_factory=review_provider_factory,
895 script_executor=script_executor,
896 )
897 return main(argv, ctx=ctx)
898 finally:
899 os.chdir(old_cwd)
900
901
902 def seed_muse_substrate(repo_root: Path) -> None:
903 """Create minimal healthy ``.muse/`` for muse-backed regime tests."""
904 muse = repo_root / ".muse"
905 muse.mkdir(parents=True, exist_ok=True)
906 (muse / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
907 (muse / "repo.json").write_text("{}", encoding="utf-8")
908 (muse / "config.toml").write_text("", encoding="utf-8")
909
910
911 def seed_freeze_repo(repo_root: Path, *, config_name: str = "config-git-only.yaml") -> Path:
912 """Write config and copy a freeze artifact fixture into a temp repo."""
913 write_config(repo_root, config_name)
914 if "muse" in config_name:
915 seed_muse_substrate(repo_root)
916 docs = repo_root / "docs"
917 docs.mkdir(parents=True, exist_ok=True)
918 artifact = docs / "FREEZE.md"
919 artifact.write_text((FIXTURES / "freeze-artifact.md").read_text(encoding="utf-8"), encoding="utf-8")
920 return artifact
921
922
923 def pass_provider_factory():
924 """Factory returning a provider that always passes."""
925 from tools.freeze_reviewer.providers.base import LocalReviewProvider
926
927 def _factory(_provider_name: str) -> LocalReviewProvider:
928 return LocalReviewProvider(scripted_findings=[])
929
930 return _factory
931
932
933 def findings_provider_factory(findings):
934 """Factory returning a provider with scripted findings."""
935 from tools.freeze_reviewer.providers.base import LocalReviewProvider
936
937 def _factory(_provider_name: str) -> LocalReviewProvider:
938 return LocalReviewProvider(scripted_findings=list(findings))
939
940 return _factory
941
942
943 def unreachable_provider_factory(cause: str = "offline"):
944 """Factory returning an unreachable local provider."""
945 from tools.freeze_reviewer.providers.base import LocalReviewProvider
946
947 def _factory(_provider_name: str) -> LocalReviewProvider:
948 return LocalReviewProvider(force_unreachable=True, unreachable_cause=cause)
949
950 return _factory
951
952
953 class FakeHttpTransport:
954 """Recording HTTP transport for API provider tests (no network)."""
955
956 def __init__(
957 self,
958 *,
959 health_status: int = 200,
960 health_body: bytes = b'{"status":"ok"}',
961 review_status: int = 200,
962 review_body: bytes | None = None,
963 fail_health: bool = False,
964 fail_review: bool = False,
965 ) -> None:
966 self.health_status = health_status
967 self.health_body = health_body
968 self.review_status = review_status
969 self.review_body = review_body or b'{"findings":[]}'
970 self.fail_health = fail_health
971 self.fail_review = fail_review
972 self.calls: list[dict] = []
973
974 def request(
975 self,
976 *,
977 method: str,
978 url: str,
979 headers: dict[str, str],
980 body: bytes | None = None,
981 timeout: float = 30.0,
982 ) -> tuple[int, bytes]:
983 from tools.freeze_reviewer.providers.api_client import ProviderTransportError
984
985 self.calls.append(
986 {
987 "method": method,
988 "url": url,
989 "headers": dict(headers),
990 "body": body,
991 }
992 )
993 if "/health" in url:
994 if self.fail_health:
995 raise ProviderTransportError("health transport failure")
996 return self.health_status, self.health_body
997 if self.fail_review:
998 raise ProviderTransportError("review transport failure")
999 return self.review_status, self.review_body
1000
1001
1002 def api_provider_factory(transport: FakeHttpTransport):
1003 """Factory returning an API provider wired to a fake HTTP transport."""
1004 from tools.freeze_reviewer.providers.api_client import ReviewApiClient
1005 from tools.freeze_reviewer.providers.base import ApiReviewProvider
1006
1007 def _factory(_provider_name: str) -> ApiReviewProvider:
1008 client = ReviewApiClient(transport=transport)
1009 return ApiReviewProvider(client=client)
1010
1011 return _factory
1012
1013
1014 def api_unreachable_provider_factory(cause: str = "API provider unavailable"):
1015 """Factory returning a forced-unreachable API provider."""
1016 from tools.freeze_reviewer.providers.base import ApiReviewProvider
1017
1018 def _factory(_provider_name: str) -> ApiReviewProvider:
1019 return ApiReviewProvider(force_unreachable=True, unreachable_cause=cause)
1020
1021 return _factory
1022
1023
1024 def seed_checkpoint_repo(repo_root: Path) -> None:
1025 """Copy checkpoint fixture pack into a temp repo and mark scripts executable."""
1026 import shutil
1027 import stat
1028
1029 (repo_root / ".overseer").mkdir(parents=True, exist_ok=True)
1030 config_src = CHECKPOINTS / "config-checkpoints-enabled.yaml"
1031 (repo_root / ".overseer" / "config.yaml").write_text(
1032 config_src.read_text(encoding="utf-8"),
1033 encoding="utf-8",
1034 )
1035 for rel in ("policy", "manifests", "scripts"):
1036 src = CHECKPOINTS / rel
1037 dest = repo_root / rel
1038 if dest.exists():
1039 shutil.rmtree(dest)
1040 shutil.copytree(src, dest)
1041 for script in (repo_root / "scripts" / "verify").glob("*.py"):
1042 script.chmod(script.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
1043
1044
1045 def load_checkpoint_config(repo_root: Path) -> OverseerConfig:
1046 """Load checkpoint-enabled config from a seeded repo."""
1047 seed_checkpoint_repo(repo_root)
1048 return load_config(repo_root / ".overseer" / "config.yaml")
1049
1050
1051 def seed_honesty_repo(repo_root: Path) -> None:
1052 """Copy honesty fixture pack into a temp repo."""
1053 import shutil
1054
1055 (repo_root / ".overseer").mkdir(parents=True, exist_ok=True)
1056 config_src = HONESTY / "config-honesty-enabled.yaml"
1057 (repo_root / ".overseer" / "config.yaml").write_text(
1058 config_src.read_text(encoding="utf-8"),
1059 encoding="utf-8",
1060 )
1061 for rel in ("artifacts", "entries"):
1062 src = HONESTY / rel
1063 dest = repo_root / rel
1064 if dest.exists():
1065 shutil.rmtree(dest)
1066 shutil.copytree(src, dest)
1067
1068
1069 def load_honesty_config(repo_root: Path) -> OverseerConfig:
1070 """Load honesty-enabled config from a seeded repo."""
1071 seed_honesty_repo(repo_root)
1072 return load_config(repo_root / ".overseer" / "config.yaml")
1073
1074
1075 def honesty_artifact_hash(repo_root: Path) -> str:
1076 """SHA-256 of the fixture sample artifact."""
1077 from tools.honesty.artifact import sha256_file_bytes
1078
1079 return sha256_file_bytes(repo_root / "artifacts" / "sample.txt")
1080
1081
1082 def load_honesty_entry(repo_root: Path, name: str, *, artifact_hash: str | None = None) -> dict:
1083 """Load a verdict entry fixture with artifact hash substituted."""
1084 import json
1085
1086 text = (HONESTY / "entries" / name).read_text(encoding="utf-8")
1087 if artifact_hash is None:
1088 artifact_hash = honesty_artifact_hash(repo_root)
1089 text = text.replace("PLACEHOLDER", artifact_hash)
1090 return json.loads(text)
1091
1092
1093 def seed_pilot_tree(
1094 repo_root: Path,
1095 *,
1096 handover_rel: str,
1097 handover_text: str = "# Hand preserved handover\n",
1098 roadmap_rel: str | None = None,
1099 roadmap_text: str | None = None,
1100 extra_cursor_rules: dict[str, str] | None = None,
1101 ) -> None:
1102 """Create a pre-existing living-doc layout for migrate fixtures."""
1103 hand = repo_root / handover_rel
1104 hand.parent.mkdir(parents=True, exist_ok=True)
1105 hand.write_text(handover_text, encoding="utf-8")
1106 if roadmap_rel is not None:
1107 road = repo_root / roadmap_rel
1108 road.parent.mkdir(parents=True, exist_ok=True)
1109 road.write_text(roadmap_text or "# Hand preserved roadmap\n", encoding="utf-8")
1110 if extra_cursor_rules:
1111 rules = repo_root / ".cursor" / "rules"
1112 rules.mkdir(parents=True, exist_ok=True)
1113 for name, text in extra_cursor_rules.items():
1114 (rules / name).write_text(text, encoding="utf-8")
1115
1116
1117 def lock_origins(repo_root: Path) -> dict[str, str]:
1118 """Return path → origin map from ``version.lock``."""
1119 from cli.version_lock import entry_origin, read_version_lock
1120
1121 lock = read_version_lock(repo_root / ".overseer" / "version.lock")
1122 return {e.path: entry_origin(e) for e in lock.footprint}
1123
1124
1125 def generate_ed25519_keypair() -> tuple[object, str]:
1126 """Return ``(private_key, ed25519:<base64> pubkey token)`` for tests only."""
1127 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
1128
1129 from tools.honesty.ed25519_util import encode_ed25519_token
1130
1131 private_key = Ed25519PrivateKey.generate()
1132 pubkey_token = encode_ed25519_token(private_key.public_key().public_bytes_raw())
1133 return private_key, pubkey_token
1134
1135
1136 def sign_entry_hash(private_key: object, entry_hash_hex: str) -> str:
1137 """Sign lowercase hex ``entry_hash_hex``; return ``ed25519:<base64>`` token (tests only)."""
1138 from tools.honesty.ed25519_util import encode_ed25519_token
1139
1140 sig_bytes = private_key.sign(entry_hash_hex.encode("utf-8")) # type: ignore[attr-defined]
1141 return encode_ed25519_token(sig_bytes)
1142
1143
1144 def attach_signed_provenance(
1145 body: dict,
1146 *,
1147 pubkey_token: str,
1148 agent_id: str = "cursor-agent",
1149 model_id: str = "gpt-5.6",
1150 human_ref: str | None = None,
1151 ) -> dict:
1152 """Return a copy of ``body`` with unsigned provenance identity fields."""
1153 provenance: dict = {"agent_id": agent_id, "model_id": model_id}
1154 if human_ref is not None:
1155 provenance["human_ref"] = human_ref
1156 signed = dict(body)
1157 signed["provenance"] = provenance
1158 signed["_test_pubkey"] = pubkey_token
1159 return signed
1160
1161
1162 def sign_append_body(
1163 body: dict,
1164 *,
1165 kind: str,
1166 prev_hash: str,
1167 private_key: object,
1168 pubkey_token: str | None = None,
1169 ) -> dict:
1170 """Validate, hash, sign, and return an append-ready body with ``provenance.sig``."""
1171 from tools.honesty.validate import validate_append_body
1172
1173 pubkey = pubkey_token or body.pop("_test_pubkey", None)
1174 if pubkey is None:
1175 raise ValueError("pubkey_token required")
1176 draft = dict(body)
1177 draft.pop("_test_pubkey", None)
1178 validated = validate_append_body(kind=kind, body=draft)
1179 preview = dict(validated)
1180 preview["prev_hash"] = prev_hash
1181 preview["provenance"] = {**validated["provenance"], "pubkey": pubkey}
1182 entry_hash = compute_entry_hash(preview)
1183 signed = dict(validated)
1184 signed["provenance"] = {
1185 **validated["provenance"],
1186 "pubkey": pubkey,
1187 "sig": sign_entry_hash(private_key, entry_hash),
1188 }
1189 return signed
1190
1191
1192 def compute_entry_hash(body: dict) -> str:
1193 """Re-export for tests building signing previews."""
1194 from tools.honesty.canonical import compute_entry_hash as _compute
1195
1196 return _compute(body)
1197
1198
1199 def finalize_signed_body(body: dict, *, private_key: object, prev_hash: str) -> dict:
1200 """Compute envelope hashes and attach ``provenance.sig`` (tests / direct ledger writes)."""
1201 from tools.honesty.canonical import compute_entry_hash
1202 from tools.honesty.genesis import utc_now_z
1203
1204 entry = dict(body)
1205 if not entry.get("ts"):
1206 entry["ts"] = utc_now_z()
1207 entry["prev_hash"] = prev_hash
1208 entry_hash = compute_entry_hash(entry)
1209 entry["entry_hash"] = entry_hash
1210 provenance = dict(entry.get("provenance", {}))
1211 provenance["sig"] = sign_entry_hash(private_key, entry_hash)
1212 entry["provenance"] = provenance
1213 return entry
1214
File History 1 commit
sha256:6abcf1fa82a7a621ccbc945f19acdba5bc0db54569599404a1452fb4a096a199 fix(ISR): default require_independent_second_reviewer to require Human minor ⚠ 1 day ago