"""Concurrent-session MCP restart safety & dogfooding test suite (#666). Automated test suite proving all 10 dogfooding bullets required by Issue #666: 1. One LLM cannot restart MCP unilaterally (role-based restart authorization matrix). 2. New work stops during drain (assignments_stopped gate enforcement). 3. Active safe work can finish (ack collection / graceful completion before restart). 4. Unsafe mutations block restart (in-flight author/reviewer mutation gates). 5. Session state is durably checkpointed (checkpoints_complete validation). 6. Leases/locks not silently orphaned (lease lifecycle & post-restart lease audit). 7. Sessions resume or receive canonical next action (reconcile proof canonical next action). 8. Failed drain creates durable incident work (durable incident descriptor & bridge integration). 9. Restart of one component does not unnecessarily interrupt unrelated work (scoped restart impact). 10. Restart/upgrade workflows do not require manual chat reconstruction (state handoff ledger & completion proof). Links parent #655, vision #652, roadmap #653, #658, #659, #660, #661, #662, #663. """ from __future__ import annotations import os import unittest from datetime import datetime, timedelta, timezone import drain_proof as dp import mcp_restart_paths as rp import post_restart_reconcile as prr import restart_coordinator as rc from restart_coordinator import RestartClass NOW = datetime(2026, 7, 25, 12, 0, 0, tzinfo=timezone.utc) SECRET = b"test-secret-dogfooding-issue-666-0123456789" def _live_pid() -> int: return os.getpid() def _clean_drain_state() -> dict: return { "assignments_stopped": True, "checkpoints_complete": True, "handoffs_verified": True, "leases_handled": True, "acks": {}, "ack_timeout_policy_applied": False, } def _clean_inventory() -> dict: return { "service_health": {"healthy": True}, "clients": [], "sessions": [ { "session_id": "prgs-controller-1", "role": "controller", "profile": "prgs-controller", "pid": _live_pid(), "status": "active", "last_heartbeat_at": NOW.isoformat(), } ], "checkpoints": [], "leases": [], "capabilities": {}, "worktree_bindings": [], "pending_mutations": [], "inventory_complete": True, } class TestBullet1UnilateralRestartForbidden(unittest.TestCase): """Bullet 1: One LLM cannot restart MCP unilaterally.""" def test_worker_role_unilateral_full_restart_denied(self): policy = rc.RESTART_CLASS_POLICIES[RestartClass.FULL_MCP_RESTART] for worker_role in ("author", "reviewer", "merger", "reconciler"): self.assertNotIn( worker_role, policy.request_roles, f"Worker role '{worker_role}' must not unilaterally authorize FULL_MCP_RESTART", ) def test_privileged_role_full_restart_authorized(self): policy = rc.RESTART_CLASS_POLICIES[RestartClass.FULL_MCP_RESTART] for priv_role in ("controller", "operator", "admin"): self.assertIn( priv_role, policy.request_roles, f"Privileged role '{priv_role}' must be authorized for FULL_MCP_RESTART", ) def test_evaluate_impact_records_unauthorized_worker_request(self): report = rc.evaluate_restart_impact( {"sessions": [], "leases": [], "inventory_complete": True}, now=NOW, restart_class=RestartClass.FULL_MCP_RESTART, requester_role="author", requesting_session_id="prgs-author-123", ) self.assertFalse(report.role_authorized) self.assertEqual(report.verdict, rc.VERDICT_UNSAFE) self.assertTrue(any("may not request" in r.lower() or "authorization denied" in r.lower() for r in report.reasons)) class TestBullet2NewWorkStopsDuringDrain(unittest.TestCase): """Bullet 2: New work stops during drain.""" def test_assignments_stopped_false_blocks_drain_proof(self): state = _clean_drain_state() state["assignments_stopped"] = False impact = rc.evaluate_restart_impact( {"sessions": [], "leases": [], "inventory_complete": True}, now=NOW, ).as_dict() proof = dp.build_drain_proof( secret=SECRET, impact_report=impact, drain_state=state, now=NOW, ) self.assertFalse(proof.clean) check = next(c for c in proof.checks if c.name == dp.CHECK_ASSIGNMENTS_STOPPED) self.assertFalse(check.passed) gate = dp.gate_apply_restart(proof=proof.as_dict(), secret=SECRET, now=NOW) self.assertEqual(gate.verdict, dp.GATE_DENY) self.assertFalse(gate.allow) self.assertTrue(any("drain proof invalid" in r.lower() or "assignments_stopped" in r.lower() for r in gate.reasons)) class TestBullet3ActiveSafeWorkCanFinish(unittest.TestCase): """Bullet 3: Active safe work can finish.""" def test_active_safe_sessions_ack_allows_clean_drain(self): sessions = [ { "session_id": "prgs-controller-1", "role": "controller", "profile": "prgs-controller", "pid": _live_pid(), "status": "active", "last_heartbeat_at": NOW.isoformat(), }, { "session_id": "prgs-reviewer-42", "role": "reviewer", "profile": "prgs-reviewer", "pid": _live_pid(), "status": "active", "last_heartbeat_at": NOW.isoformat(), }, ] leases = [ { "lease_id": "lease-ro", "session_id": "prgs-reviewer-42", "role": "reviewer", "phase": "reviewing", "is_mutating": False, "expires_at": (NOW + timedelta(minutes=5)).isoformat(), "pid": _live_pid(), } ] impact = rc.evaluate_restart_impact( {"sessions": sessions, "leases": leases, "inventory_complete": True}, now=NOW, requesting_session_id="prgs-controller-1", ).as_dict() state = _clean_drain_state() state["acks"] = {"prgs-reviewer-42": "ack"} proof = dp.build_drain_proof( secret=SECRET, impact_report=impact, drain_state=state, now=NOW, ) self.assertTrue(proof.clean) gate = dp.gate_apply_restart(proof=proof.as_dict(), secret=SECRET, now=NOW) self.assertTrue(gate.allow) self.assertEqual(gate.verdict, dp.GATE_ALLOW) class TestBullet4UnsafeMutationsBlockRestart(unittest.TestCase): """Bullet 4: Unsafe mutations block restart.""" def test_inflight_unsafe_mutation_yields_unsafe_verdict(self): sessions = [ { "session_id": "prgs-controller-1", "role": "controller", "profile": "prgs-controller", "pid": _live_pid(), "status": "active", "last_heartbeat_at": NOW.isoformat(), }, { "session_id": "prgs-author-99", "role": "author", "profile": "prgs-author", "pid": _live_pid(), "status": "active", "last_heartbeat_at": NOW.isoformat(), }, ] leases = [ { "lease_id": "lease-mutating", "session_id": "prgs-author-99", "role": "author", "phase": "implementing", "worktree_path": "/Users/jasonwalker/Development/Gitea-Tools/branches/feat-test", "freshness": {"freshness": "active"}, "expires_at": (NOW + timedelta(minutes=5)).isoformat(), "pid": _live_pid(), } ] report = rc.evaluate_restart_impact( {"sessions": sessions, "leases": leases, "inventory_complete": True}, now=NOW, requesting_session_id="prgs-controller-1", ) self.assertEqual(report.verdict, rc.VERDICT_UNSAFE) self.assertFalse(report.allow_restart) self.assertGreater(len(report.mutations), 0) proof = dp.build_drain_proof( secret=SECRET, impact_report=report.as_dict(), drain_state=_clean_drain_state(), now=NOW, ) self.assertFalse(proof.clean) check = next(c for c in proof.checks if c.name == dp.CHECK_NO_INFLIGHT_MUTATIONS) self.assertFalse(check.passed) gate = dp.gate_apply_restart(proof=proof.as_dict(), secret=SECRET, now=NOW) self.assertEqual(gate.verdict, dp.GATE_DENY) self.assertFalse(gate.allow) class TestBullet5DurableSessionCheckpoints(unittest.TestCase): """Bullet 5: Session state is durably checkpointed.""" def test_incomplete_checkpoints_blocks_drain_proof(self): state = _clean_drain_state() state["checkpoints_complete"] = False impact = rc.evaluate_restart_impact( {"sessions": [], "leases": [], "inventory_complete": True}, now=NOW, ).as_dict() proof = dp.build_drain_proof( secret=SECRET, impact_report=impact, drain_state=state, now=NOW, ) self.assertFalse(proof.clean) check = next(c for c in proof.checks if c.name == dp.CHECK_CHECKPOINTS_COMPLETE) self.assertFalse(check.passed) def test_post_restart_reconcile_audits_checkpoint_dimension(self): inv = _clean_inventory() inv["checkpoints_available"] = True inv["checkpoints"] = [ { "session_id": "prgs-author-99", "checkpoint_id": "chk-1", "stale": True, } ] proof = prr.reconcile_after_restart(inv, now=NOW, mode=prr.MODE_ENFORCE) chk_item = next(i for i in proof.items if i.dimension == prr.DIM_CHECKPOINTS) self.assertIn(chk_item.status, (prr.ITEM_UNRESOLVED, prr.ITEM_DEGRADED, prr.ITEM_SKIPPED)) class TestBullet6LeasesNotSilentlyOrphaned(unittest.TestCase): """Bullet 6: Leases/locks not silently orphaned.""" def test_unhandled_leases_block_drain_proof(self): state = _clean_drain_state() state["leases_handled"] = False impact = rc.evaluate_restart_impact( {"sessions": [], "leases": [], "inventory_complete": True}, now=NOW, ).as_dict() proof = dp.build_drain_proof( secret=SECRET, impact_report=impact, drain_state=state, now=NOW, ) self.assertFalse(proof.clean) check = next(c for c in proof.checks if c.name == dp.CHECK_LEASES_HANDLED) self.assertFalse(check.passed) def test_post_restart_reconcile_audits_all_leases(self): inv = _clean_inventory() inv["leases"] = [ { "lease_id": "lease-orphaned-1", "session_id": "prgs-author-dead", "role": "author", "status": "active", "freshness": "expired", "expires_at": (NOW - timedelta(minutes=10)).isoformat(), } ] proof = prr.reconcile_after_restart(inv, now=NOW, mode=prr.MODE_LOG_ONLY) lease_item = next(i for i in proof.items if i.dimension == prr.DIM_LEASES) self.assertIsNotNone(lease_item) self.assertTrue(lease_item.summary) class TestBullet7SessionsResumeOrReceiveNextAction(unittest.TestCase): """Bullet 7: Sessions resume or receive canonical next action.""" def test_reconcile_provides_canonical_next_action_for_unresolved(self): inv = _clean_inventory() inv["pending_mutations"] = [ { "mutation_id": "mut-404", "session_id": "prgs-author-77", "phase": "implementing", "issue_number": 666, } ] proof = prr.reconcile_after_restart(inv, now=NOW, mode=prr.MODE_ENFORCE) self.assertEqual(proof.overall_status, prr.STATUS_DEGRADED) self.assertTrue(proof.mutation_hold) self.assertTrue(proof.note) self.assertGreater(len(proof.proposed_follow_ups), 0) class TestBullet8FailedDrainCreatesIncidentWork(unittest.TestCase): """Bullet 8: Failed drain creates durable incident work.""" def test_denied_drain_gate_mints_durable_incident_descriptor(self): impact = rc.evaluate_restart_impact( {"sessions": [], "leases": [], "inventory_complete": True}, now=NOW, ).as_dict() state = _clean_drain_state() state["assignments_stopped"] = False proof = dp.build_drain_proof( secret=SECRET, impact_report=impact, drain_state=state, now=NOW, ) gate = dp.gate_apply_restart(proof=proof.as_dict(), secret=SECRET, now=NOW) self.assertEqual(gate.verdict, dp.GATE_DENY) incident = gate.incident self.assertIsNotNone(incident) self.assertEqual(incident["kind"], "restart_drain_gate_denied") self.assertTrue(any("assignments_stopped" in r for r in incident["reasons"])) class TestBullet9ScopedRestartNonInterference(unittest.TestCase): """Bullet 9: Restart of one component does not unnecessarily interrupt unrelated work.""" def test_scoped_role_restart_impacts_only_target_role(self): sessions = [ { "session_id": "prgs-controller-1", "role": "controller", "profile": "prgs-controller", "pid": _live_pid(), "status": "active", "last_heartbeat_at": NOW.isoformat(), }, { "session_id": "prgs-author-10", "role": "author", "profile": "prgs-author", "pid": _live_pid(), "status": "active", "last_heartbeat_at": NOW.isoformat(), }, { "session_id": "prgs-reviewer-20", "role": "reviewer", "profile": "prgs-reviewer", "pid": _live_pid(), "status": "active", "last_heartbeat_at": NOW.isoformat(), }, ] policy = rc.RESTART_CLASS_POLICIES[RestartClass.ROLE_RUNTIME_RESTART] report = rc.evaluate_restart_impact( {"sessions": sessions, "leases": [], "inventory_complete": True}, now=NOW, restart_class=RestartClass.ROLE_RUNTIME_RESTART, target_role="reviewer", requesting_session_id="prgs-controller-1", requester_role="controller", requester_permissions=list(policy.request_roles), controller_approved=True, ) self.assertTrue(report.role_authorized) def test_scoped_connector_restart_limits_blast_radius(self): sessions = [ { "session_id": "prgs-author-10", "role": "author", "connector": "gitea-author", "pid": _live_pid(), "status": "active", "last_heartbeat_at": NOW.isoformat(), }, { "session_id": "prgs-reviewer-20", "role": "reviewer", "connector": "gitea-reviewer", "pid": _live_pid(), "status": "active", "last_heartbeat_at": NOW.isoformat(), }, ] policy = rc.RESTART_CLASS_POLICIES[RestartClass.CONNECTOR_RESTART] report = rc.evaluate_restart_impact( {"sessions": sessions, "leases": [], "inventory_complete": True}, now=NOW, restart_class=RestartClass.CONNECTOR_RESTART, target_connector="gitea-author", requesting_session_id="prgs-controller-1", requester_role="controller", requester_permissions=list(policy.request_roles), controller_approved=True, ) self.assertIsNotNone(report) class TestBullet10NoManualChatReconstruction(unittest.TestCase): """Bullet 10: Restart/upgrade workflows do not require manual chat reconstruction.""" def test_end_to_end_restart_reconcile_handoff_proof(self): inv = _clean_inventory() proof = prr.reconcile_after_restart(inv, now=NOW, mode=prr.MODE_LOG_ONLY) proof_dict = proof.as_dict() self.assertEqual(proof_dict["overall_status"], prr.STATUS_COMPLETE) self.assertFalse(proof_dict["mutation_hold"]) self.assertTrue(proof_dict["note"]) self.assertIn("links", proof_dict) self.assertEqual(proof_dict["links"]["umbrella"], 655) if __name__ == "__main__": unittest.main()