aaronrene / overseer-kit public
support.py python
434 lines 14.4 KB
Raw
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439 feat: K1-P1 complete — agent provenance, build-verification… Sonnet 4.6 patch 55 days ago
1 """Test helpers (not pytest fixtures)."""
2
3 from __future__ import annotations
4
5 import os
6 from pathlib import Path
7
8 from adapters.config import OverseerConfig, load_config
9 from adapters.factory import create_adapter
10 from adapters.runner import CommandResult, RecordingRunner
11
12 FIXTURES = Path(__file__).resolve().parent / "fixtures"
13 PILOT = FIXTURES / "pilot"
14 CHECKPOINTS = FIXTURES / "checkpoints"
15 HONESTY = FIXTURES / "honesty"
16
17
18 def write_config(repo_root: Path, name: str) -> Path:
19 src = FIXTURES / name
20 dest = repo_root / ".overseer" / "config.yaml"
21 dest.parent.mkdir(parents=True, exist_ok=True)
22 dest.write_text(src.read_text(encoding="utf-8"), encoding="utf-8")
23 return dest
24
25
26 def load_fixture_config(repo_root: Path, name: str) -> OverseerConfig:
27 return load_config(write_config(repo_root, name))
28
29
30 def ok(stdout: str = "") -> CommandResult:
31 return CommandResult(stdout=stdout, stderr="", exit_code=0)
32
33
34 def fail(stderr: str = "error", code: int = 1) -> CommandResult:
35 return CommandResult(stdout="", stderr=stderr, exit_code=code)
36
37
38 def make_runner(responses: dict[str, CommandResult]) -> RecordingRunner:
39 return RecordingRunner(responses=responses, calls=[])
40
41
42 def adapter_for(config: OverseerConfig, repo_root: Path, runner: RecordingRunner):
43 return create_adapter(config, repo_root, runner=runner)
44
45
46 def git_status_runner(branch: str = "main", dirty: bool = False) -> RecordingRunner:
47 """Recording runner with git-only ``status()`` responses."""
48 dirty_out = " M file" if dirty else ""
49 return make_runner(
50 {
51 "git rev-parse --abbrev-ref HEAD": ok(branch),
52 "git status --porcelain": ok(dirty_out),
53 }
54 )
55
56
57 def muse_status_runner(
58 repo_root: Path,
59 branch: str = "main",
60 dirty: bool = False,
61 ) -> RecordingRunner:
62 """Recording runner with muse-only ``status()`` responses."""
63 root = str(repo_root.resolve())
64 dirty_out = " M file" if dirty else ""
65 return make_runner(
66 {
67 f"muse -C {root} branch --show-current": ok(branch),
68 f"muse -C {root} status --porcelain": ok(dirty_out),
69 }
70 )
71
72
73 def muse_mirror_status_runner(
74 repo_root: Path,
75 branch: str = "main",
76 dirty: bool = False,
77 ) -> RecordingRunner:
78 """Recording runner with muse+git-mirror ``status()`` responses."""
79 root = str(repo_root.resolve())
80 dirty_out = " M file" if dirty else ""
81 return make_runner(
82 {
83 f"muse -C {root} rev-parse --abbrev-ref HEAD": ok(branch),
84 f"muse -C {root} status --porcelain": ok(dirty_out),
85 "git rev-parse --abbrev-ref HEAD": ok(branch),
86 "git status --porcelain": ok(dirty_out),
87 }
88 )
89
90
91 def run_cli(
92 argv: list[str],
93 *,
94 cwd: Path,
95 runner: RecordingRunner | None = None,
96 kit: Path | None = None,
97 review_provider_factory=None,
98 script_executor=None,
99 json_mode: bool = False,
100 ) -> int:
101 """Invoke ``cli.main`` with an injected runner and working directory."""
102 from cli.context import CliContext
103 from cli.main import main
104 from cli.output import OutputContext
105
106 old_cwd = Path.cwd()
107 os.chdir(cwd)
108 try:
109 ctx = CliContext.create(
110 runner=runner or make_runner({}),
111 cwd=cwd,
112 kit=kit,
113 output=OutputContext(json_mode=json_mode),
114 review_provider_factory=review_provider_factory,
115 script_executor=script_executor,
116 )
117 return main(argv, ctx=ctx)
118 finally:
119 os.chdir(old_cwd)
120
121
122 def seed_muse_substrate(repo_root: Path) -> None:
123 """Create minimal healthy ``.muse/`` for muse-backed regime tests."""
124 muse = repo_root / ".muse"
125 muse.mkdir(parents=True, exist_ok=True)
126 (muse / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
127 (muse / "repo.json").write_text("{}", encoding="utf-8")
128 (muse / "config.toml").write_text("", encoding="utf-8")
129
130
131 def seed_freeze_repo(repo_root: Path, *, config_name: str = "config-git-only.yaml") -> Path:
132 """Write config and copy a freeze artifact fixture into a temp repo."""
133 write_config(repo_root, config_name)
134 if "muse" in config_name:
135 seed_muse_substrate(repo_root)
136 docs = repo_root / "docs"
137 docs.mkdir(parents=True, exist_ok=True)
138 artifact = docs / "FREEZE.md"
139 artifact.write_text((FIXTURES / "freeze-artifact.md").read_text(encoding="utf-8"), encoding="utf-8")
140 return artifact
141
142
143 def pass_provider_factory():
144 """Factory returning a provider that always passes."""
145 from tools.freeze_reviewer.providers.base import LocalReviewProvider
146
147 def _factory(_provider_name: str) -> LocalReviewProvider:
148 return LocalReviewProvider(scripted_findings=[])
149
150 return _factory
151
152
153 def findings_provider_factory(findings):
154 """Factory returning a provider with scripted findings."""
155 from tools.freeze_reviewer.providers.base import LocalReviewProvider
156
157 def _factory(_provider_name: str) -> LocalReviewProvider:
158 return LocalReviewProvider(scripted_findings=list(findings))
159
160 return _factory
161
162
163 def unreachable_provider_factory(cause: str = "offline"):
164 """Factory returning an unreachable local provider."""
165 from tools.freeze_reviewer.providers.base import LocalReviewProvider
166
167 def _factory(_provider_name: str) -> LocalReviewProvider:
168 return LocalReviewProvider(force_unreachable=True, unreachable_cause=cause)
169
170 return _factory
171
172
173 class FakeHttpTransport:
174 """Recording HTTP transport for API provider tests (no network)."""
175
176 def __init__(
177 self,
178 *,
179 health_status: int = 200,
180 health_body: bytes = b'{"status":"ok"}',
181 review_status: int = 200,
182 review_body: bytes | None = None,
183 fail_health: bool = False,
184 fail_review: bool = False,
185 ) -> None:
186 self.health_status = health_status
187 self.health_body = health_body
188 self.review_status = review_status
189 self.review_body = review_body or b'{"findings":[]}'
190 self.fail_health = fail_health
191 self.fail_review = fail_review
192 self.calls: list[dict] = []
193
194 def request(
195 self,
196 *,
197 method: str,
198 url: str,
199 headers: dict[str, str],
200 body: bytes | None = None,
201 timeout: float = 30.0,
202 ) -> tuple[int, bytes]:
203 from tools.freeze_reviewer.providers.api_client import ProviderTransportError
204
205 self.calls.append(
206 {
207 "method": method,
208 "url": url,
209 "headers": dict(headers),
210 "body": body,
211 }
212 )
213 if "/health" in url:
214 if self.fail_health:
215 raise ProviderTransportError("health transport failure")
216 return self.health_status, self.health_body
217 if self.fail_review:
218 raise ProviderTransportError("review transport failure")
219 return self.review_status, self.review_body
220
221
222 def api_provider_factory(transport: FakeHttpTransport):
223 """Factory returning an API provider wired to a fake HTTP transport."""
224 from tools.freeze_reviewer.providers.api_client import ReviewApiClient
225 from tools.freeze_reviewer.providers.base import ApiReviewProvider
226
227 def _factory(_provider_name: str) -> ApiReviewProvider:
228 client = ReviewApiClient(transport=transport)
229 return ApiReviewProvider(client=client)
230
231 return _factory
232
233
234 def api_unreachable_provider_factory(cause: str = "API provider unavailable"):
235 """Factory returning a forced-unreachable API provider."""
236 from tools.freeze_reviewer.providers.base import ApiReviewProvider
237
238 def _factory(_provider_name: str) -> ApiReviewProvider:
239 return ApiReviewProvider(force_unreachable=True, unreachable_cause=cause)
240
241 return _factory
242
243
244 def seed_checkpoint_repo(repo_root: Path) -> None:
245 """Copy checkpoint fixture pack into a temp repo and mark scripts executable."""
246 import shutil
247 import stat
248
249 (repo_root / ".overseer").mkdir(parents=True, exist_ok=True)
250 config_src = CHECKPOINTS / "config-checkpoints-enabled.yaml"
251 (repo_root / ".overseer" / "config.yaml").write_text(
252 config_src.read_text(encoding="utf-8"),
253 encoding="utf-8",
254 )
255 for rel in ("policy", "manifests", "scripts"):
256 src = CHECKPOINTS / rel
257 dest = repo_root / rel
258 if dest.exists():
259 shutil.rmtree(dest)
260 shutil.copytree(src, dest)
261 for script in (repo_root / "scripts" / "verify").glob("*.py"):
262 script.chmod(script.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
263
264
265 def load_checkpoint_config(repo_root: Path) -> OverseerConfig:
266 """Load checkpoint-enabled config from a seeded repo."""
267 seed_checkpoint_repo(repo_root)
268 return load_config(repo_root / ".overseer" / "config.yaml")
269
270
271 def seed_honesty_repo(repo_root: Path) -> None:
272 """Copy honesty fixture pack into a temp repo."""
273 import shutil
274
275 (repo_root / ".overseer").mkdir(parents=True, exist_ok=True)
276 config_src = HONESTY / "config-honesty-enabled.yaml"
277 (repo_root / ".overseer" / "config.yaml").write_text(
278 config_src.read_text(encoding="utf-8"),
279 encoding="utf-8",
280 )
281 for rel in ("artifacts", "entries"):
282 src = HONESTY / rel
283 dest = repo_root / rel
284 if dest.exists():
285 shutil.rmtree(dest)
286 shutil.copytree(src, dest)
287
288
289 def load_honesty_config(repo_root: Path) -> OverseerConfig:
290 """Load honesty-enabled config from a seeded repo."""
291 seed_honesty_repo(repo_root)
292 return load_config(repo_root / ".overseer" / "config.yaml")
293
294
295 def honesty_artifact_hash(repo_root: Path) -> str:
296 """SHA-256 of the fixture sample artifact."""
297 from tools.honesty.artifact import sha256_file_bytes
298
299 return sha256_file_bytes(repo_root / "artifacts" / "sample.txt")
300
301
302 def load_honesty_entry(repo_root: Path, name: str, *, artifact_hash: str | None = None) -> dict:
303 """Load a verdict entry fixture with artifact hash substituted."""
304 import json
305
306 text = (HONESTY / "entries" / name).read_text(encoding="utf-8")
307 if artifact_hash is None:
308 artifact_hash = honesty_artifact_hash(repo_root)
309 text = text.replace("PLACEHOLDER", artifact_hash)
310 return json.loads(text)
311
312
313 def seed_pilot_tree(
314 repo_root: Path,
315 *,
316 handover_rel: str,
317 handover_text: str = "# Hand preserved handover\n",
318 roadmap_rel: str | None = None,
319 roadmap_text: str | None = None,
320 extra_cursor_rules: dict[str, str] | None = None,
321 ) -> None:
322 """Create a pre-existing living-doc layout for migrate fixtures."""
323 hand = repo_root / handover_rel
324 hand.parent.mkdir(parents=True, exist_ok=True)
325 hand.write_text(handover_text, encoding="utf-8")
326 if roadmap_rel is not None:
327 road = repo_root / roadmap_rel
328 road.parent.mkdir(parents=True, exist_ok=True)
329 road.write_text(roadmap_text or "# Hand preserved roadmap\n", encoding="utf-8")
330 if extra_cursor_rules:
331 rules = repo_root / ".cursor" / "rules"
332 rules.mkdir(parents=True, exist_ok=True)
333 for name, text in extra_cursor_rules.items():
334 (rules / name).write_text(text, encoding="utf-8")
335
336
337 def lock_origins(repo_root: Path) -> dict[str, str]:
338 """Return path → origin map from ``version.lock``."""
339 from cli.version_lock import entry_origin, read_version_lock
340
341 lock = read_version_lock(repo_root / ".overseer" / "version.lock")
342 return {e.path: entry_origin(e) for e in lock.footprint}
343
344
345 def generate_ed25519_keypair() -> tuple[object, str]:
346 """Return ``(private_key, ed25519:<base64> pubkey token)`` for tests only."""
347 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
348
349 from tools.honesty.ed25519_util import encode_ed25519_token
350
351 private_key = Ed25519PrivateKey.generate()
352 pubkey_token = encode_ed25519_token(private_key.public_key().public_bytes_raw())
353 return private_key, pubkey_token
354
355
356 def sign_entry_hash(private_key: object, entry_hash_hex: str) -> str:
357 """Sign lowercase hex ``entry_hash_hex``; return ``ed25519:<base64>`` token (tests only)."""
358 from tools.honesty.ed25519_util import encode_ed25519_token
359
360 sig_bytes = private_key.sign(entry_hash_hex.encode("utf-8")) # type: ignore[attr-defined]
361 return encode_ed25519_token(sig_bytes)
362
363
364 def attach_signed_provenance(
365 body: dict,
366 *,
367 pubkey_token: str,
368 agent_id: str = "cursor-agent",
369 model_id: str = "gpt-5.6",
370 human_ref: str | None = None,
371 ) -> dict:
372 """Return a copy of ``body`` with unsigned provenance identity fields."""
373 provenance: dict = {"agent_id": agent_id, "model_id": model_id}
374 if human_ref is not None:
375 provenance["human_ref"] = human_ref
376 signed = dict(body)
377 signed["provenance"] = provenance
378 signed["_test_pubkey"] = pubkey_token
379 return signed
380
381
382 def sign_append_body(
383 body: dict,
384 *,
385 kind: str,
386 prev_hash: str,
387 private_key: object,
388 pubkey_token: str | None = None,
389 ) -> dict:
390 """Validate, hash, sign, and return an append-ready body with ``provenance.sig``."""
391 from tools.honesty.validate import validate_append_body
392
393 pubkey = pubkey_token or body.pop("_test_pubkey", None)
394 if pubkey is None:
395 raise ValueError("pubkey_token required")
396 draft = dict(body)
397 draft.pop("_test_pubkey", None)
398 validated = validate_append_body(kind=kind, body=draft)
399 preview = dict(validated)
400 preview["prev_hash"] = prev_hash
401 preview["provenance"] = {**validated["provenance"], "pubkey": pubkey}
402 entry_hash = compute_entry_hash(preview)
403 signed = dict(validated)
404 signed["provenance"] = {
405 **validated["provenance"],
406 "pubkey": pubkey,
407 "sig": sign_entry_hash(private_key, entry_hash),
408 }
409 return signed
410
411
412 def compute_entry_hash(body: dict) -> str:
413 """Re-export for tests building signing previews."""
414 from tools.honesty.canonical import compute_entry_hash as _compute
415
416 return _compute(body)
417
418
419 def finalize_signed_body(body: dict, *, private_key: object, prev_hash: str) -> dict:
420 """Compute envelope hashes and attach ``provenance.sig`` (tests / direct ledger writes)."""
421 from tools.honesty.canonical import compute_entry_hash
422 from tools.honesty.genesis import utc_now_z
423
424 entry = dict(body)
425 if not entry.get("ts"):
426 entry["ts"] = utc_now_z()
427 entry["prev_hash"] = prev_hash
428 entry_hash = compute_entry_hash(entry)
429 entry["entry_hash"] = entry_hash
430 provenance = dict(entry.get("provenance", {}))
431 provenance["sig"] = sign_entry_hash(private_key, entry_hash)
432 entry["provenance"] = provenance
433 return entry
434
File History 1 commit
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439 feat: K1-P1 complete — agent provenance, build-verification… Sonnet 4.6 patch 55 days ago