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.
17 kB · 465 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466#!/usr/bin/env python3"""Fixture contract for structured project operations."""
from __future__ import annotations
import jsonimport importlib.utilimport osimport subprocessimport tempfileimport unittestfrom pathlib import Path
import scenarioimport work_orders
PROJECT_STATUS_SPEC = importlib.util.spec_from_file_location( "project_status", Path(__file__).with_name("project-status.py"))assert PROJECT_STATUS_SPEC and PROJECT_STATUS_SPEC.loaderproject_status = importlib.util.module_from_spec(PROJECT_STATUS_SPEC)PROJECT_STATUS_SPEC.loader.exec_module(project_status)
SPEC = """# Spec: {title}
```Type: specStatus: {status}Stage: {stage}Work order: {task}Work priority: {priority}Work class: {work_class}Blocked by: {blocked}Exclusive keys: - tools/Design: - wiki/law.md#lawDepends on: none```
## Acceptance criteria
1. Observable.{extra}"""
class ProjectOpsFixtures(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) (self.root / "wiki/process").mkdir(parents=True) (self.root / "tools").mkdir() (self.root / "wiki/law.md").write_text( "# Law\n\n```\nType: law\n```\n\n## Law\n", encoding="utf-8" ) (self.root / "wiki/process/ROADMAP.md").write_text( "# ROADMAP\n\n```\nType: knowledge\n```\n\n## How to dispatch\n", encoding="utf-8", )
def tearDown(self) -> None: self.temp.cleanup()
def write_spec( self, name: str, *, status: str = "READY", stage: str = "B1", priority: int = 10, blocked: str = "none", extra: str = "", ) -> Path: path = self.root / f"wiki/{name}.md" path.write_text( SPEC.format( title=name, status=status, stage=stage, task=name, priority=priority, work_class="process", blocked=blocked, extra=extra, ), encoding="utf-8", ) return path
def test_lane_classification_respects_blockers_and_stage(self) -> None: base = self.write_spec("base", status="IN PROGRESS", priority=10) anchor = work_orders.slugify_heading("Spec: base") self.write_spec( "held", status="DRAFT", priority=20, blocked=f"\n - wiki/{base.name}#{anchor}", ) self.write_spec("later", stage="B2", priority=1) specs = work_orders.load_specs(self.root) self.assertEqual([], work_orders.validate(self.root, specs)) by_rel = {spec.rel: spec for spec in specs} lanes = {spec.task: work_orders.lane(spec, by_rel) for spec in specs} self.assertEqual("current", lanes["base"]) self.assertEqual("held", lanes["held"]) self.assertEqual("staged", lanes["later"])
def test_obsolete_exclusive_path_fails(self) -> None: path = self.write_spec("bad-path") text = path.read_text(encoding="utf-8").replace(" - tools/", " - src/old.rs") path.write_text(text, encoding="utf-8") errors = work_orders.validate(self.root, work_orders.load_specs(self.root)) self.assertTrue(any("does not exist" in error for error in errors))
def test_resolved_open_and_closed_blocker_fail(self) -> None: self.write_spec( "stale", extra=( "\n## Questions [OPEN]\n\n- (Resolved yesterday: use one.)\n" "\n## Notes\n\nIssue #4 closed. Criterion is BLOCKED on issue #4.\n" ), ) errors = work_orders.validate(self.root, work_orders.load_specs(self.root)) self.assertTrue(any("resolved item" in error for error in errors)) self.assertTrue(any("still marks a criterion BLOCKED" in error for error in errors))
def test_projection_check_detects_staleness(self) -> None: self.write_spec("ready") specs = work_orders.load_specs(self.root) projected = work_orders.projected_roadmap(self.root, specs) self.assertIn(work_orders.BEGIN, projected) self.assertNotEqual( projected, (self.root / "wiki/process/ROADMAP.md").read_text(encoding="utf-8"), )
def test_checked_in_scenarios_parse(self) -> None: repo = Path(__file__).resolve().parent.parent parsed = [scenario.parse(path) for path in sorted((repo / "scenarios").glob("*.agent"))] self.assertGreaterEqual(len(parsed), 3)
def test_activity_overlap_warns_without_blocking_and_task_missing_refuses(self) -> None: repo = Path(__file__).resolve().parent.parent claims = self.root / "claims" env = os.environ.copy() env["MISALIGNED_CLAIMS_DIR"] = str(claims) env["MISALIGNED_CLAIM_PID"] = str(os.getpid()) first = subprocess.run( ["bash", "tools/claim.sh", "activity", "first", "--class", "sim", "--key", "crates/misaligned-core"], cwd=repo, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) self.assertEqual(0, first.returncode, first.stdout) second = subprocess.run( ["bash", "tools/claim.sh", "claim", "second", "--class", "sim", "--key", "crates/misaligned-core/src/sim.rs"], cwd=repo, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) self.assertEqual(0, second.returncode, second.stdout) self.assertIn("overlap", second.stdout.lower()) blocked = subprocess.run( ["bash", "tools/claim.sh", "set-status", "second", "blocked"], cwd=repo, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) self.assertEqual(0, blocked.returncode, blocked.stdout) check = subprocess.run( ["bash", "tools/claim.sh", "check", "--key", "crates/misaligned-core/src/sim.rs"], cwd=repo, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) self.assertEqual(0, check.returncode, check.stdout) self.assertIn("advisory", check.stdout.lower()) missing = subprocess.run( ["bash", "tools/task.sh", "finish", "definitely-missing"], cwd=repo, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) self.assertNotEqual(0, missing.returncode) self.assertIn("does not exist", missing.stdout) planned = subprocess.run( ["bash", "tools/task.sh", "start", "wiki/process/agent-scale.md", "--dry-run"], cwd=repo, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) self.assertEqual(0, planned.returncode, planned.stdout) self.assertIn("worktree-new.sh", planned.stdout) self.assertIn("project-operations", planned.stdout)
def test_dead_pid_activity_survives_while_task_worktree_exists(self) -> None: repo = Path(__file__).resolve().parent.parent claims = self.root / "claims" claims.mkdir() worktrees = self.root / "worktrees" (worktrees / "project-operations").mkdir(parents=True) (claims / "project-operations.claim").write_text( "id=project-operations\nclass=process\nstatus=claimed\n" "pid=999999\nstarted=test\nhost=test\nkey=tools/\n", encoding="utf-8", ) env = os.environ.copy() env["MISALIGNED_CLAIMS_DIR"] = str(claims) env["MISALIGNED_WORKTREE_ROOT"] = str(worktrees) result = subprocess.run( ["bash", "tools/claim.sh", "list"], cwd=repo, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) self.assertEqual(0, result.returncode, result.stdout) self.assertIn("project-operations", result.stdout) self.assertTrue((claims / "project-operations.claim").is_file())
def test_project_status_consistency_links_live_activity_run_and_worktree(self) -> None: now = 1_720_000_000.0 payload = { "activities": [ { "record_name": "task-a", "id": "task-a", "status": "claimed", "started": "2024-07-03T09:46:00Z", } ], "runs": [ { "record_name": "task-a", "id": "task-a", "status": "running", "updated_at": "2024-07-03T09:46:00Z", "worktree": str(self.root / "task-a"), } ], "worktrees": [ { "path": str(self.root / "task-a"), "branch": "worktree-task-a", } ], } checked = project_status.consistency(payload, now=now) self.assertTrue(checked["ok"], checked) self.assertEqual([], checked["warnings"])
def test_project_status_allows_low_level_activity_without_heartbeat(self) -> None: payload = { "activities": [ { "record_name": "task-a", "id": "task-a", "status": "claimed", "started": "2024-07-03T09:46:00Z", } ], "runs": [], "worktrees": [ { "path": str(self.root / "task-a"), "branch": "worktree-task-a", } ], } checked = project_status.consistency(payload, now=1_720_000_000.0) self.assertTrue(checked["ok"], checked) self.assertEqual([], checked["warnings"])
def test_project_status_recent_broken_links_fail_but_stale_links_warn(self) -> None: now = 1_720_000_000.0 recent = { "activities": [], "runs": [ { "record_name": "lost", "id": "lost", "status": "running", "updated_at": "2024-07-03T09:46:00Z", } ], "worktrees": [], } checked = project_status.consistency(recent, now=now) self.assertFalse(checked["ok"]) self.assertTrue(any("no linked registered worktree" in error for error in checked["errors"])) self.assertTrue(any("no active activity" in error for error in checked["errors"]))
recent["runs"][0]["updated_at"] = "2024-07-03T01:00:00Z" checked = project_status.consistency(recent, now=now) self.assertTrue(checked["ok"], checked) self.assertTrue(any("heartbeat is stale" in warning for warning in checked["warnings"])) self.assertTrue(any("no linked registered worktree" in warning for warning in checked["warnings"]))
def test_project_status_rejects_invalid_and_mismatched_records(self) -> None: payload = { "activities": [ { "record_name": "expected", "id": "other", "status": "claimed", "started": "2024-07-03T09:46:00Z", } ], "runs": [ { "record_name": "broken", "status": "invalid", "parse_error": "bad json", } ], "worktrees": [], } checked = project_status.consistency(payload, now=1_720_000_000.0) self.assertFalse(checked["ok"]) self.assertTrue(any("declares id other" in error for error in checked["errors"])) self.assertTrue(any("bad json" in error for error in checked["errors"]))
def test_heartbeat_preserves_worktree_through_phase_and_end(self) -> None: repo = Path(__file__).resolve().parent.parent runs = self.root / "runs" worktree = self.root / "task-worktree" worktree.mkdir() env = os.environ.copy() env["MISALIGNED_RUNS_DIR"] = str(runs) for command in ( ["start", "task-a", "--worktree", str(worktree), "--phase", "boot"], ["phase", "task-a", "test"], ["end", "task-a", "--status", "ok"], ): result = subprocess.run( ["bash", "tools/heartbeat.sh", *command], cwd=repo, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) self.assertEqual(0, result.returncode, result.stdout) status = json.loads((runs / "task-a/status.json").read_text(encoding="utf-8")) self.assertEqual(str(worktree), status["worktree"])
def test_landing_lock_refuses_live_holder_and_reaps_dead_holder(self) -> None: repo = Path(__file__).resolve().parent.parent lock = self.root / "landing.lock" lock.mkdir() (lock / "pid").write_text(f"{os.getpid()}\n", encoding="utf-8") env = os.environ.copy() env["MISALIGNED_LANDING_LOCK"] = str(lock) env["MISALIGNED_LANDING_WAIT"] = "0" live = subprocess.run( ["bash", "-c", "source tools/task.sh; acquire_landing_lock"], cwd=repo, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) self.assertNotEqual(0, live.returncode) self.assertIn("another task is landing", live.stdout)
(lock / "pid").write_text("999999\n", encoding="utf-8") stale = subprocess.run( [ "bash", "-c", "source tools/task.sh; acquire_landing_lock; release_landing_lock", ], cwd=repo, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) self.assertEqual(0, stale.returncode, stale.stdout) self.assertIn("clearing stale landing lock", stale.stdout) self.assertFalse(lock.exists())
lock.mkdir() pidless = subprocess.run( [ "bash", "-c", "source tools/task.sh; acquire_landing_lock; release_landing_lock", ], cwd=repo, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) self.assertEqual(0, pidless.returncode, pidless.stdout) self.assertIn("no holder pid", pidless.stdout) self.assertFalse(lock.exists())
def test_publish_site_pins_the_primary_head_revision(self) -> None: repo = Path(__file__).resolve().parent.parent primary = self.root / "primary" (primary / "tools").mkdir(parents=True) subprocess.run(["git", "init", "-q", str(primary)], check=True) subprocess.run(["git", "-C", str(primary), "config", "user.name", "Fixture"], check=True) subprocess.run( ["git", "-C", str(primary), "config", "user.email", "fixture@example.com"], check=True, ) (primary / "source.txt").write_text("source\n", encoding="utf-8") subprocess.run(["git", "-C", str(primary), "add", "source.txt"], check=True) subprocess.run(["git", "-C", str(primary), "commit", "-q", "-m", "fixture"], check=True) expected = subprocess.check_output( ["git", "-C", str(primary), "rev-parse", "HEAD"], text=True ).strip() (primary / "tools/site-deploy.sh").write_text( "#!/usr/bin/env bash\nset -euo pipefail\nprintf '%s\\n' \"$SITE_SOURCE_REVISION\" > published-revision\n", encoding="utf-8", )
result = subprocess.run( [ "bash", "-c", f'source tools/task.sh; primary="{primary}"; publish_site', ], cwd=repo, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) self.assertEqual(0, result.returncode, result.stdout) self.assertEqual( expected, (primary / "published-revision").read_text(encoding="utf-8").strip(), )
if __name__ == "__main__": unittest.main()