feat(mcp): enforce strict cross-project mutation boundaries (Closes #707)

docs/incidents/670-direct-master-commit-2fa97c26.md

@@ -0,0 +1,83 @@
+# Incident #670: bare direct-to-master commit `2fa97c26` (retroactive audit)
+
+Status: verified; disposition recommendation: **accept as-is, no revert** (final
+disposition owned by controller per issue #670).
+
+## Summary
+
+Commit `2fa97c26fbda555a1a83930ca5fdcea9d8e47b50`
+(`fix(mcp): load dotenv relative to project root`) landed on `prgs/master`
+as a single-parent commit with no PR wrapper and no review record, bypassing
+the sanctioned issue → branch → PR → review → merge workflow. It was
+discovered during the PR #654 post-merge audit. PR #654 itself merged
+cleanly via the Gitea API and did **not** introduce this commit.
+
+## Verification evidence (acceptance criteria 1–3)
+
+- **AC1 — present on `prgs/master`: yes.**
+  `git merge-base --is-ancestor 2fa97c26fbda555a1a83930ca5fdcea9d8e47b50 prgs/master` → true.
+- **AC2 — no PR or review record: confirmed.**
+  The commit is a single-parent, non-merge commit sitting directly on
+  first-parent master between the #629 merge (`5ab5fe85`) and the #654
+  merge (`ec903b0d`). A PR landing on master produces a merge commit (or a
+  PR-linked head); neither exists here. The controller audit at issue-create
+  time also found no PR wrapper and no review record for this SHA.
+- **AC3 — changed files and diff summary: confirmed.**
+  `gitea_auth.py | 5 +++--` (+3/−2). Single parent
+  `5ab5fe8583c07134d55dadf09381aecb67df246e`. The change moves
+  `PROJECT_ROOT` derivation above `load_dotenv()` and loads
+  `.env` relative to the project root instead of the process CWD.
+
+## AC4 — why no immediate revert
+
+- The dotenv fix is intentional and required for correct runtime behavior:
+  without it, `load_dotenv()` resolves `.env` against the process working
+  directory, which breaks MCP server launches whose CWD is not the project
+  root.
+- The change is small (+3/−2), self-contained in `gitea_auth.py`, and has
+  been running on master without incident since 2026-07-10.
+- Reverting would re-introduce a real bug to remove a provenance defect —
+  the wrong trade. Provenance is repaired retroactively by this document,
+  issue #670, and the hardening landed under #671.
+- If the controller later judges the change unsafe, a separate
+  revert/repair issue is the sanctioned path (issue #670, recommended
+  disposition option 4).
+
+## AC5 — workflow-hardening linkage
+
+Prevention already landed: **issue #671** (closed)
+*“Block direct pushes to stable branches from MCP workflow sessions”*,
+implemented by commit `5933d87647656643a67a50331c4c7b06ea751dad`
+(`feat(guard): block direct stable-branch pushes from MCP workflow sessions`).
+
+Shipped guardrails include:
+
+- `gitea_record_stable_branch_push_attempt` — classifies proposed commands
+  for direct stable-branch push intent (`git push <remote> master`,
+  refspecs, `HEAD:master`, `--force`, dry-run intent, `:master` delete),
+  plus root/control-checkout local commits not carried by an issue branch,
+  and writes a durable `stable_branch_contamination` marker.
+- `gitea_audit_stable_branch_contamination` — reconciler-only audit/clear
+  path; a contaminated worker session cannot self-clear.
+- Review/merge/close/completion mutations fail closed while a
+  contamination marker is active.
+
+## AC6 — PR #654 was not the source
+
+- `2fa97c26` is the **first parent** of the #654 merge commit
+  `ec903b0d619e7a27d24aed272a890f4e5d381411`; it predates the #654 merge.
+- First-parent history `5ab5fe8..ec903b0`:
+  `2fa97c2 fix(mcp): load dotenv relative to project root` followed by
+  `ec903b0 Merge pull request 'feat: lifecycle role/hazard labels ... (#603)' (#654)`.
+- The #654 merger audit confirmed `ec903b0d` was a valid Gitea-API merge,
+  the `git push prgs master` attempt during that run was a no-op, and the
+  net change `2fa97c2..ec903b0` contained only the reviewed #603
+  lifecycle-label files.
+- Conclusion: #654 merged reviewed content only; the unauthorized-path
+  defect is solely the earlier bare commit `2fa97c26`.
+
+## Explicit non-actions (unchanged by this audit)
+
+- No revert of `2fa97c26`.
+- No force-push or history rewrite.
+- No master mutation from the audit session.

gitea_mcp_server.py

@@ -3325,10 +3325,16 @@ def _effective_remote(remote: str) -> str:
     return remote
 
 
-def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
+def _resolve(
+    remote: str,
+    host: str | None,
+    org: str | None,
+    repo: str | None,
+    for_mutation: bool = False,
+):
     """Resolve remote + overrides to (host, org, repo).
 
-    #714 / #530: when the caller omits org and/or repo, prefer the
+    #714 / #530 / #707: when the caller omits org and/or repo, prefer the
     workspace-aligned git remote over ``REMOTES`` defaults (e.g. bare
     ``remote=prgs`` must not force ``Timesheet`` when the checkout is
     ``Gitea-Tools``). Explicit caller org/repo always win and are validated
@@ -3414,6 +3420,7 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
         # Workspace-filled sides are intentional alignment for #530.
         org_explicit=org_explicit or filled_org,
         repo_explicit=repo_explicit or filled_repo,
+        for_mutation=for_mutation,
     )
     return (resolved_host, resolved_org, resolved_repo)
 
@@ -3475,8 +3482,9 @@ def _enforce_remote_repo_guard(
     *,
     org_explicit: bool,
     repo_explicit: bool,
+    for_mutation: bool = False,
 ) -> None:
-    """Fail closed on a remote/repo mismatch vs. the local git remote (#530).
+    """Fail closed on a remote/repo mismatch vs. the local git remote (#530/#707).
 
     Best-effort: bypassed under pytest unless ``GITEA_FORCE_REMOTE_REPO_CHECK`` is
     set, so the unit suite (which calls tools with bare remotes against mocked APIs)
@@ -3488,6 +3496,12 @@ def _enforce_remote_repo_guard(
     ):
         return
     local_remote_url = _local_git_remote_url(remote)
+    primary_org = None
+    primary_repo = None
+    ctx = session_ctx.get_session_context()
+    if ctx:
+        primary_org = ctx.get("org")
+        primary_repo = ctx.get("repository")
     assessment = remote_repo_guard.assess_remote_repo_match(
         remote=remote,
         resolved_org=resolved_org,
@@ -3495,6 +3509,9 @@ def _enforce_remote_repo_guard(
         local_remote_url=local_remote_url,
         org_explicit=org_explicit,
         repo_explicit=repo_explicit,
+        for_mutation=for_mutation,
+        primary_org=primary_org,
+        primary_repo=primary_repo,
     )
     if assessment["block"]:
         raise RuntimeError(remote_repo_guard.format_remote_repo_guard_error(assessment))
@@ -4063,7 +4080,7 @@ def gitea_lock_issue(
     resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
         worktree_path, _canonical_local_git_root()
     )
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     existing_issue_lock = _load_existing_issue_lock(
         remote=remote, org=o, repo=r, issue_number=issue_number
     )
@@ -5239,7 +5256,7 @@ def gitea_create_pr(
             org=org,
             repo=repo,
         )
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
 
     # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
     lock_data = _resolve_issue_lock_for_pr(remote=remote, org=o, repo=r, head=head)
@@ -9560,7 +9577,7 @@ def gitea_edit_pr(
                 required_permission="gitea.pr.close",
             )
 
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
 
@@ -9844,7 +9861,7 @@ def gitea_commit_files(
     verify_preflight_purity(remote=remote, worktree_path=worktree_path, task="commit_files", org=org, repo=repo)
     processed_files, source_proofs = _prepare_commit_payload_files(files)
 
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/contents"
 
@@ -9980,7 +9997,7 @@ def gitea_publish_unpublished_issue_branch(
         repo=repo,
     )
 
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     git_remote = (git_remote_name or remote or "").strip()
 
     existing_lock = issue_lock_store.load_issue_lock(
@@ -10183,7 +10200,7 @@ def gitea_bootstrap_author_issue_worktree(
         repo=repo,
     )
 
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     canonical_root = _canonical_local_git_root()
 
     import author_issue_bootstrap
@@ -12131,7 +12148,7 @@ def gitea_reconcile_merged_cleanups(
                 "blocker_kind": "invalid_pr_number",
             }
 
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
 
@@ -12521,7 +12538,7 @@ def gitea_assess_already_landed_reconciliation(
             "permission_report": _permission_block_report("gitea.read"),
         }
 
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     auth = _auth(h)
     pr = api_request("GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth)
 

remote_repo_guard.py

@@ -54,20 +54,62 @@ def assess_remote_repo_match(
     local_remote_url: str | None,
     org_explicit: bool,
     repo_explicit: bool,
+    for_mutation: bool = False,
+    primary_org: str | None = None,
+    primary_repo: str | None = None,
 ) -> dict:
     """Fail closed when the resolved org/repo disagrees with the local git remote.
 
-    The guard is intentionally conservative:
+    The guard enforces two protection levels:
 
-    * When the caller passed both ``org`` and ``repo`` explicitly, their intent is
+    1. Cross-Project Mutation Boundary (#707):
-      authoritative and the guard never blocks.
+       When ``for_mutation`` is True (codebase mutation operation: branch, commit, PR,
-    * When the local git remote URL is unavailable (``None``/empty), corroboration
+       merge, branch deletion), the target repository (``resolved_org/resolved_repo``)
-      is impossible, so the guard does not block (best-effort only).
+       MUST match the primary authorized project context (``primary_org/primary_repo``
-    * Otherwise, the resolved ``org/repo`` slug must appear in the local remote URL
+       or parsed from ``local_remote_url``). Any attempt to mutate a different project
-      (case-insensitive); if it does not, the guard blocks.
+       fails closed, even if explicit org/repo were passed. Metadata operations (such
+       as creating issues or commenting on issues) across projects remain allowed.
+
+    2. Workspace Mismatch Guard (#530):
+       When ``for_mutation`` is False (or targets match), bare remotes must resolve
+       to an org/repo slug present in the local git remote URL. When explicit org/repo
+       are supplied for non-mutation operations, the caller's intent is authoritative.
     """
     reasons: list[str] = []
 
+    eff_primary_org = primary_org
+    eff_primary_repo = primary_repo
+    if not eff_primary_org or not eff_primary_repo:
+        parsed_primary = parse_org_repo_from_remote_url(local_remote_url)
+        if parsed_primary:
+            eff_primary_org = eff_primary_org or parsed_primary[0]
+            eff_primary_repo = eff_primary_repo or parsed_primary[1]
+
+    # #707: Enforce cross-project codebase mutation boundary if for_mutation is True
+    if for_mutation and eff_primary_org and eff_primary_repo:
+        if (
+            resolved_org.lower() != eff_primary_org.lower()
+            or resolved_repo.lower() != eff_primary_repo.lower()
+        ):
+            reasons.append(
+                f"Cross-project mutation guard (#707): Attempted codebase mutation targeting "
+                f"'{resolved_org}/{resolved_repo}' outside of primary authorized project "
+                f"context '{eff_primary_org}/{eff_primary_repo}'"
+            )
+            return {
+                "proven": False,
+                "block": True,
+                "cross_project_mutation_block": True,
+                "reasons": reasons,
+                "remote": remote,
+                "resolved_org": resolved_org,
+                "resolved_repo": resolved_repo,
+                "primary_org": eff_primary_org,
+                "primary_repo": eff_primary_repo,
+                "local_remote_url": local_remote_url,
+                "remediation": f"Cross-project codebase work is forbidden. Create an issue in the target repository ('{resolved_org}/{resolved_repo}') instead.",
+            }
+
     if org_explicit and repo_explicit:
         return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)
 
@@ -88,6 +130,16 @@ def assess_remote_repo_match(
 
 def format_remote_repo_guard_error(assessment: dict) -> str:
     """Single RuntimeError message for the MCP resolver gate."""
+    if assessment.get("cross_project_mutation_block"):
+        resolved = f"{assessment.get('resolved_org')}/{assessment.get('resolved_repo')}"
+        primary = f"{assessment.get('primary_org')}/{assessment.get('primary_repo')}"
+        return (
+            f"Cross-project mutation guard (#707): Attempted codebase mutation targeting '{resolved}' "
+            f"outside of primary authorized project context '{primary}'. "
+            f"Cross-project codebase work (creating branches, committing files, creating PRs) is forbidden; "
+            f"create an issue in the target repository ('{resolved}') instead."
+        )
+
     reasons = "; ".join(
         assessment.get("reasons") or ["remote/repo resolution mismatch"]
     )
@@ -120,3 +172,4 @@ def _assessment(
         "local_remote_url": local_remote_url,
         "remediation": REMEDIATION,
     }
+

tests/test_cross_project_mutation_boundary.py

@@ -0,0 +1,94 @@
+import os
+import sys
+import unittest
+from unittest import mock
+
+import remote_repo_guard
+import gitea_mcp_server as server
+from session_context_binding import _reset_session_context_for_testing, bind_session_context
+
+
+LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+
+
+class TestCrossProjectMutationBoundary(unittest.TestCase):
+    def setUp(self):
+        os.environ["PYTEST_CURRENT_TEST"] = "test"
+        _reset_session_context_for_testing()
+
+    def tearDown(self):
+        _reset_session_context_for_testing()
+        os.environ.pop("PYTEST_CURRENT_TEST", None)
+
+    def test_cross_project_codebase_mutation_blocked(self):
+        """Codebase mutations targeting another project must fail closed (#707)."""
+        assessment = remote_repo_guard.assess_remote_repo_match(
+            remote="prgs",
+            resolved_org="Other-Org",
+            resolved_repo="Other-Repo",
+            local_remote_url=LOCAL_GITEA_TOOLS_URL,
+            org_explicit=True,
+            repo_explicit=True,
+            for_mutation=True,
+        )
+        self.assertTrue(assessment["block"])
+        self.assertTrue(assessment.get("cross_project_mutation_block"))
+        self.assertEqual(assessment["primary_org"], "Scaled-Tech-Consulting")
+        self.assertEqual(assessment["primary_repo"], "Gitea-Tools")
+
+        err_msg = remote_repo_guard.format_remote_repo_guard_error(assessment)
+        self.assertIn("Cross-project mutation guard (#707)", err_msg)
+        self.assertIn("Attempted codebase mutation targeting 'Other-Org/Other-Repo'", err_msg)
+        self.assertIn("outside of primary authorized project context 'Scaled-Tech-Consulting/Gitea-Tools'", err_msg)
+        self.assertIn("create an issue in the target repository ('Other-Org/Other-Repo') instead", err_msg)
+
+    def test_same_project_codebase_mutation_allowed(self):
+        """Codebase mutations targeting the primary project context are allowed."""
+        assessment = remote_repo_guard.assess_remote_repo_match(
+            remote="prgs",
+            resolved_org="Scaled-Tech-Consulting",
+            resolved_repo="Gitea-Tools",
+            local_remote_url=LOCAL_GITEA_TOOLS_URL,
+            org_explicit=True,
+            repo_explicit=True,
+            for_mutation=True,
+        )
+        self.assertFalse(assessment["block"])
+        self.assertFalse(assessment.get("cross_project_mutation_block", False))
+
+    def test_cross_project_metadata_operation_allowed(self):
+        """Metadata operations (issue creation, comments) across project boundaries are allowed."""
+        assessment = remote_repo_guard.assess_remote_repo_match(
+            remote="prgs",
+            resolved_org="Other-Org",
+            resolved_repo="Other-Repo",
+            local_remote_url=LOCAL_GITEA_TOOLS_URL,
+            org_explicit=True,
+            repo_explicit=True,
+            for_mutation=False,
+        )
+        self.assertTrue(assessment["proven"])
+        self.assertFalse(assessment["block"])
+
+    @mock.patch.dict(os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"})
+    @mock.patch("gitea_mcp_server._local_git_remote_url", return_value=LOCAL_GITEA_TOOLS_URL)
+    def test_mcp_server_resolve_cross_project_mutation_fails_closed(self, mock_remote):
+        """_resolve with for_mutation=True blocks cross-project targets."""
+        with self.assertRaises(RuntimeError) as ctx:
+            server._resolve("prgs", None, "Other-Org", "Other-Repo", for_mutation=True)
+        
+        err = str(ctx.exception)
+        self.assertIn("Cross-project mutation guard (#707)", err)
+        self.assertIn("create an issue in the target repository", err)
+
+    @mock.patch.dict(os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"})
+    @mock.patch("gitea_mcp_server._local_git_remote_url", return_value=LOCAL_GITEA_TOOLS_URL)
+    def test_mcp_server_resolve_cross_project_metadata_succeeds(self, mock_remote):
+        """_resolve with for_mutation=False allows explicit cross-project target."""
+        host, org, repo = server._resolve("prgs", None, "Other-Org", "Other-Repo", for_mutation=False)
+        self.assertEqual(org, "Other-Org")
+        self.assertEqual(repo, "Other-Repo")
+
+
+if __name__ == "__main__":
+    unittest.main()

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()