597 lines
32 KiB
Python
597 lines
32 KiB
Python
"""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()
|