M2: Commit-or-Abort — cycle guard with 10-minute slice rule and crash recovery (Epic #842)

allegro/tests/test_cycle_guard.py

@@ -0,0 +1,143 @@
+"""100% compliance test for Allegro Commit-or-Abort (M2, Epic #842)."""
+
+import json
+import os
+import sys
+import tempfile
+import time
+import unittest
+from datetime import datetime, timezone, timedelta
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+import cycle_guard as cg
+
+
+class TestCycleGuard(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.TemporaryDirectory()
+        self.state_path = os.path.join(self.tmpdir.name, "cycle_state.json")
+        cg.STATE_PATH = self.state_path
+
+    def tearDown(self):
+        self.tmpdir.cleanup()
+        cg.STATE_PATH = cg.DEFAULT_STATE
+
+    def test_load_empty_state(self):
+        state = cg.load_state(self.state_path)
+        self.assertEqual(state["status"], "complete")
+        self.assertIsNone(state["cycle_id"])
+
+    def test_start_cycle(self):
+        state = cg.start_cycle("M2: Commit-or-Abort", path=self.state_path)
+        self.assertEqual(state["status"], "in_progress")
+        self.assertEqual(state["target"], "M2: Commit-or-Abort")
+        self.assertIsNotNone(state["cycle_id"])
+
+    def test_start_slice_requires_in_progress(self):
+        with self.assertRaises(RuntimeError):
+            cg.start_slice("test", path=self.state_path)
+
+    def test_slice_lifecycle(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("gather", path=self.state_path)
+        state = cg.load_state(self.state_path)
+        self.assertEqual(len(state["slices"]), 1)
+        self.assertEqual(state["slices"][0]["name"], "gather")
+        self.assertEqual(state["slices"][0]["status"], "in_progress")
+
+        cg.end_slice(status="complete", artifact="artifact.txt", path=self.state_path)
+        state = cg.load_state(self.state_path)
+        self.assertEqual(state["slices"][0]["status"], "complete")
+        self.assertEqual(state["slices"][0]["artifact"], "artifact.txt")
+        self.assertIsNotNone(state["slices"][0]["ended_at"])
+
+    def test_commit_cycle(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        cg.end_slice(path=self.state_path)
+        proof = {"files": ["a.py"]}
+        state = cg.commit_cycle(proof=proof, path=self.state_path)
+        self.assertEqual(state["status"], "complete")
+        self.assertEqual(state["proof"], proof)
+        self.assertIsNotNone(state["completed_at"])
+
+    def test_commit_without_in_progress_fails(self):
+        with self.assertRaises(RuntimeError):
+            cg.commit_cycle(path=self.state_path)
+
+    def test_abort_cycle(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        state = cg.abort_cycle("manual abort", path=self.state_path)
+        self.assertEqual(state["status"], "aborted")
+        self.assertEqual(state["abort_reason"], "manual abort")
+        self.assertIsNotNone(state["aborted_at"])
+        self.assertEqual(state["slices"][-1]["status"], "aborted")
+
+    def test_slice_timeout_true(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        # Manually backdate slice start to 11 minutes ago
+        state = cg.load_state(self.state_path)
+        old = (datetime.now(timezone.utc) - timedelta(minutes=11)).isoformat()
+        state["slices"][0]["started_at"] = old
+        cg.save_state(state, self.state_path)
+        self.assertTrue(cg.check_slice_timeout(max_minutes=10, path=self.state_path))
+
+    def test_slice_timeout_false(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        self.assertFalse(cg.check_slice_timeout(max_minutes=10, path=self.state_path))
+
+    def test_resume_or_abort_keeps_fresh_cycle(self):
+        cg.start_cycle("test", path=self.state_path)
+        state = cg.resume_or_abort(path=self.state_path)
+        self.assertEqual(state["status"], "in_progress")
+
+    def test_resume_or_abort_aborts_stale_cycle(self):
+        cg.start_cycle("test", path=self.state_path)
+        # Backdate start to 31 minutes ago
+        state = cg.load_state(self.state_path)
+        old = (datetime.now(timezone.utc) - timedelta(minutes=31)).isoformat()
+        state["started_at"] = old
+        cg.save_state(state, self.state_path)
+        state = cg.resume_or_abort(path=self.state_path)
+        self.assertEqual(state["status"], "aborted")
+        self.assertIn("crash recovery", state["abort_reason"])
+
+    def test_slice_duration_minutes(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        # Backdate by 5 minutes
+        state = cg.load_state(self.state_path)
+        old = (datetime.now(timezone.utc) - timedelta(minutes=5)).isoformat()
+        state["slices"][0]["started_at"] = old
+        cg.save_state(state, self.state_path)
+        mins = cg.slice_duration_minutes(path=self.state_path)
+        self.assertAlmostEqual(mins, 5.0, delta=0.5)
+
+    def test_cli_resume_prints_status(self):
+        cg.start_cycle("test", path=self.state_path)
+        rc = cg.main(["resume"])
+        self.assertEqual(rc, 0)
+
+    def test_cli_check_timeout(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        state = cg.load_state(self.state_path)
+        old = (datetime.now(timezone.utc) - timedelta(minutes=11)).isoformat()
+        state["slices"][0]["started_at"] = old
+        cg.save_state(state, self.state_path)
+        rc = cg.main(["check"])
+        self.assertEqual(rc, 1)
+
+    def test_cli_check_ok(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        rc = cg.main(["check"])
+        self.assertEqual(rc, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()