fix(auth): prevent repository .env files from injecting workspace bindings (Closes #704)

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_auth.py

@@ -22,8 +22,62 @@ import gitea_config
 
 PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
 
-# Load standard .env if present
+# Reserved runtime-control / workspace-binding environment variables (#704).
-load_dotenv(os.path.join(PROJECT_ROOT, ".env"))
+# Repository .env files MUST NOT populate or override any of these keys.
+RESERVED_WORKTREE_ENV_KEYS: frozenset[str] = frozenset({
+    "GITEA_ACTIVE_WORKTREE",
+    "GITEA_AUTHOR_WORKTREE",
+    "GITEA_REVIEWER_WORKTREE",
+    "GITEA_MERGER_WORKTREE",
+    "GITEA_RECONCILER_WORKTREE",
+})
+
+
+def is_reserved_worktree_env_key(key: str | None) -> bool:
+    """Return True if *key* is a reserved runtime workspace binding variable (#704)."""
+    if not key:
+        return False
+    k = str(key).upper().strip()
+    return k in RESERVED_WORKTREE_ENV_KEYS or (k.startswith("GITEA_") and k.endswith("_WORKTREE"))
+
+
+def load_env_file_sanitized(
+    env_path: str,
+    *,
+    target_env: dict | os._Environ | None = None,
+) -> list[str]:
+    """Load a .env file without populating or overriding reserved workspace keys (#704).
+
+    Pre-existing process environment values retain their precedence. Reserved
+    runtime-control keys found in repository files are ignored (without logging
+    their values). Returns a list of sanitized rejection reasons.
+    """
+    if target_env is None:
+        target_env = os.environ
+    if not os.path.exists(env_path) or os.path.isdir(env_path):
+        return []
+
+    rejection_reasons: list[str] = []
+    try:
+        file_vals = dotenv_values(env_path)
+        for key, val in file_vals.items():
+            if not key or val is None:
+                continue
+            if is_reserved_worktree_env_key(key):
+                filename = os.path.basename(env_path)
+                rejection_reasons.append(
+                    f"Ignored reserved runtime workspace key '{key}' from repository {filename}"
+                )
+                continue
+            if key not in target_env:
+                target_env[key] = val
+    except Exception:
+        pass
+    return rejection_reasons
+
+
+# Load standard .env if present (sanitized to prevent repo workspace binding contamination #704)
+load_env_file_sanitized(os.path.join(PROJECT_ROOT, ".env"))
 
 # Dictionary to store configurations parsed dynamically from .env.* files
 DYNAMIC_CONFIGS = {}
@@ -37,9 +91,13 @@ for env_path in glob.glob(os.path.join(PROJECT_ROOT, ".env*")):
         continue
     try:
         config_vals = dotenv_values(env_path)
-        site = config_vals.get("GITEA_SITE") or config_vals.get("GITEA_HOST")
+        # Filter out reserved workspace keys from dynamic configs (#704)
+        sanitized_config = {
+            k: v for k, v in config_vals.items() if not is_reserved_worktree_env_key(k)
+        }
+        site = sanitized_config.get("GITEA_SITE") or sanitized_config.get("GITEA_HOST")
         if site:
-            DYNAMIC_CONFIGS[site.lower().strip()] = config_vals
+            DYNAMIC_CONFIGS[site.lower().strip()] = sanitized_config
     except Exception:
         pass
 

tests/test_issue_704_prevent_env_workspace_bindings.py

@@ -0,0 +1,166 @@
+"""Tests for Issue #704: Preventing repository .env files from injecting workspace bindings.
+
+Acceptance Criteria (#704):
+1. Repository .env loading cannot populate or override GITEA_ACTIVE_WORKTREE or any role-specific GITEA_*_WORKTREE runtime-binding variable.
+2. Runtime workspace bindings are accepted only from sanctioned managed-launch/session mechanisms.
+3. Pre-existing sanctioned process environment values retain their intended precedence.
+4. Importing gitea_auth or related modules does not mutate workspace-binding state from repository files.
+5. Reserved runtime-control keys found in .env are ignored or rejected with a sanitized actionable reason; their values are never logged.
+6. The protection applies consistently to author, reviewer, merger, and reconciler namespaces.
+7. Comprehensive test coverage for stale worktree, missing worktree, task-specific, role-specific, launcher binding, repeated imports, namespace isolation, precedence, and absence of secret leakage.
+8. Dirty-state and workspace-preflight gates cannot be bypassed by an injected missing-path binding.
+9. No environment, dotenv, offline-import, or caller-controlled path can forge native transport or mutation provenance.
+10. Cross-linked with #702, PR #703, #510.
+11. Required immediate follow-up to Issue #702 / PR #703.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+import importlib
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import gitea_auth
+from gitea_auth import is_reserved_worktree_env_key, load_env_file_sanitized
+
+
+class TestIssue704PreventEnvWorkspaceBindings(unittest.TestCase):
+    """Test suite verifying .env workspace-binding injection prevention (#704)."""
+
+    def setUp(self):
+        self.tmpdir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmpdir.cleanup)
+        self.env_dir = Path(self.tmpdir.name)
+
+    def test_is_reserved_worktree_env_key(self):
+        """Verify key classification for all role namespaces (#704 AC6)."""
+        reserved_keys = [
+            "GITEA_ACTIVE_WORKTREE",
+            "GITEA_AUTHOR_WORKTREE",
+            "GITEA_REVIEWER_WORKTREE",
+            "GITEA_MERGER_WORKTREE",
+            "GITEA_RECONCILER_WORKTREE",
+            "gitea_active_worktree",
+            "GITEA_CUSTOM_ROLE_WORKTREE",
+        ]
+        for key in reserved_keys:
+            self.assertTrue(
+                is_reserved_worktree_env_key(key),
+                f"Expected {key} to be recognized as a reserved worktree key",
+            )
+
+        unreserved_keys = [
+            "GITEA_USER",
+            "GITEA_PASS",
+            "GITEA_TOKEN",
+            "GITEA_HOST",
+            "PATH",
+        ]
+        for key in unreserved_keys:
+            self.assertFalse(
+                is_reserved_worktree_env_key(key),
+                f"Expected {key} to NOT be recognized as a reserved worktree key",
+            )
+
+    def test_load_env_file_sanitized_ignores_reserved_keys(self):
+        """Verify .env loading ignores GITEA_ACTIVE_WORKTREE and role-specific keys (#704 AC1)."""
+        env_file = self.env_dir / ".env"
+        stale_path = "/tmp/stale-worktree-path-1234"
+        env_file.write_text(
+            f"GITEA_USER=testuser\n"
+            f"GITEA_ACTIVE_WORKTREE={stale_path}\n"
+            f"GITEA_AUTHOR_WORKTREE={stale_path}\n"
+            f"GITEA_REVIEWER_WORKTREE={stale_path}\n"
+            f"GITEA_MERGER_WORKTREE={stale_path}\n"
+            f"GITEA_RECONCILER_WORKTREE={stale_path}\n"
+        )
+
+        test_env = {}
+        reasons = load_env_file_sanitized(str(env_file), target_env=test_env)
+
+        # Unreserved key loaded
+        self.assertEqual(test_env.get("GITEA_USER"), "testuser")
+
+        # Reserved keys ignored
+        self.assertNotIn("GITEA_ACTIVE_WORKTREE", test_env)
+        self.assertNotIn("GITEA_AUTHOR_WORKTREE", test_env)
+        self.assertNotIn("GITEA_REVIEWER_WORKTREE", test_env)
+        self.assertNotIn("GITEA_MERGER_WORKTREE", test_env)
+        self.assertNotIn("GITEA_RECONCILER_WORKTREE", test_env)
+
+        # Rejection reasons populated without leaking the secret value (#704 AC5)
+        self.assertTrue(len(reasons) >= 5)
+        for r in reasons:
+            self.assertNotIn(stale_path, r, "Secret/path value must not leak into rejection reason")
+
+    def test_preexisting_sanctioned_launcher_env_retained(self):
+        """Sanctioned launcher values in process env are retained (#704 AC2, AC3)."""
+        sanctioned_path = "/tmp/sanctioned-launcher-worktree"
+        test_env = {"GITEA_ACTIVE_WORKTREE": sanctioned_path}
+
+        env_file = self.env_dir / ".env"
+        env_file.write_text("GITEA_ACTIVE_WORKTREE=/tmp/injected-repo-worktree\n")
+
+        load_env_file_sanitized(str(env_file), target_env=test_env)
+
+        # Pre-existing value retained, not overwritten by .env
+        self.assertEqual(test_env.get("GITEA_ACTIVE_WORKTREE"), sanctioned_path)
+
+    def test_stale_or_missing_worktree_in_env_ignored(self):
+        """Stale or non-existent worktree path in .env file is ignored (#704 AC7)."""
+        nonexistent_path = "/nonexistent/branches/stale-issue-999"
+        env_file = self.env_dir / ".env"
+        env_file.write_text(f"GITEA_ACTIVE_WORKTREE={nonexistent_path}\n")
+
+        test_env = {}
+        load_env_file_sanitized(str(env_file), target_env=test_env)
+
+        self.assertNotIn("GITEA_ACTIVE_WORKTREE", test_env)
+
+    def test_repeated_module_import_does_not_mutate_workspace_env(self):
+        """Repeated imports of gitea_auth leave os.environ un-contaminated (#704 AC4, AC7)."""
+        # Ensure no active worktree env exists initially
+        original_val = os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
+        try:
+            importlib.reload(gitea_auth)
+            self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
+
+            importlib.reload(gitea_auth)
+            self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
+        finally:
+            if original_val is not None:
+                os.environ["GITEA_ACTIVE_WORKTREE"] = original_val
+
+    def test_namespace_isolation_all_roles_protected(self):
+        """Verify protection across author, reviewer, merger, reconciler (#704 AC6)."""
+        env_file = self.env_dir / ".env"
+        env_file.write_text(
+            "GITEA_AUTHOR_WORKTREE=/bad/author\n"
+            "GITEA_REVIEWER_WORKTREE=/bad/reviewer\n"
+            "GITEA_MERGER_WORKTREE=/bad/merger\n"
+            "GITEA_RECONCILER_WORKTREE=/bad/reconciler\n"
+        )
+        test_env = {}
+        load_env_file_sanitized(str(env_file), target_env=test_env)
+
+        self.assertEqual(test_env, {})
+
+    def test_absence_of_secret_leakage(self):
+        """Rejection reasons contain key names but never secret path values (#704 AC5)."""
+        sensitive_path = "/Users/secret/path/private_repo"
+        env_file = self.env_dir / ".env"
+        env_file.write_text(f"GITEA_ACTIVE_WORKTREE={sensitive_path}\n")
+
+        test_env = {}
+        reasons = load_env_file_sanitized(str(env_file), target_env=test_env)
+        for reason in reasons:
+            self.assertNotIn(sensitive_path, reason)
+
+
+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()