feat(lock): allow exact-owner renewal of expired author issue locks (Closes #760)

An expired author issue lease could not be renewed by the exact session that
already owned it. `assess_same_issue_lease_conflict` computed same-owner
evidence and then returned on the expired branch before consulting it, and
`assess_expired_lock_reclaim` only permits takeover on a dead PID or a missing
worktree. Because the recorded PID is the long-lived MCP daemon rather than the
authoring task, a lease that expires under a live daemon is the ordinary case
for any author task outliving the TTL — and in that case the owner's own lock
became permanently unmodifiable through sanctioned tools.

Add `issue_lock_renewal`, a pure evidence assessor for that one case, and
evaluate its disposition before the expired foreign-takeover return.

Renewal requires an exact match of remote, org, repo, issue number, operation
type, branch, realpath-normalized worktree, claimant username, and claimant
profile; a registered worktree that exists, sits on the locked branch, and is
clean; local and remote heads that agree; and, when an owning PR exists, a PR
head that agrees too. No competing live lock, competing branch claim, or other
owning PR may exist. Any missing or contradictory evidence fails closed.

PID liveness is never authorization: it is recorded as evidence and is neither
necessary nor sufficient (AC16). Only an expired lease is ever a candidate, so
a live foreign lease stays non-recoverable (AC12) and dead-PID takeover keeps
its existing #601 conditions (AC11). Renewal is never caller-declarable — the
waiver is server-computed and `gitea_lock_issue` gains no parameter (AC14).

A sanctioned renewal records prior PID, prior expiry, replacement PID, new
expiry, claimant, and its supporting proof under `lease_renewal`, and reuses
the #772 compare-and-swap so two sessions observing the same expired lease
cannot both win.

Absolute wall-clock expiry is preserved. Sliding heartbeat renewal, fencing
tokens, and the shared cross-role lifecycle remain #790's scope and are
deliberately not implemented here; #790 stays sequenced behind this change.

Tests: 40 new cases covering the positive path, every near-match and
foreign-owner refusal, the non-candidate cases, the gate-ordering regression,
the renewal record, downstream mutation ownership, and AC14/AC17. Two #772 test
doubles now forward the new keyword.

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

gitea_mcp_server.py

@@ -2017,6 +2017,7 @@ 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 issue_lock_renewal  # noqa: E402
 import stacked_pr_support  # noqa: E402
 import merge_approval_gate  # noqa: E402
 import review_quarantine  # noqa: E402  # #695 contaminated formal-review quarantine
@@ -2316,7 +2317,12 @@ def _resolve_issue_lock_for_pr(
     return lock_data
 
 
-def _save_issue_lock(data: dict, *, expected_generation: int | None = None) -> str:
+def _save_issue_lock(
+    data: dict,
+    *,
+    expected_generation: int | None = None,
+    renewal_sanctioned: bool = False,
+) -> str:
     existing = issue_lock_store.load_issue_lock(
         remote=str(data.get("remote") or ""),
         org=str(data.get("org") or ""),
@@ -2328,7 +2334,9 @@ def _save_issue_lock(data: dict, *, expected_generation: int | None = None) -> s
         raise RuntimeError(overwrite_block)
     try:
         return issue_lock_store.bind_session_lock(
-            data, expected_generation=expected_generation
+            data,
+            expected_generation=expected_generation,
+            renewal_sanctioned=renewal_sanctioned,
         )
     except Exception as e:
         raise RuntimeError(f"Could not write issue lock file: {e}") from e
@@ -2448,6 +2456,79 @@ def _evaluate_issue_lock_recovery(
     )
 
 
+def _evaluate_issue_lock_renewal(
+    existing_lock: dict,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    remote: str,
+    h: str | None,
+    o: str,
+    r: str,
+    git_state: dict,
+) -> dict:
+    """Gather evidence and decide exact-owner renewal of an expired lease (#760).
+
+    Mirrors ``_evaluate_issue_lock_recovery``: every input is durable lock state
+    or a live server-side observation (Gitea branch/PR inventory, git in the
+    declared worktree, the local lock store). Nothing is reachable from an MCP
+    caller's parameters, so no caller can assert its way into a renewal
+    (#760 AC14).
+    """
+    renewal_auth = _auth(h)
+    try:
+        renewal_branches = api_get_all(
+            f"{repo_api_url(h, o, r)}/branches", renewal_auth
+        )
+    except Exception as exc:
+        raise RuntimeError(
+            f"Could not list branches to verify exact-owner lease renewal: {exc}"
+        )
+
+    remote_head: str | None = None
+    candidates: list[str] = []
+    for entry in renewal_branches:
+        entry_name = _branch_entry_name(entry)
+        if entry_name == branch_name:
+            remote_head = _branch_entry_commit_sha(entry)
+        if issue_lock_adoption.branch_carries_issue_marker(entry_name, issue_number):
+            candidates.append(entry_name)
+
+    pr_head: str | None = None
+    pr_number: int | None = None
+    for pull in _list_open_pulls(h, o, r, renewal_auth):
+        pull_head = pull.get("head") or {}
+        if str(pull_head.get("ref") or "") == branch_name:
+            pr_head = pull_head.get("sha")
+            pr_number = pull.get("number")
+            break
+
+    claimant = _work_lease_claimant(h)
+    return issue_lock_renewal.assess_exact_owner_lease_renewal(
+        existing_lock,
+        issue_number=issue_number,
+        branch_name=branch_name,
+        worktree_path=worktree_path,
+        remote=remote,
+        org=o,
+        repo=r,
+        identity=claimant.get("username"),
+        profile=claimant.get("profile"),
+        operation_type=AUTHOR_ISSUE_WORK_LEASE,
+        current_branch=git_state.get("current_branch"),
+        porcelain_status=git_state.get("porcelain_status") or "",
+        worktree_exists=os.path.isdir(os.path.realpath(worktree_path)),
+        head_sha=git_state.get("head_sha"),
+        remote_head_sha=remote_head,
+        pr_head_sha=pr_head,
+        pr_number=pr_number,
+        competing_live_locks=issue_lock_store.list_live_locks(),
+        candidate_branches=candidates,
+        current_pid=os.getpid(),
+    )
+
+
 def _work_lease_claimant(host: str | None) -> dict:
     profile = get_profile()
     username = _IDENTITY_CACHE.get(host) if host else None
@@ -3893,15 +3974,12 @@ def gitea_lock_issue(
     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(
+    # #760: the competing-lease disposition is decided below, once the worktree
-        existing_issue_lock,
+    # and Gitea evidence an exact-owner renewal depends on has actually been
-        issue_number=issue_number,
+    # observed. Deciding it here — before any of that exists — is what made the
-        branch_name=branch_name,
+    # same-owner allowance unreachable for an expired lease. The authoritative
-        worktree_path=resolved_worktree,
+    # check still runs inside bind_session_lock under the per-issue flock, so
-        operation_type=AUTHOR_ISSUE_WORK_LEASE,
+    # moving this one later cannot widen the window for a competing writer.
-    )
-    if active_lease_block:
-        raise RuntimeError(active_lease_block)
 
     # ── Stacked-PR base declaration (opt-in, #484) ──
     # Normal work leaves stacked_base_branch None → master-equivalent path.
@@ -3967,6 +4045,53 @@ def gitea_lock_issue(
     recovery_sanctioned = bool(
         recovery_assessment and recovery_assessment.get("recovery_sanctioned")
     )
+
+    # ── Exact-owner renewal of an expired lease (#760) ──
+    # The opposite trigger from #753 above: there the lease is unexpired and the
+    # PID is dead; here the lease has expired while the recording daemon — which
+    # is the long-lived MCP server, not the authoring task — may well still be
+    # up. Only an expired lease is assessed, so a live foreign lease is never a
+    # candidate (AC12) and dead-PID takeover keeps its existing conditions
+    # (AC11). A refusal never raises: it withholds the waiver and leaves the
+    # conflict check below to fail closed exactly as before.
+    renewal_assessment: dict | None = None
+    if (
+        existing_issue_lock
+        and existing_issue_lock.get("issue_number") == issue_number
+        and issue_lock_store.is_lease_expired(existing_issue_lock)
+    ):
+        renewal_assessment = _evaluate_issue_lock_renewal(
+            existing_issue_lock,
+            issue_number=issue_number,
+            branch_name=branch_name,
+            worktree_path=resolved_worktree,
+            remote=remote,
+            h=h,
+            o=o,
+            r=r,
+            git_state=git_state,
+        )
+    renewal_sanctioned = bool(
+        renewal_assessment and renewal_assessment.get("renewal_sanctioned")
+    )
+
+    active_lease_block = issue_lock_store.assess_same_issue_lease_conflict(
+        existing_issue_lock,
+        issue_number=issue_number,
+        branch_name=branch_name,
+        worktree_path=resolved_worktree,
+        operation_type=AUTHOR_ISSUE_WORK_LEASE,
+        renewal_sanctioned=renewal_sanctioned,
+    )
+    if active_lease_block:
+        reasons = [active_lease_block]
+        # Name the exact missing evidence when this looked like a renewal, so a
+        # blocked owner sees why rather than only the generic takeover text.
+        if renewal_assessment and renewal_assessment.get("is_candidate"):
+            reasons.append(
+                issue_lock_renewal.format_renewal_refusal(renewal_assessment)
+            )
+        raise RuntimeError("; ".join(reasons))
     # #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
@@ -4070,18 +4195,35 @@ def gitea_lock_issue(
             recovery_assessment,
             recovered_at=_work_lease_timestamp(_work_lease_now()),
         )
+    if renewal_sanctioned and renewal_assessment:
+        # #760 AC9: record both sides of the transition — prior PID and expiry,
+        # replacement PID and new expiry — so a renewed lock is auditable and
+        # never reads as an original claim.
+        data["lease_renewal"] = issue_lock_renewal.build_renewal_record(
+            renewal_assessment,
+            renewed_at=_work_lease_timestamp(_work_lease_now()),
+            new_expires_at=str(work_lease.get("expires_at") or ""),
+        )
 
     # #772 AC5: a recovery replaces a claim another session already owned, so
     # its write is a compare-and-swap against the generation the assessment was
     # made on. Two replacement sessions that both observed the same dead owner
     # cannot both succeed — the second finds a moved generation and fails
     # closed. Ordinary first-time claims keep the unconditional write.
+    # #760 uses the same compare-and-swap: a renewal also replaces a claim that
+    # already existed on disk, so two sessions that both observed the same
+    # expired lease cannot both win — the second finds a moved generation and
+    # fails closed.
     expected_generation = (
         issue_lock_store.lock_generation(existing_issue_lock)
-        if recovery_sanctioned
+        if (recovery_sanctioned or renewal_sanctioned)
         else None
     )
-    lock_file_path = _save_issue_lock(data, expected_generation=expected_generation)
+    lock_file_path = _save_issue_lock(
+        data,
+        expected_generation=expected_generation,
+        renewal_sanctioned=renewal_sanctioned,
+    )
     lock_record = issue_lock_store.read_lock_file(lock_file_path) or data
     freshness = issue_lock_store.assess_lock_freshness(lock_record)
     competing = [
@@ -4150,6 +4292,16 @@ def gitea_lock_issue(
             issue_number=issue_number,
             branch_name=branch_name,
         )
+    if renewal_sanctioned:
+        # #760 AC13: the renewal is visible in the native tool result, so an
+        # owner never has to inspect the lock file to confirm what happened.
+        result["lease_renewal"] = lock_record.get("lease_renewal")
+        result["message"] = (
+            f"Renewed the expired {AUTHOR_ISSUE_WORK_LEASE} lease on issue "
+            f"#{issue_number} for its exact recorded owner, on branch "
+            f"'{branch_name}' from worktree '{resolved_worktree}' "
+            "(fail-closed check complete)."
+        )
     if agent_artifacts:
         result["warnings"] = [
             "Agent temp artifacts at repo root (delete before implementation): "

issue_lock_renewal.py

@@ -0,0 +1,425 @@
+"""Exact-owner renewal of an expired author issue lease (#760).
+
+An author issue lease carries an absolute wall-clock expiry stamped once at
+lock time. The PID recorded alongside it is the long-lived MCP daemon, not the
+authoring task, so a lease that expires while its daemon is still up is the
+ordinary case for any author task that outlives the TTL — not an anomaly.
+
+Before this module, that case was unreachable.
+``issue_lock_store.assess_same_issue_lease_conflict`` computed same-owner
+evidence and then returned on the expired branch before consulting it, and
+``assess_expired_lock_reclaim`` only permits takeover on a dead PID or a
+missing worktree. An exact owner whose daemon is alive and whose worktree is
+present satisfied neither, so its own lock became permanently unmodifiable
+through sanctioned tools.
+
+This module is the pure evidence assessor for that one narrow case. It answers
+a single question: may *this* session renew a lease it can prove it already
+owns? It performs no mutation and no network I/O, and it never trusts a caller
+assertion — every field is compared against durable lock state or a live
+observation supplied by the caller and gathered server-side.
+
+Deliberate boundaries:
+
+* **Renewal is not takeover.** A refusal here never widens what
+  ``assess_expired_lock_reclaim`` already allows; foreign expired locks keep
+  requiring a dead PID or missing worktree (#760 AC11), and a *live* foreign
+  lease stays non-recoverable by construction because only an expired lease is
+  ever a candidate (AC12).
+* **PID liveness is never authorization.** A live recorded PID proves the
+  daemon is up, nothing more. It is recorded as evidence and is neither
+  necessary nor sufficient for renewal (AC16).
+* **Absolute expiry is preserved.** Renewal issues a new absolute expiry from
+  the moment of the write. It does not introduce sliding heartbeat renewal,
+  lease generations as fencing tokens, or a shared cross-role lifecycle — that
+  is #790's scope and is deliberately not implemented here.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Iterable, Mapping, Sequence
+
+from issue_lock_store import AUTHOR_ISSUE_WORK_LEASE, is_lease_expired, is_process_alive
+from reviewer_worktree import parse_dirty_tracked_files
+
+# Outcome values.
+RENEWAL_SANCTIONED = "RENEWAL_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 _text(value: Any) -> str:
+    return str(value or "").strip()
+
+
+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 _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 _lock_lease(lock: Mapping[str, Any]) -> dict[str, Any]:
+    lease = lock.get("work_lease")
+    return dict(lease) if isinstance(lease, Mapping) else {}
+
+
+def _lock_operation_type(lock: Mapping[str, Any]) -> str:
+    lease = _lock_lease(lock)
+    return _text(lease.get("operation_type")) or AUTHOR_ISSUE_WORK_LEASE
+
+
+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 _competing_lock_reasons(
+    competing_live_locks: Iterable[Mapping[str, Any]] | None,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+) -> list[str]:
+    """Live locks that would contend with this renewal (#760 AC7).
+
+    A live lock on the *same* issue cannot coexist with this expired lease, so
+    any live entry naming this issue, branch, or worktree belongs to somebody
+    else and refuses the renewal.
+    """
+    reasons: list[str] = []
+    for entry in competing_live_locks or ():
+        if not isinstance(entry, Mapping):
+            continue
+        entry_issue = entry.get("issue_number")
+        entry_branch = _text(entry.get("branch_name"))
+        entry_worktree = _text(entry.get("worktree_path"))
+        if entry_issue == issue_number:
+            reasons.append(
+                f"a live lock already exists for issue #{issue_number} "
+                f"(pid {entry.get('pid')}); renewal would contend with it"
+            )
+            continue
+        if entry_branch and entry_branch == _text(branch_name):
+            reasons.append(
+                f"live lock for issue #{entry_issue} already holds branch "
+                f"'{branch_name}'"
+            )
+        if entry_worktree and _same_realpath(entry_worktree, worktree_path):
+            reasons.append(
+                f"live lock for issue #{entry_issue} already holds worktree "
+                f"'{worktree_path}'"
+            )
+    return reasons
+
+
+def assess_exact_owner_lease_renewal(
+    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,
+    operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
+    current_branch: str | None = None,
+    porcelain_status: str = "",
+    worktree_exists: bool = False,
+    head_sha: str | None = None,
+    remote_head_sha: str | None = None,
+    pr_head_sha: str | None = None,
+    pr_number: int | None = None,
+    competing_live_locks: Sequence[Mapping[str, Any]] | None = None,
+    candidate_branches: Sequence[str] | None = None,
+    current_pid: int | None = None,
+    now: Any = None,
+) -> dict[str, Any]:
+    """Decide whether an expired lease may be renewed by its exact owner.
+
+    Returns a disposition dict; it never raises and never mutates. A refusal
+    withholds permission, leaving every pre-existing guard to fail closed
+    exactly as before — this assessment can only ever *add* permission.
+
+    ``NO_CANDIDATE`` means the situation is not an exact-owner renewal at all
+    (no lock, different issue, different operation, or an unexpired lease) and
+    the caller should carry on with its normal path. ``REFUSED`` means it looked
+    like one but the evidence did not hold, and ``reasons`` names exactly what
+    was missing.
+    """
+    evidence: dict[str, Any] = {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "worktree_path": worktree_path,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "operation_type": operation_type,
+        "identity": identity,
+        "profile": profile,
+    }
+
+    def _result(outcome: str, reasons: list[str], **extra: Any) -> dict[str, Any]:
+        return {
+            "outcome": outcome,
+            "renewal_sanctioned": outcome == RENEWAL_SANCTIONED,
+            "is_candidate": outcome in (RENEWAL_SANCTIONED, REFUSED),
+            "reasons": reasons,
+            "evidence": {**evidence, **extra},
+        }
+
+    if not isinstance(existing_lock, Mapping) or not existing_lock:
+        return _result(NO_CANDIDATE, ["no existing lock to renew"])
+
+    if existing_lock.get("issue_number") != issue_number:
+        return _result(
+            NO_CANDIDATE,
+            [
+                f"existing lock is for issue #{existing_lock.get('issue_number')}, "
+                f"not #{issue_number}"
+            ],
+        )
+
+    existing_operation = _lock_operation_type(existing_lock)
+    if existing_operation != operation_type:
+        return _result(
+            NO_CANDIDATE,
+            [
+                f"existing lease operation '{existing_operation}' is not "
+                f"'{operation_type}'"
+            ],
+        )
+
+    # Only an *expired* lease is ever a renewal candidate. An unexpired lease —
+    # live, or stale by dead PID — is somebody else's problem: the first needs no
+    # renewal, and the second is #753's dead-session recovery. This is also what
+    # makes a live foreign lease non-recoverable here (#760 AC12).
+    if not is_lease_expired(existing_lock, now=now):
+        return _result(
+            NO_CANDIDATE,
+            ["lease has not expired; renewal does not apply"],
+        )
+
+    malformed = _malformed_reasons(existing_lock)
+    if malformed:
+        return _result(
+            REFUSED,
+            ["durable lock is missing or has unusable fields: " + ", ".join(malformed)],
+        )
+
+    lease = _lock_lease(existing_lock)
+    claimant = _lock_claimant(existing_lock)
+    recorded_pid = _recorded_pid(existing_lock)
+    prior_expires_at = _text(lease.get("expires_at"))
+
+    # #760 AC16: recorded purely as evidence. A live daemon PID is neither
+    # necessary nor sufficient for renewal, and nothing below branches on it.
+    recorded_pid_alive = is_process_alive(recorded_pid)
+
+    extra: dict[str, Any] = {
+        "prior_pid": recorded_pid,
+        "prior_pid_alive": recorded_pid_alive,
+        "prior_expires_at": prior_expires_at,
+        "replacement_pid": current_pid,
+        "recorded_claimant": claimant,
+        "head_sha": head_sha,
+        "remote_head_sha": remote_head_sha,
+        "pr_head_sha": pr_head_sha,
+        "pr_number": pr_number,
+    }
+
+    reasons: list[str] = []
+
+    # ── AC3: exact ownership identity ──
+    if _text(existing_lock.get("remote")) != _text(remote):
+        reasons.append(
+            f"recorded remote '{existing_lock.get('remote')}' does not match "
+            f"'{remote}'"
+        )
+    if _text(existing_lock.get("org")) != _text(org):
+        reasons.append(
+            f"recorded org '{existing_lock.get('org')}' does not match '{org}'"
+        )
+    if _text(existing_lock.get("repo")) != _text(repo):
+        reasons.append(
+            f"recorded repo '{existing_lock.get('repo')}' does not match '{repo}'"
+        )
+    if _text(existing_lock.get("branch_name")) != _text(branch_name):
+        reasons.append(
+            f"recorded branch '{existing_lock.get('branch_name')}' does not match "
+            f"'{branch_name}'"
+        )
+    if not _same_realpath(_text(existing_lock.get("worktree_path")), worktree_path):
+        reasons.append(
+            f"recorded worktree '{existing_lock.get('worktree_path')}' does not "
+            f"match '{worktree_path}'"
+        )
+
+    recorded_identity = _text(claimant.get("username"))
+    recorded_profile = _text(claimant.get("profile"))
+    if not recorded_identity or not recorded_profile:
+        reasons.append(
+            "durable lock does not record both a claimant username and profile"
+        )
+    if recorded_identity and recorded_identity != _text(identity):
+        reasons.append(
+            f"recorded claimant '{recorded_identity}' does not match active "
+            f"identity '{_text(identity) or 'unknown'}'"
+        )
+    if recorded_profile and recorded_profile != _text(profile):
+        reasons.append(
+            f"recorded profile '{recorded_profile}' does not match active profile "
+            f"'{_text(profile) or 'unknown'}'"
+        )
+
+    # ── AC4: the registered worktree still exists, is on the branch, and is clean ──
+    if not worktree_exists:
+        reasons.append(f"declared worktree '{worktree_path}' does not exist")
+    if _text(current_branch) != _text(branch_name):
+        reasons.append(
+            f"worktree is on branch '{_text(current_branch) or 'unknown'}', not "
+            f"'{branch_name}'"
+        )
+    dirty = parse_dirty_tracked_files(porcelain_status or "")
+    if dirty:
+        reasons.append(
+            "worktree has uncommitted tracked changes: " + ", ".join(sorted(dirty))
+        )
+
+    # ── AC5/AC6: published heads must agree ──
+    if not _text(head_sha):
+        reasons.append("local head could not be observed")
+    if not _text(remote_head_sha):
+        reasons.append(
+            "remote branch head could not be observed; an unpublished branch "
+            "cannot prove exact-owner renewal"
+        )
+    if _text(head_sha) and _text(remote_head_sha) and head_sha != remote_head_sha:
+        reasons.append(
+            f"local head {head_sha} does not equal remote head {remote_head_sha}"
+        )
+    if pr_number is not None:
+        if not _text(pr_head_sha):
+            reasons.append(f"owning PR #{pr_number} head could not be observed")
+        elif _text(head_sha) and pr_head_sha != head_sha:
+            reasons.append(
+                f"owning PR #{pr_number} head {pr_head_sha} does not equal local "
+                f"head {head_sha}"
+            )
+
+    # ── AC7: nothing else claims this work ──
+    reasons.extend(
+        _competing_lock_reasons(
+            competing_live_locks,
+            issue_number=issue_number,
+            branch_name=branch_name,
+            worktree_path=worktree_path,
+        )
+    )
+    other_branches = [
+        name
+        for name in (candidate_branches or ())
+        if _text(name) and _text(name) != _text(branch_name)
+    ]
+    if other_branches:
+        reasons.append(
+            "other branches already carry this issue marker: "
+            + ", ".join(sorted(other_branches))
+        )
+
+    if reasons:
+        return _result(REFUSED, reasons, **extra)
+
+    return _result(
+        RENEWAL_SANCTIONED,
+        [
+            f"exact owner '{recorded_identity}' ({recorded_profile}) proved "
+            f"ownership of issue #{issue_number} on branch '{branch_name}' from "
+            f"worktree '{worktree_path}'; local, remote"
+            + (f", and PR #{pr_number}" if pr_number is not None else "")
+            + f" heads all equal {head_sha}; lease expired at "
+            f"{prior_expires_at or 'unknown'}"
+        ],
+        **extra,
+    )
+
+
+def build_renewal_record(
+    assessment: Mapping[str, Any] | None,
+    *,
+    renewed_at: str,
+    new_expires_at: str,
+) -> dict[str, Any]:
+    """Durable audit record for a sanctioned renewal (#760 AC9).
+
+    Records both sides of the transition — prior PID and expiry, replacement PID
+    and new expiry — so a renewed lock is never mistakable for an original
+    claim, and so the evidence the waiver was granted on stays inspectable.
+    """
+    data = dict(assessment or {})
+    evidence = dict(data.get("evidence") or {})
+    recorded_claimant = dict(evidence.get("recorded_claimant") or {})
+    return {
+        "renewed": bool(data.get("renewal_sanctioned")),
+        "renewed_at": renewed_at,
+        "prior_pid": evidence.get("prior_pid"),
+        "prior_pid_alive": evidence.get("prior_pid_alive"),
+        "prior_expires_at": evidence.get("prior_expires_at"),
+        "replacement_pid": evidence.get("replacement_pid"),
+        "new_expires_at": new_expires_at,
+        "identity": recorded_claimant.get("username"),
+        "profile": recorded_claimant.get("profile"),
+        "branch_name": evidence.get("branch_name"),
+        "worktree_path": evidence.get("worktree_path"),
+        "head_sha": evidence.get("head_sha"),
+        "remote_head_sha": evidence.get("remote_head_sha"),
+        "pr_head_sha": evidence.get("pr_head_sha"),
+        "pr_number": evidence.get("pr_number"),
+        "reason": "expired lease renewed by its exact recorded owner",
+        "proof": list(data.get("reasons") or []),
+    }
+
+
+def format_renewal_refusal(assessment: Mapping[str, Any] | None) -> str:
+    """One-line refusal summary for a blocked caller."""
+    data = dict(assessment or {})
+    reasons = list(data.get("reasons") or [])
+    if not reasons:
+        return "exact-owner lease renewal was not available (no evidence recorded)"
+    return "exact-owner lease renewal refused: " + "; ".join(reasons)

issue_lock_store.py

@@ -168,6 +168,7 @@ def bind_session_lock(
     lock_dir: str | None = None,
     *,
     expected_generation: int | None = None,
+    renewal_sanctioned: bool = False,
 ) -> str:
     """Persist a keyed lock and bind it to the current process session.
 
@@ -220,6 +221,7 @@ def bind_session_lock(
                 issue_number=issue_number,
                 branch_name=str(record.get("branch_name") or ""),
                 worktree_path=str(record.get("worktree_path") or ""),
+                renewal_sanctioned=renewal_sanctioned,
             )
             if lease_block:
                 raise RuntimeError(lease_block)
@@ -483,9 +485,19 @@ def assess_same_issue_lease_conflict(
     branch_name: str,
     worktree_path: str,
     operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
+    renewal_sanctioned: bool = False,
     now: datetime | None = None,
 ) -> str | None:
-    """Return a fail-closed error when a competing live lease blocks acquisition."""
+    """Return a fail-closed error when a competing live lease blocks acquisition.
+
+    ``renewal_sanctioned`` is set only when
+    ``issue_lock_renewal.assess_exact_owner_lease_renewal`` has already proven,
+    from the durable lock plus live server-side observation, that this session
+    is the exact recorded owner of an *expired* lease (#760). It is never a
+    caller-supplied parameter of any MCP tool (#760 AC14): the server computes
+    it and passes it down. Left False, every pre-existing disposition is
+    unchanged.
+    """
     if not existing_lock:
         return None
 
@@ -506,6 +518,16 @@ def assess_same_issue_lease_conflict(
         and _same_realpath(str(existing_worktree or ""), worktree_path)
     )
     if is_lease_expired(existing_lock, now=now):
+        # #760 AC1/AC2: exact-owner renewal is a different disposition from
+        # foreign takeover and is evaluated first. Before this, both branches
+        # below returned unconditionally, so the same_owner allowance further
+        # down was unreachable for every expired lease — an owner could never
+        # renew its own lock once the wall clock passed, no matter how complete
+        # its ownership evidence. Requires BOTH the locally recomputed
+        # same_owner match and the server-proven renewal waiver; either alone is
+        # insufficient.
+        if same_owner and renewal_sanctioned:
+            return None
         reclaim = assess_expired_lock_reclaim(existing_lock, now=now)
         if reclaim.get("reclaim_allowed"):
             # #601: expired + dead pid / missing worktree may be reclaimed

tests/test_issue_760_exact_owner_lease_renewal.py

@@ -0,0 +1,447 @@
+"""Exact-owner renewal of an expired author issue lease (#760).
+
+Covers the renewal disposition that lets the exact recorded owner re-acquire
+its own lock after the wall-clock lease expires — including while the recording
+MCP daemon PID is still alive — plus every rejection condition that must keep
+failing closed, and the pre-existing dead-PID and live-foreign dispositions
+that must remain untouched.
+"""
+
+import inspect
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import issue_lock_renewal  # noqa: E402
+import issue_lock_store  # noqa: E402
+
+ISSUE = 5150
+BRANCH = f"fix/issue-{ISSUE}-demo"
+WORKTREE = "/scratch/wt-5150"
+HEAD = "c" * 40
+OTHER_SHA = "d" * 40
+IDENTITY = "example-user"
+PROFILE = "example-author"
+REMOTE = "prgs"
+ORG = "ExampleOrg"
+REPO = "ExampleRepo"
+
+
+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 past_ts(hours: int = 1) -> str:
+    return (
+        (datetime.now(timezone.utc) - timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def future_ts(hours: int = 4) -> str:
+    return (
+        (datetime.now(timezone.utc) + timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def make_lock(*, expires_at: str | None = None, pid: int | None = None, **overrides):
+    """An expired lock owned by a still-alive daemon PID — the #760 condition."""
+    lock = {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "worktree_path": WORKTREE,
+        "remote": REMOTE,
+        "org": ORG,
+        "repo": REPO,
+        # os.getpid() is unambiguously alive: the whole point of #760 is that
+        # daemon liveness is not evidence of an active author task.
+        "session_pid": os.getpid() if pid is None else pid,
+        "lock_generation": 3,
+        "work_lease": {
+            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+            "issue_number": ISSUE,
+            "branch": BRANCH,
+            "worktree_path": WORKTREE,
+            "claimant": {"username": IDENTITY, "profile": PROFILE},
+            "created_at": past_ts(5),
+            "expires_at": expires_at or past_ts(),
+        },
+    }
+    lease_overrides = overrides.pop("work_lease", None)
+    if lease_overrides:
+        lock["work_lease"].update(lease_overrides)
+    lock.update(overrides)
+    return lock
+
+
+def assess(lock=None, **overrides):
+    """Run the assessor with all-passing evidence unless overridden."""
+    kwargs = {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "worktree_path": WORKTREE,
+        "remote": REMOTE,
+        "org": ORG,
+        "repo": REPO,
+        "identity": IDENTITY,
+        "profile": PROFILE,
+        "current_branch": BRANCH,
+        "porcelain_status": "",
+        "worktree_exists": True,
+        "head_sha": HEAD,
+        "remote_head_sha": HEAD,
+        "pr_head_sha": None,
+        "pr_number": None,
+        "competing_live_locks": [],
+        "candidate_branches": [BRANCH],
+        "current_pid": 4242,
+    }
+    kwargs.update(overrides)
+    return issue_lock_renewal.assess_exact_owner_lease_renewal(
+        make_lock() if lock is None else lock, **kwargs
+    )
+
+
+class ExactOwnerRenewalGranted(unittest.TestCase):
+    """AC1/AC3-AC7: the positive path."""
+
+    def test_expired_lease_alive_pid_exact_owner_is_renewable(self):
+        result = assess()
+        self.assertEqual(result["outcome"], issue_lock_renewal.RENEWAL_SANCTIONED)
+        self.assertTrue(result["renewal_sanctioned"])
+        self.assertTrue(result["is_candidate"])
+
+    def test_renewal_holds_when_owning_pr_head_matches(self):
+        result = assess(pr_number=999, pr_head_sha=HEAD)
+        self.assertTrue(result["renewal_sanctioned"])
+
+    def test_evidence_records_both_sides_of_the_transition(self):
+        result = assess()
+        evidence = result["evidence"]
+        self.assertEqual(evidence["prior_pid"], os.getpid())
+        self.assertTrue(evidence["prior_pid_alive"])
+        self.assertEqual(evidence["replacement_pid"], 4242)
+        self.assertTrue(evidence["prior_expires_at"])
+
+
+class ExactOwnerRenewalRefused(unittest.TestCase):
+    """AC3-AC8: every near-match must fail closed, one reason at a time."""
+
+    def _refused(self, **overrides):
+        result = assess(**overrides)
+        self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
+        self.assertFalse(result["renewal_sanctioned"])
+        self.assertTrue(result["reasons"])
+        return result
+
+    def test_different_branch_refused(self):
+        result = self._refused(branch_name=f"fix/issue-{ISSUE}-other")
+        self.assertTrue(any("branch" in r for r in result["reasons"]))
+
+    def test_different_worktree_refused(self):
+        result = self._refused(worktree_path="/scratch/somewhere-else")
+        self.assertTrue(any("worktree" in r for r in result["reasons"]))
+
+    def test_different_claimant_refused(self):
+        result = self._refused(identity="someone-else")
+        self.assertTrue(any("claimant" in r for r in result["reasons"]))
+
+    def test_different_profile_refused(self):
+        result = self._refused(profile="other-author")
+        self.assertTrue(any("profile" in r for r in result["reasons"]))
+
+    def test_different_remote_org_or_repo_refused(self):
+        self._refused(remote="dadeschools")
+        self._refused(org="OtherOrg")
+        self._refused(repo="OtherRepo")
+
+    def test_dirty_worktree_refused(self):
+        result = self._refused(porcelain_status=" M gitea_mcp_server.py\n")
+        self.assertTrue(any("uncommitted" in r for r in result["reasons"]))
+
+    def test_missing_worktree_refused(self):
+        result = self._refused(worktree_exists=False)
+        self.assertTrue(any("does not exist" in r for r in result["reasons"]))
+
+    def test_worktree_on_wrong_branch_refused(self):
+        self._refused(current_branch="master")
+
+    def test_local_and_remote_head_mismatch_refused(self):
+        result = self._refused(remote_head_sha=OTHER_SHA)
+        self.assertTrue(
+            any("does not equal remote head" in r for r in result["reasons"])
+        )
+
+    def test_unpublished_branch_refused(self):
+        result = self._refused(remote_head_sha=None)
+        self.assertTrue(any("remote branch head" in r for r in result["reasons"]))
+
+    def test_pr_head_mismatch_refused(self):
+        result = self._refused(pr_number=999, pr_head_sha=OTHER_SHA)
+        self.assertTrue(any("does not equal local" in r for r in result["reasons"]))
+
+    def test_unobservable_pr_head_refused(self):
+        self._refused(pr_number=999, pr_head_sha=None)
+
+    def test_competing_live_lock_on_same_issue_refused(self):
+        result = self._refused(
+            competing_live_locks=[
+                {"issue_number": ISSUE, "branch_name": BRANCH, "pid": 777}
+            ]
+        )
+        self.assertTrue(any("live lock" in r for r in result["reasons"]))
+
+    def test_competing_live_lock_holding_the_branch_refused(self):
+        self._refused(
+            competing_live_locks=[
+                {"issue_number": 111, "branch_name": BRANCH, "worktree_path": ""}
+            ]
+        )
+
+    def test_competing_branch_claim_refused(self):
+        result = self._refused(candidate_branches=[BRANCH, f"feat/issue-{ISSUE}-rival"])
+        self.assertTrue(any("issue marker" in r for r in result["reasons"]))
+
+    def test_malformed_durable_lock_refused(self):
+        lock = make_lock()
+        lock["worktree_path"] = ""
+        result = assess(lock)
+        self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
+
+    def test_lock_without_recorded_claimant_refused(self):
+        lock = make_lock()
+        lock["work_lease"]["claimant"] = {}
+        result = assess(lock)
+        self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
+
+
+class NotARenewalCandidate(unittest.TestCase):
+    """AC12 and scope: situations renewal must decline to judge at all."""
+
+    def test_live_foreign_lease_is_never_a_candidate(self):
+        lock = make_lock(expires_at=future_ts())
+        result = assess(lock, identity="someone-else")
+        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
+        self.assertFalse(result["renewal_sanctioned"])
+
+    def test_unexpired_lease_is_never_a_candidate(self):
+        lock = make_lock(expires_at=future_ts())
+        result = assess(lock)
+        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
+
+    def test_dead_pid_under_unexpired_lease_stays_with_753(self):
+        """The opposite trigger; #760 must not re-own it."""
+        lock = make_lock(expires_at=future_ts(), pid=dead_pid())
+        result = assess(lock)
+        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
+
+    def test_absent_lock_is_not_a_candidate(self):
+        result = assess({})
+        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
+
+    def test_different_issue_is_not_a_candidate(self):
+        lock = make_lock()
+        lock["issue_number"] = ISSUE + 1
+        result = assess(lock)
+        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
+
+    def test_different_operation_type_is_not_a_candidate(self):
+        lock = make_lock()
+        lock["work_lease"]["operation_type"] = "review_pr_work"
+        result = assess(lock)
+        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
+
+
+class DaemonPidIsNotTaskLiveness(unittest.TestCase):
+    """AC16: a live recorded PID is never, by itself, authorization."""
+
+    def test_alive_pid_alone_does_not_authorize_renewal(self):
+        # Every ownership fact except the live PID is wrong.
+        result = assess(identity="someone-else", branch_name="fix/issue-1-nope")
+        self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
+        self.assertTrue(result["evidence"]["prior_pid_alive"])
+
+    def test_renewal_does_not_require_a_dead_pid(self):
+        result = assess()
+        self.assertTrue(result["evidence"]["prior_pid_alive"])
+        self.assertTrue(result["renewal_sanctioned"])
+
+    def test_dead_pid_does_not_block_an_otherwise_exact_owner(self):
+        lock = make_lock(pid=dead_pid())
+        result = assess(lock)
+        self.assertTrue(result["renewal_sanctioned"])
+
+
+class ConflictGateOrdering(unittest.TestCase):
+    """AC2: the same-owner allowance is reachable on an expired lease.
+
+    These cases need a worktree that genuinely exists on disk. The #601 reclaim
+    affordance already permits takeover when the recorded worktree is missing,
+    so a fictional path would satisfy the gate for the wrong reason and never
+    exercise the ordering defect this issue is about.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        cls._tmp = tempfile.TemporaryDirectory()
+        cls.worktree = cls._tmp.name
+
+    @classmethod
+    def tearDownClass(cls):
+        cls._tmp.cleanup()
+
+    def present_lock(self, **overrides):
+        return make_lock(worktree_path=self.worktree, **overrides)
+
+    def test_expired_same_owner_is_allowed_when_renewal_is_sanctioned(self):
+        block = issue_lock_store.assess_same_issue_lease_conflict(
+            self.present_lock(),
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.worktree,
+            renewal_sanctioned=True,
+        )
+        self.assertIsNone(block)
+
+    def test_expired_same_owner_still_blocks_without_the_waiver(self):
+        """Regression for the ordering defect: no waiver, no change in behavior.
+
+        Live PID and a present worktree, so the #601 reclaim affordance refuses;
+        before #760 this was the permanent dead end for an exact owner.
+        """
+        lock = self.present_lock()
+        self.assertFalse(
+            issue_lock_store.assess_expired_lock_reclaim(lock)["reclaim_allowed"]
+        )
+        block = issue_lock_store.assess_same_issue_lease_conflict(
+            lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.worktree,
+        )
+        self.assertIsNotNone(block)
+        self.assertIn("Recovery review is required", block)
+
+    def test_waiver_does_not_unlock_a_different_owner(self):
+        """AC11: the waiver is scoped by same_owner, not merely by its own flag."""
+        block = issue_lock_store.assess_same_issue_lease_conflict(
+            self.present_lock(),
+            issue_number=ISSUE,
+            branch_name=f"fix/issue-{ISSUE}-someone-else",
+            worktree_path=self.worktree,
+            renewal_sanctioned=True,
+        )
+        self.assertIsNotNone(block)
+        self.assertIn("Recovery review is required", block)
+
+    def test_live_lease_disposition_is_unchanged(self):
+        """AC12: a live foreign lease still blocks, waiver or not."""
+        block = issue_lock_store.assess_same_issue_lease_conflict(
+            self.present_lock(expires_at=future_ts()),
+            issue_number=ISSUE,
+            branch_name=f"fix/issue-{ISSUE}-someone-else",
+            worktree_path="/scratch/other",
+            renewal_sanctioned=True,
+        )
+        self.assertIsNotNone(block)
+        self.assertIn("already has an active", block)
+
+    def test_dead_pid_reclaim_path_is_unchanged(self):
+        """AC11: expired + dead PID still reclaims through the #601 affordance."""
+        lock = self.present_lock(pid=dead_pid())
+        reclaim = issue_lock_store.assess_expired_lock_reclaim(lock)
+        self.assertTrue(reclaim["reclaim_allowed"])
+        block = issue_lock_store.assess_same_issue_lease_conflict(
+            lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.worktree,
+        )
+        self.assertIsNone(block)
+
+
+class RenewalRecordAndDownstream(unittest.TestCase):
+    """AC9/AC10: durable audit trail, and a renewed lock that actually works."""
+
+    def test_record_captures_prior_and_replacement_state(self):
+        assessment = assess()
+        record = issue_lock_renewal.build_renewal_record(
+            assessment,
+            renewed_at="2026-01-01T00:00:00Z",
+            new_expires_at="2026-01-01T04:00:00Z",
+        )
+        self.assertTrue(record["renewed"])
+        self.assertEqual(record["prior_pid"], os.getpid())
+        self.assertEqual(record["new_expires_at"], "2026-01-01T04:00:00Z")
+        self.assertEqual(record["renewed_at"], "2026-01-01T00:00:00Z")
+        self.assertEqual(record["identity"], IDENTITY)
+        self.assertEqual(record["profile"], PROFILE)
+        self.assertTrue(record["prior_expires_at"])
+        self.assertTrue(record["proof"])
+
+    def test_renewed_lock_satisfies_verify_lock_for_mutation(self):
+        renewed = make_lock(expires_at=future_ts())
+        renewed["session_pid"] = os.getpid()
+        renewed["lease_renewal"] = {"renewed": True}
+        verdict = issue_lock_store.verify_lock_for_mutation(
+            renewed,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+        )
+        self.assertTrue(verdict["proven"])
+        self.assertFalse(verdict["block"])
+
+    def test_refusal_message_names_the_missing_evidence(self):
+        assessment = assess(porcelain_status=" M gitea_mcp_server.py\n")
+        message = issue_lock_renewal.format_renewal_refusal(assessment)
+        self.assertIn("refused", message)
+        self.assertIn("uncommitted", message)
+
+
+class NoCallerControlledRenewalFlag(unittest.TestCase):
+    """AC14: renewal eligibility is never declarable by a caller."""
+
+    def test_lock_issue_tool_exposes_no_renewal_parameter(self):
+        import gitea_mcp_server
+
+        target = gitea_mcp_server.gitea_lock_issue
+        target = getattr(target, "fn", getattr(target, "__wrapped__", target))
+        params = set(inspect.signature(target).parameters)
+        for forbidden in ("renewal_sanctioned", "renew", "allow_renewal", "is_owner"):
+            self.assertNotIn(forbidden, params)
+
+    def test_store_defaults_to_no_waiver(self):
+        params = inspect.signature(
+            issue_lock_store.assess_same_issue_lease_conflict
+        ).parameters
+        self.assertIs(params["renewal_sanctioned"].default, False)
+        bind_params = inspect.signature(issue_lock_store.bind_session_lock).parameters
+        self.assertIs(bind_params["renewal_sanctioned"].default, False)
+
+
+class NoIssueNumberSpecialCasing(unittest.TestCase):
+    """AC17: no repository issue or PR number is special-cased."""
+
+    def test_module_contains_no_hardcoded_issue_special_cases(self):
+        source = inspect.getsource(issue_lock_renewal)
+        code = "\n".join(
+            line for line in source.splitlines() if not line.strip().startswith("#")
+        )
+        for literal in ("757", "759", "760"):
+            self.assertNotIn(f"== {literal}", code)
+            self.assertNotIn(f"issue_number == {literal}", code)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_772_unpublished_claim_recovery.py

@@ -871,9 +871,13 @@ class TestAc6McpUnpublishedClaimRecovery(_UnpublishedMcpBase):
         save_calls: list[dict] = []
         real_save = mcp_server._save_issue_lock
 
-        def tracking_save(data, *, expected_generation=None):
+        def tracking_save(data, *, expected_generation=None, renewal_sanctioned=False):
             save_calls.append({"expected_generation": expected_generation, "data": dict(data)})
-            return real_save(data, expected_generation=expected_generation)
+            return real_save(
+                data,
+                expected_generation=expected_generation,
+                renewal_sanctioned=renewal_sanctioned,
+            )
 
         with patch("mcp_server._save_issue_lock", side_effect=tracking_save):
             result = self.run_lock_issue()
@@ -926,18 +930,33 @@ class TestAc6McpUnpublishedClaimRecovery(_UnpublishedMcpBase):
         real_bind = issue_lock_store.bind_session_lock
         bind_calls: list[int | None] = []
 
-        def racing_bind(data, lock_dir=None, expected_generation=None):
+        # #760 added the renewal waiver keyword; the double forwards it verbatim
+        # so this race still exercises the real compare-and-swap.
+        def racing_bind(
+            data, lock_dir=None, expected_generation=None, renewal_sanctioned=False
+        ):
             bind_calls.append(expected_generation)
             if expected_generation is None:
-                return real_bind(data, lock_dir=lock_dir, expected_generation=None)
+                return real_bind(
+                    data,
+                    lock_dir=lock_dir,
+                    expected_generation=None,
+                    renewal_sanctioned=renewal_sanctioned,
+                )
             # First concurrent writer wins.
             if len([c for c in bind_calls if c is not None]) == 1:
                 return real_bind(
-                    data, lock_dir=lock_dir, expected_generation=expected_generation
+                    data,
+                    lock_dir=lock_dir,
+                    expected_generation=expected_generation,
+                    renewal_sanctioned=renewal_sanctioned,
                 )
             # Second concurrent writer still holds the pre-race generation.
             return real_bind(
-                data, lock_dir=lock_dir, expected_generation=expected_generation
+                data,
+                lock_dir=lock_dir,
+                expected_generation=expected_generation,
+                renewal_sanctioned=renewal_sanctioned,
             )
 
         # First recovery succeeds and advances generation.