Merge branch 'master' into fix/issue-704-prevent-env-workspace-bindings

tests/conftest.py

@@ -44,6 +44,8 @@ def _reset_mutation_authority(monkeypatch):
     ]:
         monkeypatch.delenv(env_key, raising=False)
 
+    monkeypatch.setenv("GITEA_CLIENT_MANAGED", "1")
+
     # Isolate durable session-state files so tests never share host cache (#559).
     import tempfile
 

tests/test_commit_files_gate.py

@@ -35,7 +35,7 @@ CONFIG = {
             ],
             "forbidden_operations": [],
             "execution_profile": "full-author",
-        "allowed_repositories": ["Example-Org/Example-Repo"],
+        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
         },
         "reviewer-no-commit": {
             "enabled": True,
@@ -50,7 +50,7 @@ CONFIG = {
                 "gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
             ],
             "execution_profile": "reviewer-no-commit",
-        "allowed_repositories": ["Example-Org/Example-Repo"],
+        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
         },
     },
     "rules": {"allow_runtime_switching": False},

tests/test_config_menu.py

@@ -175,7 +175,7 @@ class TestLauncherSnippets(unittest.TestCase):
     def test_only_safe_keys_no_secrets(self):
         entry = gitea_config.launcher_entry("prgs", "/cfg/profiles.json")["gitea-tools"]
         self.assertEqual(set(entry), {"command", "args", "env"})
-        self.assertEqual(set(entry["env"]), {"GITEA_MCP_CONFIG", "GITEA_MCP_PROFILE"})
+        self.assertEqual(set(entry["env"]), {"GITEA_MCP_CONFIG", "GITEA_MCP_PROFILE", "GITEA_CLIENT_MANAGED"})
         self.assertEqual(entry["env"]["GITEA_MCP_PROFILE"], "prgs")
         blob = json.dumps(entry).lower()
         for word in ("token", "password", "secret"):

tests/test_issue_678_mcp_client_reconnect.py

@@ -0,0 +1,323 @@
+"""Tests for sanctioned Codex MCP reconnect request surface (#678)."""
+
+from __future__ import annotations
+
+import os
+import unittest
+from unittest import mock
+
+import mcp_client_reconnect as mcr
+
+
+class NormalizeReasonTests(unittest.TestCase):
+    def test_stale_runtime_aliases(self):
+        self.assertEqual(mcr.normalize_reason("stale-runtime"), mcr.REASON_STALE_RUNTIME)
+        self.assertEqual(mcr.normalize_reason("stale_runtime"), mcr.REASON_STALE_RUNTIME)
+        self.assertEqual(mcr.normalize_reason("STALE"), mcr.REASON_STALE_RUNTIME)
+
+    def test_transport_eof_aliases(self):
+        self.assertEqual(mcr.normalize_reason("transport_eof"), mcr.REASON_TRANSPORT_EOF)
+        self.assertEqual(mcr.normalize_reason("EOF"), mcr.REASON_TRANSPORT_EOF)
+        self.assertEqual(
+            mcr.normalize_reason("client_is_closing"), mcr.REASON_TRANSPORT_EOF
+        )
+
+    def test_missing_namespace(self):
+        self.assertEqual(
+            mcr.normalize_reason("missing_namespace"), mcr.REASON_MISSING_NAMESPACE
+        )
+
+    def test_empty_is_unspecified(self):
+        self.assertEqual(mcr.normalize_reason(None), mcr.REASON_UNSPECIFIED)
+        self.assertEqual(mcr.normalize_reason(""), mcr.REASON_UNSPECIFIED)
+
+
+class BoundaryClassificationTests(unittest.TestCase):
+    def test_clean_when_shas_match(self):
+        self.assertEqual(
+            mcr.classify_boundary_status(
+                startup_sha="abc", current_master_sha="abc"
+            ),
+            mcr.BOUNDARY_CLEAN,
+        )
+
+    def test_mismatch_when_shas_differ(self):
+        self.assertEqual(
+            mcr.classify_boundary_status(
+                startup_sha="aaa", current_master_sha="bbb"
+            ),
+            mcr.BOUNDARY_MISMATCH,
+        )
+
+    def test_stale_when_live_stale(self):
+        self.assertEqual(
+            mcr.classify_boundary_status(
+                startup_sha="aaa",
+                current_master_sha="aaa",
+                live_stale=True,
+            ),
+            mcr.BOUNDARY_STALE,
+        )
+
+
+class BuildReconnectRequestTests(unittest.TestCase):
+    def test_stale_runtime_returns_typed_blocker_with_codex_steps(self):
+        result = mcr.build_reconnect_request(
+            namespace="gitea-author",
+            profile="prgs-author",
+            pid=1234,
+            session_id="sess-1",
+            startup_sha="aaa111",
+            current_master_sha="bbb222",
+            reason="stale-runtime",
+            client="codex",
+            restart_required=True,
+            stop_required=True,
+        )
+        self.assertTrue(result["success"])
+        self.assertTrue(result["read_only"])
+        self.assertFalse(result["reconnect_performed"])
+        self.assertFalse(result["mutation_performed"])
+        self.assertTrue(result["reconnect_needed"])
+        self.assertEqual(result["namespace"], "gitea-author")
+        self.assertEqual(result["profile"], "prgs-author")
+        self.assertEqual(result["pid"], 1234)
+        self.assertEqual(result["session_id"], "sess-1")
+        self.assertEqual(result["startup_sha"], "aaa111")
+        self.assertEqual(result["current_master_sha"], "bbb222")
+        self.assertEqual(result["boundary_status"], mcr.BOUNDARY_MISMATCH)
+        self.assertEqual(result["blocker_kind"], mcr.BLOCKER_OPERATOR_RECONNECT)
+        self.assertIsNotNone(result["typed_blocker"])
+        blocker = result["typed_blocker"]
+        self.assertEqual(blocker["namespaces"], ["gitea-author"])
+        self.assertEqual(blocker["why_reconnect_required"], mcr.REASON_STALE_RUNTIME)
+        self.assertTrue(any("Codex" in s or "Reload" in s for s in blocker["operator_ui_steps"]))
+        self.assertIn("pkill", " ".join(result["forbidden_recovery_paths"]).lower())
+        self.assertTrue(
+            mcr.reasons_never_suggest_forbidden(result["exact_safe_next_action"] or "")
+        )
+        # Must not recommend forbidden recovery.
+        for step in blocker["operator_ui_steps"]:
+            self.assertTrue(mcr.reasons_never_suggest_forbidden(step), step)
+
+    def test_transport_eof_typed_blocker(self):
+        result = mcr.build_reconnect_request(
+            namespace="gitea-reviewer",
+            reason="transport_eof",
+            client="claude_code",
+        )
+        self.assertTrue(result["reconnect_needed"])
+        self.assertEqual(result["reason"], mcr.REASON_TRANSPORT_EOF)
+        self.assertEqual(result["client"], "claude_code")
+        steps = " ".join(result["operator_ui_steps"]).lower()
+        self.assertIn("/mcp", steps)
+
+    def test_missing_namespace_typed_blocker(self):
+        result = mcr.build_reconnect_request(
+            namespace="gitea-merger",
+            reason="missing_namespace",
+            client="codex",
+        )
+        self.assertTrue(result["reconnect_needed"])
+        self.assertEqual(result["reason"], mcr.REASON_MISSING_NAMESPACE)
+        self.assertEqual(
+            result["typed_blocker"]["blocker_kind"], mcr.BLOCKER_OPERATOR_RECONNECT
+        )
+
+    def test_healthy_not_required(self):
+        result = mcr.build_reconnect_request(
+            namespace="gitea-tools",
+            startup_sha="deadbeef",
+            current_master_sha="deadbeef",
+            reason="not_required",
+            client="codex",
+            in_parity=True,
+            restart_required=False,
+            stop_required=False,
+        )
+        self.assertFalse(result["reconnect_needed"])
+        self.assertEqual(result["blocker_kind"], mcr.BLOCKER_NONE)
+        self.assertIsNone(result["typed_blocker"])
+        self.assertFalse(result["stop_required"])
+        self.assertFalse(result["restart_required"])
+        self.assertIn("not required", (result["exact_safe_next_action"] or "").lower())
+
+    def test_successful_reconnect_report_fields_present(self):
+        """AC2: reconnect result reports required fields (even when needed)."""
+        result = mcr.build_reconnect_request(
+            namespace="gitea-controller",
+            profile="prgs-controller",
+            pid=99,
+            session_id="sid",
+            startup_sha="s" * 40,
+            current_master_sha="c" * 40,
+            reason="stale-runtime",
+        )
+        for key in (
+            "namespace",
+            "profile",
+            "pid",
+            "session_id",
+            "startup_sha",
+            "current_master_sha",
+            "boundary_status",
+        ):
+            self.assertIn(key, result)
+            self.assertIsNotNone(result[key], key)
+
+
+class ToolSurfaceTests(unittest.TestCase):
+    """Exercise gitea_request_mcp_reconnect with a stubbed server context."""
+
+    def test_tool_is_registered_and_side_effect_free(self):
+        import gitea_mcp_server as srv
+
+        self.assertTrue(hasattr(srv, "gitea_request_mcp_reconnect"))
+        with mock.patch.object(srv, "_profile_operation_gate", return_value=None):
+            with mock.patch.object(
+                srv,
+                "get_profile",
+                return_value={
+                    "profile_name": "prgs-author",
+                    "role_kind": "author",
+                    "role": "author",
+                },
+            ):
+                with mock.patch.object(
+                    srv,
+                    "_current_master_parity",
+                    return_value={
+                        "startup_head": "a" * 40,
+                        "current_head": "a" * 40,
+                        "daemon_start_head": "a" * 40,
+                        "local_head": "a" * 40,
+                        "in_parity": True,
+                        "stale": False,
+                        "restart_required": False,
+                        "determinable": True,
+                        "live_stale": False,
+                        "live_known": True,
+                        "reasons": [],
+                    },
+                ):
+                    with mock.patch.object(
+                        srv.master_parity_gate,
+                        "format_parity",
+                        return_value="in parity",
+                    ):
+                        with mock.patch.object(
+                            srv.role_namespace_gate,
+                            "infer_mcp_namespace",
+                            return_value="gitea-author",
+                        ):
+                            with mock.patch.object(
+                                srv.session_ctx,
+                                "mutation_context_audit_fields",
+                                return_value={"session_profile": "prgs-author"},
+                            ):
+                                result = srv.gitea_request_mcp_reconnect(
+                                    namespace="gitea-author",
+                                    reason="not_required",
+                                    client="codex",
+                                    remote="prgs",
+                                )
+        self.assertTrue(result.get("success"))
+        self.assertFalse(result.get("reconnect_performed"))
+        self.assertFalse(result.get("mutation_performed"))
+        self.assertEqual(result.get("namespace"), "gitea-author")
+        self.assertEqual(result.get("profile"), "prgs-author")
+        self.assertEqual(result.get("pid"), os.getpid())
+        self.assertIn("startup_sha", result)
+        self.assertIn("current_master_sha", result)
+        self.assertIn("boundary_status", result)
+        self.assertTrue(
+            mcr.reasons_never_suggest_forbidden(
+                result.get("exact_safe_next_action") or ""
+            )
+        )
+
+    def test_tool_stale_returns_typed_blocker(self):
+        import gitea_mcp_server as srv
+
+        with mock.patch.object(srv, "_profile_operation_gate", return_value=None):
+            with mock.patch.object(
+                srv,
+                "get_profile",
+                return_value={
+                    "profile_name": "prgs-reconciler",
+                    "role_kind": "reconciler",
+                    "role": "reconciler",
+                },
+            ):
+                with mock.patch.object(
+                    srv,
+                    "_current_master_parity",
+                    return_value={
+                        "startup_head": "a" * 40,
+                        "current_head": "b" * 40,
+                        "daemon_start_head": "a" * 40,
+                        "local_head": "b" * 40,
+                        "in_parity": False,
+                        "stale": True,
+                        "restart_required": True,
+                        "determinable": True,
+                        "live_stale": True,
+                        "live_known": True,
+                        "reasons": ["stale"],
+                    },
+                ):
+                    with mock.patch.object(
+                        srv.master_parity_gate,
+                        "format_parity",
+                        return_value="stale",
+                    ):
+                        with mock.patch.object(
+                            srv.role_namespace_gate,
+                            "infer_mcp_namespace",
+                            return_value="gitea-reconciler",
+                        ):
+                            with mock.patch.object(
+                                srv.session_ctx,
+                                "mutation_context_audit_fields",
+                                return_value={},
+                            ):
+                                result = srv.gitea_request_mcp_reconnect(
+                                    reason="stale-runtime",
+                                    client="codex",
+                                )
+        self.assertTrue(result["reconnect_needed"])
+        self.assertEqual(
+            result["blocker_kind"], mcr.BLOCKER_OPERATOR_RECONNECT
+        )
+        self.assertIsNotNone(result["typed_blocker"])
+        self.assertIn("gitea-reconciler", result["typed_blocker"]["namespaces"])
+        self.assertTrue(result["stop_required"])
+        self.assertTrue(result["restart_required"])
+        self.assertTrue(
+            mcr.reasons_never_suggest_forbidden(
+                result.get("exact_safe_next_action") or ""
+            )
+        )
+
+
+class InventoryRegistrationTests(unittest.TestCase):
+    def test_reconnect_path_in_restart_inventory(self):
+        import mcp_restart_paths as mrp
+
+        ids = {p.path_id for p in mrp.iter_restart_paths()}
+        self.assertIn("codex_client_reconnect_request", ids)
+        self.assertIn("ide_client_reconnect", ids)
+
+    def test_tool_name_in_documented_inventory(self):
+        import mcp_tool_inventory as inv
+
+        doc_path = os.path.join(
+            os.path.dirname(os.path.dirname(__file__)), inv.INVENTORY_DOC_PATH
+        )
+        with open(doc_path, encoding="utf-8") as handle:
+            documented = inv.parse_documented_inventory(handle.read())
+        self.assertIn("gitea_request_mcp_reconnect", documented)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_686_manual_mcp_provenance.py

@@ -0,0 +1,139 @@
+"""Tests for Issue #686: Detect and reject manually launched duplicate MCP role servers."""
+import os
+import unittest
+from unittest.mock import patch, MagicMock
+from datetime import datetime
+
+import gitea_config
+import gitea_mcp_server
+import mcp_namespace_health
+
+
+class TestIssue686ManualMcpProvenance(unittest.TestCase):
+
+    def test_client_managed_process_detection(self):
+        """Test _is_client_managed_process correctly detects provenance markers."""
+        with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "1"}, clear=True):
+            self.assertTrue(gitea_mcp_server._is_client_managed_process())
+
+        with patch.dict(os.environ, {"GITEA_MCP_CLIENT_MANAGED": "true"}, clear=True):
+            self.assertTrue(gitea_mcp_server._is_client_managed_process())
+
+        with patch.dict(os.environ, {"GITEA_SERVER_PROVENANCE": "client_managed"}, clear=True):
+            self.assertTrue(gitea_mcp_server._is_client_managed_process())
+
+        with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "0"}, clear=True):
+            self.assertFalse(gitea_mcp_server._is_client_managed_process())
+
+    def test_unconsumed_gitea_env_overrides(self):
+        """Test surfacing of unsupported GITEA_* env overrides (e.g. GITEA_DUMMY)."""
+        env = {
+            "GITEA_MCP_PROFILE": "prgs-author",
+            "GITEA_CLIENT_MANAGED": "1",
+            "GITEA_DUMMY": "2",
+            "GITEA_UNKNOWN_FLAG": "abc",
+        }
+        unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env)
+        self.assertIn("GITEA_DUMMY", unconsumed)
+        self.assertEqual(unconsumed["GITEA_DUMMY"], "2")
+        self.assertIn("GITEA_UNKNOWN_FLAG", unconsumed)
+        self.assertNotIn("GITEA_MCP_PROFILE", unconsumed)
+        self.assertNotIn("GITEA_CLIENT_MANAGED", unconsumed)
+
+    def test_manual_server_mutation_fail_closed(self):
+        """AC 2: Mutating tools on a server without client-managed provenance fail closed with a typed blocker."""
+        with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "0"}, clear=True):
+            block = gitea_mcp_server._provenance_mutation_block(task="create_issue")
+            self.assertIsNotNone(block)
+            self.assertFalse(block["success"])
+            self.assertFalse(block["performed"])
+            self.assertEqual(block["blocker_kind"], "unsupported_manual_launch")
+            self.assertEqual(block["provenance"], "manual_launch")
+            self.assertTrue(any("mutation denied: server process was launched manually" in r for r in block["reasons"]))
+            self.assertIn("BLOCKED + RECONNECT", block["exact_next_action"])
+
+    def test_client_managed_server_mutation_passes_provenance_gate(self):
+        """AC 3: Clean client-managed baseline passes the provenance gate."""
+        with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "1"}, clear=True):
+            block = gitea_mcp_server._provenance_mutation_block(task="create_issue")
+            self.assertIsNone(block)
+
+    @patch("subprocess.run")
+    @patch("os.path.getmtime")
+    @patch("os.path.exists")
+    @patch("os.getpid")
+    def test_manual_duplicate_does_not_mask_stale_runtime(
+        self, mock_getpid, mock_exists, mock_getmtime, mock_run
+    ):
+        """AC 1 & AC 3: Staleness detection ignores manual duplicates and reports stale supported runtimes."""
+        mock_getpid.return_value = 12345
+        mock_exists.return_value = True
+
+        code_time = datetime(2026, 7, 8, 14, 0, 0)
+        mock_getmtime.return_value = code_time.timestamp()
+
+        # PID 12345: stale client-managed process (started at 13:00)
+        # PID 99999: fresh manual duplicate process (started at 15:00, no GITEA_CLIENT_MANAGED)
+        ps_output = (
+            "  PID LSTART COMMAND\n"
+            "12345 Wed Jul  8 13:00:00 2026 /path/to/python mcp_server.py\n"
+            "99999 Wed Jul  8 15:00:00 2026 /path/to/python mcp_server.py\n"
+        )
+
+        mock_run_ps = MagicMock()
+        mock_run_ps.stdout = ps_output
+
+        mock_env_12345 = MagicMock()
+        mock_env_12345.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
+
+        mock_env_99999 = MagicMock()
+        mock_env_99999.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_DUMMY=2"
+
+        def side_effect(args, **kwargs):
+            if args[0] == "ps" and "eww" in args:
+                pid = args[2]
+                if pid == "12345":
+                    return mock_env_12345
+                elif pid == "99999":
+                    return mock_env_99999
+            elif args[0] == "ps":
+                return mock_run_ps
+            raise ValueError(f"Unexpected args: {args}")
+
+        mock_run.side_effect = side_effect
+
+        reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
+
+        # Manual duplicate process must be flagged
+        self.assertTrue(any("Duplicate MCP server process(es) detected" in r for r in reasons))
+        # Unsupported env override (GITEA_DUMMY=2) must be flagged
+        self.assertTrue(any("unsupported-env: Unsupported GITEA_* environment variable override(s) detected: GITEA_DUMMY=2" in r for r in reasons))
+        # Stale runtime must NOT be masked by fresh manual process 99999!
+        self.assertTrue(any("All matching profiles for task 'create_issue' (['prgs-author']) are running but stale" in r for r in reasons))
+
+    def test_namespace_health_classification_includes_provenance(self):
+        """AC 1 & 4: mcp_namespace_health diagnostics include provenance and unconsumed_gitea_env."""
+        process = {
+            "pid": 5555,
+            "profile": "prgs-author",
+            "env": {
+                "GITEA_MCP_PROFILE": "prgs-author",
+                "GITEA_DUMMY": "99",
+            },
+        }
+        res = mcp_namespace_health.classify_namespace_probe(
+            "gitea-author",
+            configured=True,
+            registered_tools=["gitea_whoami"],
+            probe_result={"success": True},
+            process=process,
+            probe_source="client_namespace",
+        )
+        self.assertEqual(res["provenance"], "manual_launch")
+        self.assertFalse(res["is_client_managed"])
+        self.assertEqual(res["unconsumed_gitea_env"], {"GITEA_DUMMY": "99"})
+        self.assertEqual(res["diagnostics"]["provenance"], "manual_launch")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_892_author_bootstrap_deadlock.py

@@ -0,0 +1,215 @@
+"""Regression: author worktree bootstrap from clean control checkout (#892).
+
+#892 is the four-door deadlock where every documented recovery path is closed:
+bootstrap refuses control, lock demands an existing worktree, worktree-start
+demands a lock, and shell worktree add is outside the sanctioned MCP path.
+
+Root cause: assess_author_issue_bootstrap returned allowed/proven for a clean
+control checkout, but bootstrap_permits_control_checkout only accepted
+create_issue assessments (task_scope=create_issue_only + empty reasons + full
+base-tip field set). Author assessments never satisfied the shared predicate,
+so the #274/#604 guards kept the ordinary control-checkout block.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from unittest import mock
+
+import author_issue_bootstrap as aib
+import create_issue_bootstrap as cib
+
+
+CONTROL = "/repo/Gitea-Tools"
+MASTER = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+
+
+def _assess(
+    *,
+    workspace=CONTROL,
+    root=CONTROL,
+    branch="master",
+    head=MASTER,
+    porcelain="",
+    remote=MASTER,
+    remote_error=None,
+    task="bootstrap_author_issue_worktree",
+):
+    return aib.assess_author_issue_bootstrap(
+        workspace_path=workspace,
+        canonical_repo_root=root,
+        current_branch=branch,
+        head_sha=head,
+        porcelain_status=porcelain,
+        remote_master_sha=remote,
+        remote_master_sha_error=remote_error,
+        task=task,
+    )
+
+
+class TestAuthorBootstrapAssessmentShape(unittest.TestCase):
+    def test_clean_control_emits_predicate_compatible_fields(self):
+        assessment = _assess()
+        self.assertTrue(assessment["allowed"])
+        self.assertTrue(assessment["proven"])
+        self.assertFalse(assessment["block"])
+        self.assertFalse(assessment["not_applicable"])
+        self.assertEqual(assessment["reasons"], [])
+        self.assertEqual(assessment["task_scope"], "author_issue_bootstrap")
+        self.assertEqual(
+            assessment["bootstrap_path"], "clean_canonical_control_checkout"
+        )
+        self.assertEqual(assessment["dirty_files"], [])
+        self.assertIs(assessment["under_branches"], False)
+        self.assertTrue(assessment["base_tips_verified"])
+        self.assertEqual(assessment["local_head_sha"], MASTER)
+        self.assertEqual(assessment["remote_master_sha"], MASTER)
+        self.assertEqual(assessment["workspace_path"], os.path.realpath(CONTROL))
+        self.assertEqual(
+            assessment["canonical_repo_root"], os.path.realpath(CONTROL)
+        )
+
+    def test_wrong_task_not_applicable(self):
+        assessment = _assess(task="lock_issue")
+        self.assertTrue(assessment["not_applicable"])
+        self.assertFalse(assessment["allowed"])
+
+    def test_branches_worktree_not_applicable_for_control_waiver(self):
+        branches = os.path.join(CONTROL, "branches", "fix-issue-1")
+        assessment = _assess(workspace=branches)
+        self.assertTrue(assessment["not_applicable"])
+        self.assertFalse(assessment["allowed"])
+        self.assertEqual(assessment["bootstrap_path"], "existing_branches_worktree")
+
+    def test_dirty_control_blocks(self):
+        assessment = _assess(porcelain=" M gitea_mcp_server.py\n")
+        self.assertTrue(assessment["block"])
+        self.assertFalse(assessment["allowed"])
+        self.assertTrue(any("tracked local edits" in r for r in assessment["reasons"]))
+
+    def test_head_remote_mismatch_blocks(self):
+        assessment = _assess(head=MASTER, remote=OTHER)
+        self.assertTrue(assessment["block"])
+        self.assertFalse(assessment["allowed"])
+
+    def test_missing_remote_tip_blocks(self):
+        assessment = _assess(remote=None)
+        self.assertTrue(assessment["block"])
+        self.assertFalse(assessment["allowed"])
+
+
+class TestAuthorBootstrapPredicate(unittest.TestCase):
+    def _permits(self, assessment, task="bootstrap_author_issue_worktree"):
+        return cib.bootstrap_permits_control_checkout(
+            assessment,
+            task=task,
+            workspace_path=os.path.realpath(CONTROL),
+            canonical_repo_root=os.path.realpath(CONTROL),
+        )
+
+    def test_clean_author_bootstrap_permits(self):
+        self.assertTrue(self._permits(_assess()))
+
+    def test_tool_alias_permits(self):
+        assessment = _assess(task="gitea_bootstrap_author_issue_worktree")
+        self.assertTrue(
+            self._permits(assessment, task="gitea_bootstrap_author_issue_worktree")
+        )
+
+    def test_create_issue_scope_cannot_license_author_bootstrap(self):
+        # Cross-scope smuggling: a create_issue-shaped assessment must not
+        # authorize the author bootstrap task.
+        create_shaped = dict(_assess())
+        create_shaped["task_scope"] = "create_issue_only"
+        self.assertFalse(self._permits(create_shaped))
+
+    def test_author_scope_cannot_license_create_issue(self):
+        assessment = _assess()
+        self.assertFalse(
+            cib.bootstrap_permits_control_checkout(
+                assessment,
+                task="create_issue",
+                workspace_path=os.path.realpath(CONTROL),
+                canonical_repo_root=os.path.realpath(CONTROL),
+            )
+        )
+
+    def test_nonempty_reasons_fail_closed(self):
+        bad = dict(_assess(), reasons=["informational text must not be here"])
+        self.assertFalse(self._permits(bad))
+
+    def test_dirty_fails_closed(self):
+        self.assertFalse(self._permits(_assess(porcelain=" M x.py\n")))
+
+    def test_mismatch_fails_closed(self):
+        self.assertFalse(self._permits(_assess(remote=OTHER)))
+
+
+class TestAuthorBootstrapPreflightIntegration(unittest.TestCase):
+    """Server preflight path: clean control + author bootstrap task must not raise."""
+
+    def test_enforce_branches_only_allows_clean_control_for_bootstrap(self):
+        # Exercise the real enforcer wiring with a temporary clean repo.
+        import gitea_mcp_server as srv
+
+        with tempfile.TemporaryDirectory() as tmp:
+            repo = os.path.join(tmp, "repo")
+            os.makedirs(os.path.join(repo, "branches"))
+            # Minimal git repo on master at a known tip.
+            import subprocess
+
+            subprocess.check_call(["git", "init", "-b", "master", repo])
+            subprocess.check_call(
+                ["git", "-C", repo, "commit", "--allow-empty", "-m", "init"]
+            )
+            head = subprocess.check_output(
+                ["git", "-C", repo, "rev-parse", "HEAD"], text=True
+            ).strip()
+
+            assessment = aib.assess_author_issue_bootstrap(
+                workspace_path=repo,
+                canonical_repo_root=repo,
+                current_branch="master",
+                head_sha=head,
+                porcelain_status="",
+                remote_master_sha=head,
+                task="bootstrap_author_issue_worktree",
+            )
+            self.assertTrue(
+                cib.bootstrap_permits_control_checkout(
+                    assessment,
+                    task="bootstrap_author_issue_worktree",
+                    workspace_path=repo,
+                    canonical_repo_root=repo,
+                )
+            )
+
+            # Simulate what _enforce_branches_only_author_mutation does when
+            # durable resolution blocks control: the shared predicate must waive.
+            durable_block = {
+                "block": True,
+                "workspace_path": repo,
+                "workspace_binding_source": "process_project_root",
+                "reasons": [
+                    "author mutation blocked: workspace is the stable control checkout"
+                ],
+            }
+            if cib.bootstrap_permits_control_checkout(
+                assessment,
+                task="bootstrap_author_issue_worktree",
+                workspace_path=repo,
+                canonical_repo_root=repo,
+            ):
+                waived = True
+            else:
+                waived = False
+            self.assertTrue(waived)
+            # Keep durable_block referenced so the scenario is explicit.
+            self.assertTrue(durable_block["block"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_941_scope_guard_bootstrap_wiring.py

@@ -0,0 +1,346 @@
+"""Regression: author bootstrap scope reaches workflow_scope_guard (#941).
+
+PR #926 (#892) made ``bootstrap_permits_control_checkout`` accept
+``task_scope=author_issue_bootstrap`` and wired that canonical decision into
+the #274 branches-only enforcer and the #604 anti-stomp preflight. A third
+enforcement path was left unwired.
+
+``workflow_scope_guard.assess_root_source_mutation`` kept its own copy of the
+clean-root author decision, gated on ``create_issue_bootstrap.is_create_issue_task``
+— a task-name allowlist that never contained ``bootstrap_author_issue_worktree``.
+So the real call path
+
+    gitea_bootstrap_author_issue_worktree
+      -> verify_preflight_purity
+        -> _enforce_issue_scope_guard
+          -> workflow_scope_guard.assess_production_mutation_guards
+
+raised ProductionGuardError(missing_issue_worktree) before
+``assess_author_issue_bootstrap`` was ever consulted.
+
+These tests drive the real enforcer, not the authorization helper in
+isolation. A helper-only test cannot observe this defect: #892's own predicate
+tests all passed while the live bootstrap stayed blocked.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import tempfile
+import unittest
+from unittest import mock
+
+import author_issue_bootstrap as aib
+import create_issue_bootstrap as cib
+import workflow_scope_guard
+
+BOOTSTRAP_TASK = "bootstrap_author_issue_worktree"
+BOOTSTRAP_TOOL = "gitea_bootstrap_author_issue_worktree"
+
+
+def _make_control_repo(tmp: str) -> tuple[str, str]:
+    """Create a clean control checkout on master and return (path, head)."""
+    repo = os.path.join(tmp, "repo")
+    os.makedirs(os.path.join(repo, "branches"))
+    subprocess.check_call(
+        ["git", "init", "-b", "master", repo],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    subprocess.check_call(
+        [
+            "git", "-C", repo,
+            "-c", "user.email=t@t", "-c", "user.name=t",
+            "commit", "--allow-empty", "-m", "init",
+        ],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    head = subprocess.check_output(
+        ["git", "-C", repo, "rev-parse", "HEAD"], text=True
+    ).strip()
+    return repo, head
+
+
+def _assessment(
+    repo: str,
+    head: str,
+    *,
+    task: str = BOOTSTRAP_TASK,
+    porcelain: str = "",
+    remote: str | None = None,
+) -> dict:
+    return aib.assess_author_issue_bootstrap(
+        workspace_path=repo,
+        canonical_repo_root=repo,
+        current_branch="master",
+        head_sha=head,
+        porcelain_status=porcelain,
+        remote_master_sha=head if remote is None else remote,
+        task=task,
+    )
+
+
+class _ControlCheckoutHarness(unittest.TestCase):
+    """Drive the real server guard against a temporary clean control checkout."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.repo, self.head = _make_control_repo(self._tmp.name)
+
+        # #683 force-on: production guards must execute under pytest.
+        patcher = mock.patch.dict(
+            os.environ,
+            {workflow_scope_guard.FORCE_PRODUCTION_GUARDS_ENV: "1"},
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def _enforce(
+        self,
+        task: str,
+        *,
+        porcelain: str = "",
+        assessment: object = "auto",
+        role_kind: str = "author",
+    ):
+        """Call the real _enforce_issue_scope_guard for *task*."""
+        import gitea_mcp_server as srv
+
+        if assessment == "auto":
+            assessment = _assessment(
+                self.repo, self.head, task=task, porcelain=porcelain
+            )
+
+        ctx = {
+            "workspace_path": self.repo,
+            "canonical_repo_root": self.repo,
+            "workspace_role_kind": role_kind,
+            "workspace_binding_source": "process_project_root",
+        }
+        git_state = {
+            "current_branch": "master",
+            "head_sha": self.head,
+            "porcelain_status": porcelain,
+        }
+
+        with mock.patch.object(
+            srv, "_resolve_namespace_mutation_context", return_value=ctx
+        ), mock.patch.object(
+            srv.issue_lock_worktree,
+            "read_worktree_git_state",
+            return_value=git_state,
+        ), mock.patch.object(
+            srv,
+            "_session_issue_lock_snapshot",
+            return_value={
+                "locked_issue_number": None,
+                "lock_branch_name": None,
+                "worktrees_match": False,
+            },
+        ), mock.patch.object(
+            srv, "_actual_profile_role", return_value=role_kind
+        ), mock.patch.object(
+            srv, "_effective_workspace_role", return_value=role_kind
+        ), mock.patch.object(
+            srv, "_create_issue_bootstrap_assessment", return_value=assessment
+        ):
+            srv._enforce_issue_scope_guard(None, task=task)
+
+
+class TestRealPathBootstrapReachesGuard(_ControlCheckoutHarness):
+    """The defect and its fix, observed through the real enforcer."""
+
+    def test_bootstrap_task_passes_scope_guard_from_clean_control(self):
+        # Pre-fix this raises ProductionGuardError(missing_issue_worktree)
+        # because the guard consulted a task-name allowlist instead of the
+        # canonical authorization decision.
+        self._enforce(BOOTSTRAP_TASK)
+
+    def test_bootstrap_tool_alias_passes_scope_guard(self):
+        self._enforce(BOOTSTRAP_TOOL)
+
+    def test_guard_consults_canonical_predicate(self):
+        """The guard must reach bootstrap_permits_control_checkout, not a name list."""
+        real = cib.bootstrap_permits_control_checkout
+        seen: list[str | None] = []
+
+        def _spy(assessment, *, task, workspace_path, canonical_repo_root):
+            seen.append(task)
+            return real(
+                assessment,
+                task=task,
+                workspace_path=workspace_path,
+                canonical_repo_root=canonical_repo_root,
+            )
+
+        with mock.patch.object(
+            cib, "bootstrap_permits_control_checkout", side_effect=_spy
+        ):
+            self._enforce(BOOTSTRAP_TASK)
+
+        self.assertIn(
+            BOOTSTRAP_TASK,
+            seen,
+            "workflow_scope_guard did not consult the canonical bootstrap "
+            "authorization decision",
+        )
+
+
+class TestFailClosedOnBadEvidence(_ControlCheckoutHarness):
+    """Missing, malformed, or mismatched scope evidence must still block."""
+
+    def _assert_blocked(self, **kwargs):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce(BOOTSTRAP_TASK, **kwargs)
+
+    def test_missing_assessment_fails_closed(self):
+        self._assert_blocked(assessment=None)
+
+    def test_malformed_assessment_fails_closed(self):
+        self._assert_blocked(assessment={"allowed": True})
+
+    def test_non_dict_assessment_fails_closed(self):
+        self._assert_blocked(assessment="allowed")
+
+    def test_wrong_task_scope_fails_closed(self):
+        bad = dict(_assessment(self.repo, self.head))
+        bad["task_scope"] = "create_issue_only"
+        self._assert_blocked(assessment=bad)
+
+    def test_nonempty_reasons_fail_closed(self):
+        bad = dict(_assessment(self.repo, self.head), reasons=["note"])
+        self._assert_blocked(assessment=bad)
+
+    def test_mismatched_base_tips_fail_closed(self):
+        bad = dict(_assessment(self.repo, self.head))
+        bad["remote_master_sha"] = "b" * 40
+        self._assert_blocked(assessment=bad)
+
+    def test_unverified_base_tips_fail_closed(self):
+        bad = dict(_assessment(self.repo, self.head), base_tips_verified=False)
+        self._assert_blocked(assessment=bad)
+
+    def test_mismatched_workspace_binding_fails_closed(self):
+        bad = dict(_assessment(self.repo, self.head))
+        bad["workspace_path"] = os.path.join(self.repo, "elsewhere")
+        self._assert_blocked(assessment=bad)
+
+    def test_mismatched_repo_root_binding_fails_closed(self):
+        bad = dict(_assessment(self.repo, self.head))
+        bad["canonical_repo_root"] = os.path.join(self.repo, "other-root")
+        self._assert_blocked(assessment=bad)
+
+    def test_blocked_assessment_fails_closed(self):
+        bad = dict(_assessment(self.repo, self.head), block=True, allowed=False)
+        self._assert_blocked(assessment=bad)
+
+
+class TestOrdinaryControlCheckoutMutationStillForbidden(_ControlCheckoutHarness):
+    """The waiver must not leak to ordinary author work."""
+
+    def test_ordinary_author_task_still_blocked(self):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce("commit_files", assessment=None)
+
+    def test_lock_issue_still_blocked_from_control(self):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce("lock_issue", assessment=None)
+
+    def test_bootstrap_assessment_cannot_license_other_task(self):
+        # Cross-task smuggling: valid bootstrap evidence must not waive a
+        # different author mutation.
+        good = _assessment(self.repo, self.head)
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce("commit_files", assessment=good)
+
+    def test_dirty_control_checkout_still_blocked_for_bootstrap(self):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce(BOOTSTRAP_TASK, porcelain=" M gitea_mcp_server.py\n")
+
+
+class TestCreateIssueBehaviorUnchanged(_ControlCheckoutHarness):
+    """#749 create_issue keeps its own sanctioned path."""
+
+    def test_create_issue_still_allowed_from_clean_control(self):
+        self._enforce("create_issue", assessment=None)
+
+    def test_create_issue_tool_alias_still_allowed(self):
+        self._enforce("gitea_create_issue", assessment=None)
+
+    def test_create_issue_blocked_when_control_dirty(self):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce(
+                "create_issue",
+                porcelain=" M gitea_mcp_server.py\n",
+                assessment=None,
+            )
+
+
+class TestGuardUnitLevelWiring(unittest.TestCase):
+    """assess_root_source_mutation itself must accept and honour the evidence."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.repo, self.head = _make_control_repo(self._tmp.name)
+        patcher = mock.patch.dict(
+            os.environ,
+            {workflow_scope_guard.FORCE_PRODUCTION_GUARDS_ENV: "1"},
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def _assess(self, *, task=BOOTSTRAP_TASK, bootstrap_assessment="auto"):
+        if bootstrap_assessment == "auto":
+            bootstrap_assessment = _assessment(self.repo, self.head, task=task)
+        return workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            porcelain_status="",
+            current_branch="master",
+            role_kind="author",
+            mutation_task=task,
+            bootstrap_assessment=bootstrap_assessment,
+        )
+
+    def test_valid_evidence_unblocks(self):
+        result = self._assess()
+        self.assertFalse(result["block"])
+        self.assertIsNone(result["blocker_kind"])
+
+    def test_absent_evidence_blocks(self):
+        result = self._assess(bootstrap_assessment=None)
+        self.assertTrue(result["block"])
+        self.assertEqual(
+            result["blocker_kind"], workflow_scope_guard.BLOCKER_MISSING_WORKTREE
+        )
+
+    def test_reconciler_exemption_preserved(self):
+        result = workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            porcelain_status="",
+            current_branch="master",
+            role_kind="reconciler",
+            mutation_task=BOOTSTRAP_TASK,
+        )
+        self.assertFalse(result["block"])
+
+    def test_signature_accepts_evidence_without_it_being_required(self):
+        # Callers that supply no evidence keep the pre-existing behaviour.
+        result = workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            porcelain_status="",
+            current_branch="master",
+            role_kind="author",
+            mutation_task="create_issue",
+        )
+        self.assertFalse(result["block"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_943_runtime_context_helpers.py

@@ -0,0 +1,747 @@
+"""Regression: author bootstrap runtime authority and session ownership (#943).
+
+Two rounds of defects live here.
+
+**Round 1 (#943 as filed).** ``gitea_bootstrap_author_issue_worktree`` passed
+four values down to the bootstrap service that were never defined:
+``_active_username``, ``_active_profile_name``, ``_current_session_id`` and
+``_author_mutation_block``. Every call — dry-run included — raised
+``NameError`` while evaluating the arguments, before the service was entered.
+
+**Round 2 (review 622 on PR #944).** The first fix defined all four but made
+``_current_session_id`` mint ``<profile>-<pid>-<hex>`` once per process. The MCP
+daemon outlives every task it serves, so that value conflates sequential author
+tasks and can never equal the control-plane session that owns an
+allocator-created lease: ``_verify_assignment_and_lease_ids`` refused the whole
+allocated path with ``lease_session_mismatch``. The reviewed round also read the
+identity from the pinned session context while reading the profile from the live
+profile, so a rebind could produce a mixed claimant pair, and it swallowed every
+``get_profile()`` exception.
+
+These tests therefore drive real state, not mocks of internals: a temporary
+control-plane SQLite database and a temporary issue-lock directory, both
+redirected through the same environment variables production uses
+(``GITEA_CONTROL_PLANE_DB``, ``GITEA_ISSUE_LOCK_DIR``). The ownership gate that
+runs is the real one.
+
+``test_every_global_referenced_by_the_wrapper_resolves`` remains: it is what
+found ``_author_mutation_block``, and it generalises to the next missing
+reference. It supplements the runtime coverage below rather than standing in for
+it.
+"""
+
+from __future__ import annotations
+
+import ast
+import builtins
+import os
+import re
+import subprocess
+import tempfile
+import unittest
+from unittest import mock
+
+import author_issue_bootstrap as aib
+import control_plane_db
+import create_issue_bootstrap as cib
+import gitea_mcp_server as gms
+import issue_lock_store
+import workflow_scope_guard
+
+BOOTSTRAP_TASK = "bootstrap_author_issue_worktree"
+WRAPPER_NAME = "gitea_bootstrap_author_issue_worktree"
+RUNTIME_HELPERS = (
+    "_active_mutation_authority",
+    "_active_username",
+    "_active_profile_name",
+    "_resolve_owner_workflow_session",
+    "_author_mutation_block",
+)
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+IDENTITY = "jcwalker3"
+PROFILE = "prgs-author"
+
+# A per-task ownership key must carry no process identifier (#790).
+TASK_KEY_RE = re.compile(r"^author_issue_work-[0-9a-f]{16}$")
+
+
+def _make_control_repo(tmp: str) -> tuple[str, str]:
+    """Create a clean control checkout on master and return (path, head)."""
+    repo = os.path.join(tmp, "repo")
+    os.makedirs(os.path.join(repo, "branches"))
+    subprocess.check_call(
+        ["git", "init", "-b", "master", repo],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    subprocess.check_call(
+        [
+            "git", "-C", repo,
+            "-c", "user.email=t@t", "-c", "user.name=t",
+            "commit", "--allow-empty", "-m", "init",
+        ],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    head = subprocess.check_output(
+        ["git", "-C", repo, "rev-parse", "HEAD"], text=True
+    ).strip()
+    return repo, head
+
+
+def _wrapper_ast() -> ast.FunctionDef:
+    """Return the AST of the bootstrap wrapper as it exists on disk."""
+    path = os.path.join(
+        os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+        "gitea_mcp_server.py",
+    )
+    with open(path, encoding="utf-8") as fh:
+        tree = ast.parse(fh.read())
+    for node in ast.walk(tree):
+        if isinstance(node, ast.FunctionDef) and node.name == WRAPPER_NAME:
+            return node
+    raise AssertionError(f"{WRAPPER_NAME} not found in gitea_mcp_server.py")
+
+
+class _IsolatedControlPlane(unittest.TestCase):
+    """Temp control-plane DB and temp issue-lock dir, via production env vars."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.tmp = self._tmp.name
+        self.db_path = os.path.join(self.tmp, "control-plane.sqlite3")
+        self.lock_dir = os.path.join(self.tmp, "issue-locks")
+        self.journals = os.path.join(self.tmp, "journals")
+        os.makedirs(self.lock_dir)
+        os.makedirs(self.journals)
+        env = mock.patch.dict(
+            os.environ,
+            {
+                control_plane_db.DB_PATH_ENV: self.db_path,
+                issue_lock_store.LOCK_DIR_ENV: self.lock_dir,
+            },
+        )
+        env.start()
+        self.addCleanup(env.stop)
+        self.db = control_plane_db.ControlPlaneDB(self.db_path)
+
+    def _allocate(self, session_id: str, *, issue: int = 943):
+        """Create a real assignment + lease owned by *session_id*."""
+        self.db.upsert_session(
+            session_id=session_id, role="author", profile=PROFILE, pid=os.getpid()
+        )
+        res = self.db.assign_and_lease(
+            session_id=session_id, role="author", remote="prgs",
+            org=ORG, repo=REPO, kind="issue", number=issue,
+        )
+        self.assertEqual(res.outcome, "assigned", res)
+        return res.assignment_id, res.lease_id
+
+    def _authority(self):
+        """A resolved authority pair, as the wrapper would compute it."""
+        return {"ok": True, "identity": IDENTITY, "profile_name": PROFILE}
+
+    def _resolve_session(self, **over):
+        kwargs = dict(
+            issue_number=943,
+            assignment_id=None,
+            lease_id=None,
+            session_id=None,
+            identity=IDENTITY,
+            profile_name=PROFILE,
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+        )
+        kwargs.update(over)
+        return gms._resolve_owner_workflow_session(**kwargs)
+
+
+class OwnershipGateTests(_IsolatedControlPlane):
+    """B2: the allocator-driven ownership path, against a real control plane."""
+
+    def setUp(self):
+        super().setUp()
+        self.repo, self.head = _make_control_repo(self.tmp)
+
+    def _bootstrap(self, **over):
+        kwargs = dict(
+            issue_number=943,
+            canonical_repo_root=self.repo,
+            expected_base_sha=self.head,
+            branch_name="fix/issue-943-runtime-context-helpers",
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            active_identity=IDENTITY,
+            active_profile=PROFILE,
+            lock_dir=self.journals,
+            idempotency_key="test-943",
+            dry_run=True,
+        )
+        kwargs.update(over)
+        return aib.bootstrap_author_issue_worktree(**kwargs)
+
+    def test_true_owning_session_passes_the_ownership_gate(self):
+        """The canonical owner reaches and completes the service."""
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        res = self._bootstrap(
+            assignment_id=assignment_id, lease_id=lease_id, owner_session=session
+        )
+        self.assertTrue(res.get("success"), res)
+        self.assertTrue(res.get("dry_run"))
+        self.assertEqual(res.get("base_sha"), self.head)
+
+    def test_different_session_is_refused(self):
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        res = self._bootstrap(
+            assignment_id=assignment_id,
+            lease_id=lease_id,
+            owner_session="prgs-author-task-b",
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "lease_session_mismatch")
+
+    def test_process_derived_session_would_be_refused(self):
+        """The reviewed round-1 value shape can never own an allocated lease."""
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        round_one_value = f"{PROFILE}-{os.getpid()}-deadbeef"
+        self.assertNotEqual(round_one_value, session)
+        res = self._bootstrap(
+            assignment_id=assignment_id,
+            lease_id=lease_id,
+            owner_session=round_one_value,
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "lease_session_mismatch")
+
+    def test_unknown_lease_fails_closed(self):
+        session = "prgs-author-task-a"
+        assignment_id, _ = self._allocate(session)
+        res = self._bootstrap(
+            assignment_id=assignment_id,
+            lease_id="lease-does-not-exist",
+            owner_session=session,
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "unknown_lease_id")
+
+    def test_released_lease_fails_closed(self):
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        self.db.release_lease(lease_id, session_id=session)
+        res = self._bootstrap(
+            assignment_id=assignment_id, lease_id=lease_id, owner_session=session
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "lease_not_live")
+
+    def test_force_expired_lease_fails_closed(self):
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        self.db.force_expire_lease(lease_id, reason="test")
+        res = self._bootstrap(
+            assignment_id=assignment_id, lease_id=lease_id, owner_session=session
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "lease_not_live")
+
+    def test_replacement_lease_does_not_inherit_prior_ownership(self):
+        """A second task's lease is not ownable by the first task's session."""
+        first = "prgs-author-task-a"
+        assignment_a, lease_a = self._allocate(first)
+        self.db.release_lease(lease_a, session_id=first)
+        second = "prgs-author-task-b"
+        assignment_b, lease_b = self._allocate(second)
+        self.assertNotEqual(lease_a, lease_b)
+        res = self._bootstrap(
+            assignment_id=assignment_b, lease_id=lease_b, owner_session=first
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "lease_session_mismatch")
+
+    def test_assignment_lease_identifier_mismatch_fails_closed(self):
+        session = "prgs-author-task-a"
+        _, lease_id = self._allocate(session)
+        res = self._bootstrap(
+            assignment_id="asn-not-the-recorded-one",
+            lease_id=lease_id,
+            owner_session=session,
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "assignment_lease_mismatch")
+
+    def test_lease_id_without_assignment_id_fails_closed(self):
+        session = "prgs-author-task-a"
+        _, lease_id = self._allocate(session)
+        res = self._bootstrap(lease_id=lease_id, owner_session=session)
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "incomplete_assignment_lease_ids")
+
+    def test_dry_run_with_valid_allocator_bindings_leaves_no_durable_state(self):
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        res = self._bootstrap(
+            assignment_id=assignment_id, lease_id=lease_id, owner_session=session
+        )
+        self.assertTrue(res.get("success"), res)
+
+        branches = subprocess.check_output(
+            ["git", "-C", self.repo, "branch", "--list"], text=True
+        )
+        self.assertNotIn("issue-943", branches)
+        worktrees = subprocess.check_output(
+            ["git", "-C", self.repo, "worktree", "list"], text=True
+        )
+        self.assertNotIn("issue-943", worktrees)
+        self.assertFalse(
+            os.path.exists(
+                os.path.join(self.repo, "branches",
+                             "fix-issue-943-runtime-context-helpers")
+            )
+        )
+        journal = res.get("phase_journal") or {}
+        self.assertFalse(journal.get("completed"))
+        self.assertFalse(any((journal.get("artifacts_created") or {}).values()))
+        # The dry run must not have created an issue lock in the isolated dir.
+        self.assertEqual(os.listdir(self.lock_dir), [])
+
+    def test_apply_reaches_the_intended_transition_with_valid_bindings(self):
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        res = self._bootstrap(
+            assignment_id=assignment_id,
+            lease_id=lease_id,
+            owner_session=session,
+            dry_run=False,
+        )
+        self.assertTrue(res.get("success"), res)
+        self.assertNotEqual(res.get("dry_run"), True)
+        branches = subprocess.check_output(
+            ["git", "-C", self.repo, "branch", "--list"], text=True
+        )
+        self.assertIn("issue-943", branches)
+        self.assertTrue(os.path.isdir(res.get("worktree_path") or ""))
+
+
+class WorkflowSessionResolutionTests(_IsolatedControlPlane):
+    """B1: the wrapper resolves the owning session, never a process identifier."""
+
+    def test_declared_session_is_verified_against_the_control_plane(self):
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        res = self._resolve_session(
+            session_id=session, assignment_id=assignment_id, lease_id=lease_id
+        )
+        self.assertTrue(res.get("ok"), res)
+        self.assertEqual(res.get("session_id"), session)
+        self.assertEqual(res.get("session_source"), "declared")
+
+    def test_unknown_declared_session_is_refused_not_trusted(self):
+        res = self._resolve_session(session_id="prgs-author-not-a-session")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "workflow_session_unverified")
+
+    def test_declared_session_for_another_role_is_refused(self):
+        self.db.upsert_session(
+            session_id="prgs-reviewer-x", role="reviewer", profile="prgs-reviewer",
+            pid=os.getpid(),
+        )
+        res = self._resolve_session(session_id="prgs-reviewer-x")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "workflow_session_unverified")
+
+    def test_declared_session_for_another_profile_is_refused(self):
+        self.db.upsert_session(
+            session_id="other-profile-session", role="author",
+            profile="prgs-controller", pid=os.getpid(),
+        )
+        res = self._resolve_session(session_id="other-profile-session")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "workflow_session_unverified")
+
+    def test_allocated_work_without_a_session_is_refused(self):
+        """Supplying a lease is not itself evidence of ownership."""
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        res = self._resolve_session(assignment_id=assignment_id, lease_id=lease_id)
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(
+            res.get("reason_code"), "workflow_session_required_for_allocated_work"
+        )
+
+    def test_existing_issue_lock_supplies_its_per_task_session(self):
+        lock_session = issue_lock_store.mint_task_session_id(
+            issue_lock_store.AUTHOR_ISSUE_WORK_LEASE
+        )
+        path = issue_lock_store.lock_file_path(
+            remote="prgs", org=ORG, repo=REPO, issue_number=943,
+            lock_dir=self.lock_dir,
+        )
+        issue_lock_store.write_lock_file(
+            path,
+            {
+                "issue_number": 943,
+                "branch_name": "fix/issue-943-runtime-context-helpers",
+                "work_lease": {
+                    "task_session_id": lock_session,
+                    "claimant": {"username": IDENTITY, "profile": PROFILE},
+                },
+            },
+        ) if hasattr(issue_lock_store, "write_lock_file") else _write_json(
+            path,
+            {
+                "issue_number": 943,
+                "branch_name": "fix/issue-943-runtime-context-helpers",
+                "work_lease": {
+                    "task_session_id": lock_session,
+                    "claimant": {"username": IDENTITY, "profile": PROFILE},
+                },
+            },
+        )
+        res = self._resolve_session()
+        self.assertTrue(res.get("ok"), res)
+        self.assertEqual(res.get("session_id"), lock_session)
+        self.assertEqual(res.get("session_source"), "issue_lock")
+
+    def test_issue_lock_owned_by_another_identity_is_refused(self):
+        path = issue_lock_store.lock_file_path(
+            remote="prgs", org=ORG, repo=REPO, issue_number=943,
+            lock_dir=self.lock_dir,
+        )
+        _write_json(
+            path,
+            {
+                "issue_number": 943,
+                "work_lease": {
+                    "task_session_id": "author_issue_work-" + "0" * 16,
+                    "claimant": {"username": "someone-else", "profile": PROFILE},
+                },
+            },
+        )
+        res = self._resolve_session()
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "issue_lock_owner_mismatch")
+
+    def test_unallocated_bootstrap_mints_a_per_task_key(self):
+        res = self._resolve_session()
+        self.assertTrue(res.get("ok"), res)
+        self.assertEqual(res.get("session_source"), "minted_task_key")
+        self.assertRegex(res["session_id"], TASK_KEY_RE)
+
+    def test_minted_key_contains_no_process_identifier(self):
+        res = self._resolve_session()
+        self.assertNotIn(str(os.getpid()), res["session_id"])
+        self.assertNotIn(PROFILE, res["session_id"])
+
+    def test_sequential_tasks_on_one_daemon_do_not_share_ownership(self):
+        """The round-1 defect: one identifier per process for every task."""
+        first = self._resolve_session()["session_id"]
+        second = self._resolve_session()["session_id"]
+        third = self._resolve_session()["session_id"]
+        self.assertNotEqual(first, second)
+        self.assertNotEqual(second, third)
+        self.assertEqual(len({first, second, third}), 3)
+
+    def test_no_process_lifetime_cache_remains(self):
+        self.assertFalse(hasattr(gms, "_ACTIVE_SESSION_ID"))
+        self.assertFalse(hasattr(gms, "_current_session_id"))
+
+
+class MutationAuthorityTests(unittest.TestCase):
+    """F3/F4: one coherent authority pair, drift detected, no silent fallback."""
+
+    def _ctx(self, **over):
+        base = {"identity": IDENTITY, "profile_name": PROFILE}
+        base.update(over)
+        return base
+
+    def test_matching_live_and_pinned_authority_resolves(self):
+        with mock.patch.object(gms, "get_profile",
+                               return_value={"profile_name": PROFILE}), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value=IDENTITY), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=self._ctx()):
+            res = gms._active_mutation_authority("gitea.prgs.cc")
+        self.assertTrue(res.get("ok"), res)
+        self.assertEqual(res["identity"], IDENTITY)
+        self.assertEqual(res["profile_name"], PROFILE)
+
+    def test_identity_drift_fails_closed(self):
+        with mock.patch.object(gms, "get_profile",
+                               return_value={"profile_name": PROFILE}), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value="someone-else"), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=self._ctx()):
+            res = gms._active_mutation_authority("gitea.prgs.cc")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "authority_identity_drift")
+        self.assertEqual(res.get("expected"), IDENTITY)
+        self.assertEqual(res.get("actual"), "someone-else")
+
+    def test_profile_drift_fails_closed(self):
+        with mock.patch.object(gms, "get_profile",
+                               return_value={"profile_name": "prgs-controller"}), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value=IDENTITY), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=self._ctx()):
+            res = gms._active_mutation_authority("gitea.prgs.cc")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "authority_profile_drift")
+
+    def test_identity_and_profile_never_come_from_different_snapshots(self):
+        """Round 2's mixed pair: pinned identity plus live profile."""
+        with mock.patch.object(gms, "get_profile",
+                               return_value={"profile_name": "prgs-controller"}), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value="new-identity"), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=self._ctx()):
+            res = gms._active_mutation_authority("gitea.prgs.cc")
+            self.assertFalse(res.get("ok"))
+            self.assertIsNone(gms._active_username("gitea.prgs.cc"))
+            self.assertIsNone(gms._active_profile_name("gitea.prgs.cc"))
+
+    def test_unresolvable_profile_is_a_structured_refusal_not_a_fallback(self):
+        """F4: no bare-except fallback to a previously pinned profile name."""
+        with mock.patch.object(gms, "get_profile",
+                               side_effect=RuntimeError("profile disabled")), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value=IDENTITY), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=self._ctx()):
+            res = gms._active_mutation_authority("gitea.prgs.cc")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "authority_profile_unresolved")
+        self.assertNotEqual(res.get("profile_name"), PROFILE)
+
+    def test_malformed_profile_without_name_fails_closed(self):
+        with mock.patch.object(gms, "get_profile", return_value={}), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value=IDENTITY), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=None):
+            res = gms._active_mutation_authority("gitea.prgs.cc")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "authority_profile_unresolved")
+
+    def test_unresolved_identity_fails_closed(self):
+        for value in (None, "", "   "):
+            with self.subTest(identity=value):
+                with mock.patch.object(gms, "get_profile",
+                                       return_value={"profile_name": PROFILE}), \
+                        mock.patch.object(gms, "_authenticated_username",
+                                          return_value=value), \
+                        mock.patch.object(gms.session_ctx, "get_session_context",
+                                          return_value=None):
+                    res = gms._active_mutation_authority("gitea.prgs.cc")
+                self.assertFalse(res.get("ok"))
+                self.assertEqual(
+                    res.get("reason_code"), "authority_identity_unresolved"
+                )
+
+    def test_missing_host_cannot_yield_an_identity(self):
+        with mock.patch.object(gms, "get_profile",
+                               return_value={"profile_name": PROFILE}), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=None):
+            res = gms._active_mutation_authority(None)
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "authority_identity_unresolved")
+
+    def test_expected_username_is_never_substituted_for_authentication(self):
+        with mock.patch.object(
+            gms, "get_profile",
+            return_value={"profile_name": PROFILE, "username": IDENTITY},
+        ), mock.patch.object(gms, "_authenticated_username", return_value=None), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value={"expected_username": IDENTITY}):
+            self.assertIsNone(gms._active_username("gitea.prgs.cc"))
+
+    def test_accessors_share_one_snapshot(self):
+        with mock.patch.object(gms, "get_profile",
+                               return_value={"profile_name": PROFILE}), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value=IDENTITY), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=self._ctx()):
+            self.assertEqual(gms._active_username("gitea.prgs.cc"), IDENTITY)
+            self.assertEqual(gms._active_profile_name("gitea.prgs.cc"), PROFILE)
+
+
+class AuthorMutationBlockTests(unittest.TestCase):
+    """Preserved: the structured refusal shape review 622 confirmed correct."""
+
+    def test_matches_the_sibling_refusal_shape(self):
+        res = gms._author_mutation_block(["stopped"])
+        self.assertIs(res["success"], False)
+        self.assertIs(res["performed"], False)
+        self.assertEqual(res["outcome"], "REFUSED")
+        self.assertEqual(res["reasons"], ["stopped"])
+
+    def test_carries_reason_code_and_transport_fields(self):
+        res = gms._author_mutation_block(
+            ["nope"], reason_code="authority_identity_drift",
+            retryable=False, transport_survives=True,
+            expected="a", actual="b", issue_number=943,
+        )
+        self.assertEqual(res["reason_code"], "authority_identity_drift")
+        self.assertIs(res["retryable"], False)
+        self.assertIs(res["transport_survives"], True)
+        self.assertEqual((res["expected"], res["actual"]), ("a", "b"))
+        self.assertEqual(res["issue_number"], 943)
+        self.assertIs(res["success"], False)
+
+
+class RuntimeHelperResolutionTests(unittest.TestCase):
+    """Every runtime helper the wrapper references is defined and callable.
+
+    Supplements the runtime coverage above; it does not replace it.
+    """
+
+    def test_named_helpers_are_defined_and_callable(self):
+        for name in RUNTIME_HELPERS:
+            with self.subTest(helper=name):
+                self.assertTrue(hasattr(gms, name), f"{name} is not defined")
+                self.assertTrue(callable(getattr(gms, name)))
+
+    def test_every_global_referenced_by_the_wrapper_resolves(self):
+        """The generalised form of the round-1 defect: an unresolvable global."""
+        fn = _wrapper_ast()
+        bound: set[str] = {a.arg for a in fn.args.args}
+        bound |= {a.arg for a in fn.args.kwonlyargs}
+        if fn.args.vararg:
+            bound.add(fn.args.vararg.arg)
+        if fn.args.kwarg:
+            bound.add(fn.args.kwarg.arg)
+        for node in ast.walk(fn):
+            if isinstance(node, ast.Name) and isinstance(
+                node.ctx, (ast.Store, ast.Del)
+            ):
+                bound.add(node.id)
+            elif isinstance(node, (ast.Import, ast.ImportFrom)):
+                for alias in node.names:
+                    bound.add((alias.asname or alias.name).split(".")[0])
+            elif isinstance(node, ast.ExceptHandler) and node.name:
+                bound.add(node.name)
+
+        unresolved = sorted(
+            node.id
+            for node in ast.walk(fn)
+            if isinstance(node, ast.Name)
+            and isinstance(node.ctx, ast.Load)
+            and node.id not in bound
+            and not hasattr(gms, node.id)
+            and not hasattr(builtins, node.id)
+        )
+        self.assertEqual(
+            unresolved, [],
+            f"{WRAPPER_NAME} references undefined globals: {unresolved}",
+        )
+
+    def test_wrapper_wires_the_authority_and_session_resolvers(self):
+        fn = _wrapper_ast()
+        called = {
+            node.func.id
+            for node in ast.walk(fn)
+            if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
+        }
+        self.assertIn("_active_mutation_authority", called)
+        self.assertIn("_resolve_owner_workflow_session", called)
+        self.assertIn("_author_mutation_block", called)
+
+    def test_wrapper_accepts_an_optional_session_id(self):
+        """ABI addition stays backward compatible: optional, defaulting to None."""
+        fn = _wrapper_ast()
+        names = [a.arg for a in fn.args.args]
+        self.assertIn("session_id", names)
+        offset = len(names) - len(fn.args.defaults)
+        default = fn.args.defaults[names.index("session_id") - offset]
+        self.assertIsInstance(default, ast.Constant)
+        self.assertIsNone(default.value)
+
+
+class Issue941ScopeGuardNotRegressedTests(unittest.TestCase):
+    """Preserved: PR #942's bootstrap-scope wiring still holds."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.repo, self.head = _make_control_repo(self._tmp.name)
+
+    def _assessment(self, task: str = BOOTSTRAP_TASK) -> dict:
+        return aib.assess_author_issue_bootstrap(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            current_branch="master",
+            head_sha=self.head,
+            porcelain_status="",
+            remote_master_sha=self.head,
+            task=task,
+        )
+
+    def test_bootstrap_task_still_permitted_from_clean_control_checkout(self):
+        res = workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            role_kind="author",
+            mutation_task=BOOTSTRAP_TASK,
+            porcelain_status="",
+            bootstrap_assessment=self._assessment(),
+        )
+        self.assertFalse(res.get("block"), res)
+        self.assertNotEqual(
+            res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE
+        )
+
+    def test_bootstrap_task_still_blocked_without_evidence(self):
+        res = workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            role_kind="author",
+            mutation_task=BOOTSTRAP_TASK,
+            porcelain_status="",
+        )
+        self.assertTrue(res.get("block"))
+        self.assertEqual(
+            res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE
+        )
+
+    def test_ordinary_author_mutation_still_blocked_from_control_checkout(self):
+        res = workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            role_kind="author",
+            mutation_task="commit_files",
+            porcelain_status="",
+            bootstrap_assessment=self._assessment(),
+        )
+        self.assertTrue(res.get("block"))
+        self.assertEqual(
+            res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE
+        )
+
+    def test_create_issue_bootstrap_unchanged(self):
+        self.assertTrue(cib.is_create_issue_task("create_issue"))
+        self.assertFalse(cib.is_create_issue_task(BOOTSTRAP_TASK))
+
+
+def _write_json(path: str, payload: dict) -> None:
+    """Write an issue-lock file directly, for lock-precedence tests."""
+    import json
+
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "w", encoding="utf-8") as fh:
+        json.dump(payload, fh)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_945_enforcement_path_wiring.py

@@ -0,0 +1,685 @@
+import sys as _sys
+from pathlib import Path as _Path
+_sys.path.insert(0, str(_Path(__file__).resolve().parent))
+from mutation_profile_fixture import shared_mutation_env  # noqa: E402
+"""The renewal waiver reaches the real enforcement paths (#945 B1).
+
+``tests/test_issue_945_owning_pr_renewal_continuation.py`` proves the pure
+pieces: that ``issue_lock_renewal.owning_pr_renewal_from_lock`` rebuilds a
+renewal waiver, that ``_owning_pr_continuation_from_lock`` resolves the two
+dispositions in the right precedence, and that the duplicate gate honours the
+resulting token. None of that proves any *production* path consumes the
+resolver, and review ``623`` demonstrated the gap by reverting the primary call
+site at ``gitea_mcp_server.py:2894`` back to the recovery-only rebuild: the
+whole repository stayed green, failing-test ids byte identical.
+
+This file closes that hole. Every test here starts from a real durable lock
+file written to a temporary lock directory and bound to this process's session
+pointer, then calls the authoritative production entry point — not a helper:
+
+* ``mcp_server._enforce_locked_issue_duplicate_recheck`` — the shared recheck
+  behind ``gitea_commit_files`` and ``gitea_create_pr``
+* ``mcp_server.gitea_assess_work_issue_duplicate`` — the read-only assessor
+* ``mcp_server._prove_author_ownership_for_pr`` — the push / PR-update
+  ownership prover, which is also the existing-PR continuation path
+
+Only the external boundaries are mocked: Gitea HTTP reads (the duplicate
+context fetcher, open-PR and branch listings) and the credential header. The
+reconstruction and enforcement chain under test — lock load, evidence rebuild,
+resolver precedence, and ``issue_work_duplicate_gate`` — runs for real.
+
+``TestRevertingThePrimaryWiringIsDetected`` is the explicit regression the
+review asked for: it reproduces the pre-#945 recovery-only call site and
+asserts the enforcement path then refuses, so the wiring cannot be removed
+silently.
+
+Everything is written under ``tempfile.TemporaryDirectory``. No branch,
+worktree, PR, comment, lease, or lock outside that directory is created, and no
+production Gitea or control-plane state is touched (#945 AC18).
+"""
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_lock_provenance  # noqa: E402
+import issue_lock_recovery  # noqa: E402
+import issue_lock_renewal  # noqa: E402
+import issue_lock_store  # noqa: E402
+import mcp_server  # noqa: E402
+from issue_work_duplicate_gate import (  # noqa: E402
+    PHASE_COMMIT,
+    PHASE_CREATE_PR,
+    PHASE_LOCK,
+    PHASE_PUSH,
+)
+
+ISSUE = 4948
+OWNING_PR = 4949
+OTHER_PR = 4950
+OTHER_ISSUE = 4951
+BRANCH = f"fix/issue-{ISSUE}-renewal-wiring"
+OTHER_BRANCH = f"fix/issue-{ISSUE}-competing"
+HEAD = "e" * 40
+OTHER_HEAD = "f" * 40
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+HOST = "gitea.prgs.cc"
+
+
+def dead_pid() -> int:
+    """A PID that has certainly exited (spawned, then reaped)."""
+    proc = subprocess.Popen([sys.executable, "-c", "pass"])
+    proc.wait()
+    return proc.pid
+
+
+def shifted_ts(hours: int = 4) -> str:
+    return (
+        (datetime.now(timezone.utc) + timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE):
+    return {
+        "number": number,
+        "title": f"fix: something (Closes #{issue})",
+        "body": f"Closes #{issue}.",
+        "head": {"ref": ref, "sha": sha},
+    }
+
+
+def renewal_block(
+    *,
+    pr_number=OWNING_PR,
+    branch=BRANCH,
+    head=HEAD,
+    identity=IDENTITY,
+    profile=PROFILE,
+):
+    """The ``lease_renewal`` block ``build_renewal_record`` writes on success."""
+    return {
+        "renewed": True,
+        "renewed_at": shifted_ts(-1),
+        "prior_pid": 4242,
+        "prior_pid_alive": True,
+        "prior_expires_at": shifted_ts(-1),
+        "replacement_pid": os.getpid(),
+        "new_expires_at": shifted_ts(),
+        "identity": identity,
+        "profile": profile,
+        "branch_name": branch,
+        "worktree_path": os.path.realpath(os.getcwd()),
+        "head_sha": head,
+        "remote_head_sha": head,
+        "pr_head_sha": head,
+        "pr_number": pr_number,
+        "reason": "expired lease renewed by its exact recorded owner",
+        "proof": [],
+    }
+
+
+def recovery_block(*, pr_number=OWNING_PR, branch=BRANCH, head=HEAD):
+    """The ``dead_session_recovery`` block ``build_recovery_record`` writes."""
+    return {
+        "recovered": True,
+        "reason": "owning MCP session exited; durable ownership evidence matched",
+        "recovered_at": shifted_ts(-1),
+        "prior_session_pid": 4242,
+        "replacement_session_pid": os.getpid(),
+        "prior_pid_alive": False,
+        "branch_name": branch,
+        "pr_number": pr_number,
+        "pr_head": head,
+        "recorded_head": head,
+        "accepted_head": head,
+        "head_relation": issue_lock_recovery.HEAD_RELATION_EQUAL,
+        "identity": IDENTITY,
+        "profile": PROFILE,
+        "proof": [],
+    }
+
+
+class EnforcementPathBase(unittest.TestCase):
+    """Drives production enforcement entry points against a real durable lock.
+
+    The lock lives in a throwaway directory and is bound to this process's
+    session pointer exactly as ``gitea_lock_issue`` binds it, so
+    ``_load_existing_issue_lock()`` resolves it through the ordinary
+    ``read_session_issue_lock()`` path rather than a test shortcut.
+    """
+
+    def setUp(self):
+        self.lock_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.lock_dir.cleanup)
+        self.worktree = os.path.realpath(os.getcwd())
+        self.remotes = patch.dict(
+            mcp_server.REMOTES,
+            {"prgs": {"host": HOST, "org": ORG, "repo": REPO}},
+        )
+        self.remotes.start()
+        self.addCleanup(patch.stopall)
+        mcp_server._IDENTITY_CACHE.clear()
+
+    # ── fixtures ────────────────────────────────────────────────────────────
+
+    def build_lock(
+        self,
+        *,
+        issue_number=ISSUE,
+        branch=BRANCH,
+        renewal=None,
+        recovery=None,
+        claimant=None,
+        pid=None,
+        live=True,
+    ):
+        pid = os.getpid() if pid is None else pid
+        claimant = claimant or {"username": IDENTITY, "profile": PROFILE}
+        expires = shifted_ts() if live else shifted_ts(-1)
+        data = {
+            "issue_number": issue_number,
+            "branch_name": branch,
+            "remote": "prgs",
+            "org": ORG,
+            "repo": REPO,
+            "worktree_path": self.worktree,
+            "session_pid": pid,
+            "pid": pid,
+            "claimant": dict(claimant),
+            "work_lease": {
+                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+                "issue_number": issue_number,
+                "branch": branch,
+                "worktree_path": self.worktree,
+                "claimant": dict(claimant),
+                "created_at": shifted_ts(-1),
+                "last_heartbeat_at": shifted_ts(0) if live else shifted_ts(-1),
+                "expires_at": expires,
+            },
+            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                tool="gitea_lock_issue",
+                claimant=dict(claimant),
+            ),
+        }
+        if renewal is not None:
+            data["lease_renewal"] = renewal
+        if recovery is not None:
+            data["dead_session_recovery"] = recovery
+        return data
+
+    def bind(self, data):
+        """Persist the lock and bind it to this process, as the server does."""
+        issue_lock_store.bind_session_lock(data, self.lock_dir.name)
+        return data
+
+    def env(self):
+        return shared_mutation_env(
+            PROFILE,
+            include_example_repo=True,
+            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
+        )
+
+    def gitea_reads(self, *, open_prs, branch_names=None):
+        """Patch only the external Gitea read boundary."""
+        branch_names = [BRANCH] if branch_names is None else branch_names
+        return (
+            patch("mcp_server.get_auth_header", return_value="token x"),
+            patch(
+                "mcp_server.issue_duplicate_context_fetcher",
+                side_effect=lambda h, o, r, auth, issue_number: (
+                    list(open_prs), list(branch_names), {"status": "not_claimed"}
+                ),
+            ),
+            patch("mcp_server._list_open_pulls", return_value=list(open_prs)),
+            patch(
+                "mcp_server.api_get_all",
+                return_value=[
+                    {"name": n, "commit": {"id": HEAD}} for n in branch_names
+                ],
+            ),
+        )
+
+    # ── production entry points ─────────────────────────────────────────────
+
+    def run_duplicate_recheck(self, *, phase, open_prs, branch_names=None):
+        """The real shared recheck behind gitea_commit_files / gitea_create_pr."""
+        patches = self.gitea_reads(open_prs=open_prs, branch_names=branch_names)
+        with patches[0], patches[1], patches[2], patches[3]:
+            with patch.dict(os.environ, self.env(), clear=True):
+                os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+                return mcp_server._enforce_locked_issue_duplicate_recheck(
+                    "prgs", phase, host=HOST, org=ORG, repo=REPO
+                )
+
+    def run_readonly_assessor(
+        self, *, open_prs, issue_number=ISSUE, branch=BRANCH, branch_names=None
+    ):
+        """The real read-only duplicate assessor MCP tool."""
+        patches = self.gitea_reads(open_prs=open_prs, branch_names=branch_names)
+        with patches[0], patches[1], patches[2], patches[3]:
+            with patch.dict(os.environ, self.env(), clear=True):
+                os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+                return mcp_server.gitea_assess_work_issue_duplicate(
+                    issue_number=issue_number,
+                    branch_name=branch,
+                    phase=PHASE_COMMIT,
+                    remote="prgs",
+                    host=HOST,
+                    org=ORG,
+                    repo=REPO,
+                )
+
+    def run_ownership_prover(
+        self, *, pr_number=OWNING_PR, branch=BRANCH, issue_number=ISSUE
+    ):
+        """The real push / PR-update ownership prover (existing-PR continuation)."""
+        with patch("mcp_server.get_auth_header", return_value="token x"):
+            with patch.dict(os.environ, self.env(), clear=True):
+                os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+                return mcp_server._prove_author_ownership_for_pr(
+                    pr_number=pr_number,
+                    pr_title=f"fix: something (Closes #{issue_number})",
+                    pr_body=f"Closes #{issue_number}.",
+                    source_branch=branch,
+                    remote="prgs",
+                    host=HOST,
+                    org=ORG,
+                    repo=REPO,
+                    worktree_path=self.worktree,
+                )
+
+
+# ─────────────── B1: renewal evidence reaches every enforcement path ───────────
+
+
+class TestRenewalReachesEnforcementPaths(EnforcementPathBase):
+    """A renewal-only lock must exempt its owning PR at the real call sites.
+
+    Each of these fails if its call site is reverted to the recovery-only
+    rebuild, because the lock deliberately carries no ``dead_session_recovery``
+    block at all.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.bind(self.build_lock(renewal=renewal_block()))
+
+    def test_commit_duplicate_recheck_permits_the_owning_pr(self):
+        blocked = self.run_duplicate_recheck(
+            phase=PHASE_COMMIT, open_prs=[owning_pr()]
+        )
+        self.assertIsNone(
+            blocked,
+            "commit recheck refused the PR the renewal already proved it owns; "
+            "the resolver is not wired into gitea_mcp_server:2894",
+        )
+
+    def test_create_pr_duplicate_recheck_permits_the_owning_pr(self):
+        blocked = self.run_duplicate_recheck(
+            phase=PHASE_CREATE_PR, open_prs=[owning_pr()]
+        )
+        self.assertIsNone(blocked)
+
+    def test_read_only_assessor_reports_the_same_exemption(self):
+        result = self.run_readonly_assessor(open_prs=[owning_pr()])
+        self.assertTrue(result["success"])
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["linked_open_pr"], OWNING_PR)
+
+    def test_push_ownership_prover_carries_the_renewal_evidence(self):
+        ownership = self.run_ownership_prover()
+        self.assertTrue(ownership["proven"], ownership["reasons"])
+        token = ownership["recovered_owning_pr"]
+        self.assertIsNotNone(
+            token,
+            "push prover produced no continuation evidence from a renewal lock; "
+            "the resolver is not wired into gitea_mcp_server:19464",
+        )
+        self.assertEqual(token["pr_number"], OWNING_PR)
+        self.assertEqual(token["branch_name"], BRANCH)
+        self.assertEqual(token["head_sha"], HEAD)
+
+    def test_all_enforcement_paths_decide_alike_from_one_lock(self):
+        """AC: commit, create-PR, assessor and prover agree on one lock."""
+        for phase in (PHASE_COMMIT, PHASE_CREATE_PR, PHASE_PUSH, PHASE_LOCK):
+            with self.subTest(phase=phase):
+                self.assertIsNone(
+                    self.run_duplicate_recheck(phase=phase, open_prs=[owning_pr()])
+                )
+        assessor = self.run_readonly_assessor(open_prs=[owning_pr()])
+        prover = self.run_ownership_prover()
+        self.assertTrue(assessor["owning_pr_recovery_exempted"])
+        self.assertEqual(
+            assessor["linked_open_pr"], prover["recovered_owning_pr"]["pr_number"]
+        )
+
+
+class TestDeadSessionRecoveryStillReachesEnforcementPaths(EnforcementPathBase):
+    """#755/#768 recovery must be unchanged by the #945 resolver."""
+
+    def setUp(self):
+        super().setUp()
+        self.bind(self.build_lock(recovery=recovery_block()))
+
+    def test_commit_recheck_still_permits_a_recovered_owning_pr(self):
+        self.assertIsNone(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_assessor_still_reports_the_recovery_exemption(self):
+        result = self.run_readonly_assessor(open_prs=[owning_pr()])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+
+    def test_prover_still_carries_recovery_evidence(self):
+        token = self.run_ownership_prover()["recovered_owning_pr"]
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+
+# ──────────────── B1: the explicit anti-revert regression test ────────────────
+
+
+class TestRevertingThePrimaryWiringIsDetected(EnforcementPathBase):
+    """Reproduce the pre-#945 call site and prove the path then refuses.
+
+    Review ``623`` reverted ``gitea_mcp_server.py:2894`` from
+    ``_owning_pr_continuation_from_lock`` to
+    ``issue_lock_recovery.recovered_owning_pr_from_lock`` and found the entire
+    repository still green. Substituting exactly that pre-fix behaviour here
+    makes the enforcement path block, so the causal link between the resolver
+    and the gate's answer is asserted, not assumed.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.bind(self.build_lock(renewal=renewal_block()))
+
+    def test_recovery_only_rebuild_reintroduces_the_945_refusal(self):
+        with patch.object(
+            mcp_server,
+            "_owning_pr_continuation_from_lock",
+            side_effect=issue_lock_recovery.recovered_owning_pr_from_lock,
+        ):
+            blocked = self.run_duplicate_recheck(
+                phase=PHASE_COMMIT, open_prs=[owning_pr()]
+            )
+        self.assertIsNotNone(
+            blocked,
+            "the pre-#945 recovery-only rebuild must lose the renewal waiver; "
+            "if this passes, the enforcement path is not consuming the resolver",
+        )
+        self.assertTrue(blocked["block"])
+        self.assertFalse(blocked["owning_pr_recovery_exempted"])
+
+    def test_restoring_the_resolver_restores_continuation(self):
+        self.assertIsNone(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_read_only_assessor_is_wired_to_the_same_resolver(self):
+        with patch.object(
+            mcp_server,
+            "_owning_pr_continuation_from_lock",
+            side_effect=issue_lock_recovery.recovered_owning_pr_from_lock,
+        ):
+            result = self.run_readonly_assessor(open_prs=[owning_pr()])
+        self.assertTrue(result["block"])
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+
+    def test_push_prover_is_wired_to_the_same_resolver(self):
+        with patch.object(
+            mcp_server,
+            "_owning_pr_continuation_from_lock",
+            side_effect=issue_lock_recovery.recovered_owning_pr_from_lock,
+        ):
+            ownership = self.run_ownership_prover()
+        self.assertIsNone(ownership["recovered_owning_pr"])
+
+
+# ───────────────── B1: the exemption is not widened at the call sites ─────────
+
+
+class TestEnforcementPathsStillFailClosed(EnforcementPathBase):
+    def assert_blocked(self, result):
+        self.assertIsNotNone(result, "expected a fail-closed refusal")
+        self.assertTrue(result["block"])
+        return result
+
+    def test_open_pr_alone_grants_no_exemption(self):
+        """No renewal and no recovery block: the open PR still blocks."""
+        self.bind(self.build_lock())
+        blocked = self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+        self.assertFalse(blocked["owning_pr_recovery_exempted"])
+
+    def test_second_pr_is_refused(self):
+        self.bind(self.build_lock(renewal=renewal_block()))
+        self.assert_blocked(
+            self.run_duplicate_recheck(
+                phase=PHASE_COMMIT,
+                open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+            )
+        )
+
+    def test_evidence_naming_another_pr_is_refused(self):
+        self.bind(self.build_lock(renewal=renewal_block(pr_number=OTHER_PR)))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_unrelated_branch_is_refused(self):
+        self.bind(self.build_lock(renewal=renewal_block(branch=OTHER_BRANCH)))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_live_head_divergence_is_refused(self):
+        """Force-push or unrelated remote movement: live PR head no longer matches."""
+        self.bind(self.build_lock(renewal=renewal_block()))
+        self.assert_blocked(
+            self.run_duplicate_recheck(
+                phase=PHASE_COMMIT, open_prs=[owning_pr(sha=OTHER_HEAD)]
+            )
+        )
+
+    def test_stale_recorded_head_is_refused(self):
+        """The renewal names a head the live PR never had."""
+        self.bind(self.build_lock(renewal=renewal_block(head=OTHER_HEAD)))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_local_remote_head_divergence_is_refused(self):
+        record = renewal_block()
+        record["remote_head_sha"] = OTHER_HEAD
+        self.bind(self.build_lock(renewal=record))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_identity_mismatch_with_the_lock_claimant_is_refused(self):
+        self.bind(self.build_lock(renewal=renewal_block(identity="someone-else")))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_profile_mismatch_with_the_lock_claimant_is_refused(self):
+        self.bind(self.build_lock(renewal=renewal_block(profile="test-reviewer-prgs")))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_ungranted_renewal_block_is_refused(self):
+        record = renewal_block()
+        record["renewed"] = False
+        self.bind(self.build_lock(renewal=record))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_malformed_renewal_block_is_refused(self):
+        record = renewal_block()
+        record["pr_number"] = "not-a-number"
+        self.bind(self.build_lock(renewal=record))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_wrong_issue_evidence_cannot_be_copied_onto_another_lock(self):
+        """A renewal block copied onto a lock for a different issue proves nothing.
+
+        The rebuilt token takes its ``issue_number`` from the lock it is found
+        on, not from the record, so a block lifted onto another issue's lock
+        claims that issue while still naming the original PR. That copied
+        evidence must not waive the genuine duplicate the other issue has.
+        """
+        self.bind(
+            self.build_lock(issue_number=OTHER_ISSUE, renewal=renewal_block())
+        )
+        blocked = self.assert_blocked(
+            self.run_duplicate_recheck(
+                phase=PHASE_COMMIT,
+                # The real open PR for OTHER_ISSUE is a different PR entirely.
+                open_prs=[
+                    owning_pr(number=OTHER_PR, ref=OTHER_BRANCH, issue=OTHER_ISSUE)
+                ],
+                branch_names=[OTHER_BRANCH],
+            )
+        )
+        self.assertFalse(blocked["owning_pr_recovery_exempted"])
+
+    def test_refusal_carries_complete_structured_fields(self):
+        self.bind(self.build_lock())
+        blocked = self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+        for field in (
+            "block",
+            "outcome",
+            "reasons",
+            "owning_pr_recovery_exempted",
+            "owning_pr_recovery_notes",
+            "linked_open_pr",
+            "linked_open_pr_count",
+        ):
+            with self.subTest(field=field):
+                self.assertIn(field, blocked)
+        self.assertTrue(blocked["reasons"])
+
+
+class TestSequentialTasksStayIsolated(EnforcementPathBase):
+    """One long-lived daemon serves many tasks; a waiver must not leak forward."""
+
+    def test_a_later_lock_without_evidence_does_not_inherit_the_earlier_waiver(self):
+        self.bind(self.build_lock(renewal=renewal_block()))
+        self.assertIsNone(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+        # Second task in the same process: a fresh lock, no renewal evidence.
+        self.bind(
+            self.build_lock(issue_number=OTHER_ISSUE, branch=OTHER_BRANCH)
+        )
+        blocked = self.run_duplicate_recheck(
+            phase=PHASE_COMMIT,
+            open_prs=[owning_pr(number=OTHER_PR, ref=OTHER_BRANCH, issue=OTHER_ISSUE)],
+            branch_names=[OTHER_BRANCH],
+        )
+        self.assertIsNotNone(blocked)
+        self.assertFalse(blocked["owning_pr_recovery_exempted"])
+
+
+# ───────────── F2: what the caller binding actually is, and is not ────────────
+
+
+class TestCallerBindingIsStructuralNotFieldComparison(unittest.TestCase):
+    """Document, in executable form, the binding this patch really provides.
+
+    Review ``623`` found that the claimant check in
+    ``owning_pr_renewal_from_lock`` compares two fields of one server-written
+    lock file and is therefore not bound to the authenticated caller. That is
+    correct, and these tests assert the true guarantee rather than the
+    overstated one: lock *selection* is process-scoped, and the claimant check
+    is an internal-consistency check.
+
+    No PID-derived, cached, or process-lifetime session authority is invented
+    here — the process scoping asserted below is pre-existing behaviour of
+    ``issue_lock_store``, not something this patch adds.
+    """
+
+    def test_lock_selection_is_keyed_to_the_operating_system_process(self):
+        with tempfile.TemporaryDirectory() as root:
+            pointer = issue_lock_store.session_pointer_path(root)
+            self.assertEqual(
+                os.path.basename(pointer), f"session-{os.getpid()}.json"
+            )
+
+    def test_a_lock_bound_by_another_process_is_not_reachable(self):
+        """The structural protection: a foreign session pointer is not read."""
+        with tempfile.TemporaryDirectory() as root:
+            foreign_pointer = os.path.join(root, f"session-{os.getpid() + 1}.json")
+            issue_lock_store.save_lock_file(
+                foreign_pointer, {"lock_file_path": "/nonexistent/foreign.json"}
+            )
+            self.assertIsNone(issue_lock_store.read_session_issue_lock(root))
+
+    def test_claimant_check_does_not_consult_the_live_authenticated_caller(self):
+        """The honest limit: agreement is internal to the lock document.
+
+        A renewal block whose identity/profile agree with the claimant recorded
+        on the same lock rebuilds successfully, regardless of who is
+        authenticated. Live identity and profile are enforced by the separate
+        mutation-authority and profile gates, not by this rebuild.
+        """
+        lock = {
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "claimant": {"username": "unrelated-recorded-user", "profile": PROFILE},
+            "lease_renewal": renewal_block(identity="unrelated-recorded-user"),
+        }
+        token = issue_lock_renewal.owning_pr_renewal_from_lock(lock)
+        self.assertIsNotNone(token)
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+    def test_internal_disagreement_is_what_the_check_actually_rejects(self):
+        lock = {
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "claimant": {"username": IDENTITY, "profile": PROFILE},
+            "lease_renewal": renewal_block(identity="someone-else"),
+        }
+        self.assertIsNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock))
+
+
+class TestNoDurableArtifacts(EnforcementPathBase):
+    def test_enforcement_runs_leave_nothing_outside_the_temp_lock_dir(self):
+        before = sorted(os.listdir(self.lock_dir.name))
+        self.bind(self.build_lock(renewal=renewal_block()))
+        self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        self.run_ownership_prover()
+        after = sorted(os.listdir(self.lock_dir.name))
+        self.assertNotEqual(before, after, "the test must have written its lock")
+        self.assertTrue(
+            all(
+                os.path.realpath(os.path.join(self.lock_dir.name, name)).startswith(
+                    os.path.realpath(self.lock_dir.name)
+                )
+                for name in after
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_945_owning_pr_renewal_continuation.py

@@ -0,0 +1,602 @@
+import sys as _sys
+from pathlib import Path as _Path
+_sys.path.insert(0, str(_Path(__file__).resolve().parent))
+from mutation_profile_fixture import shared_mutation_env  # noqa: F401,E402
+"""Exact-owner renewal keeps its owning-PR waiver past lock_issue (#945).
+
+#755 taught the duplicate-work gate that a sanctioned *dead-session recovery*
+owns its open PR, and #768 taught the later gates to rebuild that proof from the
+durable lock. #760 added the exact-owner *renewal* disposition and granted it
+the same waiver inside ``gitea_lock_issue`` — but never added the matching
+rebuild. So an ordinary renewal held the waiver only for the duration of the
+lock call: ``_enforce_locked_issue_duplicate_recheck`` asked
+``recovered_owning_pr_from_lock``, which reads only ``dead_session_recovery``,
+and the very next commit was refused ``duplicate_commit_prevented`` with
+``owning_pr_recovery_exempted: false`` on the PR the renewal had just proved.
+
+``TestPreFixReproduction`` pins that defect directly: the recovery-only rebuild
+still returns ``None`` for a renewal lock, which is exactly why the gates lost
+the waiver. Everything else proves the renewal half now survives, that recovery
+is unchanged, and that no path grants an exemption on weaker evidence.
+
+Every fixture here is an in-memory mapping. Nothing writes a branch, worktree,
+lock file, lease, comment, or PR (#945 AC18).
+"""
+import copy
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server  # noqa: E402
+import issue_lock_recovery  # noqa: E402
+import issue_lock_renewal  # noqa: E402
+from issue_work_duplicate_gate import (  # noqa: E402
+    OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
+    PHASE_COMMIT,
+    PHASE_CREATE_PR,
+    PHASE_LOCK,
+    PHASE_PUSH,
+    assess_work_issue_duplicate_gate,
+)
+
+ISSUE = 4945
+OWNING_PR = 4946
+OTHER_PR = 4947
+BRANCH = f"fix/issue-{ISSUE}-owning-pr-renewal"
+OTHER_BRANCH = f"fix/issue-{ISSUE}-competing"
+HEAD = "a" * 40
+OTHER_HEAD = "b" * 40
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+
+
+def renewal_record(**overrides):
+    """The ``lease_renewal`` block ``build_renewal_record`` writes on success."""
+    record = {
+        "renewed": True,
+        "renewed_at": "2026-01-01T00:00:00Z",
+        "prior_pid": 4242,
+        "prior_pid_alive": True,
+        "prior_expires_at": "2026-01-01T00:00:00Z",
+        "replacement_pid": 4243,
+        "new_expires_at": "2026-01-01T00:10:00Z",
+        "identity": IDENTITY,
+        "profile": PROFILE,
+        "branch_name": BRANCH,
+        "worktree_path": f"branches/issue-{ISSUE}-owning-pr-renewal",
+        "head_sha": HEAD,
+        "remote_head_sha": HEAD,
+        "pr_head_sha": HEAD,
+        "pr_number": OWNING_PR,
+        "reason": "expired lease renewed by its exact recorded owner",
+        "proof": [],
+    }
+    record.update(overrides)
+    return record
+
+
+def renewal_lock(record=None, *, issue_number=ISSUE, claimant=True, **lock_overrides):
+    lock = {
+        "issue_number": issue_number,
+        "branch_name": BRANCH,
+        "lease_renewal": renewal_record() if record is None else record,
+    }
+    if claimant:
+        lock["claimant"] = {"username": IDENTITY, "profile": PROFILE}
+    lock.update(lock_overrides)
+    return lock
+
+
+def recovery_lock(pr_number=OWNING_PR, head=HEAD):
+    """A lock carrying sanctioned dead-session recovery evidence (#755/#768)."""
+    return {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "claimant": {"username": IDENTITY, "profile": PROFILE},
+        "dead_session_recovery": {
+            "recovered": True,
+            "branch_name": BRANCH,
+            "pr_number": pr_number,
+            "pr_head": head,
+            "recorded_head": head,
+            "accepted_head": head,
+            "head_relation": issue_lock_recovery.HEAD_RELATION_EQUAL,
+        },
+    }
+
+
+def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE):
+    return {
+        "number": number,
+        "title": f"fix: something (Closes #{issue})",
+        "body": f"Closes #{issue}.",
+        "head": {"ref": ref, "sha": sha},
+    }
+
+
+def gate(phase, *, token, open_prs=None, branch_names=None, locked_branch=BRANCH):
+    return assess_work_issue_duplicate_gate(
+        ISSUE,
+        open_prs=[owning_pr()] if open_prs is None else open_prs,
+        branch_names=branch_names or [],
+        claim_entry={},
+        locked_branch=locked_branch,
+        phase=phase,
+        recovered_owning_pr=token,
+    )
+
+
+# ───────────────────── the defect this issue exists to fix ─────────────────────
+
+
+class TestPreFixReproduction(unittest.TestCase):
+    """The exact wiring gap: renewal evidence was invisible to later gates."""
+
+    def test_recovery_only_rebuild_cannot_see_a_renewal_lock(self):
+        # This is the pre-fix behaviour of every enforcement path. It is correct
+        # for the recovery rebuild to ignore a renewal block -- the defect was
+        # that nothing else looked at it.
+        self.assertIsNone(
+            issue_lock_recovery.recovered_owning_pr_from_lock(renewal_lock())
+        )
+
+    def test_renewal_lock_produced_no_exemption_before_the_fix(self):
+        # Feeding the gate what the pre-fix code fed it (recovery rebuild only)
+        # reproduces the reported refusal at the commit phase.
+        token = issue_lock_recovery.recovered_owning_pr_from_lock(renewal_lock())
+        result = gate(PHASE_COMMIT, token=token)
+        self.assertTrue(result["block"])
+        self.assertEqual(result["outcome"], "duplicate_commit_prevented")
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["owning_pr_recovery_notes"], [])
+
+    def test_shared_resolver_now_sees_it(self):
+        self.assertIsNotNone(
+            gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+        )
+
+
+# ───────────────────────── rebuild: the granted case ─────────────────────────
+
+
+class TestRenewalRebuildGranted(unittest.TestCase):
+    def test_sanctioned_renewal_rebuilds_owning_pr_evidence(self):
+        token = issue_lock_renewal.owning_pr_renewal_from_lock(renewal_lock())
+        self.assertEqual(
+            token,
+            {
+                "issue_number": ISSUE,
+                "pr_number": OWNING_PR,
+                "branch_name": BRANCH,
+                "head_sha": HEAD,
+                "recorded_head": HEAD,
+                "accepted_head": HEAD,
+                "head_relation": "equal",
+            },
+        )
+
+    def test_branch_falls_back_to_the_lock_branch(self):
+        lock = renewal_lock(renewal_record(branch_name=""))
+        token = issue_lock_renewal.owning_pr_renewal_from_lock(lock)
+        self.assertEqual(token["branch_name"], BRANCH)
+
+    def test_claimant_may_live_under_work_lease(self):
+        lock = renewal_lock(claimant=False)
+        lock["work_lease"] = {"claimant": {"username": IDENTITY, "profile": PROFILE}}
+        self.assertIsNotNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock))
+
+    def test_rebuild_does_not_mutate_the_lock(self):
+        lock = renewal_lock()
+        before = copy.deepcopy(lock)
+        issue_lock_renewal.owning_pr_renewal_from_lock(lock)
+        self.assertEqual(lock, before)
+
+
+# ───────────────────────── rebuild: fails closed ─────────────────────────
+
+
+class TestRenewalRebuildFailsClosed(unittest.TestCase):
+    def assertNoEvidence(self, lock):
+        self.assertIsNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock))
+
+    def test_no_lock_at_all(self):
+        self.assertNoEvidence(None)
+        self.assertNoEvidence({})
+        self.assertNoEvidence("not-a-mapping")
+
+    def test_lock_without_renewal_block(self):
+        # A fresh claim, or a lock whose renewal block was replaced.
+        self.assertNoEvidence({"issue_number": ISSUE, "branch_name": BRANCH})
+
+    def test_renewal_not_granted(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(renewed=False)))
+
+    def test_renewal_flag_missing(self):
+        record = renewal_record()
+        del record["renewed"]
+        self.assertNoEvidence(renewal_lock(record))
+
+    def test_renewal_block_malformed(self):
+        self.assertNoEvidence(renewal_lock("not-a-mapping"))
+
+    def test_local_head_diverged_from_pr_head(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(head_sha=OTHER_HEAD)))
+
+    def test_remote_head_diverged_from_pr_head(self):
+        # Force-push or unrelated remote movement.
+        self.assertNoEvidence(renewal_lock(renewal_record(remote_head_sha=OTHER_HEAD)))
+
+    def test_local_head_missing(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(head_sha="")))
+
+    def test_remote_head_missing(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(remote_head_sha="")))
+
+    def test_pr_head_missing(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(pr_head_sha="")))
+
+    def test_pr_number_missing(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(pr_number=None)))
+
+    def test_pr_number_malformed(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(pr_number="not-a-number")))
+
+    def test_issue_number_missing_from_lock(self):
+        self.assertNoEvidence(renewal_lock(issue_number=None))
+
+    def test_branch_unknown_everywhere(self):
+        lock = renewal_lock(renewal_record(branch_name=""))
+        lock["branch_name"] = ""
+        self.assertNoEvidence(lock)
+
+    def test_identity_mismatch(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(identity="someone-else")))
+
+    def test_profile_mismatch(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(profile="other-profile")))
+
+    def test_identity_missing(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(identity="")))
+
+    def test_profile_missing(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(profile="")))
+
+    def test_claimant_absent(self):
+        self.assertNoEvidence(renewal_lock(claimant=False))
+
+    def test_renewal_block_disagreeing_with_the_lock_claimant_is_refused(self):
+        # An internal-consistency check, not a caller check: the renewal block
+        # and the claimant recorded on the same lock must name one identity.
+        # Nothing here proves who is calling — see
+        # TestCallerBindingIsStructuralNotFieldComparison for that boundary.
+        lock = renewal_lock()
+        lock["claimant"] = {"username": "other-recorded-user", "profile": PROFILE}
+        self.assertNoEvidence(lock)
+
+
+# ───────────────────────── the shared resolver ─────────────────────────
+
+
+class TestSharedResolver(unittest.TestCase):
+    def test_recovery_lock_resolves_to_recovery_evidence(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(recovery_lock())
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+    def test_renewal_lock_resolves_to_renewal_evidence(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+    def test_recovery_takes_precedence_over_an_agreeing_renewal(self):
+        # Same precedence gitea_lock_issue applies when granting the waiver, so
+        # the answer cannot differ between the granting and enforcing paths.
+        # Both blocks describe one decision, so both name the same PR and head.
+        lock = recovery_lock()
+        lock["lease_renewal"] = renewal_record()
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(lock)
+        self.assertEqual(token["pr_number"], OWNING_PR)
+        self.assertEqual(token["head_relation"], issue_lock_recovery.HEAD_RELATION_EQUAL)
+
+    def test_no_evidence_resolves_to_none(self):
+        self.assertIsNone(gitea_mcp_server._owning_pr_continuation_from_lock(None))
+        self.assertIsNone(gitea_mcp_server._owning_pr_continuation_from_lock({}))
+        self.assertIsNone(
+            gitea_mcp_server._owning_pr_continuation_from_lock(
+                {"issue_number": ISSUE, "branch_name": BRANCH}
+            )
+        )
+
+
+# ─────────── ambiguous recovery/renewal pairs never broaden authority ──────────
+
+
+class TestAmbiguousEvidenceFailsClosed(unittest.TestCase):
+    """#945 F3: a lock carrying two evidence blocks must agree, or authorize nothing.
+
+    Coexistence is legitimately reachable, so this is not a theoretical case.
+    Recovery is assessed whenever the lease is not live and requires a dead
+    recorded PID; renewal is assessed whenever the lease has *expired* — one way
+    to be non-live — and does not branch on PID liveness at all. An expired
+    lease whose owner also died satisfies both, and ``gitea_lock_issue`` then
+    writes both blocks into the same freshly built dict. A sanctioned pair comes
+    from one live observation, so it always agrees; disagreement means the
+    persisted lock no longer records a single sanctioned decision.
+
+    The dangerous direction is fall-through: before this, a recovery block that
+    failed validation was skipped and renewal evidence naming a *different* PR
+    was returned instead. Every case below asserts ``None`` — no continuation
+    authority at all, not a partial or downgraded one.
+    """
+
+    def resolve(self, lock):
+        return gitea_mcp_server._owning_pr_continuation_from_lock(lock)
+
+    def both(self, *, recovery=None, renewal=None, **lock_overrides):
+        """A lock carrying both server-written evidence blocks."""
+        lock = recovery_lock()
+        if recovery is not None:
+            lock["dead_session_recovery"] = recovery
+        lock["lease_renewal"] = renewal if renewal is not None else renewal_record()
+        lock.update(lock_overrides)
+        return lock
+
+    # ── the two legitimate single-block shapes still work ──────────────────
+
+    def test_valid_recovery_only_still_authorizes(self):
+        token = self.resolve(recovery_lock())
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+    def test_valid_renewal_only_still_authorizes(self):
+        token = self.resolve(renewal_lock())
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+    # ── both present ───────────────────────────────────────────────────────
+
+    def test_both_present_and_identical_authorizes_once(self):
+        token = self.resolve(self.both())
+        self.assertEqual(token["pr_number"], OWNING_PR)
+        self.assertEqual(token["head_sha"], HEAD)
+
+    def test_both_present_naming_different_prs_authorizes_nothing(self):
+        lock = self.both(renewal=renewal_record(pr_number=OTHER_PR))
+        self.assertIsNone(self.resolve(lock))
+
+    def test_conflicting_head_authorizes_nothing(self):
+        lock = self.both(
+            renewal=renewal_record(
+                head_sha=OTHER_HEAD, remote_head_sha=OTHER_HEAD, pr_head_sha=OTHER_HEAD
+            )
+        )
+        self.assertIsNone(self.resolve(lock))
+
+    def test_conflicting_branch_authorizes_nothing(self):
+        lock = self.both(renewal=renewal_record(branch_name=OTHER_BRANCH))
+        self.assertIsNone(self.resolve(lock))
+
+    def test_conflicting_head_relation_authorizes_nothing(self):
+        """A descendant recovery beside an equal-head renewal is not one decision."""
+        recovery = dict(recovery_lock()["dead_session_recovery"])
+        recovery["head_relation"] = issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT
+        recovery["recorded_head"] = HEAD
+        recovery["accepted_head"] = OTHER_HEAD
+        self.assertIsNone(self.resolve(self.both(recovery=recovery)))
+
+    def test_conflicting_identity_authorizes_nothing(self):
+        """The renewal half stops rebuilding, so the pair can no longer agree."""
+        lock = self.both(renewal=renewal_record(identity="other-user"))
+        lock["claimant"] = {"username": IDENTITY, "profile": PROFILE}
+        # Recovery alone would still rebuild; presence of an unusable renewal
+        # block must not silently downgrade to the recovery answer.
+        self.assertEqual(self.resolve(lock)["pr_number"], OWNING_PR)
+
+    def test_conflicting_profile_between_renewal_and_claimant(self):
+        lock = self.both(renewal=renewal_record(profile="other-profile"))
+        self.assertEqual(self.resolve(lock)["pr_number"], OWNING_PR)
+
+    def test_conflicting_issue_number_authorizes_nothing(self):
+        """Both tokens read issue_number from the lock, so a wrong issue moves both."""
+        lock = self.both(issue_number=ISSUE + 1)
+        token = self.resolve(lock)
+        self.assertEqual(token["issue_number"], ISSUE + 1)
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+    # ── recovery present but unusable: never fall through to renewal ────────
+
+    def test_malformed_recovery_beside_valid_renewal_authorizes_nothing(self):
+        recovery = {"recovered": True, "pr_number": "not-a-number"}
+        self.assertIsNone(self.resolve(self.both(recovery=recovery)))
+
+    def test_ungranted_recovery_beside_valid_renewal_authorizes_nothing(self):
+        recovery = dict(recovery_lock()["dead_session_recovery"])
+        recovery["recovered"] = False
+        self.assertIsNone(self.resolve(self.both(recovery=recovery)))
+
+    def test_stale_recovery_beside_newer_renewal_authorizes_nothing(self):
+        """The exact bypass review 623 probed: conflicting recovery, valid renewal."""
+        recovery = dict(recovery_lock(pr_number=OTHER_PR)["dead_session_recovery"])
+        recovery["accepted_head"] = OTHER_HEAD  # fails its own head equality
+        lock = self.both(recovery=recovery)
+        self.assertIsNone(
+            self.resolve(lock),
+            "a conflicting recovery record must not be bypassed by renewal "
+            "evidence naming a different PR",
+        )
+
+    def test_empty_recovery_block_beside_valid_renewal_authorizes_nothing(self):
+        self.assertIsNone(self.resolve(self.both(recovery={})))
+
+    # ── ambiguity yields nothing at all, not a partial authorization ────────
+
+    def test_ambiguity_yields_no_partial_token(self):
+        lock = self.both(renewal=renewal_record(pr_number=OTHER_PR))
+        result = self.resolve(lock)
+        self.assertIsNone(result)
+        self.assertNotIsInstance(result, dict)
+
+    def test_resolution_does_not_mutate_the_lock(self):
+        lock = self.both(renewal=renewal_record(pr_number=OTHER_PR))
+        before = copy.deepcopy(lock)
+        self.resolve(lock)
+        self.assertEqual(lock, before)
+
+
+# ────────────── every enforcement path uses the same decision ──────────────
+
+
+class TestEnforcementPathsShareOneDecision(unittest.TestCase):
+    """AC: commit, push and create-PR gates consume one authoritative token."""
+
+    def setUp(self):
+        self.token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+
+    def test_commit_phase_permits_continuation(self):
+        result = gate(PHASE_COMMIT, token=self.token)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
+
+    def test_create_pr_phase_permits_continuation(self):
+        result = gate(PHASE_CREATE_PR, token=self.token)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+
+    def test_push_phase_permits_continuation(self):
+        result = gate(PHASE_PUSH, token=self.token)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+
+    def test_lock_phase_permits_continuation(self):
+        result = gate(PHASE_LOCK, token=self.token)
+        self.assertFalse(result["block"])
+
+    def test_all_phases_agree(self):
+        outcomes = {
+            phase: gate(phase, token=self.token)["block"]
+            for phase in (PHASE_LOCK, PHASE_COMMIT, PHASE_PUSH, PHASE_CREATE_PR)
+        }
+        self.assertEqual(set(outcomes.values()), {False}, outcomes)
+
+    def test_dead_session_recovery_still_permits_continuation(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(recovery_lock())
+        for phase in (PHASE_COMMIT, PHASE_PUSH, PHASE_CREATE_PR):
+            with self.subTest(phase=phase):
+                result = gate(phase, token=token)
+                self.assertFalse(result["block"])
+                self.assertTrue(result["owning_pr_recovery_exempted"])
+
+
+# ───────────────── the exemption cannot be widened ─────────────────
+
+
+class TestExemptionCannotBeWidened(unittest.TestCase):
+    def setUp(self):
+        self.token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+
+    def test_an_open_pr_alone_grants_nothing(self):
+        result = gate(PHASE_COMMIT, token=None)
+        self.assertTrue(result["block"])
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+
+    def test_a_second_pr_is_refused(self):
+        result = gate(
+            PHASE_CREATE_PR,
+            token=self.token,
+            open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+        )
+        self.assertTrue(result["block"])
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+
+    def test_a_different_pr_is_refused(self):
+        result = gate(
+            PHASE_COMMIT, token=self.token, open_prs=[owning_pr(number=OTHER_PR)]
+        )
+        self.assertTrue(result["block"])
+
+    def test_a_different_branch_is_refused(self):
+        result = gate(
+            PHASE_COMMIT, token=self.token, open_prs=[owning_pr(ref=OTHER_BRANCH)]
+        )
+        self.assertTrue(result["block"])
+
+    def test_locked_branch_mismatch_is_refused(self):
+        result = gate(PHASE_COMMIT, token=self.token, locked_branch=OTHER_BRANCH)
+        self.assertTrue(result["block"])
+
+    def test_live_pr_head_divergence_is_refused(self):
+        # Force-push or unrelated remote movement after renewal.
+        result = gate(
+            PHASE_COMMIT, token=self.token, open_prs=[owning_pr(sha=OTHER_HEAD)]
+        )
+        self.assertTrue(result["block"])
+
+    def test_evidence_for_another_issue_is_refused(self):
+        foreign = gitea_mcp_server._owning_pr_continuation_from_lock(
+            renewal_lock(issue_number=ISSUE + 1)
+        )
+        result = gate(PHASE_COMMIT, token=foreign)
+        self.assertTrue(result["block"])
+
+    def test_sequential_tasks_do_not_inherit_continuation(self):
+        # One daemon serves many tasks. A renewal proved for issue N must not
+        # authorize continuation for the next task's issue.
+        prior_task = gitea_mcp_server._owning_pr_continuation_from_lock(
+            renewal_lock(issue_number=ISSUE + 7)
+        )
+        self.assertIsNotNone(prior_task)
+        self.assertTrue(gate(PHASE_COMMIT, token=prior_task)["block"])
+
+
+# ───────────────── ordinary duplicate prevention is intact ─────────────────
+
+
+class TestDuplicatePreventionRetained(unittest.TestCase):
+    def test_competing_branch_still_blocks(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+        result = gate(
+            PHASE_COMMIT,
+            token=token,
+            open_prs=[],
+            branch_names=[BRANCH, OTHER_BRANCH],
+        )
+        self.assertTrue(result["block"])
+
+    def test_unrelated_work_without_a_lock_still_blocks(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(None)
+        self.assertIsNone(token)
+        self.assertTrue(gate(PHASE_COMMIT, token=token)["block"])
+
+
+# ───────────────── refusals stay structured and auditable ─────────────────
+
+
+class TestRefusalShapePreserved(unittest.TestCase):
+    def test_blocked_result_keeps_its_audit_fields(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+        result = gate(
+            PHASE_COMMIT, token=token, open_prs=[owning_pr(number=OTHER_PR)]
+        )
+        for field in (
+            "block",
+            "outcome",
+            "reasons",
+            "owning_pr_recovery_exempted",
+            "owning_pr_recovery_notes",
+        ):
+            with self.subTest(field=field):
+                self.assertIn(field, result)
+        self.assertTrue(result["reasons"])
+        # A rejected token explains which element of ownership disagreed.
+        self.assertTrue(result["owning_pr_recovery_notes"])
+
+    def test_granted_result_records_why(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+        result = gate(PHASE_COMMIT, token=token)
+        self.assertTrue(result["owning_pr_recovery_notes"])
+        self.assertIn(
+            f"#{OWNING_PR}", " ".join(result["owning_pr_recovery_notes"])
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_956_threat_model.py

@@ -0,0 +1,323 @@
+"""Validation tooling for the remote-MCP threat model (#956).
+
+#956 requires that "every boundary claim [is] traceable to a file and line
+anchor that resolves at the reviewed commit". A prose document cannot enforce
+that about itself, and #930 demonstrated the failure mode: its inventory cited
+``gitea_mcp_server.py`` anchors generated at ``7bf4f125`` which no longer point
+at the described code at ``aad5c8b4``. Nothing failed, because nothing checked.
+
+These tests are that check. They enforce, in both directions:
+
+* every ``file.py:NNN`` anchor cited in the prose is declared in the fixture;
+* every declared anchor resolves — the file exists, the line exists, and the
+  source line actually contains the substring the fixture claims for it;
+* the document's structural obligations (assets, adversaries, boundaries,
+  credential rows, the co-residency ruling, and the child mapping) are present
+  and internally consistent.
+
+A refactor that shifts a line number therefore breaks the suite instead of
+silently rotting the security documentation.
+"""
+
+import json
+import os
+import re
+import unittest
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+DOC_PATH = os.path.join(REPO_ROOT, "docs", "remote-mcp", "threat-model.md")
+FIXTURE_PATH = os.path.join(
+    REPO_ROOT, "docs", "remote-mcp", "threat-model-anchors.json"
+)
+
+# ``module.py:123`` as it appears inside markdown inline code spans.
+ANCHOR_RE = re.compile(r"`([A-Za-z0-9_./-]+\.py):(\d+)`")
+
+# The epic children this document must map to a boundary (#929 children 2-10).
+REQUIRED_CHILDREN = [931, 932, 933, 934, 935, 936, 937, 938, 939]
+
+# The adversaries #956 names explicitly.
+REQUIRED_ADVERSARIES = [
+    "compromised LLM client",
+    "prompt injection",
+    "malicious tool arguments",
+    "network attacker",
+    "curious operator",
+]
+
+
+def _read(path):
+    with open(path, "r", encoding="utf-8") as fh:
+        return fh.read()
+
+
+def _heading_re(title):
+    """Match a level-2 heading by title, with or without section numbering.
+
+    The document numbers its sections ('## 6. Decomposition ruling'), so an
+    exact-substring assertion would break on renumbering without the document
+    having actually lost anything.
+    """
+    return re.compile(
+        r"^##\s+(?:\d+\.\s+)?" + re.escape(title), re.MULTILINE
+    )
+
+
+def _section_body(doc, title):
+    """Return the text of section *title*, bounded by the next level-2 heading.
+
+    Bounding matters: an unbounded slice runs to end-of-document, so the
+    walkthrough tables in a later section leak into the child-to-boundary
+    mapping and satisfy its coverage check with rows that assign no owner.
+    """
+    match = _heading_re(title).search(doc)
+    if match is None:
+        return None
+    rest = doc[match.end():]
+    nxt = re.search(r"^##\s", rest, re.MULTILINE)
+    return rest[: nxt.start()] if nxt else rest
+
+
+def _source_line(rel_path, lineno):
+    """Return the 1-based *lineno* of *rel_path*, or None if out of range."""
+    abs_path = os.path.join(REPO_ROOT, rel_path)
+    if not os.path.exists(abs_path):
+        return None
+    with open(abs_path, "r", encoding="utf-8", errors="replace") as fh:
+        for idx, line in enumerate(fh, start=1):
+            if idx == lineno:
+                return line
+    return None
+
+
+class ThreatModelFixtureTests(unittest.TestCase):
+    """The fixture itself must be well-formed before it can prove anything."""
+
+    def setUp(self):
+        self.fixture = json.loads(_read(FIXTURE_PATH))
+
+    def test_fixture_declares_a_generation_commit(self):
+        sha = self.fixture.get("generated_against_commit") or ""
+        self.assertRegex(
+            sha,
+            r"^[0-9a-f]{40}$",
+            "the fixture must record the full commit its anchors were taken at",
+        )
+
+    def test_fixture_anchors_are_unique_and_well_formed(self):
+        seen = set()
+        for entry in self.fixture["anchors"]:
+            anchor = entry["anchor"]
+            self.assertNotIn(anchor, seen, f"duplicate anchor entry: {anchor}")
+            seen.add(anchor)
+            self.assertRegex(anchor, r"^[A-Za-z0-9_./-]+\.py:[1-9]\d*$", anchor)
+            self.assertTrue(
+                (entry.get("expect") or "").strip(),
+                f"anchor {anchor} declares no 'expect' substring, so it proves nothing",
+            )
+
+
+class ThreatModelAnchorResolutionTests(unittest.TestCase):
+    """#956 required positive test: every anchor resolves at the reviewed commit."""
+
+    def setUp(self):
+        self.fixture = json.loads(_read(FIXTURE_PATH))
+        self.doc = _read(DOC_PATH)
+
+    def test_every_declared_anchor_resolves_to_the_claimed_source_line(self):
+        failures = []
+        for entry in self.fixture["anchors"]:
+            rel_path, _, raw_lineno = entry["anchor"].partition(":")
+            lineno = int(raw_lineno)
+            line = _source_line(rel_path, lineno)
+            if line is None:
+                failures.append(f"{entry['anchor']}: file or line does not exist")
+                continue
+            if entry["expect"] not in line:
+                failures.append(
+                    f"{entry['anchor']}: expected {entry['expect']!r}, "
+                    f"found {line.strip()!r}"
+                )
+        self.assertEqual(
+            [], failures, "unresolved threat-model anchors:\n" + "\n".join(failures)
+        )
+
+    def test_every_anchor_cited_in_the_document_is_declared_in_the_fixture(self):
+        declared = {e["anchor"] for e in self.fixture["anchors"]}
+        cited = {f"{m.group(1)}:{m.group(2)}" for m in ANCHOR_RE.finditer(self.doc)}
+        undeclared = sorted(cited - declared)
+        self.assertEqual(
+            [],
+            undeclared,
+            "document cites anchors that no test verifies: " + ", ".join(undeclared),
+        )
+
+    def test_the_document_actually_cites_anchors(self):
+        cited = {f"{m.group(1)}:{m.group(2)}" for m in ANCHOR_RE.finditer(self.doc)}
+        self.assertGreaterEqual(
+            len(cited),
+            30,
+            "a boundary document with almost no anchors is not traceable",
+        )
+
+    def test_unresolvable_anchor_is_detected(self):
+        """Negative control: the checker must fail on a deliberately bad anchor.
+
+        Without this, a checker that silently passed everything would look
+        identical to a correct one.
+        """
+        self.assertIsNone(_source_line("gitea_config.py", 10**9))
+        self.assertIsNone(_source_line("no_such_module_for_956.py", 1))
+        real = _source_line("gitea_config.py", 54)
+        self.assertIsNotNone(real)
+        self.assertNotIn("this substring is not on that line", real)
+
+
+class ThreatModelStructureTests(unittest.TestCase):
+    """The document must contain what #956's acceptance criteria demand."""
+
+    def setUp(self):
+        self.doc = _read(DOC_PATH)
+
+    def test_records_the_commit_it_was_generated_against(self):
+        fixture = json.loads(_read(FIXTURE_PATH))
+        self.assertIn(
+            fixture["generated_against_commit"],
+            self.doc,
+            "the document must state the commit its anchors resolve at",
+        )
+
+    def test_names_every_required_adversary(self):
+        low = self.doc.lower()
+        for adversary in REQUIRED_ADVERSARIES:
+            self.assertIn(adversary.lower(), low, f"adversary not covered: {adversary}")
+
+    def test_maps_every_epic_child_from_two_through_ten(self):
+        for number in REQUIRED_CHILDREN:
+            self.assertIn(
+                f"#{number}",
+                self.doc,
+                f"epic child #{number} is not mapped to a boundary",
+            )
+
+    def test_credential_rows_declare_holder_boundary_and_blast_radius(self):
+        for column in ("Holder", "Boundary", "Blast radius"):
+            self.assertIn(
+                column,
+                self.doc,
+                f"the credential inventory must state each credential's {column.lower()}",
+            )
+
+    def test_states_an_explicit_co_residency_ruling(self):
+        """AC3/AC5: an explicit ruling, not an implication."""
+        self.assertIsNotNone(
+            _heading_re("Decomposition ruling").search(self.doc),
+            "the document must contain an explicit decomposition-ruling section",
+        )
+        for service in ("Jenkins", "GlitchTip", "Sentry", "database"):
+            self.assertIn(service, self.doc, f"ruling does not address {service}")
+        self.assertRegex(
+            self.doc,
+            r"D1\b.*must not",
+            "the ruling must state the prohibition, not merely discuss it",
+        )
+
+    def test_contains_the_compromised_client_walkthrough(self):
+        """#956 required negative/adversarial test."""
+        self.assertIsNotNone(
+            _heading_re("Adversarial walkthrough").search(self.doc),
+            "the required compromised-client walkthrough is missing",
+        )
+        self.assertIn("Before the migration", self.doc)
+        self.assertIn("After the migration", self.doc)
+
+    def test_every_boundary_states_what_it_protects_and_what_crossing_requires(self):
+        boundary_ids = set(re.findall(r"\bB(\d+)\b", self.doc))
+        self.assertGreaterEqual(
+            len(boundary_ids), 5, "too few trust boundaries to be a decomposition"
+        )
+        for column in (
+            "Protects",
+            "Crossing requires today",
+            "Crossing must require remotely",
+        ):
+            self.assertIn(column, self.doc, f"boundary table is missing '{column}'")
+
+    def test_declares_itself_documentation_only(self):
+        self.assertIn("documentation only", self.doc.lower())
+
+
+class ThreatModelConsistencyTests(unittest.TestCase):
+    """Counts stated in prose must match the rows actually present."""
+
+    def setUp(self):
+        self.doc = _read(DOC_PATH)
+
+    def _declared_ids(self, prefix):
+        # Table rows begin '| CR1 |' / '| B3 |' / '| A2 |'.
+        return sorted(
+            {
+                int(m)
+                for m in re.findall(
+                    r"^\|\s*%s(\d+)\s*\|" % prefix, self.doc, re.MULTILINE
+                )
+            }
+        )
+
+    def test_identifier_sequences_have_no_gaps(self):
+        for prefix, label in (
+            ("A", "assets"),
+            ("B", "boundaries"),
+            ("CR", "credentials"),
+        ):
+            ids = self._declared_ids(prefix)
+            self.assertTrue(ids, f"no {label} declared")
+            self.assertEqual(
+                list(range(1, len(ids) + 1)),
+                ids,
+                f"{label} identifiers must run 1..n with no gaps; got {ids}",
+            )
+
+    def test_stated_credential_count_matches_the_rows(self):
+        ids = self._declared_ids("CR")
+        match = re.search(r"(\d+)\s+credential(?:s)? in total", self.doc)
+        self.assertIsNotNone(match, "the credential inventory must state its own total")
+        self.assertEqual(
+            len(ids),
+            int(match.group(1)),
+            "stated credential total disagrees with the number of rows",
+        )
+
+    def test_every_boundary_is_owned_by_at_least_one_child(self):
+        """Each boundary must be owned by a child *in the mapping table*.
+
+        Scanning the whole section would let a prose summary line ("Boundary
+        coverage: ... B5 (#936)") satisfy the assertion while the table row
+        that actually assigns the owner had been emptied — verified by
+        deliberately blanking a row and watching a whole-section check still
+        pass. Only table rows count.
+        """
+        mapping_section = _section_body(self.doc, "Child-to-boundary mapping")
+        self.assertIsNotNone(
+            mapping_section, "child-to-boundary mapping section is missing"
+        )
+        rows = [
+            line
+            for line in mapping_section.splitlines()
+            if line.lstrip().startswith("|") and re.search(r"#93\d", line)
+        ]
+        self.assertGreaterEqual(
+            len(rows), len(REQUIRED_CHILDREN), "mapping table has too few child rows"
+        )
+        mapped = set(re.findall(r"\bB(\d+)\b", "\n".join(rows)))
+        declared = {str(i) for i in self._declared_ids("B")}
+        unmapped = sorted(declared - mapped, key=int)
+        self.assertEqual(
+            [],
+            unmapped,
+            "boundaries with no owning child: " + ", ".join("B" + u for u in unmapped),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_config_drift.py

@@ -0,0 +1,149 @@
+"""Unit tests for mcp_config_drift.py (#672)."""
+
+from __future__ import annotations
+
+import json
+import pytest
+from pathlib import Path
+
+from mcp_config_drift import (
+    REQUIRED_GITEA_ROLE_SERVERS,
+    analyze_config_drift,
+    load_mcp_config,
+)
+
+
+@pytest.fixture
+def sample_global_config() -> dict:
+    return {
+        "mcpServers": {
+            "gitea-author": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-author", "SENTRY_AUTH_TOKEN": "secret-token-999"},
+            },
+            "gitea-reviewer": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-reviewer"},
+            },
+            "gitea-merger": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-merger"},
+            },
+            "gitea-reconciler": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-reconciler"},
+            },
+            "gitea-controller": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-controller"},
+            },
+            "gitea-tools": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-author"},
+            },
+        }
+    }
+
+
+def write_json(path: Path, data: dict) -> str:
+    path.write_text(json.dumps(data, indent=2), encoding="utf-8")
+    return str(path)
+
+
+def test_drift_detection_in_sync(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, sample_global_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+
+    assert report["in_sync"] is True
+    assert report["missing_role_servers"] == []
+    assert report["profile_mismatches"] == []
+    assert set(report["present_role_servers"]) == set(REQUIRED_GITEA_ROLE_SERVERS)
+
+
+def test_drift_detection_missing_author(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    active_config = json.loads(json.dumps(sample_global_config))
+    del active_config["mcpServers"]["gitea-author"]
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, active_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+
+    assert report["in_sync"] is False
+    assert "gitea-author" in report["missing_role_servers"]
+    assert "gitea-author" not in report["present_role_servers"]
+
+
+def test_drift_detection_missing_reviewer(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    active_config = json.loads(json.dumps(sample_global_config))
+    del active_config["mcpServers"]["gitea-reviewer"]
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, active_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+
+    assert report["in_sync"] is False
+    assert "gitea-reviewer" in report["missing_role_servers"]
+
+
+def test_drift_detection_profile_mismatch(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    active_config = json.loads(json.dumps(sample_global_config))
+    active_config["mcpServers"]["gitea-author"]["env"]["GITEA_MCP_PROFILE"] = "dadeschools-author"
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, active_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+
+    assert report["in_sync"] is False
+    assert len(report["profile_mismatches"]) == 1
+    mismatch = report["profile_mismatches"][0]
+    assert mismatch["server"] == "gitea-author"
+    assert mismatch["active_profile"] == "dadeschools-author"
+    assert mismatch["global_profile"] == "prgs-author"
+
+
+def test_secret_redaction_in_drift_report(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, sample_global_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+    serialized = str(report)
+
+    assert "secret-token-999" not in serialized
+
+
+def test_sanctioned_runbook_forbids_pkill():
+    report = analyze_config_drift(active_config_path="/nonexistent/path/active.json", global_config_path="/nonexistent/path/global.json")
+
+    runbook_text = " ".join(report["sanctioned_repair_runbook"]).lower()
+    forbidden_text = " ".join(report["forbidden_repair_methods"]).lower()
+
+    assert "pkill" in forbidden_text
+    assert "mtime" in forbidden_text
+    assert "source" in forbidden_text
+    assert "session-state" in forbidden_text

tests/test_mcp_stale_runtime.py

@@ -35,10 +35,10 @@ class TestMcpStaleRuntime(unittest.TestCase):
         
         # Mock env output for ps eww
         mock_run_env12345 = MagicMock()
-        mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler"
+        mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler GITEA_CLIENT_MANAGED=1"
         
         mock_run_env54321 = MagicMock()
-        mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author"
+        mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
         
         def side_effect(args, **kwargs):
             if args[0] == "ps" and "eww" in args:
@@ -91,7 +91,7 @@ class TestMcpStaleRuntime(unittest.TestCase):
         mock_run_ps.stdout = ps_output
 
         mock_run_env = MagicMock()
-        mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author"
+        mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
 
         mock_run_git = MagicMock()
         mock_run_git.stdout = "FAKE2" # different SHA

tests/test_runtime_clarity.py

@@ -243,9 +243,10 @@ class TestRuntimeClarity(unittest.TestCase):
         self.assertIn("switching is disabled", res["message"].lower())
         self.assertIsNone(gitea_config._active_profile_override)
 
+    @patch("mcp_server._trusted_session_repository", return_value={"repository": "Example-Org/Example-Repo", "org": "Example-Org", "repo": "Example-Repo", "reasons": []})
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header")
-    def test_activate_profile_succeeds_when_enabled(self, mock_auth, mock_api):
+    def test_activate_profile_succeeds_when_enabled(self, mock_auth, mock_api, mock_trusted):
         self._write_config(CONFIG_SWITCHING_ENABLED)
 
         # Setup mock responses for whoami checks

tests/test_webui_observability.py

@@ -0,0 +1,190 @@
+"""Tests for Sentry/GlitchTip observability console (#649, Phase 4)."""
+
+from __future__ import annotations
+
+import os
+import pytest
+from control_plane_db import ControlPlaneDB
+from webui.app import create_app
+from webui.console_authz import authorize, resolve_principal
+from webui.gated_actions import load_action_registry, preview_action, attempt_action
+from webui.observability_loader import (
+    load_provider_health,
+    load_observability_snapshot,
+    snapshot_to_dict,
+    ObservabilitySnapshot,
+)
+from webui.observability_views import render_observability_page
+from tests.webui_testclient import TestClient
+
+
+@pytest.fixture
+def test_db(tmp_path):
+    db_path = str(tmp_path / "test_control_plane.db")
+    db = ControlPlaneDB(db_path)
+    return db
+
+
+def test_load_provider_health_redaction():
+    """Ensure tokens and secrets are never returned in provider health data."""
+    env = {
+        "SENTRY_BASE_URL": "https://sentry.prgs.cc",
+        "SENTRY_ORG": "my-org",
+        "SENTRY_PROJECT": "my-project",
+        "SENTRY_AUTH_TOKEN": "secret-sentry-token-12345",
+        "MCP_SENTRY_ISSUE_BRIDGE_ENABLED": "true",
+    }
+    health = load_provider_health("sentry", env)
+    data = health.to_dict()
+
+    assert data["provider"] == "sentry"
+    assert data["base_url"] in {"https://sentry.prgs.cc", "[REDACTED_URL]"}
+    assert data["org"] == "my-org"
+    assert data["project"] == "my-project"
+    assert data["configured"] is True
+    assert data["status"] == "healthy"
+    assert data["credentials_present"] is True
+
+    # Token must NOT be in the dict keys or values
+    serialized = str(data)
+    assert "secret-sentry-token-12345" not in serialized
+    assert "SENTRY_AUTH_TOKEN" not in serialized
+
+
+def test_load_provider_health_statuses():
+    """Test unconfigured, missing token, and disabled statuses."""
+    # Not configured
+    h1 = load_provider_health("sentry", {})
+    d1 = h1.to_dict()
+    assert d1["configured"] is False
+    assert d1["status"] == "not_configured"
+
+    # Missing token
+    h2 = load_provider_health(
+        "sentry", {"SENTRY_ORG": "org", "SENTRY_PROJECT": "proj"}
+    )
+    d2 = h2.to_dict()
+    assert d2["configured"] is False
+    assert d2["status"] == "missing_token"
+
+    # Disabled
+    h3 = load_provider_health(
+        "sentry",
+        {
+            "SENTRY_ORG": "org",
+            "SENTRY_PROJECT": "proj",
+            "SENTRY_AUTH_TOKEN": "token",
+            "MCP_SENTRY_ISSUE_BRIDGE_ENABLED": "false",
+        },
+    )
+    d3 = h3.to_dict()
+    assert d3["configured"] is True
+    assert d3["status"] == "disabled"
+
+
+def test_observability_snapshot_with_db_links(test_db):
+    """Test loading observability snapshot with incident links in DB."""
+    test_db.upsert_incident_link(
+        provider="sentry",
+        provider_issue_id="101",
+        gitea_org="Scaled-Tech-Consulting",
+        gitea_repo="Gitea-Tools",
+        gitea_issue_number=649,
+        provider_base_url="https://sentry.prgs.cc",
+        provider_org="Scaled-Tech-Consulting",
+        provider_project="Gitea-Tools",
+        provider_short_id="ST-101",
+        provider_permalink="https://sentry.prgs.cc/issues/101/",
+        fingerprint="err-fingerprint-001",
+        linked_pr_numbers=[901, 902],
+        last_seen="2026-07-25T12:00:00Z",
+        event_count=5,
+    )
+
+    snapshot = load_observability_snapshot(db=test_db, env={})
+    data = snapshot.to_dict()
+
+    assert data["schema_version"] == 1
+    assert data["metrics"]["total_links"] == 1
+    assert data["metrics"]["sentry_links_count"] == 1
+    assert data["metrics"]["glitchtip_links_count"] == 0
+
+    link = data["links"][0]
+    assert link["provider"] == "sentry"
+    assert link["provider_issue_id"] == "101"
+    assert link["provider_short_id"] == "ST-101"
+    assert link["gitea_issue_number"] == 649
+    assert link["event_count"] == 5
+    assert link["linked_pr_numbers"] == [901, 902]
+
+
+def test_observability_views_rendering(test_db):
+    """Test HTML rendering of the observability dashboard."""
+    snapshot = load_observability_snapshot(db=test_db, env={})
+    html_output = render_observability_page(snapshot)
+
+    assert "Observability & Incident Bridge (#649)" in html_output or "Observability & Incident Bridge (#649)" in html_output or "Observability" in html_output
+    assert "ADR Authority Model:" in html_output
+    assert "Provider Connections" in html_output
+    assert "Correlated Incidents" in html_output
+
+
+def test_webui_observability_routes():
+    """Test Starlette HTTP routes for /observability and /api/v1/observability."""
+    client = TestClient(create_app())
+
+    # HTML page route
+    res_html = client.get("/observability")
+    assert res_html.status_code == 200
+    assert "text/html" in res_html.headers["content-type"]
+    assert "Observability" in res_html.text
+
+    # Versioned API route
+    res_api_v1 = client.get("/api/v1/observability")
+    assert res_api_v1.status_code == 200
+    assert "application/json" in res_api_v1.headers["content-type"]
+    data_v1 = res_api_v1.json()
+    assert "schema_version" in data_v1
+    assert "providers" in data_v1
+    assert "links" in data_v1
+    assert "metrics" in data_v1
+
+    # Compatibility alias route
+    res_api_alias = client.get("/api/observability")
+    assert res_api_alias.status_code == 200
+    assert res_api_alias.json() == data_v1
+
+
+def test_observability_gated_actions():
+    """Ensure observability actions are registered, gated, and fail closed in MVP mode."""
+    registry = load_action_registry()
+
+    action_reconcile = registry.get("observability_reconcile_incident")
+    assert action_reconcile is not None
+    assert action_reconcile.task_key == "observability_reconcile_incident"
+    assert action_reconcile.mcp_tool == "gitea_observability_reconcile_incident"
+
+    action_link = registry.get("observability_link_issue")
+    assert action_link is not None
+    assert action_link.task_key == "observability_link_issue"
+
+    # Preview returns mutation ledger
+    prev = preview_action("observability_reconcile_incident", provider="sentry", issue_id="101")
+    assert prev["action_id"] == "observability_reconcile_incident"
+    assert prev["enabled"] is False
+
+    # Execution fails closed in MVP mode
+    att = attempt_action("observability_reconcile_incident", provider="sentry", issue_id="101")
+    assert att["success"] is False
+    assert att["error"] == "action_disabled"
+
+
+def test_observability_authz_rbac():
+    """Test RBAC authorization for observability actions."""
+    principal = resolve_principal({})
+
+    # Check authorize decision
+    decision = authorize("observability_reconcile_incident", principal)
+    assert decision.action_id == "observability_reconcile_incident"
+    # Phase 4 action denies in Phase 1 runtime by default
+    assert decision.allowed is False