"""Behavioral control tests. All mutations are confined to disposable campaign roots.""" from concurrent.futures import ProcessPoolExecutor from copy import deepcopy from datetime import datetime, timedelta, timezone import json import os from pathlib import Path import shutil import subprocess import sys import tempfile import unittest from unittest.mock import patch REPO = Path(__file__).resolve().parents[2] sys.path.insert(0, str(REPO / "tools")) from campaign import Campaign, ControlError, MODELS, atomic_json, file_hash, read_json import run_agent OWNER = dict(actor="worker", role="implementer", model=MODELS["implementer"]) LEAD = dict(actor="lead", role="lead", model=MODELS["lead"]) VERIFIER = dict(actor="independent", role="verifier", model=MODELS["verifier"]) def contender(root): try: return Campaign(root).lease("acquire", "shared-vm", **OWNER)["token"] except ControlError: return None class Controls(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix="sots-controls-", dir=os.environ.get("SOTS_TEST_TMPDIR", "/tmp/opencode")) self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) / "re" (self.root / "campaign/contracts").mkdir(parents=True) for filename in ("models.json", "contract.schema.json"): shutil.copyfile(REPO / "campaign" / filename, self.root / "campaign" / filename) self.c = Campaign(str(self.root)) self.contract = read_json(REPO / "campaign/contracts/controls-bootstrap.json") self.contract.update(id="slice", status="proposed", checkpoint=None, evidence=[], owner={"name": "worker", "role": "implementer"}) self.contract["baseline"]["re"]["path"] = str(self.root) self.engine = Path(self.temp.name) / "engine" self.engine.mkdir() self.contract["baseline"]["engine"]["path"] = str(self.engine) for kind, repo in (("re", self.root), ("engine", self.engine)): (repo / "source.txt").write_text("source A\n") subprocess.run(["git", "init", "-q", str(repo)], check=True) subprocess.run(["git", "-C", str(repo), "add", "source.txt"], check=True) subprocess.run(["git", "-C", str(repo), "-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid", "commit", "-qm", "fixture"], check=True) self.contract["baseline"][kind]["commit"] = run_agent.git(repo, "rev-parse", "HEAD") self.c.save(self.contract) atomic_json(self.root / "campaign/measurement.json", {"tests": "executed", "distinct_states": ["a", "b"]}) self.baseline_patch = patch.object(Campaign, "check_baseline") self.baseline_patch.start() self.addCleanup(self.baseline_patch.stop) def checkpoint(self, session="worker-session"): return self.c.checkpoint("slice", **OWNER, session=session, summary="Observed branches a and b; decision: collect independent reproduction.", artifacts=["campaign/measurement.json"], next_action="Reproduce the two states independently.") def implementing(self): self.c.transition("slice", "ready", **OWNER) self.c.transition("slice", "implementing", **OWNER) def verification(self): self.implementing() self.checkpoint() self.c.transition("slice", "verification", **OWNER) self.c.evidence("slice", self.evidence(), **OWNER) def evidence(self, integrated=False): artifact = self.c.artifact("campaign/measurement.json") return dict(id="e1" if not integrated else "integrated", axis="control-tests", path="campaign/measurement.json", sha256=file_hash(self.root / "campaign/measurement.json"), source=deepcopy(self.contract["baseline"]), integrated=integrated, source_binding=self.c.source_binding(self.contract), binaries=[artifact], inputs=[artifact], outcomes=[{"criterion": a["id"], "status": "pass", "artifact": artifact} for a in self.contract["acceptance"]]) def verdict(self): return self.c.verdict("slice", **VERIFIER, session="verify-session", verdict="pass", explanation="Independently reproduced both states and reviewed package.") def test_schema_rejects_unknown_missing_types_and_ids(self): for mutate in (lambda c: c.update(unexpected=True), lambda c: c.pop("effects"), lambda c: c.update(inputs="not-an-array"), lambda c: c.update(id="../escape"), lambda c: c["owner"].update(model="claimed"), lambda c: c["baseline"]["engine"].update(commit="HEAD"), lambda c: c.update(acceptance=[{"id": "x", "axis": "a", "criterion": "c", "extra": 1}])): value = deepcopy(self.contract) mutate(value) with self.assertRaises(ControlError): self.c.validate(value) def test_duplicate_json_keys_and_nonfinite_numbers(self): for text in ('{"id":"a","id":"b"}', '{"value":NaN}'): path = self.root / "campaign/bad.json" path.write_text(text) with self.assertRaises(ControlError): read_json(path) def test_state_root_and_path_escape(self): with self.assertRaises(ControlError): Campaign("relative") with self.assertRaises(ControlError): self.c.path("campaign/../../outside") (self.root / "campaign/escape").symlink_to(self.root.parent, target_is_directory=True) with self.assertRaises(ControlError): self.c.path("campaign/escape/outside") wrong = deepcopy(self.contract) wrong["baseline"]["re"]["path"] = str(self.root / "other") with self.assertRaises(ControlError): self.c.validate(wrong) def test_lifecycle_no_bypass_or_missing_inputs(self): for target in ("implementing", "verification", "integration", "accepted"): with self.assertRaises(ControlError): self.c.transition("slice", target, **LEAD) self.contract["inputs"] = [] self.c.save(self.contract) with self.assertRaisesRegex(ControlError, "ready requires"): self.c.transition("slice", "ready", **OWNER) def test_identity_and_dependency_fail_closed(self): with self.assertRaises(ControlError): self.c.transition("slice", "ready", actor="impostor", role="implementer", model=MODELS["implementer"]) with self.assertRaises(ControlError): self.c.transition("slice", "ready", actor="lead", role="lead", model=MODELS["resolver"]) self.contract["dependencies"] = ["missing"] self.c.save(self.contract) with self.assertRaises(ControlError): self.c.transition("slice", "ready", **OWNER) def test_missing_and_stale_checkpoint_and_artifact(self): self.implementing() with self.assertRaisesRegex(ControlError, "missing durable"): self.c.transition("slice", "verification", **OWNER) cp = self.checkpoint() path = self.c.path(self.c.load("slice")["checkpoint"]) cp["timestamp"] = (datetime.now(timezone.utc) - timedelta(minutes=16)).isoformat() atomic_json(path, cp) with self.assertRaisesRegex(ControlError, "stale"): self.c.transition("slice", "verification", **OWNER) self.checkpoint() atomic_json(self.root / "campaign/measurement.json", {"changed": True}) with self.assertRaisesRegex(ControlError, "hash mismatch"): self.c.transition("slice", "verification", **OWNER) def test_checkpoint_basis_session_model_and_bounds(self): self.implementing() self.checkpoint() c = self.c.load("slice") for kwargs in ({"session": "different"}, {"model": MODELS["resolver"]}, {"since": "2999-01-01T00:00:00+00:00"}): with self.assertRaises(ControlError): self.c.check_checkpoint(c, **kwargs) c["effects"].append("Changed scope") with self.assertRaisesRegex(ControlError, "basis"): self.c.check_checkpoint(c) with self.assertRaises(ControlError): self.c.checkpoint("slice", **OWNER, session="x", summary="x" * 6001, artifacts=[], next_action="probe") with self.assertRaisesRegex(ControlError, "own mutable contract"): self.c.checkpoint("slice", **OWNER, session="x", summary="checkpoint", artifacts=["campaign/contracts/slice.json"], next_action="probe") def test_wip_bound(self): self.implementing() for cid in ("second", "third"): c = deepcopy(self.contract) c["id"] = cid self.c.save(c) self.c.transition(cid, "ready", **OWNER) self.c.transition("second", "implementing", **OWNER) with self.assertRaisesRegex(ControlError, "WIP"): self.c.transition("third", "implementing", **OWNER) self.assertEqual(self.c.load("third")["status"], "ready") def test_surprise_blocks_and_resolution_invalidates(self): self.verification() self.verdict() s = self.c.surprise("slice", **OWNER, summary="Unexpected RNG draw", probe="Run boundary fixture") self.assertEqual(self.c.load("slice")["status"], "blocked") with self.assertRaisesRegex(ControlError, "unresolved"): self.c.transition("slice", "needs-revision", **LEAD) with self.assertRaises(ControlError): self.c.resolve(s["id"], **OWNER, explanation="Maybe fine", probe="Test") with self.assertRaises(ControlError): self.c.resolve(s["id"], **LEAD, explanation="", probe="Test") self.c.resolve(s["id"], **LEAD, explanation="Boundary assumption false; revise adapter", probe="Measure rejection draw at twist") c = self.c.load("slice") self.assertEqual(c["status"], "needs-revision") self.assertEqual(c["evidence"], []) self.assertIsNone(c["checkpoint"]) self.assertEqual(read_json(self.root / "campaign/runtime/verdicts/slice.json")["verdict"], "invalidated") with self.assertRaises(ControlError): self.c.transition("slice", "accepted", **LEAD) def test_multiple_surprises_remain_blocked_until_all_resolved(self): records = [self.c.surprise("slice", **OWNER, summary=str(i), probe="Probe") for i in range(2)] self.c.resolve(records[0]["id"], **LEAD, explanation="Revise", probe="Probe") self.assertEqual(self.c.load("slice")["status"], "blocked") self.c.resolve(records[1]["id"], **LEAD, explanation="Revise", probe="Probe") self.assertEqual(self.c.load("slice")["status"], "needs-revision") def test_acceptance_requires_independent_final_integrated_package(self): self.verification() with self.assertRaises(ControlError): self.c.transition("slice", "integration", **LEAD) with self.assertRaisesRegex(ControlError, "independent"): self.c.verdict("slice", actor="worker", role="verifier", model=MODELS["verifier"], session="separate", verdict="pass", explanation="Reviewed") with self.assertRaisesRegex(ControlError, "overlaps"): self.c.verdict("slice", **VERIFIER, session="worker-session", verdict="pass", explanation="Reviewed") self.verdict() self.c.transition("slice", "integration", **LEAD) with self.assertRaisesRegex(ControlError, "axes"): self.c.transition("slice", "accepted", **LEAD) self.c.evidence("slice", self.evidence(integrated=True), **LEAD) with self.assertRaisesRegex(ControlError, "verdict"): self.c.transition("slice", "accepted", **LEAD) self.verdict() self.c.transition("slice", "accepted", **LEAD) self.assertEqual(self.c.load("slice")["status"], "accepted") def test_evidence_source_tampering_and_revision(self): self.verification() bad = self.evidence() bad["source"]["engine"]["commit"] = "f" * 40 with self.assertRaisesRegex(ControlError, "source"): self.c.evidence("slice", bad, **OWNER) with self.assertRaises(ControlError): self.c.evidence("slice", self.evidence(integrated=True), **OWNER) self.verdict() atomic_json(self.root / "campaign/measurement.json", {"forged": True}) with self.assertRaisesRegex(ControlError, "hash"): self.c.transition("slice", "integration", **LEAD) self.c.transition("slice", "needs-revision", **LEAD) self.assertIsNone(self.c.load("slice")["checkpoint"]) self.assertEqual(self.c.load("slice")["evidence"], []) def test_lease_concurrent_acquisition_exactly_one_winner(self): with ProcessPoolExecutor(max_workers=4) as pool: results = list(pool.map(contender, [str(self.root)] * 8)) self.assertEqual(sum(r is not None for r in results), 1) self.assertEqual(self.c.lease("show", "shared-vm")["status"], "held") def test_lease_token_owner_and_explicit_stale_release(self): lease = self.c.lease("acquire", "vm", **OWNER) for kwargs in ({**OWNER, "token": "bad"}, {**LEAD, "token": lease["token"]}, {**OWNER, "lead_release": True, "reason": "stale"}, {**LEAD, "lead_release": True}): with self.assertRaises(ControlError): self.c.lease("release", "vm", **kwargs) path = self.root / "campaign/runtime/leases/vm.json" lease["timestamp"] = "2000-01-01T00:00:00+00:00" atomic_json(path, lease) with self.assertRaises(ControlError): self.c.lease("acquire", "vm", **LEAD) self.c.lease("release", "vm", **LEAD, lead_release=True, reason="Operator stopped; access verified") replacement = self.c.lease("acquire", "vm", **OWNER) self.assertNotEqual(lease["token"], replacement["token"]) self.c.lease("release", "vm", **OWNER, token=replacement["token"]) def test_model_registry_no_fallback(self): model_path = self.root / "campaign/models.json" models = read_json(model_path) models["roles"]["implementer"] = MODELS["resolver"] atomic_json(model_path, models) with self.assertRaisesRegex(ControlError, "model policy"): Campaign(str(self.root)) def test_runner_role_status_surprise_and_missing_checkpoint(self): with self.assertRaises(ControlError): run_agent.check_launch(self.c, self.c.load("slice"), "implementer", "worker") self.implementing() with self.assertRaisesRegex(ControlError, "missing durable"): run_agent.check_launch(self.c, self.c.load("slice"), "implementer", "worker") self.checkpoint() run_agent.check_launch(self.c, self.c.load("slice"), "implementer", "worker") with self.assertRaises(ControlError): run_agent.check_launch(self.c, self.c.load("slice"), "implementer", "other") self.c.surprise("slice", **OWNER, summary="Unmodeled input", probe="Measure it") self.assertTrue(run_agent.check_launch(self.c, self.c.load("slice"), "lead", "lead")) self.assertTrue(run_agent.check_launch(self.c, self.c.load("slice"), "resolver", "resolver")) with self.assertRaises(ControlError): run_agent.check_launch(self.c, self.c.load("slice"), "implementer", "worker") def test_final_integrated_verifier_launch_preserves_guards(self): self.verification() run_agent.check_launch(self.c, self.c.load("slice"), "verifier", "independent") self.verdict() self.c.transition("slice", "integration", **LEAD) self.c.evidence("slice", self.evidence(integrated=True), **LEAD) run_agent.check_launch(self.c, self.c.load("slice"), "verifier", "independent") with self.assertRaises(ControlError): run_agent.check_launch(self.c, self.c.load("slice"), "verifier", "worker") (self.engine / "source.txt").write_text("changed at same HEAD") with self.assertRaises(ControlError): run_agent.check_launch(self.c, self.c.load("slice"), "verifier", "independent") (self.engine / "source.txt").write_text("source A\n") self.c.surprise("slice", **LEAD, summary="new contradictory evidence", probe="repeat control") with self.assertRaises(ControlError): run_agent.check_launch(self.c, self.c.load("slice"), "verifier", "independent") def test_old_recovery_checks_integrity_but_not_age(self): self.implementing() cp = self.checkpoint() cp["timestamp"] = "2000-01-01T00:00:00+00:00" atomic_json(self.c.path(self.c.load("slice")["checkpoint"]), cp) run_agent.check_launch(self.c, self.c.load("slice"), "implementer", "worker") with self.assertRaisesRegex(ControlError, "stale"): self.c.check_checkpoint(self.c.load("slice")) (self.root / "campaign/measurement.json").write_text("changed") with self.assertRaisesRegex(ControlError, "hash mismatch"): run_agent.check_launch(self.c, self.c.load("slice"), "implementer", "worker") def test_same_head_source_mutation_rejects_verdict_and_promotion(self): self.verification() self.verdict() before = run_agent.git(self.engine, "rev-parse", "HEAD") (self.engine / "source.txt").write_text("source B") self.assertEqual(before, run_agent.git(self.engine, "rev-parse", "HEAD")) for check in (lambda: self.c.check_verdict(self.c.load("slice")), lambda: self.c.transition("slice", "integration", **LEAD)): with self.assertRaisesRegex(ControlError, "source content"): check() def test_integrated_re_source_mutation_rejects_acceptance(self): self.verification() self.verdict() self.c.transition("slice", "integration", **LEAD) self.c.evidence("slice", self.evidence(True), **LEAD) self.verdict() (self.root / "source.txt").write_text("changed RE tooling") with self.assertRaisesRegex(ControlError, "source content"): self.c.transition("slice", "accepted", **LEAD) def test_immutable_inputs_and_criterion_coverage(self): self.implementing() for field in ("inputs", "binaries", "outcomes"): e = self.evidence() e[field] = [] with self.assertRaises(ControlError): self.c.evidence("slice", e, **OWNER) e = self.evidence() e["outcomes"][0]["status"] = "blocked" with self.assertRaisesRegex(ControlError, "did not pass"): self.c.evidence("slice", e, **OWNER) def test_candidate_worktree_binding_cannot_claim_integration(self): tree = Path(self.temp.name) / "candidate-engine" subprocess.run(["git", "-C", str(self.engine), "worktree", "add", "--quiet", "--detach", str(tree), "HEAD"], check=True) self.verification() self.verdict() self.c.transition("slice", "integration", **LEAD) e = self.evidence(True) e["source_binding"] = self.c.source_binding(self.contract, {"engine": str(tree), "re": str(self.root)}) with self.assertRaisesRegex(ControlError, "canonical"): self.c.evidence("slice", e, **LEAD) e["integrated"] = False self.c.evidence("slice", e, **LEAD) (tree / "source.txt").write_text("candidate changed at unchanged HEAD") with self.assertRaisesRegex(ControlError, "source content"): self.c.check_evidence(self.c.load("slice")) def test_source_binding_add_delete_mode_and_escape(self): before = self.c.source_binding(self.contract) p = self.engine / "new-source.txt" p.write_text("new untracked source") self.assertNotEqual(before, self.c.source_binding(self.contract)) p.unlink() self.assertEqual(before, self.c.source_binding(self.contract)) (self.engine / "source.txt").chmod(0o755) self.assertNotEqual(before, self.c.source_binding(self.contract)) (self.engine / "source.txt").unlink() self.assertNotEqual(before, self.c.source_binding(self.contract)) (self.engine / "escape").symlink_to(self.root / "source.txt") with self.assertRaisesRegex(ControlError, "escapes"): self.c.source_binding(self.contract) def test_scientific_artifacts_and_secret_path_escapes(self): path = self.root / "verify/results/oracle.bin" path.parent.mkdir(parents=True) path.write_bytes(b"immutable oracle") self.c.check_artifact(self.c.artifact("verify/results/oracle.bin")) (self.root / "outside").symlink_to(self.root.parent, target_is_directory=True) (self.root / "secrets").mkdir() (self.root / "secrets/token").write_text("fixture") (self.root / "alias").symlink_to(self.root / "secrets/token") for name in ("../escape", "outside/engine/source.txt", "secrets/token", "alias", ".env", ".git/config", "/etc/passwd"): with self.assertRaises(ControlError): self.c.artifact(name) class Runner(Controls): """Actual subprocess launch/log checks; Git identity has separate real-worktree tests.""" # Inherit fixture helpers without running the Controls tests a second time. def setUp(self): super().setUp() self.implementing() self.checkpoint() self.trees = {"engine": self.root / "engine-tree", "re": self.root / "re-tree"} for path in self.trees.values(): path.mkdir() config = {"small_model": run_agent.COMPACTION_MODEL, "agent": {"sots-implementer": {"model": MODELS["implementer"], "steps": 40}, "compaction": {"model": run_agent.COMPACTION_MODEL}}} atomic_json(self.root / "opencode.json", config) self.fake = self.root / "fake-opencode" self.args = run_agent.parser().parse_args(["--state-root", str(self.root), "--contract", "slice", "--role", "implementer", "--actor", "worker", "--engine-worktree", str(self.trees["engine"]), "--re-worktree", str(self.trees["re"]), "--cwd", "engine", "--opencode", str(self.fake)]) self.tree_patch = patch.object(run_agent, "check_worktrees", return_value=self.trees) self.tree_patch.start() self.addCleanup(self.tree_patch.stop) self.source_patch = patch.object(run_agent, "source_identity", return_value={"engine": {"manifest_sha256": "a" * 64}, "re": {"manifest_sha256": "b" * 64}}) self.source_patch.start() self.addCleanup(self.source_patch.stop) def fake_process(self, checkpoint=True, available=True, exitcode=0, observed_model=None, event_type="step_finish", reason="stop", session_id="actual-opencode-session", mutate=""): script = f'''#!{sys.executable} import json, os, re, sys sys.path.insert(0, {str(REPO / "tools")!r}) from campaign import Campaign if sys.argv[1] == 'models': print('\\n'.join({(sorted(set(MODELS.values())) if available else [MODELS['resolver']])!r})) sys.exit(0) if sys.argv[1:3] == ['debug', 'config']: print(os.environ['OPENCODE_CONFIG_CONTENT']) sys.exit(0) assert '--model' in sys.argv and sys.argv[sys.argv.index('--model') + 1] == {MODELS['implementer']!r} assert '--agent' in sys.argv and '--continue' not in sys.argv and '--session' not in sys.argv assert os.environ['OPENCODE_CONFIG'] == {str(self.root / 'opencode.json')!r} assert json.loads(os.environ['OPENCODE_CONFIG_CONTENT'])['agent']['sots-implementer']['steps'] == 40 session = re.search(r'checkpoint session=(run-[a-f0-9]+)', sys.argv[-1]).group(1) if {checkpoint!r}: Campaign({str(self.root)!r}).checkpoint('slice', actor='worker', role='implementer', model={MODELS['implementer']!r}, session=session, summary='Quantum finished; measurements in artifact.', artifacts=['campaign/measurement.json'], next_action='Independently reproduce fixture.') {mutate} event = {{'type': {event_type!r}, 'sessionID': {session_id!r}, 'part': {{'reason': {reason!r}}}}} if {observed_model!r}: provider, model = {observed_model!r}.split('/', 1) event.update(providerID=provider, modelID=model) print(json.dumps(event)) sys.exit({exitcode}) ''' self.fake.write_text(script) self.fake.chmod(0o700) def test_run_success_captures_actual_events_and_checkpoint(self): self.fake_process(observed_model=MODELS["implementer"]) result = run_agent.run(self.args) self.assertEqual(result["status"], "complete") self.assertEqual(result["observed_sessions"], ["actual-opencode-session"]) self.assertEqual(result["requested_model"], MODELS["implementer"]) self.assertTrue(self.c.path(result["log"]).is_file()) self.assertEqual(self.c.check_checkpoint(self.c.load("slice"))["session"], result["id"]) def test_run_missing_end_checkpoint_is_incomplete(self): self.fake_process(checkpoint=False) with self.assertRaisesRegex(ControlError, "incomplete"): run_agent.run(self.args) runs = list((self.root / "campaign/runtime/runs").glob("run-*.json")) self.assertEqual(len(runs), 1) self.assertEqual(read_json(runs[0])["status"], "incomplete") def test_unavailable_model_and_config_bound(self): self.fake_process(available=False) self.args.dry_run = True with self.assertRaisesRegex(ControlError, "unavailable"): run_agent.run(self.args) self.fake_process() config = read_json(self.root / "opencode.json") config["agent"]["sots-implementer"]["steps"] = 100 atomic_json(self.root / "opencode.json", config) with self.assertRaisesRegex(ControlError, "40-step"): run_agent.run(self.args) def test_dry_run_writes_no_run_and_no_fallback(self): self.fake_process() self.args.dry_run = True result = run_agent.run(self.args) self.assertEqual(result["status"], "dry-run") self.assertFalse((self.root / "campaign/runtime/runs").exists()) def test_model_mismatch_and_nonzero_exit_fail(self): self.fake_process(observed_model=MODELS["resolver"]) with self.assertRaisesRegex(ControlError, "observed runner model"): run_agent.run(self.args) self.fake_process(exitcode=3) with self.assertRaisesRegex(ControlError, "exited 3"): run_agent.run(self.args) def test_running_reservation_is_not_stolen(self): self.fake_process() atomic_json(self.root / "campaign/runtime/runs/active-slice.json", {"contract": "slice", "run": "run-interrupted", "status": "running"}) with self.assertRaisesRegex(ControlError, "already has a running"): run_agent.run(self.args) def test_zero_exit_error_and_incomplete_events_fail(self): for kwargs in ({"event_type": "error"}, {"reason": "tool-calls"}, {"reason": "length"}, {"session_id": ""}, {"event_type": "text"}): self.fake_process(**kwargs) with self.assertRaisesRegex(ControlError, "successful completed step"): run_agent.run(self.args) def test_config_mutation_during_zero_exit_success_fails(self): self.fake_process(mutate=f"from pathlib import Path\np=Path({str(self.root / 'opencode.json')!r})\np.write_text(p.read_text()+'\\n')") with self.assertRaisesRegex(ControlError, "config changed"): run_agent.run(self.args) def test_error_followed_by_success_still_fails(self): self.fake_process(mutate="print(json.dumps({'type': 'error', 'sessionID': 'actual-opencode-session'}))") with self.assertRaisesRegex(ControlError, "runner error"): run_agent.run(self.args) def test_resolution_permission_scope_and_architecture_authority(self): config = read_json(self.root / "opencode.json") for role in ("lead", "resolver"): config["agent"]["sots-" + role] = {"model": MODELS[role], "steps": 40} atomic_json(self.root / "opencode.json", config) self.fake_process() self.args.role = "lead" self.args.actor = "lead" self.args.dry_run = True normal = run_agent.run(self.args) self.assertIn("Sol loop authority", normal["command"][-1]) self.assertNotIn("Workers cannot change architecture", normal["command"][-1]) self.c.surprise("slice", **OWNER, summary="Scope false", probe="Read evidence") blocked = self.c.load("slice") blocked["checkpoint"] = None self.c.save(blocked) self.args.role = "resolver" result = run_agent.run(self.args) self.assertTrue(result["resolution_only"]) self.assertIn("RESOLUTION-ONLY", result["command"][-1]) permissions = result["config"]["overlay"]["agent"]["sots-resolver"]["permission"] self.assertEqual(permissions["edit"]["*"], "deny") self.assertEqual(permissions["bash"]["*"], "deny") self.assertEqual(permissions["task"], "deny") def test_baseline_rechecked_under_reservation(self): self.fake_process() with patch.object(run_agent, "check_worktrees", side_effect=[self.trees, ControlError("baseline changed")]) as check: with self.assertRaisesRegex(ControlError, "baseline changed"): run_agent.run(self.args) self.assertEqual(check.call_count, 2) def test_permissions_expanded_prompt_and_provenance(self): prompt = self.root / "campaign/worker.md" prompt.write_text("Owned bounded worker instructions") config = read_json(self.root / "opencode.json") config["agent"]["sots-implementer"].update(prompt="{file:campaign/worker.md}", permission={"edit": {"*": "allow", "*AGENTS.md": "deny"}, "task": "deny"}) atomic_json(self.root / "opencode.json", config) _, env, provenance = run_agent.configuration(self.c, "implementer", self.trees) agent = json.loads(env["OPENCODE_CONFIG_CONTENT"])["agent"]["sots-implementer"] self.assertEqual(agent["prompt"], prompt.read_text()) self.assertEqual(agent["permission"]["external_directory"][str(self.root) + "/*"], "allow") self.assertEqual(agent["permission"]["external_directory"][str(self.root)], "allow") self.assertEqual(agent["permission"]["bash"], "allow") self.assertEqual(agent["permission"]["edit"]["*AGENTS.md"], "deny") prompt.write_text("Changed expanded instructions") self.assertNotEqual(provenance["expanded_prompt_sha256"], run_agent.configuration(self.c, "implementer", self.trees)[2]["expanded_prompt_sha256"]) # unittest inherits tests; explicitly build the suite so shared fixtures are not duplicate cases. def load_tests(loader, tests, pattern): suite = unittest.TestSuite() for cls in (Controls, Runner, Worktrees): for name in cls.__dict__: if name.startswith("test_"): suite.addTest(cls(name)) return suite class Worktrees(unittest.TestCase): def test_actual_worktree_identity_and_baseline_without_source_commits(self): # Clone existing local history without checkout; never commit/stage in source or fixtures. with tempfile.TemporaryDirectory(prefix="sots-worktrees-", dir=os.environ.get("SOTS_TEST_TMPDIR", "/tmp/opencode")) as directory: root = Path(directory) sources = {kind: root / (kind + "-source") for kind in ("engine", "re")} trees = {kind: root / (kind + "-tree") for kind in ("engine", "re")} for kind in sources: subprocess.run(["git", "clone", "--quiet", "--shared", "--no-checkout", str(REPO), str(sources[kind])], check=True) subprocess.run(["git", "-C", str(sources[kind]), "worktree", "add", "--quiet", "--detach", "--no-checkout", str(trees[kind]), "HEAD"], check=True) (sources["re"] / "campaign/contracts").mkdir(parents=True) for filename in ("models.json", "contract.schema.json"): shutil.copyfile(REPO / "campaign" / filename, sources["re"] / "campaign" / filename) c = Campaign(str(sources["re"])) contract = read_json(REPO / "campaign/contracts/controls-bootstrap.json") contract["baseline"] = {kind: {"path": str(path), "commit": run_agent.git(path, "rev-parse", "HEAD")} for kind, path in sources.items()} Campaign.check_baseline(contract) self.assertEqual(run_agent.check_worktrees(c, contract, str(trees["engine"]), str(trees["re"])), trees) for engine, re_tree in ((sources["engine"], trees["re"]), (trees["engine"], trees["engine"]), (trees["re"], trees["engine"])): with self.assertRaises(ControlError): run_agent.check_worktrees(c, contract, str(engine), str(re_tree)) contract["baseline"]["engine"]["commit"] = "f" * 40 with self.assertRaisesRegex(ControlError, "baseline"): run_agent.check_worktrees(c, contract, str(trees["engine"]), str(trees["re"])) with self.assertRaisesRegex(ControlError, "baseline"): Campaign.check_baseline(contract) if __name__ == "__main__": unittest.main()