Merge branch 'master' into feat/issue-420-server-code-parity

tests/conftest.py

@@ -18,14 +18,25 @@ def _reset_mutation_authority(monkeypatch):
     explicitly.
     """
     monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
+    # Isolate durable session-state files so tests never share host cache (#559).
+    import tempfile
+
+    _state_tmp = tempfile.TemporaryDirectory(prefix="gitea-session-state-")
+    monkeypatch.setenv("GITEA_MCP_SESSION_STATE_DIR", _state_tmp.name)
     try:
         import mcp_server
     except Exception:
+        _state_tmp.cleanup()
         yield
         return
     monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
     monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
     monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
+    try:
+        import review_workflow_load
+        review_workflow_load._REVIEW_WORKFLOW_LOAD = None
+    except Exception:
+        pass
     try:
         import capability_stop_terminal
         capability_stop_terminal.clear()
@@ -37,3 +48,13 @@ def _reset_mutation_authority(monkeypatch):
         capability_stop_terminal.clear()
     except Exception:
         pass
+    try:
+        import review_workflow_load
+        review_workflow_load._REVIEW_WORKFLOW_LOAD = None
+    except Exception:
+        pass
+    try:
+        mcp_server._REVIEW_DECISION_LOCK = None
+    except Exception:
+        pass
+    _state_tmp.cleanup()

tests/test_bootstrap_review_path_docs.py

@@ -0,0 +1,81 @@
+"""Doc-contract checks for the bootstrap review path (#557).
+
+Self-hosted MCP workflow fixes can deadlock live review daemons. These tests
+pin the narrow controller-authorized bootstrap path so it cannot silently
+regress into a general gate bypass.
+"""
+
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+
+DOC = REPO_ROOT / "docs" / "bootstrap-review-path.md"
+RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md"
+SKILL = REPO_ROOT / "skills" / "llm-project-workflow" / "SKILL.md"
+
+
+def _normalize(text: str) -> str:
+    return " ".join(text.split())
+
+
+def test_bootstrap_doc_exists_and_names_authorization_marker():
+    text = DOC.read_text(encoding="utf-8")
+    assert "BOOTSTRAP REVIEW AUTHORIZATION (#557)" in text
+    assert "bootstrap deadlock" in text.lower() or "Bootstrap deadlock" in text
+
+
+def test_bootstrap_doc_requires_validation_and_audits():
+    text = _normalize(DOC.read_text(encoding="utf-8"))
+    for phrase in (
+        "git diff --check",
+        "Duplicate-PR audit",
+        "Mutation ledger audit",
+        "Contamination audit",
+        "workflow-contaminated",
+    ):
+        assert phrase in text, f"bootstrap doc missing: {phrase!r}"
+
+
+def test_bootstrap_doc_lists_allowed_and_forbidden_actions():
+    text = _normalize(DOC.read_text(encoding="utf-8"))
+    lower = text.lower()
+    for phrase in (
+        "Allowed verification actions",
+        "Forbidden actions",
+        "project root",
+        "direct Python import",
+        "branches/",
+    ):
+        assert phrase in text, f"bootstrap doc missing: {phrase!r}"
+    assert "curl" in lower
+    assert "raw" in lower
+
+
+def test_bootstrap_doc_forbids_general_gate_weakening():
+    text = _normalize(DOC.read_text(encoding="utf-8")).lower()
+    assert "never weakens normal" in text or "does not weaken normal" in text
+    assert "general bypass" in text or "general gate" in text
+
+
+def test_bootstrap_doc_post_land_restart_and_verify():
+    text = _normalize(DOC.read_text(encoding="utf-8"))
+    for phrase in (
+        "Restart MCP daemons",
+        "canonical tools",
+        "gitea_whoami",
+    ):
+        assert phrase in text, f"post-bootstrap guidance missing: {phrase!r}"
+
+
+def test_runbook_links_bootstrap_path():
+    text = _normalize(RUNBOOK.read_text(encoding="utf-8"))
+    assert "Bootstrap Review Path for self-hosted MCP fixes (#557)" in text
+    assert "bootstrap-review-path.md" in text
+    assert "BOOTSTRAP REVIEW AUTHORIZATION (#557)" in text
+
+
+def test_skill_links_bootstrap_path():
+    text = _normalize(SKILL.read_text(encoding="utf-8"))
+    assert "Bootstrap Review Path (#557)" in text
+    assert "bootstrap-review-path.md" in text
+    assert "BLOCKED + DIAGNOSE" in text or "BLOCKED" in text

tests/test_final_report_validator.py

@@ -35,6 +35,9 @@ def _review_handoff(**overrides):
         "- External-state mutations: none",
         "- Read-only diagnostics: issue and PR metadata inspected",
         "- Current status: PR open",
+        "- Next actor: author",
+        "- Next action: address review feedback",
+        "- Next prompt: Fix PR #203 per reviewer request_changes and push updated head",
         "- Blockers: none",
         "- Next: await author",
         "- Safety: no self-review; no self-merge",
@@ -92,6 +95,9 @@ def _reconcile_handoff(drop=(), **overrides):
         "- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
         "- Reconciliation mutations: PR comment posted",
         "- Current status: reconciliation complete",
+        "- Next actor: reconciler",
+        "- Next action: close PR #99 after capability proof",
+        "- Next prompt: Reconcile already-landed PR #99 with prgs-reconciler profile",
         "- Blocker: missing gitea.pr.close capability",
         "- Safe next action: hand off to a profile with gitea.pr.close",
         "- No review/merge confirmation: confirmed",
@@ -498,6 +504,81 @@ class TestCanonicalReconcileSchema(unittest.TestCase):
         )
 
 
+class TestReconcilerCloseProof(unittest.TestCase):
+    """Issue #306: a reconciler PR close must carry its proof fields."""
+
+    def _closed_report(self, **kwargs):
+        # A report that claims PR #99 was closed via the reconciler path.
+        report = (
+            _reconcile_handoff(**kwargs)
+            .replace(
+                "- Capabilities proven: gitea.read, gitea.pr.comment",
+                "- Capabilities proven: gitea.read, gitea.pr.comment, gitea.pr.close",
+            )
+            .replace("- Missing capabilities: gitea.pr.close", "- Missing capabilities: none")
+            .replace("- PRs closed: none", "- PRs closed: #99")
+            .replace(
+                "- Blocker: missing gitea.pr.close capability", "- Blocker: none"
+            )
+        )
+        return report
+
+    def test_full_proof_close_passes(self):
+        result = assess_final_report_validator(
+            self._closed_report(), "reconcile_already_landed"
+        )
+        self.assertEqual(result["grade"], "A")
+        self.assertFalse(
+            any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
+        )
+
+    def test_close_without_capability_proof_blocks(self):
+        # Claims PRs closed: #99 but never proves gitea.pr.close capability.
+        report = _reconcile_handoff().replace("- PRs closed: none", "- PRs closed: #99")
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
+        )
+
+    def test_close_without_ancestor_proof_blocks(self):
+        report = self._closed_report(drop=("Ancestor proof",))
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
+        )
+
+    def test_close_without_linked_issue_result_blocks(self):
+        report = self._closed_report(drop=("Linked issue live status", "Issues closed"))
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
+        )
+
+    def test_close_lock_requires_proof_even_if_text_says_none(self):
+        # Session lock proves a PR was closed; the report omits close proof.
+        result = assess_final_report_validator(
+            _reconcile_handoff(),
+            "reconcile_already_landed",
+            reconciler_close_lock={"pr_closed": True},
+        )
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
+        )
+
+    def test_comment_only_reconcile_needs_no_close_proof(self):
+        result = assess_final_report_validator(
+            _reconcile_handoff(), "reconcile_already_landed"
+        )
+        self.assertEqual(result["grade"], "A")
+        self.assertFalse(
+            any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"])
+        )
+
+
 class TestEntryPoint(unittest.TestCase):
     def test_unknown_task_kind_blocks(self):
         result = assess_final_report_validator("report", "unknown_mode")

tests/test_issue_acceptance_gate.py

@@ -0,0 +1,183 @@
+"""Tests for controller issue-acceptance gate (#500)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_acceptance_gate  # noqa: E402
+from final_report_validator import assess_final_report_validator  # noqa: E402
+
+
+def _accepted_comment(**overrides):
+    lines = [
+        "## Controller Issue Acceptance",
+        "",
+        "STATE:",
+        "accepted",
+        "",
+        "WHO_IS_NEXT:",
+        "controller",
+        "",
+        "NEXT_ACTION:",
+        "Close the tracker follow-up after verification.",
+        "",
+        "NEXT_PROMPT:",
+        "Verify deployment checklist for issue #500 and post acceptance.",
+        "",
+        "ISSUE:",
+        "#500",
+        "",
+        "MERGED_PR:",
+        "#503",
+        "",
+        "MERGE_COMMIT:",
+        "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "",
+        "ACCEPTANCE_CRITERIA_CHECKED:",
+        "- [x] documentation added",
+        "- [x] validator tests added",
+        "",
+        "VALIDATION_REVIEWED:",
+        "pytest tests/test_issue_acceptance_gate.py -q",
+        "",
+        "CONTROLLER_DECISION:",
+        "accepted",
+        "",
+        "WHY:",
+        "All acceptance criteria satisfied with proof.",
+        "",
+        "MISSING_WORK:",
+        "none",
+        "",
+        "FOLLOW_UP_ISSUES:",
+        "none",
+        "",
+        "BLOCKERS:",
+        "none",
+        "",
+        "LAST_UPDATED_BY:",
+        "jcwalker3 / prgs-controller / 2026-07-08",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        text = text.replace(f"{key}:\n", f"{key}:\n{value}\n", 1)
+    return text
+
+
+def _rejection_comment(state="needs-tests"):
+    text = _accepted_comment()
+    text = text.replace("STATE:\naccepted", f"STATE:\n{state}")
+    text = text.replace(
+        "CONTROLLER_DECISION:\naccepted",
+        "CONTROLLER_DECISION:\nrejected",
+    )
+    text = text.replace(
+        "ACCEPTANCE_CRITERIA_CHECKED:\n- [x] documentation added\n- [x] validator tests added",
+        "ACCEPTANCE_CRITERIA_CHECKED:\n- [ ] regression tests for rejection paths",
+    )
+    text = text.replace(
+        "MISSING_WORK:\nnone",
+        "MISSING_WORK:\nAdd regression tests for controller rejection paths.",
+    )
+    text = text.replace(
+        "NEXT_PROMPT:\nVerify deployment checklist for issue #500 and post acceptance.",
+        "NEXT_PROMPT:\nImplement the missing tests for issue #500 and reopen the PR.",
+    )
+    return text
+
+
+class TestControllerAcceptanceComment(unittest.TestCase):
+    def test_accepted_comment_valid(self):
+        result = issue_acceptance_gate.validate_controller_acceptance_comment(
+            _accepted_comment()
+        )
+        self.assertTrue(result["valid"], result["reasons"])
+
+    def test_missing_who_is_next_rejected(self):
+        text = _accepted_comment().replace("WHO_IS_NEXT:\ncontroller\n", "WHO_IS_NEXT:\n\n")
+        result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
+        self.assertFalse(result["valid"])
+        self.assertTrue(any("WHO_IS_NEXT" in r for r in result["reasons"]))
+
+    def test_missing_next_prompt_on_rejection(self):
+        text = _rejection_comment()
+        text = text.replace(
+            "NEXT_PROMPT:\nImplement the missing tests for issue #500 and reopen the PR.\n",
+            "NEXT_PROMPT:\n\n",
+        )
+        result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
+        self.assertFalse(result["valid"])
+        self.assertTrue(any("NEXT_PROMPT" in r for r in result["reasons"]))
+
+    def test_vague_merge_only_completion_detected(self):
+        text = "PR merged. Issue complete."
+        self.assertTrue(issue_acceptance_gate.claims_merge_only_issue_complete(text))
+
+    def test_rejection_paths_require_missing_work(self):
+        for state in (
+            "more-work-required",
+            "needs-tests",
+            "needs-docs",
+            "needs-feature-enhancement",
+            "needs-follow-up-issue",
+        ):
+            text = _rejection_comment(state=state).replace(
+                "MISSING_WORK:\nAdd regression tests for controller rejection paths.\n",
+                "MISSING_WORK:\n\n",
+            )
+            result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
+            self.assertFalse(result["valid"], state)
+            self.assertTrue(any("MISSING_WORK" in r for r in result["reasons"]))
+
+    def test_accepted_requires_checked_criteria(self):
+        text = _accepted_comment().replace(
+            "ACCEPTANCE_CRITERIA_CHECKED:\n- [x] documentation added\n- [x] validator tests added\n",
+            "ACCEPTANCE_CRITERIA_CHECKED:\n- [ ] documentation added\n",
+        )
+        result = issue_acceptance_gate.validate_controller_acceptance_comment(text)
+        self.assertFalse(result["valid"])
+        self.assertTrue(any("checked acceptance criterion" in r for r in result["reasons"]))
+
+
+class TestFinalReportIntegration(unittest.TestCase):
+    def test_merge_only_complete_blocks_work_issue_report(self):
+        report = (
+            "## Controller Handoff\n\n"
+            "- Task: work issue #500\n"
+            "- Merge result: merged\n"
+            "- Current status: PR merged; issue complete\n"
+        )
+        result = assess_final_report_validator(report, "work_issue")
+        self.assertTrue(
+            any(
+                f["rule_id"] == "shared.issue_acceptance_gate"
+                for f in result["findings"]
+            ),
+            result["findings"],
+        )
+
+    def test_pending_acceptance_allowed(self):
+        report = (
+            "## Controller Handoff\n\n"
+            "- Task: work issue #500\n"
+            "- Merge result: merged\n"
+            "- Current status: controller acceptance pending\n"
+        )
+        result = assess_final_report_validator(report, "work_issue")
+        self.assertFalse(
+            any(
+                f["rule_id"] == "shared.issue_acceptance_gate"
+                for f in result["findings"]
+            ),
+            result["findings"],
+        )
+
+    def test_accepted_block_allows_complete_claim(self):
+        report = _accepted_comment() + "\n\nIssue is complete."
+        gate = issue_acceptance_gate.validate_final_report_issue_acceptance(report)
+        self.assertTrue(gate["valid"], gate["reasons"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_session_state.py

@@ -0,0 +1,207 @@
+"""Tests for durable MCP session state shared across daemon processes (#559)."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys_path_root = str(Path(__file__).resolve().parent.parent)
+import sys
+
+if sys_path_root not in sys.path:
+    sys.path.insert(0, sys_path_root)
+
+import mcp_session_state
+import review_workflow_load
+import mcp_server
+
+
+class TestMcpSessionStateStore(unittest.TestCase):
+    def setUp(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self.state_dir = self._tmpdir.name
+        self._env = patch.dict(
+            os.environ,
+            {
+                mcp_session_state.STATE_DIR_ENV: self.state_dir,
+                mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
+                "GITEA_MCP_PROFILE": "prgs-reviewer",
+            },
+            clear=False,
+        )
+        self._env.start()
+
+    def tearDown(self):
+        self._env.stop()
+        self._tmpdir.cleanup()
+
+    def test_round_trip_same_identity(self):
+        saved = mcp_session_state.save_state(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            payload={"loaded": True, "workflow_hash": "abc123def456"},
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        self.assertIsNotNone(saved)
+        self.assertEqual(saved["workflow_hash"], "abc123def456")
+        self.assertIn("recorded_at", saved)
+
+        loaded = mcp_session_state.load_state(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        self.assertIsNotNone(loaded)
+        self.assertEqual(loaded["workflow_hash"], "abc123def456")
+        self.assertEqual(loaded["profile_identity"], "prgs-reviewer")
+
+    def test_profile_mismatch_returns_none(self):
+        mcp_session_state.save_state(
+            kind=mcp_session_state.KIND_DECISION_LOCK,
+            payload={"final_review_decision_ready": True, "remote": "prgs"},
+            remote="prgs",
+        )
+        with patch.dict(
+            os.environ,
+            {mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-author"},
+            clear=False,
+        ):
+            loaded = mcp_session_state.load_state(
+                kind=mcp_session_state.KIND_DECISION_LOCK,
+                remote="prgs",
+            )
+        self.assertIsNone(loaded)
+
+    def test_clear_removes_file(self):
+        mcp_session_state.save_state(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            payload={"loaded": True},
+            remote="prgs",
+        )
+        path = mcp_session_state.state_file_path(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            remote="prgs",
+            profile_identity="prgs-reviewer",
+            state_dir=self.state_dir,
+        )
+        self.assertTrue(os.path.exists(path))
+        mcp_session_state.clear_state(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            remote="prgs",
+            profile_identity="prgs-reviewer",
+        )
+        self.assertFalse(os.path.exists(path))
+
+    def test_files_are_private_mode(self):
+        mcp_session_state.save_state(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            payload={"loaded": True},
+            remote="prgs",
+        )
+        path = mcp_session_state.state_file_path(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            remote="prgs",
+            profile_identity="prgs-reviewer",
+            state_dir=self.state_dir,
+        )
+        mode = os.stat(path).st_mode & 0o777
+        self.assertEqual(mode, 0o600)
+
+
+class TestWorkflowLoadCrossProcess(unittest.TestCase):
+    def setUp(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self._env = patch.dict(
+            os.environ,
+            {
+                mcp_session_state.STATE_DIR_ENV: self._tmpdir.name,
+                mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
+                "GITEA_MCP_PROFILE": "prgs-reviewer",
+                "GITEA_MCP_REMOTE": "prgs",
+            },
+            clear=False,
+        )
+        self._env.start()
+        review_workflow_load.clear_review_workflow_load()
+
+    def tearDown(self):
+        review_workflow_load.clear_review_workflow_load()
+        self._env.stop()
+        self._tmpdir.cleanup()
+
+    def test_different_pid_same_profile_accepted(self):
+        root = str(Path(__file__).resolve().parent.parent)
+        recorded = review_workflow_load.record_review_workflow_load(root)
+        self.assertTrue(recorded["loaded"])
+
+        # Simulate another daemon process: clear memory, keep durable file,
+        # and change apparent PID identity only (profile remains the same).
+        review_workflow_load._REVIEW_WORKFLOW_LOAD = None
+        status = review_workflow_load.workflow_load_status(root)
+        self.assertTrue(status["workflow_load_proof_present"])
+        self.assertTrue(
+            status["workflow_load_valid"],
+            msg=status.get("reasons"),
+        )
+
+    def test_profile_mismatch_blocks(self):
+        root = str(Path(__file__).resolve().parent.parent)
+        review_workflow_load.record_review_workflow_load(root)
+        review_workflow_load._REVIEW_WORKFLOW_LOAD = None
+        with patch.dict(
+            os.environ,
+            {mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-author"},
+            clear=False,
+        ):
+            # Durable load uses active identity; mismatched profile key misses.
+            status = review_workflow_load.workflow_load_status(root)
+        self.assertFalse(status["workflow_load_proof_present"])
+
+
+class TestDecisionLockCrossProcess(unittest.TestCase):
+    def setUp(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self._env = patch.dict(
+            os.environ,
+            {
+                mcp_session_state.STATE_DIR_ENV: self._tmpdir.name,
+                mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
+            },
+            clear=False,
+        )
+        self._env.start()
+        mcp_server._save_review_decision_lock(None)
+        review_workflow_load.clear_review_workflow_load()
+
+    def tearDown(self):
+        mcp_server._save_review_decision_lock(None)
+        review_workflow_load.clear_review_workflow_load()
+        self._env.stop()
+        self._tmpdir.cleanup()
+
+    def test_decision_lock_survives_memory_clear(self):
+        with patch.object(mcp_server, "get_profile", return_value={
+            "profile_name": "prgs-reviewer",
+        }):
+            mcp_server.init_review_decision_lock("prgs", "review_pr", force=True)
+            lock = mcp_server._load_review_decision_lock()
+            self.assertIsNotNone(lock)
+            self.assertEqual(lock["remote"], "prgs")
+            self.assertFalse(lock["final_review_decision_ready"])
+
+            # New process: memory empty, durable state remains.
+            mcp_server._REVIEW_DECISION_LOCK = None
+            restored = mcp_server._load_review_decision_lock()
+            self.assertIsNotNone(restored)
+            self.assertEqual(restored["remote"], "prgs")
+            reasons = mcp_server._review_decision_session_reasons(restored)
+            self.assertEqual(reasons, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_stale_runtime.py

@@ -0,0 +1,71 @@
+import os
+import unittest
+from unittest.mock import patch, MagicMock
+from datetime import datetime
+
+import gitea_mcp_server
+
+class TestMcpStaleRuntime(unittest.TestCase):
+    @patch("subprocess.run")
+    @patch("os.path.getmtime")
+    @patch("os.path.exists")
+    @patch("os.getpid")
+    @patch.dict("os.environ", {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"})
+    def test_stale_and_missing_runtimes(self, mock_getpid, mock_exists, mock_getmtime, mock_run):
+        # Setup mocks
+        mock_getpid.return_value = 12345
+        mock_exists.return_value = True
+        
+        # Code modification time: Jul 8 2026, 14:00:00
+        code_time = datetime(2026, 7, 8, 14, 0, 0)
+        mock_getmtime.return_value = code_time.timestamp()
+        
+        # Mock ps -ax output
+        # PID 12345 is self (started at 13:00:00 - stale)
+        # PID 54321 is prgs-author (started at 15:00:00 - fresh)
+        # prgs-reviewer is missing
+        ps_output = (
+            "  PID LSTART COMMAND\n"
+            "12345 Wed Jul  8 13:00:00 2026 /path/to/python mcp_server.py\n"
+            "54321 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 output for ps eww
+        mock_run_env12345 = MagicMock()
+        mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler"
+        
+        mock_run_env54321 = MagicMock()
+        mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author"
+        
+        def side_effect(args, **kwargs):
+            if args[0] == "ps" and "eww" in args:
+                pid = args[2]
+                if pid == "12345":
+                    return mock_run_env12345
+                elif pid == "54321":
+                    return mock_run_env54321
+            elif args[0] == "ps":
+                return mock_run_ps
+            raise ValueError(f"Unexpected subprocess run args: {args}")
+            
+        mock_run.side_effect = side_effect
+        
+        # Test 1: required reviewer role matching prgs-reviewer (which is missing)
+        reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("review_pr", ["prgs-reviewer"])
+        
+        self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons))
+        self.assertTrue(any("stale-runtime: None of the matching profiles for task" in r for r in reasons))
+        
+        # Test 2: required author role matching prgs-author (which is fresh)
+        reasons_author = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
+        
+        # Still contains self stale error
+        self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons_author))
+        # But does NOT contain missing author profile error
+        self.assertFalse(any("None of the matching profiles for task" in r for r in reasons_author))
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_merged_cleanup_reconcile.py

@@ -4,7 +4,7 @@ import os
 import sys
 import tempfile
 import unittest
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
@@ -154,6 +154,290 @@ class TestMergedCleanupReport(unittest.TestCase):
         finally:
             os.remove(lock_path)
 
+    @patch("subprocess.run")
+    def test_discover_local_worktrees(self, mock_run):
+        mock_output = (
+            "worktree /repo/Gitea-Tools\n"
+            "bare\n\n"
+            "worktree /repo/Gitea-Tools/branches/custom-name\n"
+            "branch refs/heads/feat/issue-11-x\n"
+            "HEAD cafebabe\n\n"
+        )
+        mock_run.return_value = MagicMock(returncode=0, stdout=mock_output)
+        res = mcr.discover_local_worktrees("/repo/Gitea-Tools")
+        self.assertEqual(res, {"feat/issue-11-x": "/repo/Gitea-Tools/branches/custom-name"})
+
+    @patch("subprocess.run")
+    @patch("merged_cleanup_reconcile.read_local_worktree_state")
+    def test_build_report_discovers_non_canonical_worktree_by_branch(self, mock_state, mock_run):
+        mock_output = (
+            "worktree /repo/Gitea-Tools\n"
+            "bare\n\n"
+            "worktree /repo/Gitea-Tools/branches/custom-name\n"
+            "branch refs/heads/feat/issue-11-x\n"
+            "HEAD cafebabe\n\n"
+        )
+        mock_run.return_value = MagicMock(returncode=0, stdout=mock_output)
+        mock_state.return_value = {
+            "exists": True,
+            "clean": True,
+            "current_branch": "feat/issue-11-x",
+            "head_sha": "cafebabe",
+            "dirty_files": [],
+        }
+
+        report = mcr.build_reconciliation_report(
+            project_root="/repo/Gitea-Tools",
+            closed_prs=[
+                {
+                    "number": 11,
+                    "title": "merged",
+                    "body": "Closes #11",
+                    "head": {"ref": "feat/issue-11-x"},
+                    "merged_at": "2026-07-06T12:00:00Z",
+                    "merge_commit_sha": "abc123",
+                }
+            ],
+            open_prs=[],
+            remote_branch_exists={"feat/issue-11-x": True},
+            head_on_master={11: True},
+            delete_capability_allowed=True,
+        )
+        self.assertEqual(report["merged_pr_count"], 1)
+        entry = report["entries"][0]
+        local = entry["local_worktree"]
+        self.assertEqual(local["worktree_path"], "/repo/Gitea-Tools/branches/custom-name")
+        self.assertEqual(local["match_type"], "branch")
+
+    @patch("subprocess.run")
+    def test_remove_local_worktree_uses_discovered_path(self, mock_run):
+        mock_output = (
+            "worktree /repo/Gitea-Tools\n"
+            "bare\n\n"
+            "worktree /repo/Gitea-Tools/branches/custom-name\n"
+            "branch refs/heads/feat/issue-11-x\n"
+            "HEAD cafebabe\n\n"
+        )
+        # First call is discover_local_worktrees, second is git worktree remove
+        mock_run.side_effect = [
+            MagicMock(returncode=0, stdout=mock_output),
+            MagicMock(returncode=0, stdout="", stderr=""),
+        ]
+
+        with patch("os.path.isdir", return_value=True):
+            result = mcr.remove_local_worktree("/repo/Gitea-Tools", "feat/issue-11-x")
+        self.assertTrue(result["success"])
+        self.assertTrue(result["performed"])
+        self.assertEqual(result["worktree_path"], "/repo/Gitea-Tools/branches/custom-name")
+        # Assert git worktree remove was called with the custom path
+        mock_run.assert_any_call(
+            ["git", "-C", "/repo/Gitea-Tools", "worktree", "remove", "/repo/Gitea-Tools/branches/custom-name"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
+
+class TestReviewerScratchCleanup(unittest.TestCase):
+    """#534: authorized cleanup path for reviewer scratch worktrees."""
+
+    def test_parse_reviewer_scratch_folder(self):
+        self.assertEqual(mcr.parse_reviewer_scratch_folder("review-pr512-submit"), 512)
+        self.assertEqual(mcr.parse_reviewer_scratch_folder("review-pr99"), 99)
+        self.assertIsNone(mcr.parse_reviewer_scratch_folder("feat-issue-11-x"))
+        self.assertIsNone(mcr.parse_reviewer_scratch_folder("review-feat-issue-11"))
+        self.assertIsNone(mcr.parse_reviewer_scratch_folder("review-pr512-extra-tail"))
+
+    def test_assess_safe_clean_detached_after_merge(self):
+        assessment = mcr.assess_reviewer_scratch_cleanup(
+            pr_number=512,
+            worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
+            folder_name="review-pr512-submit",
+            pr_merged=True,
+            pr_closed=True,
+            worktree_state={
+                "exists": True,
+                "clean": True,
+                "current_branch": None,
+                "head_sha": "abc",
+                "dirty_files": [],
+            },
+            active_reviewer_lease=False,
+        )
+        self.assertTrue(assessment["safe_to_remove_worktree"])
+        self.assertFalse(assessment["author_worktree_cleanup"])
+        self.assertEqual(
+            assessment["recommended_action"], "remove_reviewer_scratch_worktree"
+        )
+        self.assertEqual(assessment["worktree_kind"], "reviewer_scratch")
+
+    def test_assess_blocks_dirty_scratch(self):
+        assessment = mcr.assess_reviewer_scratch_cleanup(
+            pr_number=512,
+            worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
+            folder_name="review-pr512-submit",
+            pr_merged=True,
+            pr_closed=True,
+            worktree_state={
+                "exists": True,
+                "clean": False,
+                "current_branch": None,
+                "head_sha": "abc",
+                "dirty_files": ["foo.py"],
+            },
+            active_reviewer_lease=False,
+        )
+        self.assertFalse(assessment["safe_to_remove_worktree"])
+        self.assertTrue(
+            any("tracked edits" in r for r in assessment["block_reasons"])
+        )
+
+    def test_assess_blocks_active_lease(self):
+        assessment = mcr.assess_reviewer_scratch_cleanup(
+            pr_number=512,
+            worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
+            folder_name="review-pr512-submit",
+            pr_merged=True,
+            pr_closed=True,
+            worktree_state={
+                "exists": True,
+                "clean": True,
+                "current_branch": None,
+                "head_sha": "abc",
+                "dirty_files": [],
+            },
+            active_reviewer_lease=True,
+        )
+        self.assertFalse(assessment["safe_to_remove_worktree"])
+        self.assertTrue(
+            any("active reviewer lease" in r for r in assessment["block_reasons"])
+        )
+
+    def test_assess_blocks_open_pr(self):
+        assessment = mcr.assess_reviewer_scratch_cleanup(
+            pr_number=512,
+            worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
+            folder_name="review-pr512-submit",
+            pr_merged=False,
+            pr_closed=False,
+            worktree_state={
+                "exists": True,
+                "clean": True,
+                "current_branch": None,
+                "head_sha": "abc",
+                "dirty_files": [],
+            },
+            active_reviewer_lease=False,
+        )
+        self.assertFalse(assessment["safe_to_remove_worktree"])
+        self.assertTrue(any("still open" in r for r in assessment["block_reasons"]))
+
+    @patch("subprocess.run")
+    @patch("merged_cleanup_reconcile.read_local_worktree_state")
+    def test_report_lists_scratch_separately_from_author(self, mock_state, mock_run):
+        mock_output = (
+            "worktree /repo/Gitea-Tools\n"
+            "bare\n\n"
+            "worktree /repo/Gitea-Tools/branches/feat-issue-11-x\n"
+            "branch refs/heads/feat/issue-11-x\n"
+            "HEAD deadbeef\n\n"
+            "worktree /repo/Gitea-Tools/branches/review-pr512-submit\n"
+            "HEAD abcdef0\n"
+            "detached\n\n"
+        )
+        mock_run.return_value = MagicMock(returncode=0, stdout=mock_output)
+
+        def _state(path):
+            if "review-pr512" in path:
+                return {
+                    "exists": True,
+                    "clean": True,
+                    "current_branch": None,
+                    "head_sha": "abcdef0",
+                    "dirty_files": [],
+                }
+            return {
+                "exists": True,
+                "clean": True,
+                "current_branch": "feat/issue-11-x",
+                "head_sha": "deadbeef",
+                "dirty_files": [],
+            }
+
+        mock_state.side_effect = _state
+        report = mcr.build_reconciliation_report(
+            project_root="/repo/Gitea-Tools",
+            closed_prs=[
+                {
+                    "number": 11,
+                    "title": "merged author",
+                    "body": "Closes #11",
+                    "head": {"ref": "feat/issue-11-x"},
+                    "merged_at": "2026-07-06T12:00:00Z",
+                },
+                {
+                    "number": 512,
+                    "title": "merged reviewed",
+                    "body": "Closes #512",
+                    "head": {"ref": "feat/issue-512-x"},
+                    "merged_at": "2026-07-06T12:00:00Z",
+                },
+            ],
+            open_prs=[],
+            remote_branch_exists={
+                "feat/issue-11-x": False,
+                "feat/issue-512-x": False,
+            },
+            head_on_master={11: True, 512: True},
+            delete_capability_allowed=True,
+            active_reviewer_leases={512: False},
+        )
+        self.assertEqual(report["merged_pr_count"], 2)
+        self.assertEqual(report["reviewer_scratch_count"], 1)
+        scratch = report["reviewer_scratch_entries"][0]
+        self.assertEqual(scratch["pr_number"], 512)
+        self.assertEqual(scratch["worktree_kind"], "reviewer_scratch")
+        self.assertFalse(scratch["author_worktree_cleanup"])
+        self.assertTrue(scratch["safe_to_remove_worktree"])
+        # Author entries must not swallow the scratch path as local_worktree.
+        author_paths = {
+            (e.get("local_worktree") or {}).get("worktree_path")
+            for e in report["entries"]
+        }
+        self.assertNotIn(
+            "/repo/Gitea-Tools/branches/review-pr512-submit", author_paths
+        )
+
+    @patch("subprocess.run")
+    def test_remove_reviewer_scratch_worktree_path_guard(self, mock_run):
+        mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
+        with patch("os.path.isdir", return_value=True):
+            bad = mcr.remove_reviewer_scratch_worktree(
+                "/repo/Gitea-Tools",
+                "/repo/Gitea-Tools/branches/feat-issue-11-x",
+            )
+            self.assertFalse(bad["performed"])
+            good = mcr.remove_reviewer_scratch_worktree(
+                "/repo/Gitea-Tools",
+                "/repo/Gitea-Tools/branches/review-pr512-submit",
+            )
+        self.assertTrue(good["success"])
+        self.assertTrue(good["performed"])
+        self.assertFalse(good["author_worktree_cleanup"])
+        mock_run.assert_any_call(
+            [
+                "git",
+                "-C",
+                "/repo/Gitea-Tools",
+                "worktree",
+                "remove",
+                "/repo/Gitea-Tools/branches/review-pr512-submit",
+            ],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_premerge_baseline_proof.py

@@ -0,0 +1,112 @@
+import unittest
+
+from premerge_baseline_proof import (
+    CLEAN_PASS,
+    PREMERGE_BASELINE_PROVEN_FAILURE,
+    UNRESOLVED_REGRESSION_RISK,
+    assess_premerge_baseline_proof,
+)
+from final_report_validator import assess_final_report_validator
+
+_FIELDS = (
+    "Pre-merge base commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
+    "Tested commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
+    "Command: venv/bin/python -m pytest tests/test_foo.py -q\n"
+    "Exit status: 1\n"
+    "Failure signature: AssertionError: None is not true\n"
+)
+
+# Current-master-only reproduction carries no pre-merge base proof (that is the
+# whole defect #533 guards against): a command/exit/signature but no base commit.
+_CURRENT_ONLY_FIELDS = (
+    "Command: venv/bin/python -m pytest tests/test_foo.py -q\n"
+    "Exit status: 1\n"
+    "Failure signature: AssertionError: None is not true\n"
+)
+
+
+class TestPremergeBaselineProof(unittest.TestCase):
+    def test_clean_pass_not_blocked(self):
+        result = assess_premerge_baseline_proof(
+            "Validation: full suite passed, exit status 0. Clean pass."
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["skipped"])
+        self.assertEqual(result["label"], CLEAN_PASS)
+
+    def test_nonzero_exit_not_claimed_baseline_not_blocked(self):
+        result = assess_premerge_baseline_proof(
+            "Full suite: 1 failed. Investigating the regression; not yet classified."
+        )
+        self.assertFalse(result["block"])
+        self.assertEqual(result["label"], UNRESOLVED_REGRESSION_RISK)
+
+    def test_current_master_only_reproduction_rejected(self):
+        report = (
+            "Full suite: 1 failed. This is a pre-existing baseline failure — "
+            "reproduced on current master after the PR merged.\n" + _CURRENT_ONLY_FIELDS
+        )
+        result = assess_premerge_baseline_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("current-master reproduction only" in r for r in result["reasons"])
+        )
+
+    def test_baseline_claim_missing_fields_blocked(self):
+        report = (
+            "Full suite: 1 failed. This failure is baseline / pre-existing. "
+            "Verified against the pre-merge base commit."
+        )
+        result = assess_premerge_baseline_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("missing required proof field" in r for r in result["reasons"])
+        )
+
+    def test_premerge_base_proof_accepted(self):
+        report = (
+            "Full suite: 1 failed. This is a pre-existing baseline failure, proven "
+            "on the PR pre-merge base commit.\n" + _FIELDS
+        )
+        result = assess_premerge_baseline_proof(report)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["proven"])
+        self.assertEqual(result["label"], PREMERGE_BASELINE_PROVEN_FAILURE)
+
+    def test_documented_known_failure_record_accepted(self):
+        report = (
+            "Full suite: 1 failed. Baseline failure: cited known-failure record #529 "
+            "which predates the PR.\n" + _FIELDS
+        )
+        result = assess_premerge_baseline_proof(report)
+        self.assertFalse(result["block"])
+        self.assertEqual(result["label"], PREMERGE_BASELINE_PROVEN_FAILURE)
+
+
+class TestPremergeBaselineProofValidatorIntegration(unittest.TestCase):
+    def _has_rule(self, report):
+        result = assess_final_report_validator(report, "review_pr")
+        return any(
+            f["rule_id"] == "reviewer.premerge_baseline_proof"
+            for f in result["findings"]
+        )
+
+    def test_review_report_current_master_only_flagged(self):
+        report = (
+            "## Reviewer final report\n"
+            "Full suite: 1 failed. Pre-existing baseline failure — reproduced on "
+            "post-merge master.\n" + _CURRENT_ONLY_FIELDS
+        )
+        self.assertTrue(self._has_rule(report))
+
+    def test_review_report_premerge_proof_clean(self):
+        report = (
+            "## Reviewer final report\n"
+            "Full suite: 1 failed. Pre-existing baseline failure proven on the "
+            "pre-merge base commit.\n" + _FIELDS
+        )
+        self.assertFalse(self._has_rule(report))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_review_workflow_load.py

@@ -33,12 +33,38 @@ class TestReviewWorkflowLoadModule(unittest.TestCase):
         self.assertTrue(status["workflow_load_proof_present"])
         self.assertTrue(status["workflow_load_valid"])
 
-    def test_stale_session_pid_blocks(self):
+    def test_profile_identity_mismatch_blocks(self):
+        """#559: different daemon PIDs are OK; profile identity mismatch is not."""
+        import tempfile
+        from unittest.mock import patch
+
+        import mcp_session_state
+
         root = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
-        review_workflow_load.record_review_workflow_load(root)
-        review_workflow_load._REVIEW_WORKFLOW_LOAD["session_pid"] = 0
-        blockers = review_workflow_load.review_workflow_load_blockers(root)
-        self.assertTrue(any("different process" in b for b in blockers))
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch.dict(
+                os.environ,
+                {
+                    mcp_session_state.STATE_DIR_ENV: tmp,
+                    mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
+                },
+                clear=False,
+            ):
+                review_workflow_load.clear_review_workflow_load()
+                review_workflow_load.record_review_workflow_load(root)
+                # Corrupt the in-memory profile identity while keeping PID.
+                review_workflow_load._REVIEW_WORKFLOW_LOAD[
+                    "session_profile_lock"
+                ] = "other-profile"
+                review_workflow_load._REVIEW_WORKFLOW_LOAD[
+                    "profile_identity"
+                ] = "other-profile"
+                blockers = review_workflow_load.review_workflow_load_blockers(root)
+                self.assertTrue(
+                    any("profile identity mismatch" in b for b in blockers),
+                    msg=blockers,
+                )
+                review_workflow_load.clear_review_workflow_load()
 
     def test_prompt_conflict_detected(self):
         conflict, reasons = review_workflow_load.assess_prompt_conflict(

tests/test_state_handoff_ledger.py

@@ -0,0 +1,243 @@
+"""Tests for canonical state handoff ledger (#494)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from final_report_validator import assess_final_report_validator  # noqa: E402
+from state_handoff_ledger import (  # noqa: E402
+    ISSUE_STATE_FIELDS,
+    assess_contradictory_state_handoff,
+    assess_final_report_next_action_handoff,
+    assess_state_comment,
+    assess_state_handoff_ledger_report,
+    parse_state_comment,
+    render_issue_state_comment,
+    render_pr_state_comment,
+    render_queue_controller_report,
+)
+
+
+def _work_issue_handoff(**overrides):
+    lines = [
+        "## Controller Handoff",
+        "",
+        "- Task: work-issue",
+        "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
+        "- Role: author",
+        "- Identity: jcwalker3 / prgs-author",
+        "- Selected issue: #494",
+        "- Current status: implementation complete; PR opened awaiting review",
+        "- Next actor: reviewer",
+        "- Next action: review PR at pinned head",
+        "- Next prompt: Review PR #500 for issue #494 at current head in reviewer worktree",
+        "- Safe next action: switch to reviewer session",
+        "- External-state mutations: none",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        needle = f"- {key}:"
+        if needle in text:
+            prefix = text.split(needle)[0]
+            rest = text.split(needle, 1)[1]
+            suffix = rest.split("\n", 1)[1] if "\n" in rest else ""
+            text = f"{prefix}{needle} {value}\n{suffix}".rstrip()
+        else:
+            text += f"\n- {key}: {value}"
+    return text
+
+
+class TestStateCommentTemplates(unittest.TestCase):
+    def test_render_issue_state_includes_all_fields(self):
+        body = render_issue_state_comment(
+            STATE="issue_ready_for_author",
+            NEXT_ACTOR="author",
+            NEXT_ACTION="lock and implement",
+            NEXT_PROMPT="Find issue #494 and open a PR",
+            LAST_UPDATED_BY="jcwalker3",
+        )
+        parsed = parse_state_comment(body)
+        for field in ISSUE_STATE_FIELDS:
+            self.assertIn(field, parsed)
+
+    def test_render_pr_state_parses(self):
+        body = render_pr_state_comment(
+            STATE="needs_review",
+            NEXT_ACTOR="reviewer",
+            NEXT_ACTION="review at head",
+            NEXT_PROMPT="Review PR #500",
+            ISSUE="#494",
+            HEAD_SHA="a" * 40,
+            LAST_UPDATED_BY="sysadmin",
+        )
+        parsed = parse_state_comment(body)
+        self.assertEqual(parsed["STATE"], "needs_review")
+        self.assertEqual(parsed["NEXT_ACTOR"], "reviewer")
+
+    def test_queue_controller_template(self):
+        body = render_queue_controller_report(
+            QUEUE_STATE="open_prs_need_review",
+            SELECTED_OBJECT="PR #500",
+            SELECTED_OBJECT_TYPE="pr",
+            NEXT_ACTOR="reviewer",
+            NEXT_ACTION="start review session",
+            NEXT_PROMPT="Review PR #500",
+            LAST_UPDATED_BY="controller",
+        )
+        result = assess_state_comment(body, kind="queue_controller")
+        self.assertTrue(result["complete"])
+
+
+class TestStateCommentValidation(unittest.TestCase):
+    def test_complete_issue_state_comment(self):
+        body = render_issue_state_comment(
+            STATE="in_progress",
+            NEXT_ACTOR="author",
+            NEXT_ACTION="push branch",
+            NEXT_PROMPT="Continue issue #494",
+            STATUS="open",
+            RELATED_PRS="none",
+            BRANCH="feat/issue-494-state-handoff-ledger",
+            HEAD_SHA="b" * 40,
+            VALIDATION="pytest passed",
+            BLOCKERS="none",
+            DUPLICATES_OR_SUPERSEDES="none",
+            LAST_UPDATED_BY="jcwalker3",
+        )
+        result = assess_state_comment(body, kind="issue_state")
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["block"])
+
+    def test_missing_fields_blocked(self):
+        body = "STATE: open\nNEXT_ACTOR: author"
+        result = assess_state_comment(body, kind="issue_state")
+        self.assertFalse(result["complete"])
+        self.assertIn("NEXT_ACTION", result["missing_fields"])
+
+    def test_discussion_requires_five_comments_by_default(self):
+        body = render_issue_state_comment(
+            STATE="collecting_feedback",
+            NEXT_ACTOR="controller",
+            NEXT_ACTION="facilitate discussion",
+            NEXT_PROMPT="Add acceptance criteria",
+            STATUS="open",
+            RELATED_PRS="none",
+            BRANCH="none",
+            HEAD_SHA="none",
+            VALIDATION="none",
+            BLOCKERS="none",
+            DUPLICATES_OR_SUPERSEDES="none",
+            LAST_UPDATED_BY="controller",
+        )
+        # Re-map to discussion fields manually for comment-count test
+        from state_handoff_ledger import render_discussion_state_comment
+
+        disc = render_discussion_state_comment(
+            STATE="collecting_feedback",
+            NEXT_ACTOR="controller",
+            NEXT_ACTION="wait for comments",
+            NEXT_PROMPT="Continue discussion",
+            COMMENT_COUNT="2",
+            SUBSTANTIVE_COMMENTS="yes",
+            URGENCY="normal",
+            BLOCKERS="none",
+            LAST_UPDATED_BY="controller",
+        )
+        result = assess_state_comment(disc, kind="discussion_state")
+        self.assertFalse(result["complete"])
+        self.assertTrue(any("5" in reason for reason in result["reasons"]))
+
+    def test_discussion_urgent_exception(self):
+        from state_handoff_ledger import render_discussion_state_comment
+
+        disc = render_discussion_state_comment(
+            STATE="ready_for_summary",
+            NEXT_ACTOR="controller",
+            NEXT_ACTION="summarize",
+            NEXT_PROMPT="Post summary",
+            COMMENT_COUNT="2",
+            SUBSTANTIVE_COMMENTS="yes",
+            URGENCY="urgent",
+            BLOCKERS="none",
+            LAST_UPDATED_BY="controller",
+        )
+        result = assess_state_comment(disc, kind="discussion_state")
+        self.assertTrue(result["complete"])
+
+
+class TestFinalReportNextAction(unittest.TestCase):
+    def test_complete_next_action_handoff(self):
+        report = _work_issue_handoff()
+        result = assess_final_report_next_action_handoff(report)
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["block"])
+
+    def test_missing_next_prompt_blocked(self):
+        report = _work_issue_handoff(**{"Next prompt": "none"})
+        result = assess_final_report_next_action_handoff(report)
+        self.assertFalse(result["complete"])
+        self.assertIn("Next prompt", result["missing_fields"])
+
+    def test_missing_handoff_section_blocked(self):
+        result = assess_final_report_next_action_handoff("no handoff here")
+        self.assertTrue(result["block"])
+
+
+class TestContradictoryState(unittest.TestCase):
+    def test_ready_to_merge_without_approval_blocked(self):
+        report = _work_issue_handoff(
+            **{
+                "Current status": "PR ready to merge",
+                "Review decision": "not attempted",
+            }
+        )
+        result = assess_contradictory_state_handoff(report)
+        self.assertTrue(result["block"])
+
+    def test_ready_to_merge_with_approval_allowed(self):
+        report = _work_issue_handoff(
+            **{
+                "Current status": "PR ready to merge",
+                "Review decision": "approve",
+            }
+        )
+        result = assess_contradictory_state_handoff(report)
+        self.assertFalse(result["block"])
+
+    def test_issue_done_without_pr_proof_blocked(self):
+        report = _work_issue_handoff(**{"Current status": "issue complete"})
+        result = assess_contradictory_state_handoff(report)
+        self.assertTrue(result["block"])
+
+    def test_external_none_with_lock_activity_blocked(self):
+        report = _work_issue_handoff(
+            **{
+                "External-state mutations": "none",
+            }
+        )
+        report += "\n- Issue mutations: gitea_lock_issue adopted branch"
+        result = assess_contradictory_state_handoff(report)
+        self.assertTrue(result["block"])
+
+    def test_discussion_complete_without_summary_blocked(self):
+        report = _work_issue_handoff(**{"Current status": "discussion complete"})
+        result = assess_contradictory_state_handoff(report)
+        self.assertTrue(result["block"])
+
+
+class TestFinalReportValidatorIntegration(unittest.TestCase):
+    def test_work_issue_validator_flags_missing_next_prompt(self):
+        report = _work_issue_handoff(**{"Next prompt": ""})
+        result = assess_final_report_validator(report, "work_issue")
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("shared.state_handoff_next_action", rule_ids)
+
+    def test_work_issue_validator_accepts_complete_handoff(self):
+        report = _work_issue_handoff()
+        result = assess_state_handoff_ledger_report(report)
+        self.assertTrue(result["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_lease_visibility.py

@@ -0,0 +1,94 @@
+"""Tests for web UI lease visibility (#433)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from starlette.testclient import TestClient
+
+from webui.app import create_app
+from webui.lease_loader import (
+    _duplicate_branch_warnings,
+    _duplicate_pr_warnings,
+    load_lease_snapshot,
+    parse_reviewer_lease_comment,
+    snapshot_to_dict,
+)
+
+
+class TestLeaseLoader(unittest.TestCase):
+    def test_parse_reviewer_lease_comment(self):
+        body = (
+            "<!-- mcp-review-lease:v1 -->\n"
+            "pr: #42\n"
+            "reviewer_identity: sysadmin\n"
+            "profile: prgs-reviewer\n"
+            "phase: validating\n"
+            "expires_at: 2026-07-07T20:00:00Z\n"
+        )
+        parsed = parse_reviewer_lease_comment(body)
+        self.assertIsNotNone(parsed)
+        assert parsed is not None
+        self.assertEqual(parsed["pr_number"], 42)
+        self.assertEqual(parsed["phase"], "validating")
+
+    def test_duplicate_pr_warnings(self):
+        raw_prs = [
+            {"number": 1, "title": "Closes #99", "body": ""},
+            {"number": 2, "title": "fixes #99", "body": ""},
+        ]
+        warnings = _duplicate_pr_warnings(raw_prs)
+        self.assertEqual(len(warnings), 1)
+        self.assertEqual(warnings[0].issue_number, 99)
+        self.assertEqual(warnings[0].pr_numbers, (1, 2))
+
+    def test_duplicate_branch_warnings(self):
+        warnings = _duplicate_branch_warnings([
+            "feat/issue-12-a",
+            "feat/issue-12-b",
+            "master",
+        ])
+        self.assertEqual(len(warnings), 1)
+        self.assertEqual(warnings[0].issue_number, 12)
+
+    def test_load_snapshot_with_injected_fetch(self):
+        def fetch_prs(_h, _o, _r, _a):
+            return ([], None)
+
+        def fetch_issues(_h, _o, _r, _a):
+            return ([], None)
+
+        snapshot = load_lease_snapshot(
+            fetch_prs=fetch_prs,
+            fetch_issues=fetch_issues,
+            fetch_comments=lambda *_a, **_k: [],
+            issue_lock_path=str(Path("/nonexistent/lock.json")),
+        )
+        data = snapshot_to_dict(snapshot)
+        self.assertIn("claim_inventory", data)
+        self.assertIn("collision_history", data)
+
+
+class TestLeaseRoutes(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_leases_page_renders(self):
+        response = self.client.get("/leases")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Leases", response.text)
+        self.assertIn("Collision warnings", response.text)
+        self.assertNotIn("child issue", response.text.lower())
+
+    def test_api_leases_json(self):
+        response = self.client.get("/api/leases")
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertIn("claim_inventory", data)
+        self.assertIn("duplicate_prs", data)
+        self.assertIn("reviewer_leases", data)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_skeleton.py

@@ -46,10 +46,14 @@ class TestWebuiSkeleton(unittest.TestCase):
         self.assertEqual(response.status_code, 200)
         self.assertIn("Gitea-Tools", response.text)
 
+    def test_leases_is_implemented(self):
+        response = self.client.get("/leases")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Leases", response.text)
+        self.assertIn("Collision warnings", response.text)
+
     def test_extra_stub_routes(self):
-        for path in ("/worktrees", "/leases"):
-            with self.subTest(path=path):
-                self.assertEqual(self.client.get(path).status_code, 200)
+        self.assertEqual(self.client.get("/worktrees").status_code, 200)
 
     def test_post_is_rejected(self):
         response = self.client.post("/health")
@@ -62,10 +66,10 @@ class TestWebuiSkeleton(unittest.TestCase):
         self.assertIn("Live queue", response.text)
 
     def test_nav_links_on_all_pages(self):
-        for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit"):
+        for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
             with self.subTest(path=path):
                 text = self.client.get(path).text
-                for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit"):
+                for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
                     self.assertIn(f'href="{href}"', text)