fix(mcp): let a sanctioned recovery keep its own owning PR (Closes #755)

#753 added the dead-session author-lock recovery assessor, but no sanctioned
recovery could ever reach the lock write. A dead-session lock is by construction
a lock for work that already has an open PR, and the #400 duplicate-work gate
blocked unconditionally on any linked open PR. gitea_lock_issue computed
recovery_sanctioned, then discarded it one gate later:

    open PR #750 already covers issue #749 (fail closed)

Because the recovered lock was never persisted, it stayed non-live, so
_prove_author_ownership_for_pr found no live author lock and
gitea_update_pr_branch_by_merge failed closed too. Both blockers had a single
cause, so only the first one is fixed here.

issue_lock_recovery.owning_pr_recovery_evidence distils a granted assessment
into the minimum evidence the duplicate gate needs: issue, branch, owning PR
number, and head. It returns None unless the outcome is RECOVERY_SANCTIONED and
the assessment's own evidence agrees that local head == remote head == PR head,
so a partial or hand-built assessment cannot authorize anything.

The duplicate gate takes that evidence and re-checks every element against the
live PR list it was handed: exactly one linked open PR, matching number, head
branch, head SHA, and locked branch. Only that exact self-owned PR is exempt.
A different PR, several linked PRs, a different branch or head, or evidence for
another issue all keep failing closed, with a diagnostic naming which element
disagreed. The competing-branch, claim-lease and commit/push/create_pr arms are
untouched, and with no evidence the gate behaves exactly as before.

Ownership proof needed no change: once the recovered lock persists under the
live session pid, _prove_author_ownership_for_pr matches it as before.

Out of scope: the PHASE_COMMIT/PHASE_PUSH/PHASE_CREATE_PR rechecks still block
on a linked open PR for every author, recovered or not. That is pre-existing
#400 behavior, unrelated to lock recovery, and #755 does not cover it.

Tests
- tests/test_issue_755_owning_pr_recovery.py (new, 31 tests) drives the real
  mcp_server.gitea_lock_issue handler, not only the pure assessor: sanctioned
  recovery relocks, persists a live lease with truthful dead_session_recovery
  provenance, and satisfies _prove_author_ownership_for_pr; competing PR,
  multiple linked PRs, mismatched head/branch/worktree, dirty worktree, live
  prior pid, and a fresh claim with no prior lock all stay blocked.
- Neutralizing owning_pr_recovery_evidence regresses all three success tests to
  the exact production error above, so the coverage is load-bearing.
- Focused suites (755, 753, duplicate gates, lock registration, provenance) —
  102 passed.
- Lock/ownership/worktree/handoff suites — 172 passed, 16 subtests.
- Full suite tests/ — 3529 passed, 6 skipped, 2 failed, 365 subtests.
- The same 2 failures reproduce on a pristine detached worktree at
  08f67007c5 (3498 passed, 6 skipped, 2 failed):
  test_issue_702_review_findings_f1_f6 F1 recovery-before-probe and
  test_reconciler_supersession_close org/repo forwarding. Pre-existing.
- git diff --check clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_017BNsk3KUuFchaZyPJsCxjk

gitea_mcp_server.py

@@ -1650,6 +1650,7 @@ import issue_lock_worktree  # noqa: E402
 import issue_lock_provenance  # noqa: E402
 import issue_lock_store  # noqa: E402
 import issue_lock_adoption  # noqa: E402
+import issue_lock_recovery  # noqa: E402
 import stacked_pr_support  # noqa: E402
 import merge_approval_gate  # noqa: E402
 import review_quarantine  # noqa: E402  # #695 contaminated formal-review quarantine
@@ -2080,6 +2081,7 @@ def _assess_issue_duplicate_gate(
     auth: str,
     locked_branch: str | None = None,
     phase: str,
+    recovered_owning_pr: dict | None = None,
 ) -> dict:
     open_prs, branch_names, claim_entry = _collect_issue_duplicate_context(
         h, o, r, auth, issue_number
@@ -2091,6 +2093,7 @@ def _assess_issue_duplicate_gate(
         claim_entry=claim_entry,
         locked_branch=locked_branch,
         phase=phase,
+        recovered_owning_pr=recovered_owning_pr,
     )
 
 
@@ -3155,8 +3158,11 @@ def gitea_lock_issue(
         worktree_path, _canonical_local_git_root()
     )
     h, o, r = _resolve(remote, host, org, repo)
+    existing_issue_lock = _load_existing_issue_lock(
+        remote=remote, org=o, repo=r, issue_number=issue_number
+    )
     active_lease_block = issue_lock_store.assess_same_issue_lease_conflict(
-        _load_existing_issue_lock(remote=remote, org=o, repo=r, issue_number=issue_number),
+        existing_issue_lock,
         issue_number=issue_number,
         branch_name=branch_name,
         worktree_path=resolved_worktree,
@@ -3198,6 +3204,83 @@ def gitea_lock_issue(
         org=org,
         repo=repo,
     )
+    # ── Dead-session lock recovery assessment (#753) ──
+    # When the MCP session that took a lock exits, the lock goes non-live
+    # (stale by dead PID) even inside its lease TTL, and the branch it owns is
+    # ahead of its base by construction — so the base-equivalence gate below
+    # makes normal re-lock unreachable for every PR that already exists.
+    #
+    # This grants a waiver ONLY for that case, proven against the durable lock
+    # record plus live git/Gitea observation. A refused assessment never raises:
+    # it simply withholds the waiver, leaving the pre-existing guard to fail
+    # closed exactly as before. Recovery can only ever add permission.
+    recovery_assessment: dict | None = None
+    if (
+        existing_issue_lock
+        and existing_issue_lock.get("issue_number") == issue_number
+        and not issue_lock_store.is_lease_live(existing_issue_lock)
+    ):
+        recovery_auth = _auth(h)
+        try:
+            recovery_branches = api_get_all(
+                f"{repo_api_url(h, o, r)}/branches", recovery_auth
+            )
+        except Exception as exc:
+            raise RuntimeError(
+                f"Could not list branches to verify issue-lock recovery: {exc}"
+            )
+        recovery_remote_head: str | None = None
+        recovery_candidates: list[str] = []
+        for entry in recovery_branches:
+            entry_name = _branch_entry_name(entry)
+            if entry_name == branch_name:
+                recovery_remote_head = _branch_entry_commit_sha(entry)
+            if issue_lock_adoption.branch_carries_issue_marker(entry_name, issue_number):
+                recovery_candidates.append(entry_name)
+        recovery_pr_head: str | None = None
+        recovery_pr_number: int | None = None
+        for pull in _list_open_pulls(h, o, r, recovery_auth):
+            pull_head = pull.get("head") or {}
+            if str(pull_head.get("ref") or "") == branch_name:
+                recovery_pr_head = pull_head.get("sha")
+                recovery_pr_number = pull.get("number")
+                break
+        recovery_claimant = _work_lease_claimant(h)
+        recovery_assessment = issue_lock_recovery.assess_dead_session_lock_recovery(
+            existing_issue_lock,
+            issue_number=issue_number,
+            branch_name=branch_name,
+            worktree_path=resolved_worktree,
+            remote=remote,
+            org=o,
+            repo=r,
+            identity=recovery_claimant.get("username"),
+            profile=recovery_claimant.get("profile"),
+            current_branch=git_state.get("current_branch"),
+            porcelain_status=git_state.get("porcelain_status") or "",
+            head_sha=git_state.get("head_sha"),
+            remote_head_sha=recovery_remote_head,
+            pr_head_sha=recovery_pr_head,
+            pr_number=recovery_pr_number,
+            competing_live_locks=issue_lock_store.list_live_locks(),
+            candidate_branches=recovery_candidates,
+            current_pid=os.getpid(),
+        )
+
+    recovery_sanctioned = bool(
+        recovery_assessment and recovery_assessment.get("recovery_sanctioned")
+    )
+    # #755: a sanctioned dead-session recovery always has an owning open PR —
+    # that is what makes it a recovery rather than a fresh claim. Carry the
+    # server-derived owning-PR evidence into the duplicate-work gate below so
+    # the PR this lock already owns is not mistaken for competing duplicate
+    # work. Withheld (None) unless recovery was granted, so the ordinary
+    # duplicate blocker is untouched on every other path.
+    recovered_owning_pr = (
+        issue_lock_recovery.owning_pr_recovery_evidence(recovery_assessment)
+        if recovery_sanctioned
+        else None
+    )
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -3205,10 +3288,20 @@ def gitea_lock_issue(
         base_equivalent=git_state.get("base_equivalent"),
         inspected_git_root=git_state.get("inspected_git_root"),
         base_branch=git_state.get("base_branch"),
+        recovery_sanctioned=recovery_sanctioned,
     )
     if lock_assessment["block"]:
+        reasons = list(lock_assessment.get("reasons") or [])
+        # Surface why recovery was unavailable, so a blocked caller sees the
+        # exact missing evidence instead of only the base-equivalence text.
+        if recovery_assessment and recovery_assessment.get("is_candidate"):
+            reasons.append(
+                issue_lock_recovery.format_recovery_refusal(recovery_assessment)
+            )
         raise RuntimeError(
-            issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
+            issue_lock_worktree.format_issue_lock_worktree_error(
+                {**lock_assessment, "reasons": reasons}
+            )
         )
 
     auth = _auth(h)
@@ -3220,6 +3313,7 @@ def gitea_lock_issue(
         auth=auth,
         locked_branch=branch_name,
         phase=issue_work_duplicate_gate.PHASE_LOCK,
+        recovered_owning_pr=recovered_owning_pr,
     )
     if duplicate_gate.get("block"):
         raise ValueError("; ".join(duplicate_gate.get("reasons") or [
@@ -3271,6 +3365,14 @@ def gitea_lock_issue(
     }
     if stacked_approved:
         data["approved_stacked_base"] = stacked_approved
+    if recovery_sanctioned and recovery_assessment:
+        # #753 AC2/AC6: record that this claim was recovered after session
+        # death, with the prior and replacement session identity, so the
+        # takeover is auditable and never looks like an original claim.
+        data["dead_session_recovery"] = issue_lock_recovery.build_recovery_record(
+            recovery_assessment,
+            recovered_at=_work_lease_timestamp(_work_lease_now()),
+        )
 
     lock_file_path = _save_issue_lock(data)
     lock_record = issue_lock_store.read_lock_file(lock_file_path) or data
@@ -3304,6 +3406,14 @@ def gitea_lock_issue(
         "lock_freshness": freshness,
         "lock_proof": lock_proof,
     }
+    if recovery_sanctioned and recovery_assessment:
+        result["dead_session_recovery"] = data["dead_session_recovery"]
+        result["message"] = (
+            f"Recovered the durable lock for issue #{issue_number} on branch "
+            f"'{branch_name}' after the owning MCP session (pid "
+            f"{recovery_assessment['evidence'].get('prior_session_pid')}) exited; "
+            "ownership evidence matched exactly (fail-closed check complete)."
+        )
     if stacked_approved:
         result["approved_stacked_base"] = stacked_approved
         result["message"] = (

issue_lock_adoption.py

@@ -85,6 +85,15 @@ def _branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool:
     return re.search(pattern, name) is not None
 
 
+def branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool:
+    """Public accessor for the exact issue-marker match (#753).
+
+    Dead-session lock recovery needs the same word-boundary matcher to detect
+    ambiguous branch claims, so it is exposed rather than reached into.
+    """
+    return _branch_carries_issue_marker(branch_name, issue_number)
+
+
 def assess_own_branch_adoption(
     *,
     issue_number: int,

issue_lock_recovery.py

@@ -0,0 +1,451 @@
+"""Dead-session author issue-lock recovery (#753).
+
+A durable author issue lock records the PID of the MCP session that took it.
+When that process exits, ``issue_lock_store.assess_lock_freshness`` classifies
+the lock as ``stale`` (``live=False``) even while its lease is still within TTL,
+so every ownership check that requires a *live* lock fails closed.
+
+Re-taking the lock through ``gitea_lock_issue`` is unreachable for real work:
+``issue_lock_worktree.assess_issue_lock_worktree`` demands the worktree be
+base-equivalent to ``master``/``main``/``dev``, and a branch that already
+carries commits is ahead of its base by construction. The existing
+``assess_expired_lock_reclaim`` affordance does not apply either, because
+``assess_same_issue_lease_conflict`` only consults it once the lease has
+*expired* — a dead PID under an unexpired lease never reaches it.
+
+This module is the pure evidence assessor for that one narrow case. It grants
+recovery only when every element of durable ownership still matches exactly and
+the recorded process is demonstrably dead. It never trusts caller assertions:
+every field is compared against durable lock state or live observation supplied
+by the caller. It performs no mutation and no network I/O.
+
+Recovery deliberately does **not** relax base-equivalence for brand-new issue
+claims — only for a lock whose own prior record already proves the branch,
+worktree, head, and author.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Iterable, Mapping, Sequence
+
+from issue_lock_store import is_process_alive
+from reviewer_worktree import parse_dirty_tracked_files
+
+# Outcome values
+RECOVERY_SANCTIONED = "RECOVERY_SANCTIONED"
+NO_CANDIDATE = "NO_CANDIDATE"
+REFUSED = "REFUSED"
+
+# Durable fields a lock must carry before it can be considered at all.
+REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path")
+
+
+def _same_realpath(left: str | None, right: str | None) -> bool:
+    if not left or not right:
+        return False
+    try:
+        return os.path.realpath(left) == os.path.realpath(right)
+    except OSError:
+        return left == right
+
+
+def _text(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def _lock_claimant(lock: Mapping[str, Any]) -> dict[str, Any]:
+    claimant = lock.get("claimant")
+    if not isinstance(claimant, Mapping):
+        lease = lock.get("work_lease")
+        claimant = lease.get("claimant") if isinstance(lease, Mapping) else None
+    return dict(claimant) if isinstance(claimant, Mapping) else {}
+
+
+def _recorded_pid(lock: Mapping[str, Any]) -> Any:
+    pid = lock.get("session_pid")
+    if pid is None:
+        pid = lock.get("pid")
+    return pid
+
+
+def _malformed_reasons(lock: Mapping[str, Any]) -> list[str]:
+    """Names of durable fields that are missing or unusable."""
+    missing: list[str] = []
+    for field in REQUIRED_LOCK_FIELDS:
+        if not _text(lock.get(field)):
+            missing.append(field)
+    pid = _recorded_pid(lock)
+    if pid is None or _text(pid) == "":
+        missing.append("session_pid/pid")
+    else:
+        try:
+            if int(pid) <= 0:
+                missing.append("session_pid/pid")
+        except (TypeError, ValueError):
+            missing.append("session_pid/pid")
+    return missing
+
+
+def assess_dead_session_lock_recovery(
+    existing_lock: Mapping[str, Any] | None,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    remote: str,
+    org: str,
+    repo: str,
+    identity: str | None,
+    profile: str | None,
+    current_branch: str | None,
+    porcelain_status: str,
+    head_sha: str | None,
+    remote_head_sha: str | None,
+    pr_head_sha: str | None = None,
+    pr_number: int | None = None,
+    competing_live_locks: Sequence[Mapping[str, Any]] | None = None,
+    candidate_branches: Iterable[str] | None = None,
+    current_pid: int | None = None,
+) -> dict[str, Any]:
+    """Decide whether a dead-session author lock may be natively recovered.
+
+    Returns a dict with ``recovery_sanctioned`` (bool), ``outcome``, ``reasons``
+    (why it was refused, or the positive proof when sanctioned), and
+    ``evidence`` (a redaction-safe record for auditing).
+
+    ``NO_CANDIDATE`` means no recovery was attempted at all — there is no
+    existing lock, or the lock does not describe this issue. The caller must
+    treat that exactly as it treated the pre-#753 world. ``REFUSED`` means a
+    candidate existed but the evidence did not agree; the caller fails closed.
+    """
+    reasons: list[str] = []
+    evidence: dict[str, Any] = {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "worktree_path": worktree_path,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+    }
+
+    if not existing_lock:
+        return _result(
+            NO_CANDIDATE, False, ["no existing durable lock for this issue"], evidence
+        )
+
+    lock = dict(existing_lock)
+
+    # ── Candidate identification ────────────────────────────────────────────
+    # Recovery only ever applies to a lock that already claims THIS issue.
+    # Anything else is not a recovery candidate and must not be reinterpreted.
+    if lock.get("issue_number") != issue_number:
+        return _result(
+            NO_CANDIDATE,
+            False,
+            [
+                f"existing lock targets issue #{lock.get('issue_number')}, "
+                f"not #{issue_number}; not a recovery candidate"
+            ],
+            evidence,
+        )
+
+    # A malformed/incomplete durable record can never prove ownership.
+    missing = _malformed_reasons(lock)
+    if missing:
+        return _result(
+            REFUSED,
+            False,
+            [
+                "durable lock record is incomplete and cannot prove ownership "
+                f"(missing/unusable: {', '.join(missing)})"
+            ],
+            evidence,
+        )
+
+    recorded_pid = _recorded_pid(lock)
+    evidence["prior_session_pid"] = recorded_pid
+    evidence["replacement_session_pid"] = (
+        current_pid if current_pid is not None else os.getpid()
+    )
+
+    # ── Repository scope ────────────────────────────────────────────────────
+    for field, expected in (("remote", remote), ("org", org), ("repo", repo)):
+        actual = _text(lock.get(field))
+        if actual != _text(expected):
+            reasons.append(
+                f"lock {field} '{actual}' does not match requested '{_text(expected)}'"
+            )
+
+    # ── Branch identity ─────────────────────────────────────────────────────
+    locked_branch = _text(lock.get("branch_name"))
+    if locked_branch != _text(branch_name):
+        reasons.append(
+            f"lock branch '{locked_branch}' does not match requested "
+            f"'{_text(branch_name)}'"
+        )
+    evidence["locked_branch"] = locked_branch
+
+    # The worktree must actually be sitting on the locked branch. Without this
+    # a clean worktree parked elsewhere could stand in for the real work.
+    checked_out = _text(current_branch)
+    if not checked_out:
+        reasons.append(
+            "worktree is not on a named branch (detached HEAD); locked-branch "
+            "occupancy could not be proven"
+        )
+    elif checked_out != locked_branch:
+        reasons.append(
+            f"worktree is on branch '{checked_out}', not the locked branch "
+            f"'{locked_branch}'"
+        )
+
+    # ── Worktree identity ───────────────────────────────────────────────────
+    locked_worktree = _text(lock.get("worktree_path"))
+    if not _same_realpath(locked_worktree, worktree_path):
+        reasons.append(
+            f"lock worktree '{locked_worktree}' does not match declared "
+            f"'{_text(worktree_path)}'"
+        )
+    evidence["locked_worktree_path"] = locked_worktree
+
+    # ── Cleanliness (never waived) ──────────────────────────────────────────
+    dirty_files = parse_dirty_tracked_files(porcelain_status)
+    if dirty_files:
+        reasons.append(
+            "worktree has tracked local edits; recovery requires a clean "
+            f"worktree (dirty files: {', '.join(dirty_files)})"
+        )
+    evidence["dirty_files"] = dirty_files
+
+    # ── Head agreement: local == remote == PR ───────────────────────────────
+    local_head = _text(head_sha)
+    remote_head = _text(remote_head_sha)
+    if not local_head:
+        reasons.append("local head SHA could not be determined")
+    if not remote_head:
+        reasons.append(
+            f"remote head for branch '{locked_branch}' could not be determined"
+        )
+    if local_head and remote_head and local_head != remote_head:
+        reasons.append(
+            f"local head {local_head} does not match remote branch head {remote_head}"
+        )
+    evidence["local_head"] = local_head or None
+    evidence["remote_head"] = remote_head or None
+
+    pr_head = _text(pr_head_sha)
+    if pr_head:
+        evidence["pr_head"] = pr_head
+        evidence["pr_number"] = pr_number
+        if local_head and pr_head != local_head:
+            reasons.append(
+                f"open PR #{pr_number} head {pr_head} does not match local head "
+                f"{local_head}"
+            )
+
+    # ── Author identity ─────────────────────────────────────────────────────
+    claimant = _lock_claimant(lock)
+    locked_identity = _text(claimant.get("username"))
+    locked_profile = _text(claimant.get("profile"))
+    evidence["locked_identity"] = locked_identity or None
+    evidence["locked_profile"] = locked_profile or None
+    if not locked_identity or not locked_profile:
+        reasons.append(
+            "durable lock does not record a claimant identity/profile; "
+            "author ownership could not be proven"
+        )
+    if not _text(identity) or not _text(profile):
+        reasons.append(
+            "active session identity/profile is unknown; author ownership "
+            "could not be proven"
+        )
+    if locked_identity and _text(identity) and locked_identity != _text(identity):
+        reasons.append(
+            f"lock claimant '{locked_identity}' does not match active identity "
+            f"'{_text(identity)}'"
+        )
+    if locked_profile and _text(profile) and locked_profile != _text(profile):
+        reasons.append(
+            f"lock profile '{locked_profile}' does not match active profile "
+            f"'{_text(profile)}'"
+        )
+
+    # ── The defining condition: the recorded owner must be dead ─────────────
+    prior_alive = is_process_alive(recorded_pid)
+    evidence["prior_pid_alive"] = prior_alive
+    if prior_alive:
+        reasons.append(
+            f"prior owner pid {recorded_pid} is still alive; this is not a "
+            "dead-session recovery"
+        )
+    if current_pid is not None and recorded_pid is not None:
+        try:
+            if int(recorded_pid) == int(current_pid):
+                reasons.append(
+                    "recorded pid is the current session; nothing to recover"
+                )
+        except (TypeError, ValueError):
+            pass
+
+    # ── Competing ownership ─────────────────────────────────────────────────
+    competing: list[dict[str, Any]] = []
+    for entry in competing_live_locks or ():
+        if not isinstance(entry, Mapping):
+            continue
+        same_issue = entry.get("issue_number") == issue_number
+        same_branch = _text(entry.get("branch_name")) == locked_branch
+        if not (same_issue or same_branch):
+            continue
+        # The lock we are recovering is not competition with itself.
+        if (
+            same_issue
+            and same_branch
+            and _same_realpath(_text(entry.get("worktree_path")), worktree_path)
+        ):
+            continue
+        competing.append(
+            {
+                "issue_number": entry.get("issue_number"),
+                "branch_name": entry.get("branch_name"),
+                "worktree_path": entry.get("worktree_path"),
+                "pid": entry.get("pid"),
+            }
+        )
+    if competing:
+        described = ", ".join(
+            f"issue #{c['issue_number']} branch '{c['branch_name']}'" for c in competing
+        )
+        reasons.append(f"competing live lock or lease exists ({described})")
+    evidence["competing_live_locks"] = competing
+
+    # ── Ambiguous branch claims ─────────────────────────────────────────────
+    others = [
+        name
+        for name in (candidate_branches or ())
+        if _text(name) and _text(name) != locked_branch
+    ]
+    if others:
+        reasons.append(
+            "multiple branches claim this issue "
+            f"({', '.join(sorted(set(others)))}); ownership is ambiguous"
+        )
+    evidence["other_candidate_branches"] = sorted(set(others))
+
+    if reasons:
+        return _result(REFUSED, False, reasons, evidence)
+
+    return _result(
+        RECOVERY_SANCTIONED,
+        True,
+        [
+            f"durable lock for issue #{issue_number} matches branch "
+            f"'{locked_branch}', worktree '{locked_worktree}', head {local_head}, "
+            f"and claimant '{locked_identity}'; recorded pid {recorded_pid} is dead"
+        ],
+        evidence,
+    )
+
+
+def _result(
+    outcome: str,
+    sanctioned: bool,
+    reasons: list[str],
+    evidence: dict[str, Any],
+) -> dict[str, Any]:
+    return {
+        "outcome": outcome,
+        "recovery_sanctioned": sanctioned,
+        "is_candidate": outcome != NO_CANDIDATE,
+        "reasons": reasons,
+        "evidence": evidence,
+    }
+
+
+def owning_pr_recovery_evidence(
+    assessment: Mapping[str, Any] | None,
+) -> dict[str, Any] | None:
+    """Server-derived proof of the open PR a sanctioned recovery already owns (#755).
+
+    A dead-session recovery is, by construction, recovery of work that already
+    has an open PR — so the duplicate-work gate's linked-open-PR blocker would
+    otherwise discard every sanctioned recovery. This distils the completed
+    assessment into the minimum evidence that gate needs to tell "the PR this
+    lock already owns" apart from "a competing duplicate PR".
+
+    Returns ``None`` unless recovery was actually granted and the assessment's
+    own evidence names exactly one owning PR whose head agrees with the local
+    and remote heads. Nothing here is caller-supplied: every field is copied
+    from evidence the assessor built out of durable lock state plus live
+    git/Gitea observation, so a caller cannot manufacture an exemption.
+    """
+    if not isinstance(assessment, Mapping):
+        return None
+    if assessment.get("outcome") != RECOVERY_SANCTIONED:
+        return None
+    if not assessment.get("recovery_sanctioned"):
+        return None
+
+    evidence = assessment.get("evidence") or {}
+    branch_name = _text(evidence.get("locked_branch"))
+    pr_head = _text(evidence.get("pr_head"))
+    local_head = _text(evidence.get("local_head"))
+    remote_head = _text(evidence.get("remote_head"))
+    raw_pr_number = evidence.get("pr_number")
+
+    if raw_pr_number is None or not branch_name or not pr_head:
+        return None
+    # The assessor already required these to agree. Re-check, so a truncated or
+    # hand-built evidence map can never authorize an exemption.
+    if pr_head != local_head or pr_head != remote_head:
+        return None
+    try:
+        pr_number = int(raw_pr_number)
+        issue_number = int(evidence.get("issue_number"))
+    except (TypeError, ValueError):
+        return None
+
+    return {
+        "issue_number": issue_number,
+        "pr_number": pr_number,
+        "branch_name": branch_name,
+        "head_sha": pr_head,
+    }
+
+
+def build_recovery_record(
+    assessment: Mapping[str, Any],
+    *,
+    recovered_at: str,
+) -> dict[str, Any]:
+    """Durable, secret-free provenance for a completed recovery (#753 AC2/AC6)."""
+    evidence = dict(assessment.get("evidence") or {})
+    return {
+        "recovered": True,
+        "reason": "owning MCP session exited; durable ownership evidence matched",
+        "recovered_at": recovered_at,
+        "prior_session_pid": evidence.get("prior_session_pid"),
+        "replacement_session_pid": evidence.get("replacement_session_pid"),
+        "prior_pid_alive": evidence.get("prior_pid_alive"),
+        "branch_name": evidence.get("locked_branch"),
+        "worktree_path": evidence.get("locked_worktree_path"),
+        "local_head": evidence.get("local_head"),
+        "remote_head": evidence.get("remote_head"),
+        "pr_head": evidence.get("pr_head"),
+        "pr_number": evidence.get("pr_number"),
+        "identity": evidence.get("locked_identity"),
+        "profile": evidence.get("locked_profile"),
+        "proof": list(assessment.get("reasons") or []),
+    }
+
+
+def format_recovery_refusal(assessment: Mapping[str, Any]) -> str:
+    """Single fail-closed message for a refused recovery attempt."""
+    reasons = list(assessment.get("reasons") or []) or [
+        "dead-session lock recovery evidence did not agree"
+    ]
+    return (
+        "Dead-session issue-lock recovery refused: "
+        + "; ".join(reasons)
+        + " (fail closed)"
+    )

issue_lock_worktree.py

@@ -92,8 +92,19 @@ def assess_issue_lock_worktree(
     inspected_git_root: str | None = None,
     base_branch: str | None = None,
     base_branches: frozenset[str] | None = None,
+    recovery_sanctioned: bool = False,
 ) -> dict:
-    """Fail closed when lock preconditions are not met on the declared worktree."""
+    """Fail closed when lock preconditions are not met on the declared worktree.
+
+    ``recovery_sanctioned`` is set only when ``issue_lock_recovery`` has already
+    proven, from the durable lock itself, that this is a dead-session recovery of
+    an existing claim (#753): same issue, branch, worktree, author, and head, with
+    the recording process dead. In that one case the base-equivalence requirement
+    is waived, because a branch that already carries the work is ahead of its base
+    by construction and could never satisfy it. Every other precondition —
+    notably worktree cleanliness — still applies unchanged, and brand-new issue
+    claims keep the full base-equivalence requirement.
+    """
     bases = base_branches or BASE_BRANCHES
     reasons: list[str] = []
     path = (worktree_path or "").strip()
@@ -111,7 +122,11 @@ def assess_issue_lock_worktree(
             f"(dirty files: {', '.join(dirty_files)})"
         )
 
-    if base_equivalent is False:
+    if recovery_sanctioned:
+        # Base-equivalence intentionally not evaluated: ownership was proven
+        # against the durable lock record instead (#753).
+        pass
+    elif base_equivalent is False:
         reasons.append(
             "issue lock worktree must be base-equivalent to one of "
             f"{_base_list(bases)} before implementation work; inspected "
@@ -139,6 +154,7 @@ def assess_issue_lock_worktree(
         inspected_git_root=inspected_git_root,
         base_branch=base_branch,
         base_equivalent=base_equivalent,
+        recovery_sanctioned=recovery_sanctioned,
     )
 
 
@@ -197,6 +213,7 @@ def _assessment(
     inspected_git_root: str | None = None,
     base_branch: str | None = None,
     base_equivalent: bool | None = None,
+    recovery_sanctioned: bool = False,
 ) -> dict:
     return {
         "proven": proven,
@@ -208,6 +225,8 @@ def _assessment(
         "dirty_files": dirty_files,
         "base_branch": base_branch,
         "base_equivalent": base_equivalent,
+        "recovery_sanctioned": recovery_sanctioned,
+        "base_equivalence_waived": bool(recovery_sanctioned),
     }
 
 

issue_work_duplicate_gate.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Mapping
 
 import issue_claim_heartbeat as claim_hb
 
@@ -27,6 +27,116 @@ def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None:
     return claim_hb._linked_open_pr(issue_number, open_prs)
 
 
+def _pr_links_issue(issue_number: int, pr: Mapping[str, Any]) -> bool:
+    """Same linkage rule ``claim_hb._linked_open_pr`` applies, per PR.
+
+    ``_linked_open_pr`` only yields the *first* match, which cannot answer
+    "is there exactly one linked PR?" — a question the owning-PR exemption
+    below must answer before it can trust any of them.
+    """
+    pattern = _issue_pattern(issue_number)
+    head = (pr.get("head") or {}).get("ref") or ""
+    text = f"{pr.get('title', '')} {pr.get('body', '')}".lower()
+    if pattern in head.lower():
+        return True
+    return (
+        f"closes #{int(issue_number)}" in text
+        or f"fixes #{int(issue_number)}" in text
+    )
+
+
+def _all_linked_open_prs(
+    issue_number: int, open_prs: list[dict]
+) -> list[Mapping[str, Any]]:
+    return [pr for pr in (open_prs or []) if _pr_links_issue(issue_number, pr)]
+
+
+def _assess_owning_pr_exemption(
+    issue_number: int,
+    *,
+    linked_open_prs: list[Mapping[str, Any]],
+    locked_branch: str | None,
+    recovered_owning_pr: Mapping[str, Any] | None,
+) -> tuple[bool, list[str]]:
+    """Is the linked open PR provably the one a sanctioned recovery owns (#755)?
+
+    ``recovered_owning_pr`` is produced by
+    ``issue_lock_recovery.owning_pr_recovery_evidence`` from a completed
+    server-side recovery assessment — it is never a caller-supplied field.
+    Every element is re-checked here against the live PR list this gate was
+    given, so a stale or partial token cannot widen the exemption.
+
+    Returns ``(exempt, diagnostic_reasons)``. Diagnostics are only emitted when
+    a token was offered and rejected, so a blocked caller can see which element
+    of ownership disagreed.
+    """
+    if not recovered_owning_pr:
+        return False, []
+
+    notes: list[str] = []
+    token_issue = recovered_owning_pr.get("issue_number")
+    token_pr = recovered_owning_pr.get("pr_number")
+    token_branch = str(recovered_owning_pr.get("branch_name") or "").strip()
+    token_head = str(recovered_owning_pr.get("head_sha") or "").strip()
+    locked = (locked_branch or "").strip()
+
+    if token_issue is not None and int(token_issue) != int(issue_number):
+        notes.append(
+            f"recovery evidence is for issue #{token_issue}, not "
+            f"#{issue_number} (no owning-PR exemption)"
+        )
+        return False, notes
+    if not locked or not token_branch or locked != token_branch:
+        notes.append(
+            f"recovery evidence branch '{token_branch or 'unknown'}' does not "
+            f"match the branch being locked '{locked or 'unknown'}' "
+            "(no owning-PR exemption)"
+        )
+        return False, notes
+    if len(linked_open_prs) != 1:
+        numbers = ", ".join(
+            f"#{pr.get('number')}" for pr in linked_open_prs
+        ) or "none"
+        notes.append(
+            f"{len(linked_open_prs)} open PRs link issue #{issue_number} "
+            f"({numbers}); recovery may only own exactly one "
+            "(no owning-PR exemption)"
+        )
+        return False, notes
+
+    only = linked_open_prs[0]
+    head_obj = only.get("head") or {}
+    only_number = only.get("number")
+    only_ref = str(head_obj.get("ref") or "").strip()
+    only_sha = str(head_obj.get("sha") or "").strip()
+
+    if token_pr is None or only_number is None or int(only_number) != int(token_pr):
+        notes.append(
+            f"linked open PR #{only_number} is not the recovered owning PR "
+            f"#{token_pr} (no owning-PR exemption)"
+        )
+        return False, notes
+    if only_ref != token_branch:
+        notes.append(
+            f"open PR #{only_number} head branch '{only_ref}' does not match "
+            f"the recovered branch '{token_branch}' (no owning-PR exemption)"
+        )
+        return False, notes
+    if not token_head or not only_sha or only_sha != token_head:
+        notes.append(
+            f"open PR #{only_number} head {only_sha or 'unknown'} does not "
+            f"match the recovered head {token_head or 'unknown'} "
+            "(no owning-PR exemption)"
+        )
+        return False, notes
+
+    return True, [
+        f"open PR #{only_number} is the exact PR already owned by the "
+        f"recovering lock for issue #{issue_number} (branch '{token_branch}', "
+        f"head {token_head}); not duplicate work"
+    ]
+
+
 def _matching_branches(
     issue_number: int,
     branch_names: list[str],
@@ -52,8 +162,15 @@ def assess_work_issue_duplicate_gate(
     claim_entry: dict | None = None,
     locked_branch: str | None = None,
     phase: str = PHASE_LOCK,
+    recovered_owning_pr: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
-    """Fail closed when duplicate work is already in flight for an issue."""
+    """Fail closed when duplicate work is already in flight for an issue.
+
+    ``recovered_owning_pr`` (#755) is server-derived evidence that a sanctioned
+    dead-session lock recovery already owns one specific open PR. It exempts
+    *only* that exact PR from the linked-open-PR blocker; every other duplicate
+    signal, and every mismatch, keeps failing closed.
+    """
     reasons: list[str] = []
     outcome = OUTCOME_DUPLICATE_WORK_NOT_PREVENTED
     prs = list(open_prs or [])
@@ -61,12 +178,23 @@ def assess_work_issue_duplicate_gate(
     pattern = _issue_pattern(issue_number)
 
     linked = _linked_open_pr(issue_number, prs)
+    linked_open_prs = _all_linked_open_prs(issue_number, prs)
+    owning_pr_exempted = False
+    exemption_notes: list[str] = []
     if linked:
-        reasons.append(
-            f"open PR #{linked.get('number')} already covers issue "
-            f"#{issue_number} (fail closed)"
+        owning_pr_exempted, exemption_notes = _assess_owning_pr_exemption(
+            issue_number,
+            linked_open_prs=linked_open_prs,
+            locked_branch=locked_branch,
+            recovered_owning_pr=recovered_owning_pr,
         )
-        outcome = OUTCOME_DUPLICATE_PR_PREVENTED
+        if not owning_pr_exempted:
+            reasons.append(
+                f"open PR #{linked.get('number')} already covers issue "
+                f"#{issue_number} (fail closed)"
+            )
+            reasons.extend(exemption_notes)
+            outcome = OUTCOME_DUPLICATE_PR_PREVENTED
 
     conflicting_branches = _matching_branches(
         issue_number, branches, locked_branch=locked_branch
@@ -122,6 +250,9 @@ def assess_work_issue_duplicate_gate(
         "phase": phase,
         "outcome": outcome,
         "linked_open_pr": linked.get("number") if linked else entry.get("linked_open_pr"),
+        "linked_open_pr_count": len(linked_open_prs),
+        "owning_pr_recovery_exempted": owning_pr_exempted,
+        "owning_pr_recovery_notes": list(exemption_notes),
         "conflicting_branches": conflicting_branches,
         "claim_status": status or None,
         "reasons": reasons,

tests/test_issue_753_dead_pid_lock_recovery.py

@@ -0,0 +1,364 @@
+"""Dead-session author issue-lock recovery (#753).
+
+Covers the narrow recovery path that lets an author re-acquire a durable lock
+after the MCP session that recorded it exits, plus every rejection condition
+that must keep failing closed.
+"""
+
+import os
+import subprocess
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import issue_lock_recovery  # noqa: E402
+import issue_lock_store  # noqa: E402
+import issue_lock_worktree  # noqa: E402
+
+ISSUE = 4242
+BRANCH = f"fix/issue-{ISSUE}-demo"
+WORKTREE = "/scratch/wt"
+HEAD = "a" * 40
+OTHER_SHA = "b" * 40
+IDENTITY = "example-user"
+PROFILE = "example-author"
+
+
+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 future_ts(hours: int = 4) -> str:
+    return (
+        (datetime.now(timezone.utc) + timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def make_lock(**overrides):
+    lock = {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "worktree_path": WORKTREE,
+        "remote": "prgs",
+        "org": "ExampleOrg",
+        "repo": "ExampleRepo",
+        "session_pid": dead_pid(),
+        "work_lease": {
+            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+            "issue_number": ISSUE,
+            "branch": BRANCH,
+            "worktree_path": WORKTREE,
+            "claimant": {"username": IDENTITY, "profile": PROFILE},
+            "expires_at": future_ts(),
+        },
+    }
+    lock.update(overrides)
+    return lock
+
+
+def assess(lock=None, **overrides):
+    kwargs = {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "worktree_path": WORKTREE,
+        "remote": "prgs",
+        "org": "ExampleOrg",
+        "repo": "ExampleRepo",
+        "identity": IDENTITY,
+        "profile": PROFILE,
+        "current_branch": BRANCH,
+        "porcelain_status": "",
+        "head_sha": HEAD,
+        "remote_head_sha": HEAD,
+        "pr_head_sha": HEAD,
+        "pr_number": 99,
+        "competing_live_locks": [],
+        "candidate_branches": [BRANCH],
+        "current_pid": os.getpid(),
+    }
+    kwargs.update(overrides)
+    return issue_lock_recovery.assess_dead_session_lock_recovery(
+        make_lock() if lock is None else lock, **kwargs
+    )
+
+
+class TestDeadSessionRecoveryGranted(unittest.TestCase):
+    def test_dead_pid_with_exact_evidence_recovers(self):
+        result = assess()
+        self.assertTrue(result["recovery_sanctioned"], result["reasons"])
+        self.assertEqual(result["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED)
+
+    def test_recovery_still_granted_when_no_open_pr_exists(self):
+        # A locked branch need not have a PR yet; absence must not block.
+        result = assess(pr_head_sha=None, pr_number=None)
+        self.assertTrue(result["recovery_sanctioned"], result["reasons"])
+
+    def test_lease_expiry_is_not_required_for_recovery(self):
+        # The defining condition is PID death, not TTL expiry (the #601 gap).
+        lock = make_lock()
+        self.assertFalse(issue_lock_store.is_lease_expired(lock))
+        self.assertFalse(issue_lock_store.assess_lock_freshness(lock)["live"])
+        self.assertTrue(assess(lock)["recovery_sanctioned"])
+
+
+class TestDeadSessionRecoveryRefused(unittest.TestCase):
+    def assert_refused(self, result, needle):
+        self.assertFalse(result["recovery_sanctioned"])
+        self.assertEqual(result["outcome"], issue_lock_recovery.REFUSED)
+        self.assertTrue(
+            any(needle in reason for reason in result["reasons"]),
+            f"expected {needle!r} in {result['reasons']}",
+        )
+
+    def test_live_prior_pid_refused(self):
+        lock = make_lock(session_pid=os.getpid(), pid=os.getpid())
+        # Distinct current pid so the refusal is attributable to liveness.
+        self.assert_refused(assess(lock, current_pid=os.getpid() + 1), "still alive")
+
+    def test_different_author_identity_refused(self):
+        self.assert_refused(
+            assess(identity="someone-else"), "does not match active identity"
+        )
+
+    def test_different_profile_refused(self):
+        self.assert_refused(
+            assess(profile="other-profile"), "does not match active profile"
+        )
+
+    def test_different_branch_refused(self):
+        lock = make_lock(branch_name=f"fix/issue-{ISSUE}-other")
+        self.assert_refused(assess(lock), "does not match requested")
+
+    def test_worktree_parked_on_another_branch_refused(self):
+        self.assert_refused(assess(current_branch="master"), "not the locked branch")
+
+    def test_detached_head_worktree_refused(self):
+        self.assert_refused(assess(current_branch=None), "detached HEAD")
+
+    def test_different_worktree_refused(self):
+        self.assert_refused(
+            assess(worktree_path="/scratch/elsewhere"), "does not match declared"
+        )
+
+    def test_dirty_worktree_refused(self):
+        self.assert_refused(
+            assess(porcelain_status=" M gitea_mcp_server.py\n"), "requires a clean"
+        )
+
+    def test_local_head_differing_from_remote_refused(self):
+        self.assert_refused(
+            assess(remote_head_sha=OTHER_SHA), "does not match remote branch head"
+        )
+
+    def test_pr_head_differing_refused(self):
+        self.assert_refused(assess(pr_head_sha=OTHER_SHA), "does not match local head")
+
+    def test_missing_remote_head_refused(self):
+        self.assert_refused(assess(remote_head_sha=None), "remote head")
+
+    def test_competing_live_lock_refused(self):
+        competing = [
+            {
+                "issue_number": ISSUE,
+                "branch_name": BRANCH,
+                "worktree_path": "/scratch/other-wt",
+                "pid": os.getpid(),
+            }
+        ]
+        self.assert_refused(
+            assess(competing_live_locks=competing), "competing live lock"
+        )
+
+    def test_unrelated_live_lock_does_not_block(self):
+        unrelated = [
+            {
+                "issue_number": 999,
+                "branch_name": "fix/issue-999-unrelated",
+                "worktree_path": "/scratch/unrelated",
+                "pid": os.getpid(),
+            }
+        ]
+        self.assertTrue(assess(competing_live_locks=unrelated)["recovery_sanctioned"])
+
+    def test_multiple_candidate_branches_refused(self):
+        self.assert_refused(
+            assess(candidate_branches=[BRANCH, f"feat/issue-{ISSUE}-rival"]),
+            "multiple branches claim this issue",
+        )
+
+    def test_repository_scope_mismatch_refused(self):
+        self.assert_refused(assess(repo="OtherRepo"), "does not match requested")
+
+    def test_malformed_lock_missing_worktree_refused(self):
+        lock = make_lock()
+        lock.pop("worktree_path")
+        self.assert_refused(assess(lock), "incomplete")
+
+    def test_malformed_lock_missing_pid_refused(self):
+        lock = make_lock()
+        lock.pop("session_pid", None)
+        lock.pop("pid", None)
+        self.assert_refused(assess(lock), "incomplete")
+
+    def test_lock_without_claimant_refused(self):
+        lock = make_lock()
+        lock["work_lease"] = dict(lock["work_lease"])
+        lock["work_lease"].pop("claimant")
+        self.assert_refused(assess(lock), "claimant identity/profile")
+
+
+class TestNotACandidate(unittest.TestCase):
+    def test_absent_lock_is_not_a_candidate(self):
+        result = issue_lock_recovery.assess_dead_session_lock_recovery(
+            None,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=WORKTREE,
+            remote="prgs",
+            org="ExampleOrg",
+            repo="ExampleRepo",
+            identity=IDENTITY,
+            profile=PROFILE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            head_sha=HEAD,
+            remote_head_sha=HEAD,
+        )
+        self.assertEqual(result["outcome"], issue_lock_recovery.NO_CANDIDATE)
+        self.assertFalse(result["recovery_sanctioned"])
+        self.assertFalse(result["is_candidate"])
+
+    def test_lock_for_a_different_issue_is_not_a_candidate(self):
+        result = assess(make_lock(issue_number=7777))
+        self.assertEqual(result["outcome"], issue_lock_recovery.NO_CANDIDATE)
+        self.assertFalse(result["recovery_sanctioned"])
+
+
+class TestWorktreeGateWaiver(unittest.TestCase):
+    def test_new_issue_claim_still_requires_base_equivalence(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path=WORKTREE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            base_equivalent=False,
+        )
+        self.assertTrue(result["block"])
+        self.assertFalse(result["base_equivalence_waived"])
+
+    def test_sanctioned_recovery_waives_base_equivalence(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path=WORKTREE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            base_equivalent=False,
+            recovery_sanctioned=True,
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertTrue(result["base_equivalence_waived"])
+
+    def test_recovery_never_waives_cleanliness(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path=WORKTREE,
+            current_branch=BRANCH,
+            porcelain_status=" M gitea_mcp_server.py\n",
+            base_equivalent=False,
+            recovery_sanctioned=True,
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("tracked file edits" in reason for reason in result["reasons"])
+        )
+
+    def test_unproven_base_equivalence_still_blocks_without_recovery(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path=WORKTREE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            base_equivalent=None,
+        )
+        self.assertTrue(result["block"])
+
+
+class TestRecoveryRecordAndDownstream(unittest.TestCase):
+    def test_recovery_record_preserves_truthful_provenance(self):
+        assessment = assess()
+        prior = assessment["evidence"]["prior_session_pid"]
+        record = issue_lock_recovery.build_recovery_record(
+            assessment, recovered_at="2026-07-18T23:21:40Z"
+        )
+        self.assertTrue(record["recovered"])
+        self.assertEqual(record["prior_session_pid"], prior)
+        self.assertEqual(record["replacement_session_pid"], os.getpid())
+        self.assertNotEqual(
+            record["prior_session_pid"], record["replacement_session_pid"]
+        )
+        self.assertFalse(record["prior_pid_alive"])
+        self.assertEqual(record["recovered_at"], "2026-07-18T23:21:40Z")
+        self.assertEqual(record["branch_name"], BRANCH)
+        self.assertEqual(record["local_head"], HEAD)
+        self.assertEqual(record["identity"], IDENTITY)
+        self.assertTrue(record["proof"])
+
+    def test_recovery_record_carries_no_secret_material(self):
+        record = issue_lock_recovery.build_recovery_record(
+            assess(), recovered_at="2026-07-18T23:21:40Z"
+        )
+        blob = repr(record).lower()
+        for banned in ("token", "password", "authorization", "secret", "api_key"):
+            self.assertNotIn(banned, blob)
+
+    def test_recovered_lock_satisfies_update_by_merge_ownership(self):
+        # After recovery the lock is rebound to the live session, so the
+        # ownership re-check used by gitea_update_pr_branch_by_merge passes.
+        assessment = assess()
+        recovered_lock = make_lock(session_pid=os.getpid(), pid=os.getpid())
+        recovered_lock["dead_session_recovery"] = (
+            issue_lock_recovery.build_recovery_record(
+                assessment, recovered_at="2026-07-18T23:21:40Z"
+            )
+        )
+        freshness = issue_lock_store.assess_lock_freshness(recovered_lock)
+        self.assertTrue(freshness["live"], freshness)
+
+        verdict = issue_lock_store.verify_lock_for_mutation(
+            recovered_lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=WORKTREE,
+        )
+        self.assertTrue(verdict["proven"], verdict["reasons"])
+        self.assertFalse(verdict["block"])
+
+    def test_pre_recovery_lock_fails_ownership_check(self):
+        # Guards against a false positive above: the dead-PID lock must fail.
+        verdict = issue_lock_store.verify_lock_for_mutation(
+            make_lock(),
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=WORKTREE,
+        )
+        self.assertTrue(verdict["block"])
+        self.assertTrue(any("not live" in reason for reason in verdict["reasons"]))
+
+    def test_no_manual_file_seeding_required(self):
+        # The whole decision is reachable from the durable record plus live
+        # observation; nothing is written to disk to reach a verdict.
+        self.assertTrue(assess()["recovery_sanctioned"])
+
+    def test_refusal_message_is_fail_closed(self):
+        message = issue_lock_recovery.format_recovery_refusal(
+            assess(porcelain_status=" M x.py\n")
+        )
+        self.assertIn("fail closed", message)
+        self.assertIn("recovery refused", message.lower())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_755_owning_pr_recovery.py

@@ -0,0 +1,609 @@
+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
+"""Dead-session lock recovery when the issue already owns an open PR (#755).
+
+#753 added the recovery *assessor*, but the production ``gitea_lock_issue``
+path still rejected every sanctioned recovery: a dead-session lock is by
+construction a lock for work that already has an open PR, and the #400
+duplicate-work gate blocked unconditionally on any linked open PR. These tests
+drive the real MCP handler, not just the pure assessor, so that gap cannot
+reopen.
+"""
+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_store  # noqa: E402
+import mcp_server  # noqa: E402
+from issue_work_duplicate_gate import (  # noqa: E402
+    OUTCOME_DUPLICATE_PR_PREVENTED,
+    OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
+    PHASE_LOCK,
+    assess_work_issue_duplicate_gate,
+)
+
+ISSUE = 4755
+BRANCH = f"fix/issue-{ISSUE}-owning-pr"
+OTHER_BRANCH = f"fix/issue-{ISSUE}-competing"
+HEAD = "c" * 40
+OTHER_HEAD = "d" * 40
+OWNING_PR = 4756
+OTHER_PR = 4757
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+
+
+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 sanctioned_token(
+    issue_number=ISSUE, pr_number=OWNING_PR, branch=BRANCH, head=HEAD
+):
+    """The evidence shape the server derives from a granted recovery."""
+    return {
+        "issue_number": issue_number,
+        "pr_number": pr_number,
+        "branch_name": branch,
+        "head_sha": head,
+    }
+
+
+# ───────────────────────── duplicate gate: exemption ─────────────────────────
+
+
+class TestOwningPrExemptionGranted(unittest.TestCase):
+    def test_exact_owning_pr_is_not_duplicate_work(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr()],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
+        self.assertEqual(result["linked_open_pr"], OWNING_PR)
+        self.assertEqual(result["linked_open_pr_count"], 1)
+
+    def test_unrelated_open_pr_alongside_owning_pr_is_ignored(self):
+        unrelated = {
+            "number": 999,
+            "title": "chore: unrelated",
+            "body": "no linkage",
+            "head": {"ref": "chore/unrelated", "sha": OTHER_HEAD},
+        }
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[unrelated, owning_pr()],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["linked_open_pr_count"], 1)
+
+
+class TestOwningPrExemptionRefused(unittest.TestCase):
+    def assert_blocked(self, result):
+        self.assertTrue(result["block"])
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED)
+
+    def test_no_recovery_evidence_keeps_ordinary_blocker(self):
+        self.assert_blocked(
+            assess_work_issue_duplicate_gate(
+                ISSUE,
+                open_prs=[owning_pr()],
+                branch_names=[BRANCH],
+                claim_entry={"status": "not_claimed"},
+                locked_branch=BRANCH,
+                phase=PHASE_LOCK,
+            )
+        )
+
+    def test_competing_pr_number_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr(number=OTHER_PR)],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+    def test_multiple_linked_open_prs_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+        self.assertEqual(result["linked_open_pr_count"], 2)
+
+    def test_different_branch_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr(ref=OTHER_BRANCH)],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+    def test_locked_branch_differing_from_evidence_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr()],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=OTHER_BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+    def test_different_head_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr(sha=OTHER_HEAD)],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+    def test_evidence_for_another_issue_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr()],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(issue_number=ISSUE + 1),
+        )
+        self.assert_blocked(result)
+
+    def test_missing_head_in_live_pr_refused(self):
+        pr = owning_pr()
+        pr["head"] = {"ref": BRANCH}
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[pr],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+
+class TestOrdinaryDuplicateBehaviorUnchanged(unittest.TestCase):
+    def test_clean_issue_still_passes(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[],
+            branch_names=["feat/other-issue-99"],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+        )
+        self.assertFalse(result["block"])
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+
+    def test_competing_branch_still_blocks_even_with_owning_pr_evidence(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr()],
+            branch_names=[BRANCH, OTHER_BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        # The owning PR is exempt, but the competing branch is not.
+        self.assertTrue(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertIn(OTHER_BRANCH, result["conflicting_branches"])
+
+
+# ─────────────────── server-derived evidence cannot be forged ───────────────────
+
+
+class TestOwningPrEvidenceDerivation(unittest.TestCase):
+    def granted(self, **evidence_overrides):
+        evidence = {
+            "issue_number": ISSUE,
+            "locked_branch": BRANCH,
+            "local_head": HEAD,
+            "remote_head": HEAD,
+            "pr_head": HEAD,
+            "pr_number": OWNING_PR,
+        }
+        evidence.update(evidence_overrides)
+        return {
+            "outcome": issue_lock_recovery.RECOVERY_SANCTIONED,
+            "recovery_sanctioned": True,
+            "is_candidate": True,
+            "reasons": [],
+            "evidence": evidence,
+        }
+
+    def test_granted_recovery_yields_evidence(self):
+        token = issue_lock_recovery.owning_pr_recovery_evidence(self.granted())
+        self.assertEqual(token, sanctioned_token())
+
+    def test_none_assessment_yields_nothing(self):
+        self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(None))
+
+    def test_refused_assessment_yields_nothing(self):
+        refused = self.granted()
+        refused["outcome"] = issue_lock_recovery.REFUSED
+        refused["recovery_sanctioned"] = False
+        self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(refused))
+
+    def test_sanctioned_flag_without_outcome_yields_nothing(self):
+        forged = self.granted()
+        forged["outcome"] = "SOMETHING_ELSE"
+        self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(forged))
+
+    def test_missing_pr_number_yields_nothing(self):
+        self.assertIsNone(
+            issue_lock_recovery.owning_pr_recovery_evidence(
+                self.granted(pr_number=None)
+            )
+        )
+
+    def test_head_disagreement_in_evidence_yields_nothing(self):
+        self.assertIsNone(
+            issue_lock_recovery.owning_pr_recovery_evidence(
+                self.granted(remote_head=OTHER_HEAD)
+            )
+        )
+        self.assertIsNone(
+            issue_lock_recovery.owning_pr_recovery_evidence(
+                self.granted(local_head=OTHER_HEAD)
+            )
+        )
+
+    def test_real_refused_assessment_yields_nothing(self):
+        """End-to-end against the real assessor, not a hand-built dict."""
+        lock = {
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "worktree_path": "/scratch/wt",
+            "remote": "prgs",
+            "org": "Example-Org",
+            "repo": "Example-Repo",
+            "session_pid": os.getpid(),  # alive → must refuse
+            "work_lease": {
+                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+                "issue_number": ISSUE,
+                "branch": BRANCH,
+                "worktree_path": "/scratch/wt",
+                "claimant": {"username": IDENTITY, "profile": PROFILE},
+                "expires_at": shifted_ts(),
+            },
+        }
+        assessment = issue_lock_recovery.assess_dead_session_lock_recovery(
+            lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path="/scratch/wt",
+            remote="prgs",
+            org="Example-Org",
+            repo="Example-Repo",
+            identity=IDENTITY,
+            profile=PROFILE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            head_sha=HEAD,
+            remote_head_sha=HEAD,
+            pr_head_sha=HEAD,
+            pr_number=OWNING_PR,
+            competing_live_locks=[],
+            candidate_branches=[BRANCH],
+            current_pid=os.getpid(),
+        )
+        self.assertFalse(assessment["recovery_sanctioned"])
+        self.assertIsNone(
+            issue_lock_recovery.owning_pr_recovery_evidence(assessment)
+        )
+
+
+# ──────────────────── end-to-end: the real gitea_lock_issue ────────────────────
+
+
+class LockIssueEndToEndBase(unittest.TestCase):
+    """Drives ``mcp_server.gitea_lock_issue`` with live git/Gitea observation
+    stubbed at the module boundary — the production gate chain itself runs."""
+
+    def setUp(self):
+        self.lock_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.lock_dir.cleanup)
+        self.worktree = os.path.realpath(os.getcwd())
+        # Bind host/org/repo to what the ``test-author-prgs`` fixture profile is
+        # pinned to, so the session-context gate under test is the real one and
+        # not a cross-host denial. The issue number and lock dir stay synthetic.
+        self.remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {
+                "host": "gitea.prgs.cc",
+                "org": ORG,
+                "repo": REPO,
+            },
+        })
+        self.remotes.start()
+        self.addCleanup(patch.stopall)
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def write_durable_lock(self, *, pid, branch=BRANCH, worktree=None):
+        path = issue_lock_store.lock_file_path(
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+        claimant = {"username": IDENTITY, "profile": PROFILE}
+        data = {
+            "issue_number": ISSUE,
+            "branch_name": branch,
+            "remote": "prgs",
+            "org": ORG,
+            "repo": REPO,
+            "worktree_path": worktree or self.worktree,
+            "session_pid": pid,
+            "pid": pid,
+            "work_lease": {
+                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+                "issue_number": ISSUE,
+                "branch": branch,
+                "worktree_path": worktree or self.worktree,
+                "claimant": claimant,
+                "created_at": shifted_ts(-1),
+                "last_heartbeat_at": shifted_ts(-1),
+                "expires_at": shifted_ts(),
+            },
+            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                tool="gitea_lock_issue",
+                claimant=claimant,
+            ),
+        }
+        issue_lock_store.save_lock_file(path, data)
+        return path
+
+    def run_lock(
+        self,
+        *,
+        open_prs,
+        porcelain="",
+        current_branch=BRANCH,
+        base_equivalent=False,
+        branch_names=None,
+        head_sha=HEAD,
+        remote_head=HEAD,
+    ):
+        branch_names = branch_names if branch_names is not None else [BRANCH]
+        branch_entries = [
+            {"name": name, "commit": {"id": remote_head}} for name in branch_names
+        ]
+        env = shared_mutation_env(
+            "test-author-prgs",
+            include_example_repo=True,
+            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
+        )
+        with patch(
+            "mcp_server.api_get_all", return_value=branch_entries
+        ), patch(
+            "mcp_server._list_open_pulls", return_value=list(open_prs)
+        ), patch(
+            "mcp_server.get_auth_header", return_value="token x"
+        ), patch(
+            "mcp_server._work_lease_claimant",
+            return_value={"username": IDENTITY, "profile": PROFILE},
+        ), patch(
+            "mcp_server.issue_lock_worktree.read_worktree_git_state",
+            return_value={
+                "current_branch": current_branch,
+                "porcelain_status": porcelain,
+                "base_equivalent": base_equivalent,
+                "head_sha": head_sha,
+                "inspected_git_root": self.worktree,
+                "base_branch": "master",
+            },
+        ), 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"}
+            ),
+        ):
+            with patch.dict(os.environ, env, clear=True):
+                os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+                return mcp_server.gitea_lock_issue(
+                    issue_number=ISSUE,
+                    branch_name=BRANCH,
+                    remote="prgs",
+                    worktree_path=self.worktree,
+                )
+
+
+class TestRecoveryWithOwningPrSucceeds(LockIssueEndToEndBase):
+    def test_dead_session_recovery_with_owning_pr_relocks(self):
+        self.write_durable_lock(pid=dead_pid())
+        result = self.run_lock(open_prs=[owning_pr()])
+
+        self.assertTrue(result["success"])
+        self.assertEqual(result["issue_number"], ISSUE)
+        self.assertEqual(result["branch_name"], BRANCH)
+        self.assertTrue(result["lock_freshness"]["live"])
+        self.assertTrue(result["lock_freshness"]["pid_alive"])
+
+    def test_recovered_lock_records_truthful_provenance(self):
+        prior = dead_pid()
+        self.write_durable_lock(pid=prior)
+        self.run_lock(open_prs=[owning_pr()])
+
+        lock = issue_lock_store.load_issue_lock(
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+        record = lock.get("dead_session_recovery") or {}
+        self.assertTrue(record.get("recovered"))
+        self.assertEqual(record.get("prior_session_pid"), prior)
+        self.assertEqual(record.get("replacement_session_pid"), os.getpid())
+        self.assertFalse(record.get("prior_pid_alive"))
+        self.assertEqual(record.get("pr_number"), OWNING_PR)
+        self.assertEqual(record.get("branch_name"), BRANCH)
+        self.assertEqual(record.get("identity"), IDENTITY)
+
+    def test_recovered_lock_is_live_and_proves_pr_ownership(self):
+        """AC6: the persisted lock satisfies update-by-merge's ownership prover."""
+        self.write_durable_lock(pid=dead_pid())
+        self.run_lock(open_prs=[owning_pr()])
+
+        lock = issue_lock_store.load_issue_lock(
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+        self.assertTrue(issue_lock_store.is_lease_live(lock))
+
+        env = shared_mutation_env(
+            "test-author-prgs",
+            include_example_repo=True,
+            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
+        )
+        with patch.dict(os.environ, env, clear=True):
+            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+            ownership = mcp_server._prove_author_ownership_for_pr(
+                pr_number=OWNING_PR,
+                pr_title=f"fix: something (Closes #{ISSUE})",
+                pr_body=f"Closes #{ISSUE}.",
+                source_branch=BRANCH,
+                remote="prgs",
+                host="gitea.prgs.cc",
+                org=ORG,
+                repo=REPO,
+                worktree_path=self.worktree,
+            )
+        self.assertTrue(ownership["proven"], ownership["reasons"])
+        self.assertTrue(ownership["has_author_lock"])
+        self.assertEqual(ownership["matched_issue"], ISSUE)
+
+
+class TestRecoveryRejectionsEndToEnd(LockIssueEndToEndBase):
+    def assert_lock_refused(self, **kwargs):
+        with self.assertRaises((ValueError, RuntimeError)) as ctx:
+            self.run_lock(**kwargs)
+        return str(ctx.exception)
+
+    def test_competing_pr_still_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        message = self.assert_lock_refused(
+            open_prs=[owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+            branch_names=[BRANCH],
+        )
+        self.assertIn("already covers issue", message)
+
+    def test_multiple_linked_open_prs_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        message = self.assert_lock_refused(
+            open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+        )
+        self.assertIn("already covers issue", message)
+
+    def test_owning_pr_on_a_different_head_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        self.assert_lock_refused(open_prs=[owning_pr(sha=OTHER_HEAD)])
+
+    def test_lock_registered_to_a_different_worktree_blocked(self):
+        self.write_durable_lock(
+            pid=dead_pid(), worktree=os.path.join(self.worktree, "elsewhere")
+        )
+        self.assert_lock_refused(open_prs=[owning_pr()])
+
+    def test_dirty_worktree_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        self.assert_lock_refused(
+            open_prs=[owning_pr()], porcelain=" M gitea_mcp_server.py"
+        )
+
+    def test_worktree_parked_on_another_branch_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        self.assert_lock_refused(
+            open_prs=[owning_pr()], current_branch="master"
+        )
+
+    def test_local_head_differing_from_remote_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        self.assert_lock_refused(
+            open_prs=[owning_pr()], head_sha=OTHER_HEAD
+        )
+
+    def test_live_prior_pid_blocked(self):
+        self.write_durable_lock(pid=os.getpid())
+        self.assert_lock_refused(open_prs=[owning_pr()])
+
+    def test_new_claim_without_prior_lock_still_requires_base_equivalence(self):
+        """AC10: no durable lock → no recovery → base-equivalence still rules."""
+        self.assert_lock_refused(open_prs=[], branch_names=[])
+
+
+if __name__ == "__main__":
+    unittest.main()