feat(allegro): complete M2 Commit-or-Abort with stop_guard, env fix, 26 tests (#845)

- Adds allegro/stop_guard.py (M1 reusable stop protocol)
- Fixes dynamic ALLEGRO_CYCLE_STATE env-var resolution in cycle_guard.py
- Expands test_cycle_guard.py to 26 pytest cases covering lifecycle,
  slices, crash recovery, CLI commands, and edge cases
- Adds allegro/__init__.py package marker

Closes #845

allegro/__init__.py

@@ -0,0 +1 @@
+"""Allegro self-improvement guard modules for Epic #842."""

allegro/cycle_guard.py

@@ -14,7 +14,10 @@ from datetime import datetime, timezone, timedelta
 from pathlib import Path
 
 DEFAULT_STATE = Path("/root/.hermes/allegro-cycle-state.json")
-STATE_PATH = Path(os.environ.get("ALLEGRO_CYCLE_STATE", DEFAULT_STATE))
+
+
+def _state_path() -> Path:
+    return Path(os.environ.get("ALLEGRO_CYCLE_STATE", DEFAULT_STATE))
 
 # Crash-recovery threshold: if a cycle has been in_progress for longer than
 # this many minutes, resume_or_abort() will auto-abort it.
@@ -26,7 +29,7 @@ def _now_iso() -> str:
 
 
 def load_state(path: Path | str | None = None) -> dict:
-    p = Path(path) if path else Path(STATE_PATH)
+    p = Path(path) if path else _state_path()
     if not p.exists():
         return _empty_state()
     try:
@@ -37,7 +40,7 @@ def load_state(path: Path | str | None = None) -> dict:
 
 
 def save_state(state: dict, path: Path | str | None = None) -> None:
-    p = Path(path) if path else Path(STATE_PATH)
+    p = Path(path) if path else _state_path()
     p.parent.mkdir(parents=True, exist_ok=True)
     state["last_updated"] = _now_iso()
     with open(p, "w") as f:

allegro/stop_guard.py

@@ -0,0 +1,186 @@
+#!/usr/bin/env python3
+"""Allegro Stop Guard — hard-interrupt gate for the Stop Protocol (M1, Epic #842).
+
+Usage:
+    python stop_guard.py check <target>       # exit 1 if stopped, 0 if clear
+    python stop_guard.py record <target>      # record a stop + log STOP_ACK
+    python stop_guard.py cleanup              # remove expired locks
+"""
+
+import argparse
+import json
+import os
+import sys
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+
+DEFAULT_REGISTRY = Path("/root/.hermes/allegro-hands-off-registry.json")
+DEFAULT_STOP_LOG = Path("/root/.hermes/burn-logs/allegro.log")
+
+REGISTRY_PATH = Path(os.environ.get("ALLEGRO_STOP_REGISTRY", DEFAULT_REGISTRY))
+STOP_LOG_PATH = Path(os.environ.get("ALLEGRO_STOP_LOG", DEFAULT_STOP_LOG))
+
+
+class StopInterrupted(Exception):
+    """Raised when a stop signal blocks an operation."""
+    pass
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def load_registry(path: Path | str | None = None) -> dict:
+    p = Path(path) if path else Path(REGISTRY_PATH)
+    if not p.exists():
+        return {
+            "version": 1,
+            "last_updated": _now_iso(),
+            "locks": [],
+            "rules": {
+                "default_lock_duration_hours": 24,
+                "auto_extend_on_stop": True,
+                "require_explicit_unlock": True,
+            },
+        }
+    try:
+        with open(p, "r") as f:
+            return json.load(f)
+    except Exception:
+        return {
+            "version": 1,
+            "last_updated": _now_iso(),
+            "locks": [],
+            "rules": {
+                "default_lock_duration_hours": 24,
+                "auto_extend_on_stop": True,
+                "require_explicit_unlock": True,
+            },
+        }
+
+
+def save_registry(registry: dict, path: Path | str | None = None) -> None:
+    p = Path(path) if path else Path(REGISTRY_PATH)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    registry["last_updated"] = _now_iso()
+    with open(p, "w") as f:
+        json.dump(registry, f, indent=2)
+
+
+def log_stop_ack(target: str, context: str, log_path: Path | str | None = None) -> None:
+    p = Path(log_path) if log_path else Path(STOP_LOG_PATH)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    ts = _now_iso()
+    entry = f"[{ts}] STOP_ACK — target='{target}' context='{context}'\n"
+    with open(p, "a") as f:
+        f.write(entry)
+
+
+def is_stopped(target: str, registry: dict | None = None) -> bool:
+    """Return True if target (or global '*') is currently stopped."""
+    reg = registry if registry is not None else load_registry()
+    now = datetime.now(timezone.utc)
+    for lock in reg.get("locks", []):
+        expires = lock.get("expires_at")
+        if expires:
+            try:
+                expires_dt = datetime.fromisoformat(expires)
+                if now > expires_dt:
+                    continue
+            except Exception:
+                continue
+        if lock.get("entity") == target or lock.get("entity") == "*":
+            return True
+    return False
+
+
+def assert_not_stopped(target: str, registry: dict | None = None) -> None:
+    """Raise StopInterrupted if target is stopped."""
+    if is_stopped(target, registry):
+        raise StopInterrupted(f"Stop signal active for '{target}'. Halt immediately.")
+
+
+def record_stop(
+    target: str,
+    context: str,
+    duration_hours: int | None = None,
+    registry_path: Path | str | None = None,
+) -> dict:
+    """Record a stop for target, log STOP_ACK, and save registry."""
+    reg = load_registry(registry_path)
+    rules = reg.get("rules", {})
+    duration = duration_hours or rules.get("default_lock_duration_hours", 24)
+    now = datetime.now(timezone.utc)
+    expires = (now + timedelta(hours=duration)).isoformat()
+
+    # Remove existing lock for same target
+    reg["locks"] = [l for l in reg.get("locks", []) if l.get("entity") != target]
+
+    lock = {
+        "entity": target,
+        "reason": context,
+        "locked_at": now.isoformat(),
+        "expires_at": expires,
+        "unlocked_by": None,
+    }
+    reg["locks"].append(lock)
+    save_registry(reg, registry_path)
+    log_stop_ack(target, context, log_path=STOP_LOG_PATH)
+    return lock
+
+
+def cleanup_expired(registry_path: Path | str | None = None) -> int:
+    """Remove expired locks and return remaining active count."""
+    reg = load_registry(registry_path)
+    now = datetime.now(timezone.utc)
+    kept = []
+    for lock in reg.get("locks", []):
+        expires = lock.get("expires_at")
+        if expires:
+            try:
+                expires_dt = datetime.fromisoformat(expires)
+                if now > expires_dt:
+                    continue
+            except Exception:
+                continue
+        kept.append(lock)
+    reg["locks"] = kept
+    save_registry(reg, registry_path)
+    return len(reg["locks"])
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Allegro Stop Guard")
+    sub = parser.add_subparsers(dest="cmd")
+
+    p_check = sub.add_parser("check", help="Check if target is stopped")
+    p_check.add_argument("target")
+
+    p_record = sub.add_parser("record", help="Record a stop")
+    p_record.add_argument("target")
+    p_record.add_argument("--context", default="manual stop")
+    p_record.add_argument("--hours", type=int, default=24)
+
+    sub.add_parser("cleanup", help="Remove expired locks")
+
+    args = parser.parse_args(argv)
+
+    if args.cmd == "check":
+        stopped = is_stopped(args.target)
+        print("STOPPED" if stopped else "CLEAR")
+        return 1 if stopped else 0
+    elif args.cmd == "record":
+        record_stop(args.target, args.context, args.hours)
+        print(f"Recorded stop for {args.target} ({args.hours}h)")
+        return 0
+    elif args.cmd == "cleanup":
+        remaining = cleanup_expired()
+        print(f"Cleanup complete. {remaining} active locks.")
+        return 0
+    else:
+        parser.print_help()
+        return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

allegro/tests/test_cycle_guard.py

@@ -1,143 +1,260 @@
-"""100% compliance test for Allegro Commit-or-Abort (M2, Epic #842)."""
+"""Tests for allegro.cycle_guard — Commit-or-Abort discipline, M2 Epic #842."""
 
 import json
 import os
-import sys
 import tempfile
-import time
-import unittest
 from datetime import datetime, timezone, timedelta
+from pathlib import Path
 
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+import pytest
 
-import cycle_guard as cg
+from allegro.cycle_guard import (
+    _empty_state,
+    _now_iso,
+    _parse_dt,
+    abort_cycle,
+    check_slice_timeout,
+    commit_cycle,
+    end_slice,
+    load_state,
+    resume_or_abort,
+    save_state,
+    slice_duration_minutes,
+    start_cycle,
+    start_slice,
+)
 
 
-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
+class TestStateLifecycle:
+    def test_empty_state(self):
+        state = _empty_state()
+        assert state["status"] == "complete"
+        assert state["cycle_id"] is None
+        assert state["version"] == 1
 
-    def tearDown(self):
-        self.tmpdir.cleanup()
-        cg.STATE_PATH = cg.DEFAULT_STATE
+    def test_load_state_missing_file(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "missing.json"
+            state = load_state(path)
+            assert state["status"] == "complete"
 
-    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_save_and_load_roundtrip(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            state = start_cycle("test-target", "details", path)
+            loaded = load_state(path)
+            assert loaded["cycle_id"] == state["cycle_id"]
+            assert loaded["status"] == "in_progress"
 
+    def test_load_state_malformed_json(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "bad.json"
+            path.write_text("not json")
+            state = load_state(path)
+            assert state["status"] == "complete"
+
+
+class TestCycleOperations:
     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"])
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            state = start_cycle("M2: Commit-or-Abort", "test details", path)
+            assert state["status"] == "in_progress"
+            assert state["target"] == "M2: Commit-or-Abort"
+            assert state["started_at"] is not None
 
-    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_start_cycle_overwrites_prior(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            old = start_cycle("old", path=path)
+            new = start_cycle("new", path=path)
+            assert new["target"] == "new"
+            loaded = load_state(path)
+            assert loaded["target"] == "new"
 
     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"])
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            start_cycle("test", path=path)
+            proof = {"files": ["a.py"], "tests": "passed"}
+            state = commit_cycle(proof, path)
+            assert state["status"] == "complete"
+            assert state["proof"]["files"] == ["a.py"]
+            assert state["completed_at"] is not None
 
-    def test_commit_without_in_progress_fails(self):
-        with self.assertRaises(RuntimeError):
-            cg.commit_cycle(path=self.state_path)
+    def test_commit_cycle_not_in_progress_raises(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            with pytest.raises(RuntimeError, match="not in_progress"):
+                commit_cycle(path=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")
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            start_cycle("test", path=path)
+            state = abort_cycle("timeout", path)
+            assert state["status"] == "aborted"
+            assert state["abort_reason"] == "timeout"
+            assert state["aborted_at"] is not None
 
-    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_abort_cycle_not_in_progress_raises(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            with pytest.raises(RuntimeError, match="not in_progress"):
+                abort_cycle("reason", path=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")
+class TestSliceOperations:
+    def test_start_and_end_slice(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            start_cycle("test", path=path)
+            start_slice("research", path)
+            state = end_slice("complete", "found answer", path)
+            assert len(state["slices"]) == 1
+            assert state["slices"][0]["name"] == "research"
+            assert state["slices"][0]["status"] == "complete"
+            assert state["slices"][0]["artifact"] == "found answer"
+            assert state["slices"][0]["ended_at"] is not None
 
-    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_start_slice_without_cycle_raises(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            with pytest.raises(RuntimeError, match="in_progress"):
+                start_slice("name", path)
+
+    def test_end_slice_without_slice_raises(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            start_cycle("test", path=path)
+            with pytest.raises(RuntimeError, match="No active slice"):
+                end_slice(path=path)
 
     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)
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            start_cycle("test", path=path)
+            start_slice("long", path)
+            state = load_state(path)
+            state["slices"][0]["started_at"] = (datetime.now(timezone.utc) - timedelta(minutes=5)).isoformat()
+            save_state(state, path)
+            minutes = slice_duration_minutes(path)
+            assert minutes is not None
+            assert minutes >= 4.9
 
-    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_check_slice_timeout_true(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            start_cycle("test", path=path)
+            start_slice("old", path)
+            state = load_state(path)
+            state["slices"][0]["started_at"] = (datetime.now(timezone.utc) - timedelta(minutes=15)).isoformat()
+            save_state(state, path)
+            assert check_slice_timeout(max_minutes=10.0, path=path) is True
 
-    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)
+    def test_check_slice_timeout_false(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            start_cycle("test", path=path)
+            start_slice("fresh", path)
+            assert check_slice_timeout(max_minutes=10.0, path=path) is False
 
 
-if __name__ == "__main__":
-    unittest.main()
+class TestCrashRecovery:
+    def test_resume_or_abort_aborts_stale_cycle(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            start_cycle("test", path=path)
+            state = load_state(path)
+            state["started_at"] = (datetime.now(timezone.utc) - timedelta(minutes=60)).isoformat()
+            save_state(state, path)
+            result = resume_or_abort(path)
+            assert result["status"] == "aborted"
+            assert "stale cycle" in result["abort_reason"]
+
+    def test_resume_or_abort_keeps_fresh_cycle(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            start_cycle("test", path=path)
+            result = resume_or_abort(path)
+            assert result["status"] == "in_progress"
+
+    def test_resume_or_abort_no_op_when_complete(self):
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            state = _empty_state()
+            save_state(state, path)
+            result = resume_or_abort(path)
+            assert result["status"] == "complete"
+
+
+class TestDateParsing:
+    def test_parse_dt_with_z(self):
+        dt = _parse_dt("2026-04-06T12:00:00Z")
+        assert dt.tzinfo is not None
+
+    def test_parse_dt_with_offset(self):
+        iso = "2026-04-06T12:00:00+00:00"
+        dt = _parse_dt(iso)
+        assert dt.tzinfo is not None
+
+
+class TestCLI:
+    def test_cli_resume(self, capsys):
+        from allegro.cycle_guard import main
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            start_cycle("cli", path=path)
+            os.environ["ALLEGRO_CYCLE_STATE"] = str(path)
+            rc = main(["resume"])
+            captured = capsys.readouterr()
+            assert rc == 0
+            assert captured.out.strip() == "in_progress"
+
+    def test_cli_start(self, capsys):
+        from allegro.cycle_guard import main
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            os.environ["ALLEGRO_CYCLE_STATE"] = str(path)
+            rc = main(["start", "target", "--details", "d"])
+            captured = capsys.readouterr()
+            assert rc == 0
+            assert "Cycle started" in captured.out
+
+    def test_cli_commit(self, capsys):
+        from allegro.cycle_guard import main
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            os.environ["ALLEGRO_CYCLE_STATE"] = str(path)
+            main(["start", "t"])
+            rc = main(["commit", "--proof", '{"ok": true}'])
+            captured = capsys.readouterr()
+            assert rc == 0
+            assert "Cycle committed" in captured.out
+
+    def test_cli_check_timeout(self, capsys):
+        from allegro.cycle_guard import main
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            start_cycle("t", path=path)
+            start_slice("s", path=path)
+            state = load_state(path)
+            state["slices"][0]["started_at"] = (datetime.now(timezone.utc) - timedelta(minutes=15)).isoformat()
+            save_state(state, path)
+            os.environ["ALLEGRO_CYCLE_STATE"] = str(path)
+            rc = main(["check"])
+            captured = capsys.readouterr()
+            assert rc == 1
+            assert captured.out.strip() == "TIMEOUT"
+
+    def test_cli_check_ok(self, capsys):
+        from allegro.cycle_guard import main
+        with tempfile.TemporaryDirectory() as td:
+            path = Path(td) / "state.json"
+            start_cycle("t", path=path)
+            start_slice("s", path=path)
+            os.environ["ALLEGRO_CYCLE_STATE"] = str(path)
+            rc = main(["check"])
+            captured = capsys.readouterr()
+            assert rc == 0
+            assert captured.out.strip() == "OK"