Fix #723: Prevent poisoned role stamp during task capability denial

This fixes #723 Defect B where attempting to acquire a reviewer lease from a merger session caused the session role to be incorrectly stamped as 'reviewer', leading to downstream workspace binding exceptions.

This commit moves the preflight capability recording to only apply when the task is allowed, and adds a try/except downstream to correctly fail-closed instead of throwing RuntimeError.

gitea_mcp_server.py

@@ -3344,6 +3344,11 @@ def _save_review_decision_lock(data):
     )
     payload["session_profile_lock"] = binding["session_profile_lock"]
     payload["profile_identity"] = binding["profile_identity"]
+    # #720: durable decision locks are recovery-critical terminal provenance,
+    # not generic TTL session cache. Stamp kind + recovery_critical for
+    # pre-existing readers and identity_match_reasons flag-based exempt.
+    payload["kind"] = mcp_session_state.KIND_DECISION_LOCK
+    payload["recovery_critical"] = True
     if binding.get("remote") and not payload.get("remote"):
         payload["remote"] = binding["remote"]
     # #695 AC6: stamp native transport provenance on durable decision locks.
@@ -4091,9 +4096,6 @@ def _evaluate_pr_review_submission(
     worktree_path: str | None = None,
 ) -> dict:
     """Shared gate chain for live submit and dry-run review tools."""
-    _verify_role_mutation_workspace(
-        remote, worktree_path=worktree_path, task="review_pr"
-    )
     action = (action or "").strip().lower()
     workflow_blockers = _review_workflow_load_gate_reasons() if live else []
     result = {
@@ -4110,6 +4112,13 @@ def _evaluate_pr_review_submission(
         "remote": remote if remote in REMOTES else None,
         "reasons": [],
     }
+    try:
+        _verify_role_mutation_workspace(
+            remote, worktree_path=worktree_path, task="review_pr"
+        )
+    except RuntimeError as e:
+        result["reasons"].append(str(e))
+        return result
     reasons = result["reasons"]
     if workflow_blockers:
         reasons.extend(workflow_blockers)
@@ -5027,6 +5036,12 @@ def gitea_cleanup_stale_review_decision_lock(
     binding = _decision_lock_binding()
     active_identity = binding.get("profile_identity")
     lock = _load_review_decision_lock()
+    # #720: when normal load yields no lock, inspect disk so assessment does not
+    # silently report "absent" while an expired/non-critical envelope remains.
+    disk_inspect = mcp_session_state.inspect_state_envelope(
+        kind=mcp_session_state.KIND_DECISION_LOCK,
+        profile_identity=active_identity,
+    )
     last = stale_review_decision_lock.last_terminal_mutation(lock)
     pr_live = None
     pr_lookup_error = None
@@ -5048,6 +5063,26 @@ def gitea_cleanup_stale_review_decision_lock(
         pr_lookup_error=pr_lookup_error,
         active_profile_identity=active_identity,
     )
+    if lock is None and disk_inspect.get("on_disk"):
+        assessment = dict(assessment)
+        assessment["reasons"] = list(assessment.get("reasons") or []) + [
+            "decision-lock file is present on disk but not loadable via normal "
+            f"TTL/identity gates: {disk_inspect.get('summary')} "
+            "(do not rm session-state files; #720)"
+        ]
+        assessment["disk_inspect"] = {
+            k: disk_inspect.get(k)
+            for k in (
+                "on_disk",
+                "has_payload",
+                "age_hours",
+                "age_exceeds_default_ttl",
+                "recovery_critical",
+                "ttl_exempt",
+                "would_ttl_reject",
+                "summary",
+            )
+        }
 
     # Optional pin: refuse apply against a different terminal PR than expected.
     if (
@@ -5102,6 +5137,19 @@ def gitea_cleanup_stale_review_decision_lock(
         "pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
         "merge_commit_sha": assessment.get("merge_commit_sha"),
         "lock_summary": assessment.get("lock_summary"),
+        "disk_inspect": assessment.get("disk_inspect") or {
+            k: disk_inspect.get(k)
+            for k in (
+                "on_disk",
+                "has_payload",
+                "age_hours",
+                "age_exceeds_default_ttl",
+                "recovery_critical",
+                "ttl_exempt",
+                "would_ttl_reject",
+                "summary",
+            )
+        },
         "audit": audit,
         "audit_comment_id": None,
         "reasons": list(assessment.get("reasons") or []),
@@ -9974,9 +10022,20 @@ def gitea_adopt_merger_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.merge"),
         }
 
-    _verify_role_mutation_workspace(
-        remote, worktree=worktree, task="adopt_merger_pr_lease"
-    )
+    try:
+        _verify_role_mutation_workspace(
+            remote, worktree=worktree, task="adopt_merger_pr_lease"
+        )
+    except RuntimeError as e:
+        return {
+            "success": False,
+            "adopted": False,
+            "pr_number": pr_number,
+            "reasons": [str(e)],
+            "active_lease": None,
+            "expected_head_sha": expected_head_sha,
+            "live_head_sha": None,
+        }
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -13877,8 +13936,6 @@ def gitea_resolve_task_capability(
                         "exact_safe_next_action": next_safe_action,
                     }
 
-    record_preflight_check("capability", required_role, resolved_task=task)
-
     # Try automatic dispatch switching
     _ensure_matching_profile(required_permission, required_role, remote, host)
 
@@ -13912,6 +13969,9 @@ def gitea_resolve_task_capability(
         permission_allowed_in_current_session and role_matches_current_session
     )
 
+    if allowed_in_current_session:
+        record_preflight_check("capability", required_role, resolved_task=task)
+
     switching = gitea_config.is_runtime_switching_enabled()
     available_in_session = allowed_in_current_session
     configured = False

mcp_session_state.py

@@ -46,8 +46,15 @@ KIND_IRRECOVERABLE_DECISION_PROVENANCE = "irrecoverable_decision_provenance"
 KIND_IRRECOVERABLE_PROVENANCE_AUTH = "irrecoverable_provenance_authorization"
 
 # Kinds that must survive the default session-state TTL (forensic / recovery).
+#
+# KIND_DECISION_LOCK is recovery-critical (#720): terminal review provenance is
+# not disposable cache. A generic four-hour TTL must not drop old-head evidence
+# or make ``fresh_review_on_current_head_allowed`` unreachable. Same-head / same-
+# run #332 protections still apply once the ledger is loadable. Other session
+# kinds (workflow load, drafts, etc.) remain TTL-bound.
 RECOVERY_CRITICAL_KINDS = frozenset(
     {
+        KIND_DECISION_LOCK,
         KIND_DECISION_LOCK_ARCHIVE,
         KIND_POST_MERGE_DECISION_RECOVERY,
         KIND_IRRECOVERABLE_DECISION_PROVENANCE,
@@ -245,6 +252,16 @@ def _write_json(path: str, data: dict[str, Any]) -> None:
                 pass
 
 
+def is_recovery_critical_record(record: dict[str, Any] | None, kind: str | None = None) -> bool:
+    """True when a durable record must outlive the generic session-state TTL."""
+    if not record and not kind:
+        return False
+    record_kind = ((record or {}).get("kind") or kind or "").strip()
+    if record_kind in RECOVERY_CRITICAL_KINDS:
+        return True
+    return bool((record or {}).get("recovery_critical"))
+
+
 def identity_match_reasons(
     record: dict[str, Any] | None,
     *,
@@ -288,9 +305,7 @@ def identity_match_reasons(
     else:
         age = _now_utc() - recorded_at
         kind = (record.get("kind") or "").strip()
-        ttl_exempt = kind in RECOVERY_CRITICAL_KINDS or bool(
-            record.get("recovery_critical")
-        )
+        ttl_exempt = is_recovery_critical_record(record, kind=kind)
         if age > timedelta(hours=ttl_hours()) and not ttl_exempt:
             reasons.append(
                 f"session state expired after {ttl_hours():g}h (fail closed)"
@@ -300,6 +315,113 @@ def identity_match_reasons(
     return reasons
 
 
+def inspect_state_envelope(
+    *,
+    kind: str,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    profile_identity: str | None = None,
+    state_dir: str | None = None,
+) -> dict[str, Any]:
+    """Read-only disk inspection for assessment when TTL would otherwise hide state (#720).
+
+    Does **not** apply identity/TTL rejection to the returned presence flags.
+    Callers use this to distinguish:
+      * no file on disk
+      * file present but TTL would reject a non-critical kind
+      * recovery-critical ledger (e.g. KIND_DECISION_LOCK) still loadable
+    Never mutates files. Never returns secrets.
+    """
+    profile = current_profile_identity(profile_identity=profile_identity)
+    path = state_file_path(
+        kind=kind,
+        remote=remote,
+        org=org,
+        repo=repo,
+        profile_identity=profile,
+        state_dir=state_dir,
+    )
+    result: dict[str, Any] = {
+        "kind": kind,
+        "profile_identity": profile,
+        "path_basename": os.path.basename(path),
+        "on_disk": False,
+        "has_payload": False,
+        "recorded_at": None,
+        "updated_at": None,
+        "age_hours": None,
+        "ttl_hours": ttl_hours(),
+        "age_exceeds_default_ttl": False,
+        "recovery_critical": kind in RECOVERY_CRITICAL_KINDS,
+        "ttl_exempt": kind in RECOVERY_CRITICAL_KINDS,
+        "would_ttl_reject": False,
+        "identity_reasons": [],
+        "summary": "no session-state file on disk",
+    }
+    if not path or not os.path.exists(path):
+        return result
+    result["on_disk"] = True
+    envelope = _read_json(path)
+    if not envelope:
+        result["summary"] = "session-state file present but unreadable or empty"
+        return result
+    payload = envelope.get("payload")
+    merged: dict[str, Any] = dict(payload) if isinstance(payload, dict) else {}
+    result["has_payload"] = isinstance(payload, dict)
+    for key in (
+        "kind",
+        "remote",
+        "org",
+        "repo",
+        "profile_identity",
+        "session_profile_lock",
+        "recorded_at",
+        "updated_at",
+        "writer_pid",
+        "recovery_critical",
+    ):
+        if key in envelope and key not in merged:
+            merged[key] = envelope[key]
+    if not merged.get("kind"):
+        merged["kind"] = kind
+    recorded_at = _parse_iso(merged.get("recorded_at") or merged.get("updated_at"))
+    result["recorded_at"] = merged.get("recorded_at") or merged.get("updated_at")
+    result["updated_at"] = merged.get("updated_at") or merged.get("recorded_at")
+    if recorded_at is not None:
+        age = _now_utc() - recorded_at
+        age_hours = age.total_seconds() / 3600.0
+        result["age_hours"] = age_hours
+        result["age_exceeds_default_ttl"] = age > timedelta(hours=ttl_hours())
+    ttl_exempt = is_recovery_critical_record(merged, kind=kind)
+    result["recovery_critical"] = ttl_exempt
+    result["ttl_exempt"] = ttl_exempt
+    identity_reasons = identity_match_reasons(
+        merged,
+        remote=remote,
+        org=org,
+        repo=repo,
+        profile_identity=profile,
+    )
+    result["identity_reasons"] = list(identity_reasons)
+    result["would_ttl_reject"] = any("expired" in r for r in identity_reasons)
+    if result["has_payload"] and ttl_exempt:
+        result["summary"] = (
+            "recovery-critical session-state present on disk and TTL-exempt; "
+            "load via load_state for full payload"
+        )
+    elif result["has_payload"] and result["would_ttl_reject"]:
+        result["summary"] = (
+            "session-state file present on disk but generic TTL would reject load "
+            f"(age_hours={result.get('age_hours')!r}, ttl={ttl_hours():g}h)"
+        )
+    elif result["has_payload"]:
+        result["summary"] = "session-state file present and within TTL / identity gates"
+    else:
+        result["summary"] = "session-state file present without a dict payload"
+    return result
+
+
 def load_state(
     *,
     kind: str,

tests/test_issue_720_expired_decision_lock.py

@@ -0,0 +1,499 @@
+"""#720: Expired old-head KIND_DECISION_LOCK must not block fresh review.
+
+Reproduction shape (PR #616):
+  * REQUEST_CHANGES terminal at head A
+  * open PR advanced to head B
+  * durable decision lock age > default 4h TTL
+  * fresh_review_on_current_head_allowed is true but unreachable under TTL-first reject
+  * mark_final fails with "session state expired after 4h"
+  * assessment may report "no lock" while the file remains on disk
+
+Security invariants preserved:
+  * same-head second terminal remains fail-closed (#332/#620)
+  * REQUEST_CHANGES is never treated as approval
+  * historical mutations remain on the ledger
+  * merged/closed moot cleanup still allowed (#594)
+  * irrecoverable provenance still requires #709 authorization
+  * ordinary profiles do not gain gitea.decision_lock.irrecoverable_recovery
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+import sys
+
+ROOT = str(Path(__file__).resolve().parent.parent)
+if ROOT not in sys.path:
+    sys.path.insert(0, ROOT)
+
+import mcp_session_state as ss
+import mcp_server
+import stale_review_decision_lock as srdl
+import task_capability_map
+
+HEAD_A = "a0fffae576673ba7df7456b32e6aec916581bdfb"
+HEAD_B = "a6a2243aad9c3e385fc70509f4941a0f0ec33162"
+HEAD_SAME = HEAD_A
+
+RC_616_A = {
+    "pr_number": 616,
+    "action": "request_changes",
+    "review_id": 443,
+    "review_state": "request_changes",
+    "head_sha": HEAD_A,
+}
+
+
+def _lock(mutations=None, **kwargs):
+    base = {
+        "task": "review_pr",
+        "kind": ss.KIND_DECISION_LOCK,
+        "remote": "prgs",
+        "org": "Scaled-Tech-Consulting",
+        "repo": "Gitea-Tools",
+        "session_pid": os.getpid(),
+        "session_profile": "prgs-reviewer",
+        "session_profile_lock": "prgs-reviewer",
+        "profile_identity": "prgs-reviewer",
+        "final_review_decision_ready": False,
+        "ready_pr_number": kwargs.get("ready_pr"),
+        "ready_action": kwargs.get("ready_action"),
+        "ready_expected_head_sha": kwargs.get("ready_head"),
+        "ready_remote": "prgs" if kwargs.get("ready_pr") else None,
+        "ready_org": "Scaled-Tech-Consulting" if kwargs.get("ready_pr") else None,
+        "ready_repo": "Gitea-Tools" if kwargs.get("ready_pr") else None,
+        "live_mutations": list(mutations or []),
+        "correction_authorized": False,
+        "correction_reason": None,
+    }
+    return base
+
+
+def _open_pr(pr_number=616, head=HEAD_B, merged=False, closed=False):
+    state = "closed" if closed or merged else "open"
+    return {
+        "number": pr_number,
+        "state": state,
+        "merged": merged,
+        "merged_at": "2026-07-16T12:00:00Z" if merged else None,
+        "merge_commit_sha": "m" * 40 if merged else None,
+        "head": {"sha": head},
+    }
+
+
+def _no_lease():
+    return {"block": False, "reasons": [], "mutation_allowed": True}
+
+
+def _feedback(blocking=False, stale=True):
+    return {
+        "success": True,
+        "has_blocking_change_requests": blocking,
+        "review_feedback_stale": stale,
+        "current_head_sha": HEAD_B,
+    }
+
+
+def _age_lock_payload(lock: dict, hours: float = 5.0) -> dict:
+    """Rewrite timestamps to *hours* ago (simulates durable age without touching prod)."""
+    aged = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat().replace(
+        "+00:00", "Z"
+    )
+    out = dict(lock)
+    out["recorded_at"] = aged
+    out["updated_at"] = aged
+    return out
+
+
+class TestIssue720ExpiredDecisionLockLifecycle(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.env = patch.dict(
+            os.environ,
+            {
+                ss.STATE_DIR_ENV: self._tmp.name,
+                ss.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
+                "GITEA_MCP_PROFILE": "prgs-reviewer",
+                "GITEA_PROFILE_NAME": "prgs-reviewer",
+            },
+            clear=False,
+        )
+        self.env.start()
+        mcp_server._REVIEW_DECISION_LOCK = None
+        import review_workflow_load
+
+        review_workflow_load.clear_review_workflow_load()
+        review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
+        mcp_server.gitea_load_review_workflow()
+
+    def tearDown(self):
+        mcp_server._REVIEW_DECISION_LOCK = None
+        import review_workflow_load
+
+        review_workflow_load.clear_review_workflow_load()
+        self.env.stop()
+        self._tmp.cleanup()
+
+    def _persist_aged_lock(self, mutations, hours: float = 5.0, **kwargs):
+        """Write decision lock aged > TTL to the temp durable store (test-only)."""
+        payload = _age_lock_payload(_lock(mutations, **kwargs), hours=hours)
+        saved = ss.save_state(
+            kind=ss.KIND_DECISION_LOCK,
+            payload=payload,
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            profile_identity="prgs-reviewer",
+        )
+        # save_state refreshes updated_at; re-age the on-disk envelope for TTL tests.
+        path = ss.state_file_path(
+            kind=ss.KIND_DECISION_LOCK,
+            profile_identity="prgs-reviewer",
+            state_dir=self._tmp.name,
+        )
+        aged = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat().replace(
+            "+00:00", "Z"
+        )
+        import json
+
+        with open(path, encoding="utf-8") as fh:
+            envelope = json.load(fh)
+        envelope["recorded_at"] = aged
+        envelope["updated_at"] = aged
+        if isinstance(envelope.get("payload"), dict):
+            envelope["payload"]["recorded_at"] = aged
+            envelope["payload"]["updated_at"] = aged
+        with open(path, "w", encoding="utf-8") as fh:
+            json.dump(envelope, fh, indent=2, sort_keys=True)
+            fh.write("\n")
+        # Drop memory so subsequent loads hit durable store.
+        mcp_server._REVIEW_DECISION_LOCK = None
+        return saved
+
+    def _mark(self, pr, action, head, feedback=None):
+        with patch("mcp_server._list_pr_lease_comments", return_value=[]), patch(
+            "mcp_server._pr_work_lease_reviewer_block", return_value=_no_lease()
+        ), patch.object(
+            mcp_server,
+            "gitea_get_pr_review_feedback",
+            return_value=feedback or _feedback(blocking=False, stale=True),
+        ), patch.object(
+            mcp_server,
+            "gitea_check_pr_eligibility",
+            return_value={"eligible": True, "head_sha": head},
+        ), patch.object(
+            mcp_server.mcp_daemon_guard,
+            "assert_sanctioned_mutation_runtime",
+            return_value=None,
+        ), patch.object(
+            mcp_server.mcp_daemon_guard,
+            "assert_no_direct_import_bypass",
+            return_value=None,
+        ):
+            return mcp_server.gitea_mark_final_review_decision(
+                pr_number=pr,
+                action=action,
+                expected_head_sha=head,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+
+    # --- AC regression matrix ---
+
+    def test_under_four_hours_fresh_review_at_b_allowed(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=1.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        res = self._mark(616, "approve", HEAD_B)
+        self.assertTrue(res.get("marked_ready"), res)
+
+    def test_over_four_hours_fresh_review_at_b_still_allowed(self):
+        """Primary #720 defect: age > TTL must not block head-B mark_final."""
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        # Prove durable load is possible for recovery-critical decision locks.
+        loaded = ss.load_state(
+            kind=ss.KIND_DECISION_LOCK,
+            profile_identity="prgs-reviewer",
+        )
+        self.assertIsNotNone(
+            loaded,
+            "expired KIND_DECISION_LOCK must remain loadable (not generic TTL cache)",
+        )
+        res = self._mark(616, "approve", HEAD_B)
+        self.assertTrue(
+            res.get("marked_ready"),
+            f"expected mark_ready on head B after >4h; got {res}",
+        )
+        reasons = " ".join(res.get("reasons") or [])
+        self.assertNotIn("session state expired", reasons)
+
+    def test_historical_review_at_a_preserved_and_stale(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        res = self._mark(616, "approve", HEAD_B)
+        self.assertTrue(res.get("marked_ready"), res)
+        lock = mcp_server._load_review_decision_lock()
+        self.assertIsNotNone(lock)
+        hist = [m for m in lock["live_mutations"] if m.get("review_id") == 443]
+        self.assertEqual(len(hist), 1)
+        self.assertEqual(hist[0]["head_sha"], HEAD_A)
+        self.assertEqual(hist[0]["action"], "request_changes")
+        a = srdl.assess_stale_review_decision_lock(
+            lock, pr_live=_open_pr(616, HEAD_B)
+        )
+        self.assertTrue(a["stale_by_head"])
+        self.assertTrue(a["fresh_review_on_current_head_allowed"])
+        self.assertEqual(a["locked_head_sha"], HEAD_A)
+
+    def test_no_reuse_approval_from_head_a(self):
+        """REQUEST_CHANGES at A is never an approval credential for B."""
+        self._persist_aged_lock([RC_616_A], hours=5.0)
+        lock = ss.load_state(
+            kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer"
+        )
+        last = srdl.last_terminal_mutation(lock)
+        self.assertEqual(last.get("action"), "request_changes")
+        self.assertNotEqual(last.get("action"), "approve")
+        # Gate must still require a new mark/submit for head B; prior RC is not approve.
+        reasons = mcp_server.terminal_review_hard_stop_reasons(
+            616, "mark_ready", expected_head_sha=HEAD_B
+        )
+        self.assertEqual(reasons, [])
+
+    def test_second_terminal_mutation_at_b_same_run_blocked(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        res = self._mark(616, "approve", HEAD_B)
+        self.assertTrue(res.get("marked_ready"), res)
+        # Record terminal approve at B (same run).
+        lock = mcp_server._load_review_decision_lock()
+        lock["final_review_decision_ready"] = True
+        lock["ready_pr_number"] = 616
+        lock["ready_action"] = "approve"
+        lock["ready_expected_head_sha"] = HEAD_B
+        mcp_server._save_review_decision_lock(lock)
+        mcp_server.record_live_review_mutation(616, "approve", review_id=999)
+        # Second terminal at same head B must hard-stop.
+        hard = mcp_server.terminal_review_hard_stop_reasons(
+            616, "mark_ready", expected_head_sha=HEAD_B
+        )
+        self.assertTrue(hard)
+        res2 = self._mark(616, "approve", HEAD_B)
+        self.assertFalse(res2.get("marked_ready"))
+
+    def test_open_pr_same_head_expired_still_fail_closed(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        res = self._mark(616, "approve", HEAD_SAME)
+        self.assertFalse(res.get("marked_ready"), res)
+        self.assertTrue(
+            any("#332" in r or "already consumed" in r for r in (res.get("reasons") or [])),
+            res,
+        )
+
+    def test_merged_pr_moot_cleanup_still_allowed(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        lock = ss.load_state(
+            kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer"
+        )
+        a = srdl.assess_stale_review_decision_lock(
+            lock, pr_live=_open_pr(616, HEAD_A, merged=True)
+        )
+        self.assertTrue(a["is_moot"])
+        self.assertTrue(a["cleanup_allowed"])
+        self.assertFalse(a["fresh_review_on_current_head_allowed"])
+
+    def test_assessment_reports_expired_old_head_not_absent(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        # Disk presence + load after lifecycle fix.
+        path = ss.state_file_path(
+            kind=ss.KIND_DECISION_LOCK,
+            profile_identity="prgs-reviewer",
+            state_dir=self._tmp.name,
+        )
+        self.assertTrue(os.path.exists(path))
+        loaded = ss.load_state(
+            kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer"
+        )
+        self.assertIsNotNone(loaded)
+        inspect = ss.inspect_state_envelope(
+            kind=ss.KIND_DECISION_LOCK,
+            profile_identity="prgs-reviewer",
+            state_dir=self._tmp.name,
+        )
+        self.assertTrue(inspect.get("on_disk"))
+        self.assertTrue(inspect.get("has_payload"))
+        self.assertTrue(inspect.get("recovery_critical") or inspect.get("ttl_exempt"))
+        self.assertTrue(inspect.get("age_hours", 0) >= 4.0)
+        a = srdl.assess_stale_review_decision_lock(
+            loaded, pr_live=_open_pr(616, HEAD_B)
+        )
+        self.assertTrue(a["has_lock"])
+        self.assertNotIn("no review decision lock present", " ".join(a["reasons"]))
+        self.assertTrue(a["stale_by_head"])
+        self.assertEqual(a["last_terminal_action"], "request_changes")
+
+    def test_existing_serialized_records_compatible(self):
+        """Pre-fix ledgers without recovery_critical flag still load via kind."""
+        payload = _age_lock_payload(_lock([RC_616_A]), hours=8.0)
+        payload.pop("recovery_critical", None)
+        # Manually write pre-#720-shaped envelope (kind only on envelope).
+        path = ss.state_file_path(
+            kind=ss.KIND_DECISION_LOCK,
+            profile_identity="prgs-reviewer",
+            state_dir=self._tmp.name,
+        )
+        import json
+
+        os.makedirs(self._tmp.name, exist_ok=True)
+        aged = payload["recorded_at"]
+        envelope = {
+            "kind": ss.KIND_DECISION_LOCK,
+            "remote": "prgs",
+            "org": "Scaled-Tech-Consulting",
+            "repo": "Gitea-Tools",
+            "profile_identity": "prgs-reviewer",
+            "session_profile_lock": "prgs-reviewer",
+            "recorded_at": aged,
+            "updated_at": aged,
+            "writer_pid": os.getpid(),
+            "payload": payload,
+        }
+        with open(path, "w", encoding="utf-8") as fh:
+            json.dump(envelope, fh, indent=2, sort_keys=True)
+            fh.write("\n")
+        loaded = ss.load_state(
+            kind=ss.KIND_DECISION_LOCK, profile_identity="prgs-reviewer"
+        )
+        self.assertIsNotNone(loaded)
+        self.assertEqual(loaded["live_mutations"][0]["review_id"], 443)
+
+    def test_decision_lock_is_recovery_critical_kind(self):
+        self.assertIn(ss.KIND_DECISION_LOCK, ss.RECOVERY_CRITICAL_KINDS)
+
+    def test_workflow_load_still_ttl_expires(self):
+        """Do not make every session-state kind permanently TTL-exempt."""
+        aged = (datetime.now(timezone.utc) - timedelta(hours=5)).isoformat().replace(
+            "+00:00", "Z"
+        )
+        rec = {
+            "kind": ss.KIND_WORKFLOW_LOAD,
+            "recorded_at": aged,
+            "updated_at": aged,
+            "profile_identity": "prgs-reviewer",
+            "session_profile_lock": "prgs-reviewer",
+        }
+        reasons = ss.identity_match_reasons(
+            rec, profile_identity="prgs-reviewer"
+        )
+        self.assertTrue(any("expired" in r for r in reasons), reasons)
+
+    def test_irrecoverable_permission_not_on_ordinary_profiles(self):
+        perm = "gitea.decision_lock.irrecoverable_recovery"
+        # Capability map must not map ordinary author/reviewer/merger tasks to it.
+        for task in (
+            "create_issue",
+            "comment_issue",
+            "review_pr",
+            "merge_pr",
+            "lock_issue",
+            "create_pr",
+        ):
+            req = task_capability_map.required_permission(task)
+            self.assertNotEqual(req, perm, msg=task)
+
+    def test_structured_error_when_same_head_expired(self):
+        self._persist_aged_lock(
+            [RC_616_A],
+            hours=5.0,
+            ready_pr=616,
+            ready_head=HEAD_A,
+            ready_action="request_changes",
+        )
+        res = self._mark(616, "approve", HEAD_A)
+        self.assertFalse(res.get("marked_ready"))
+        reasons = res.get("reasons") or []
+        self.assertTrue(reasons)
+        blob = " ".join(reasons)
+        # Actionable recovery text from hard-stop (not generic internal_error).
+        self.assertIn("#332", blob)
+        self.assertTrue(
+            "#620" in blob or "head moved" in blob or "new expected_head_sha" in blob
+            or "already consumed" in blob
+        )
+        self.assertNotIn("internal_error", blob.lower())
+
+
+class TestIssue720InspectEnvelope(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.env = patch.dict(
+            os.environ,
+            {
+                ss.STATE_DIR_ENV: self._tmp.name,
+                ss.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
+            },
+            clear=False,
+        )
+        self.env.start()
+
+    def tearDown(self):
+        self.env.stop()
+        self._tmp.cleanup()
+
+    def test_inspect_missing_file(self):
+        info = ss.inspect_state_envelope(
+            kind=ss.KIND_DECISION_LOCK,
+            profile_identity="prgs-reviewer",
+            state_dir=self._tmp.name,
+        )
+        self.assertFalse(info["on_disk"])
+        self.assertFalse(info["has_payload"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_resolve_task_capability.py

@@ -442,5 +442,23 @@ class TestResolveTaskCapability(unittest.TestCase):
                 with self.assertRaises(ValueError):
                     mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs")
 
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_task_capability_does_not_poison_role_stamp_on_denial(self, _auth, _api):
+        # Issue #723: attempting to acquire a reviewer lease from a merger profile
+        # should fail without poisoning the session role stamp as 'reviewer'.
+        with patch.dict(os.environ, self._env("merger-profile")):
+            # Initially no preflight check is recorded
+            self.assertEqual(mcp_server._preflight_resolved_role, None)
+            
+            # Request reviewer lease task which should be denied for merger
+            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            
+            # The task is denied
+            self.assertFalse(res["allowed_in_current_session"])
+            
+            # The session role stamp should NOT be poisoned
+            self.assertNotEqual(mcp_server._preflight_resolved_role, "reviewer")
+
 if __name__ == "__main__":
     unittest.main()