gitea_mcp_server.py

@@ -2781,6 +2781,87 @@ def _collect_issue_duplicate_context(
     return issue_duplicate_context_fetcher(h, o, r, auth, issue_number)
 
 
+# Every field of the owning-PR continuation token is security relevant: the
+# issue and PR it names, the branch it is scoped to, and each head the waiver
+# was measured against. Two evidence blocks that disagree on any of them cannot
+# both describe the single sanctioned decision the lock is supposed to record.
+_CONTINUATION_EVIDENCE_BINDINGS = (
+    "issue_number",
+    "pr_number",
+    "branch_name",
+    "head_sha",
+    "recorded_head",
+    "accepted_head",
+    "head_relation",
+)
+
+
+def _continuation_evidence_agrees(recovered: dict, renewed: dict) -> bool:
+    """Do two rebuilt continuation tokens bind to exactly the same thing (#945)?"""
+    return all(
+        recovered.get(field) == renewed.get(field)
+        for field in _CONTINUATION_EVIDENCE_BINDINGS
+    )
+
+
+def _owning_pr_continuation_from_lock(lock_record: dict | None) -> dict | None:
+    """Owning-PR continuation evidence a persisted lock still proves (#945).
+
+    ``gitea_lock_issue`` grants the duplicate-work waiver from either a
+    sanctioned dead-session recovery (#755) or a sanctioned exact-owner renewal
+    (#760), in that precedence. Every later enforcement path — commit,
+    create-PR, push-ownership, and the read-only duplicate assessor — re-derives
+    ownership from the durable lock instead of that live assessment.
+
+    Until #945 only the recovery half was rebuilt there, so an ordinary
+    exact-owner renewal lost its waiver the moment ``gitea_lock_issue``
+    returned: the author renewed successfully and was then refused
+    ``duplicate_commit_prevented`` with ``owning_pr_recovery_exempted: false``
+    on the very PR the renewal had just proved it owned.
+
+    Resolving both halves here, in the same precedence the lock path applies,
+    keeps the answer from drifting between the gate that grants the waiver and
+    the gates that enforce it. This only decides which server-written block the
+    token is rebuilt from — the token is still re-validated against live PR
+    state by ``issue_work_duplicate_gate._assess_owning_pr_exemption``, which
+    remains the single authoritative policy for whether an exemption applies.
+
+    **Both blocks present is a reachable, legitimate state, and it must agree.**
+    The two dispositions are not mutually exclusive at the writer. Recovery is
+    assessed whenever the lease is not live and requires the recorded PID to be
+    dead; renewal is assessed whenever the lease has *expired* — which is itself
+    one way to be non-live — and deliberately does not branch on PID liveness
+    (#760 AC16). An expired lease whose recorded owner has also died therefore
+    satisfies both, and ``gitea_lock_issue`` writes ``dead_session_recovery``
+    and ``lease_renewal`` into the same freshly built ``data`` dict. Because a
+    sanctioned pair was derived from one live observation in one call, it always
+    describes the same issue, PR, branch and head. Disagreement means the
+    persisted lock is no longer a faithful record of a single sanctioned
+    decision, so no continuation authority is returned.
+
+    Ambiguity never broadens authority. A ``dead_session_recovery`` block that
+    is present but does not rebuild — conflicting, stale, malformed, or only
+    partially valid — fails closed here rather than falling through to renewal:
+    otherwise a recovery record naming one PR could be bypassed by valid-looking
+    renewal evidence naming another. Recovery-only and renewal-only locks keep
+    their existing behaviour exactly, and provenance stays server-controlled —
+    this still only ever re-reads blocks the server itself wrote.
+    """
+    if not lock_record:
+        return None
+    recovery_present = isinstance(lock_record.get("dead_session_recovery"), dict)
+    recovered = issue_lock_recovery.recovered_owning_pr_from_lock(lock_record)
+    # Present but unusable recovery evidence is ambiguous, not absent.
+    if recovery_present and not recovered:
+        return None
+    renewed = issue_lock_renewal.owning_pr_renewal_from_lock(lock_record)
+    if recovered and renewed and not _continuation_evidence_agrees(recovered, renewed):
+        return None
+    if recovered:
+        return recovered
+    return renewed
+
+
 def _assess_issue_duplicate_gate(
     issue_number: int,
     *,
@@ -2833,6 +2914,11 @@ def _enforce_locked_issue_duplicate_recheck(
     commit and create-PR phases run in their own calls, long after the recovery
     assessment ended, so without this they re-block the very PR the recovery
     already proved belongs to this author.
+
+    #945: an exact-owner *renewal* (#760) owns its open PR for exactly the same
+    reason, and ``gitea_lock_issue`` already waives the blocker for both. Both
+    halves are resolved together here so the renewal waiver survives past the
+    lock call instead of expiring with it.
     """
     lock_data = _load_existing_issue_lock()
     if not lock_data:
@@ -2856,9 +2942,7 @@ def _enforce_locked_issue_duplicate_recheck(
         auth=auth,
         locked_branch=locked_branch,
         phase=phase,
-        recovered_owning_pr=issue_lock_recovery.recovered_owning_pr_from_lock(
+        recovered_owning_pr=_owning_pr_continuation_from_lock(lock_data),
-            lock_data
-        ),
     )
     if gate.get("block"):
         return gate
@@ -5140,9 +5224,10 @@ def gitea_assess_work_issue_duplicate(
     recovered_owning_pr = None
     lock_data = _load_existing_issue_lock()
     if lock_data and int(lock_data.get("issue_number") or 0) == int(issue_number):
-        recovered_owning_pr = issue_lock_recovery.recovered_owning_pr_from_lock(
+        # #945: rebuilt from a sanctioned recovery *or* a sanctioned exact-owner
-            lock_data
+        # renewal, so this read-only assessor reports the same disposition the
-        )
+        # commit and create-PR gates will enforce.
+        recovered_owning_pr = _owning_pr_continuation_from_lock(lock_data)
     gate = _assess_issue_duplicate_gate(
         issue_number,
         h=h,
@@ -19424,7 +19509,10 @@ def _prove_author_ownership_for_pr(
     # advance is the one recovery already sanctioned, not a foreign head.
     recovered_owning_pr = None
     if proven and lock_record:
-        candidate = issue_lock_recovery.recovered_owning_pr_from_lock(lock_record)
+        # #945: a sanctioned exact-owner renewal proves the same ownership of
+        # the same PR, so the push gate resolves both halves rather than seeing
+        # only the recovery one.
+        candidate = _owning_pr_continuation_from_lock(lock_record)
         if candidate and int(candidate.get("pr_number") or 0) == int(pr_number):
             recovered_owning_pr = candidate
     return {

issue_lock_renewal.py

@@ -436,6 +436,117 @@ def owning_pr_renewal_evidence(
     }
 
 
+def owning_pr_renewal_from_lock(
+    lock_record: Mapping[str, Any] | None,
+) -> dict[str, Any] | None:
+    """Rebuild owning-PR renewal evidence from a persisted lock (#945).
+
+    The renewal mirror of ``issue_lock_recovery.recovered_owning_pr_from_lock``.
+    ``owning_pr_renewal_evidence`` supplies the waiver for the duration of the
+    ``gitea_lock_issue`` call only. The commit, push, create-PR, and
+    duplicate-assessment gates run later in their own calls and re-derive
+    ownership from the durable lock instead — so without this the open PR that
+    renewal already proved belongs to this author reappears there as competing
+    duplicate work, and the exact owner is refused with
+    ``duplicate_commit_prevented`` despite complete matching evidence.
+
+    This reads only the ``lease_renewal`` block that the server itself writes,
+    on a lock the caller must already own. Like the recovery mirror it is a
+    re-read of server-derived state, never a fresh assertion: a caller able to
+    forge it could equally forge the lock file every other ownership gate
+    already treats as authoritative.
+
+    Renewal has no descendant case — the assessor required the local, remote and
+    PR heads to be equal — so that equality is re-checked here, and the record
+    must still name the claimant the lock records.
+
+    **What the claimant check below is, and what it is not.** It compares
+    ``lease_renewal.identity``/``profile`` against the claimant recorded on the
+    *same* lock file. Both sides are server-written fields of one document, so
+    this is an internal-consistency check: it rejects a lock whose renewal block
+    and claimant disagree. It does **not** consult the live authenticated caller
+    and therefore does not, on its own, prove that the session invoking a later
+    gate is the session the renewal was granted to.
+
+    The binding that actually keeps one session from using another's renewal is
+    structural, and it lives in the caller rather than here. The enforcement
+    paths load the lock through ``_load_existing_issue_lock()`` with no issue
+    coordinates, which resolves ``issue_lock_store.read_session_issue_lock()``
+    → the session pointer at ``session-{os.getpid()}.json``. Lock *selection* is
+    scoped to the operating-system process, so a caller cannot aim the recheck
+    at a lock some other process bound. Its limits follow from what that scope
+    is: it is per-process, not per-authenticated-user; it says nothing about a
+    lock reached by explicit issue coordinates rather than the session pointer,
+    and nothing about two roles sharing one process. Live identity and profile
+    are enforced separately, by the mutation-authority and profile gates each
+    mutating path already runs — not by this rebuild.
+
+    This function is therefore strictly a re-read with an added consistency
+    requirement. It narrows what a persisted lock can authorize; it never widens
+    it, and it never substitutes for a caller-identity gate.
+    """
+    if not isinstance(lock_record, Mapping):
+        return None
+    record = lock_record.get("lease_renewal")
+    if not isinstance(record, Mapping) or not record.get("renewed"):
+        return None
+
+    branch_name = _text(record.get("branch_name")) or _text(
+        lock_record.get("branch_name")
+    )
+    pr_head = _text(record.get("pr_head_sha"))
+    local_head = _text(record.get("head_sha"))
+    remote_head = _text(record.get("remote_head_sha"))
+    raw_pr_number = record.get("pr_number")
+    raw_issue_number = lock_record.get("issue_number")
+
+    if raw_pr_number is None or raw_issue_number is None:
+        return None
+    if not branch_name or not pr_head:
+        return None
+    # The assessor required all three heads to agree before it granted renewal.
+    # Re-check, so a truncated, drifted, or hand-built record cannot widen the
+    # exemption past the single head the renewal disposition actually proved.
+    if not local_head or not remote_head:
+        return None
+    if pr_head != local_head or pr_head != remote_head:
+        return None
+    # Renewal is refused outright unless the durable lock records both a
+    # claimant username and profile, so a sanctioned record always carries them.
+    # Requiring them to still agree rejects a lock whose renewal block and
+    # claimant disagree. Both values are read from this one server-written
+    # document: this is internal consistency, not a check against the live
+    # authenticated caller — see the docstring for the binding that is.
+    claimant = lock_record.get("claimant")
+    if not isinstance(claimant, Mapping):
+        lease = lock_record.get("work_lease")
+        claimant = lease.get("claimant") if isinstance(lease, Mapping) else None
+    if not isinstance(claimant, Mapping):
+        return None
+    identity = _text(record.get("identity"))
+    profile = _text(record.get("profile"))
+    if not identity or identity != _text(claimant.get("username")):
+        return None
+    if not profile or profile != _text(claimant.get("profile")):
+        return None
+
+    try:
+        pr_number = int(raw_pr_number)
+        issue_number = int(raw_issue_number)
+    except (TypeError, ValueError):
+        return None
+
+    return {
+        "issue_number": issue_number,
+        "pr_number": pr_number,
+        "branch_name": branch_name,
+        "head_sha": pr_head,
+        "recorded_head": pr_head,
+        "accepted_head": pr_head,
+        "head_relation": "equal",
+    }
+
+
 def build_renewal_record(
     assessment: Mapping[str, Any] | None,
     *,

tests/test_issue_945_enforcement_path_wiring.py

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

tests/test_issue_945_owning_pr_renewal_continuation.py

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