diff --git a/tests/test_mcp_restart_governance.py b/tests/test_mcp_restart_governance.py new file mode 100644 index 0000000..06ff034 --- /dev/null +++ b/tests/test_mcp_restart_governance.py @@ -0,0 +1,478 @@ +"""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()