Something went wrong. Try again.
A video game where you play as a misaligned AI, deceiving and building power. An experiment in spec-driven development. forked from cameron.stream/misaligned
Something went wrong. Try again.
7.7 kB · 223 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224#!/usr/bin/env python3"""Run deterministic agent-protocol scenarios and emit evidence bundles."""
from __future__ import annotations
import argparseimport difflibimport hashlibimport jsonimport shutilimport subprocessimport sysimport tempfilefrom dataclasses import dataclassfrom pathlib import Path
@dataclassclass Scenario: path: Path name: str commands: list[str] expects: list[str] rejects: list[str] bevy_shot: str | None
def parse(path: Path) -> Scenario: commands: list[str] = [] expects: list[str] = [] rejects: list[str] = [] name = path.stem bevy_shot: str | None = None errors: list[str] = [] for number, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): line = raw.strip() if not line: continue if line.startswith("#"): directive = line[1:].strip() key, sep, value = directive.partition(":") if not sep: continue key, value = key.strip().upper(), value.strip() if key == "NAME": name = value elif key == "EXPECT": expects.append(value) elif key == "REJECT": rejects.append(value) elif key == "BEVY_SHOT": bevy_shot = value else: errors.append(f"{path}:{number}: unknown directive {key}") continue commands.append(line) if not commands: errors.append(f"{path}: has no commands") if "quit" not in (command.split()[0].lower() for command in commands): errors.append(f"{path}: must include quit") if not expects: errors.append(f"{path}: must declare at least one EXPECT") if errors: raise ValueError("\n".join(errors)) return Scenario(path, name, commands, expects, rejects, bevy_shot)
def scenario_paths(root: Path, requested: list[Path]) -> list[Path]: if requested: return [path if path.is_absolute() else root / path for path in requested] return sorted((root / "scenarios").glob("*.agent"))
def binary(root: Path, requested: Path | None, no_build: bool) -> Path: if requested: result = requested if requested.is_absolute() else root / requested else: result = root / "target/debug/misaligned" if not no_build: subprocess.run( ["cargo", "build", "--quiet", "-p", "misaligned-terminal", "--bin", "misaligned"], cwd=root, check=True, ) if not result.is_file(): raise FileNotFoundError(f"agent binary not found: {result}") return result
def invoke(bin_path: Path, scenario: Scenario, seed: int, root: Path) -> str: script = "\n".join(scenario.commands) + "\n" result = subprocess.run( [str(bin_path), "--agent", "--seed", str(seed)], cwd=root, text=True, input=script, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=45, check=False, ) if result.returncode != 0: raise RuntimeError( f"{scenario.path}: agent process exited {result.returncode}\n{result.stdout}" ) return result.stdout
def assert_output(scenario: Scenario, transcript: str) -> None: errors: list[str] = [] if "\x1b" in transcript: errors.append("transcript contains ANSI escape bytes") for expected in scenario.expects: if expected not in transcript: errors.append(f"missing EXPECT: {expected}") for rejected in scenario.rejects: if rejected in transcript: errors.append(f"matched REJECT: {rejected}") if errors: raise AssertionError(f"{scenario.path}: " + "; ".join(errors))
def write_evidence(output: Path, scenario: Scenario, transcript: str, seed: int) -> dict: output.mkdir(parents=True, exist_ok=True) digest = hashlib.sha256(transcript.encode("utf-8")).hexdigest() shutil.copyfile(scenario.path, output / "commands.agent") (output / "transcript.txt").write_text(transcript, encoding="utf-8") (output / "sha256.txt").write_text(digest + "\n", encoding="utf-8") payload = { "scenario": scenario.name, "source": str(scenario.path), "seed": seed, "sha256": digest, "expects": scenario.expects, "rejects": scenario.rejects, "bevy_shot": scenario.bevy_shot, "status": "ok", } (output / "result.json").write_text( json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) return payload
def run_one( root: Path, scenario: Scenario, bin_path: Path, seed: int, output: Path | None, with_bevy: bool,) -> dict: first = invoke(bin_path, scenario, seed, root) second = invoke(bin_path, scenario, seed, root) if first != second: diff = "\n".join( difflib.unified_diff( first.splitlines(), second.splitlines(), fromfile="run-1", tofile="run-2", lineterm="" ) ) raise AssertionError(f"{scenario.path}: same-seed output drifted\n{diff}") assert_output(scenario, first) payload = { "scenario": scenario.name, "seed": seed, "sha256": hashlib.sha256(first.encode("utf-8")).hexdigest(), "status": "ok", } if output: payload = write_evidence(output, scenario, first, seed) if with_bevy and scenario.bevy_shot: if output: subprocess.run( ["bash", "tools/bevy-headless.sh", scenario.bevy_shot, "--output", str(output / "bevy.png")], cwd=root, check=True, ) else: with tempfile.TemporaryDirectory(prefix="misaligned-scenario-") as directory: subprocess.run( ["bash", "tools/bevy-headless.sh", scenario.bevy_shot, "--output", str(Path(directory) / "bevy.png")], cwd=root, check=True, ) return payload
def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("scenarios", nargs="*", type=Path) parser.add_argument("--check-definitions", action="store_true") parser.add_argument("--seed", type=int, default=1) parser.add_argument("--output", type=Path) parser.add_argument("--bin", type=Path) parser.add_argument("--no-build", action="store_true") parser.add_argument("--with-bevy", action="store_true") parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parent.parent) args = parser.parse_args(argv) root = args.root.resolve() try: paths = scenario_paths(root, args.scenarios) parsed = [parse(path) for path in paths] if not parsed: raise ValueError("no scenarios found") if args.check_definitions: print(f"scenarios: OK ({len(parsed)} definitions)") return 0 bin_path = binary(root, args.bin, args.no_build) results = [] for scenario in parsed: output = None if args.output: output = args.output / scenario.path.stem if len(parsed) > 1 else args.output results.append(run_one(root, scenario, bin_path, args.seed, output, args.with_bevy)) print(f"scenario: OK {scenario.name} sha256={results[-1]['sha256'][:12]}") return 0 except (OSError, ValueError, RuntimeError, AssertionError, subprocess.SubprocessError) as exc: print(f"FAIL: {exc}", file=sys.stderr) return 1
if __name__ == "__main__": raise SystemExit(main())