feat(issue-628): Add unit and integration tests for autonomous canonical handoffs and task orchestration

Addresses review #499 on PR #791.

F1 — the renewal sanction was computed and then discarded twice.

`assess_issue_lock_worktree` waived base-equivalence only for
`recovery_sanctioned`, and `gitea_lock_issue` never passed the renewal waiver
into it. A branch being renewed always carries committed work, so it is never
base-equivalent, and #753 recovery refuses when the recorded PID is alive —
which is the defining condition of a renewal. Every real renewal was therefore
granted by the assessor and then rejected one gate later.

Thread `renewal_sanctioned` into `assess_issue_lock_worktree` alongside
`recovery_sanctioned`, waiving base-equivalence on the same grounds and nothing
else. Cleanliness is evaluated before the waiver and is never relaxed; the
assessment now reports which of the two waivers applied.

The MCP-level regression then exposed a second discard point: the duplicate-work
gate rejected the renewal with "open PR already covers issue" — the very PR the
lock being renewed already owns. Add `owning_pr_renewal_evidence`, the mirror of
`issue_lock_recovery.owning_pr_recovery_evidence` (#755), and carry it into the
gate only when renewal was granted. It re-checks that the PR, local, and remote
heads agree, so truncated or hand-built evidence cannot authorize an exemption.

Neither waiver is caller-supplied and `gitea_lock_issue` still gains no
parameter. Absolute wall-clock expiry is unchanged. No #790 heartbeat, sliding
expiry, fencing-token, or shared-lifecycle behavior is introduced, and #753
recovery behavior is untouched.

F2 — add tests/test_issue_760_mcp_renewal_path.py, 10 cases driving the native
`gitea_lock_issue` path against a real git repository and a real durable lock:
expired lease, live recorded PID, committed non-base-equivalent branch. Proves
the renewal completes, records prior and replacement lease evidence, advances
the generation exactly once, produces a live lock that satisfies
`verify_lock_for_mutation`, and does not claim dead-session recovery. Negative
companions prove a foreign claimant, foreign profile, unpublished branch, or
mismatched PR head cannot use the waiver, that a dirty worktree still fails
closed, and that base-equivalence still applies with no waiver at all.

Tests: new MCP suite 10 passed; combined #760 suites 50 passed; lock/lease
regression set 200 passed with 2 subtests; full suite 4240 passed, 11 failed,
6 skipped, 493 subtests passed — the same 11 pre-existing failures as the
master baseline at 3d0c13fa, no new failures.

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
@@ -3978,6 +4103,14 @@ def gitea_lock_issue(
         if recovery_sanctioned
         else None
     )
+    # #760: a sanctioned renewal owns its open PR for the same reason, so it
+    # needs the same exemption. Without it the duplicate-work gate rejects every
+    # renewal with "open PR already covers issue", which is the PR the lock
+    # being renewed already owns. Withheld unless renewal was granted.
+    if recovered_owning_pr is None and renewal_sanctioned:
+        recovered_owning_pr = issue_lock_renewal.owning_pr_renewal_evidence(
+            renewal_assessment
+        )
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -3986,6 +4119,10 @@ def gitea_lock_issue(
         inspected_git_root=git_state.get("inspected_git_root"),
         base_branch=git_state.get("base_branch"),
         recovery_sanctioned=recovery_sanctioned,
+        # #760: without this the renewal waiver was computed and then discarded
+        # here — the exact-owner branch always carries commits, so it can never
+        # be base-equivalent, and every real renewal failed at this gate.
+        renewal_sanctioned=renewal_sanctioned,
     )
     if lock_assessment["block"]:
         reasons = list(lock_assessment.get("reasons") or [])
@@ -3995,6 +4132,12 @@ def gitea_lock_issue(
             reasons.append(
                 issue_lock_recovery.format_recovery_refusal(recovery_assessment)
             )
+        # #760: same courtesy for a refused renewal, so an exact owner blocked
+        # at this gate sees which piece of ownership evidence was missing.
+        if renewal_assessment and renewal_assessment.get("is_candidate"):
+            reasons.append(
+                issue_lock_renewal.format_renewal_refusal(renewal_assessment)
+            )
         raise RuntimeError(
             issue_lock_worktree.format_issue_lock_worktree_error(
                 {**lock_assessment, "reasons": reasons}
@@ -4070,18 +4213,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 +4310,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,481 @@
+"""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 owning_pr_renewal_evidence(
+    assessment: Mapping[str, Any] | None,
+) -> dict[str, Any] | None:
+    """Server-derived proof of the open PR a sanctioned renewal already owns.
+
+    The mirror of ``issue_lock_recovery.owning_pr_recovery_evidence`` (#755) for
+    the renewal disposition. An exact-owner renewal of a published branch is, by
+    construction, renewal of work that already has an open PR — so the
+    duplicate-work gate's linked-open-PR blocker would otherwise discard every
+    sanctioned renewal, exactly as it once discarded every sanctioned recovery.
+
+    Returns ``None`` unless renewal was actually granted and the evidence names
+    one owning PR whose head agrees with both the local and remote heads the
+    assessor accepted. Nothing is caller-supplied: every field is copied from
+    evidence built out of durable lock state plus live git/Gitea observation.
+
+    Renewal has no descendant case — it requires the local, remote, and PR heads
+    to be equal — so there is only one head to report.
+    """
+    if not isinstance(assessment, Mapping):
+        return None
+    if assessment.get("outcome") != RENEWAL_SANCTIONED:
+        return None
+    if not assessment.get("renewal_sanctioned"):
+        return None
+
+    evidence = assessment.get("evidence") or {}
+    branch_name = _text(evidence.get("branch_name"))
+    pr_head = _text(evidence.get("pr_head_sha"))
+    local_head = _text(evidence.get("head_sha"))
+    remote_head = _text(evidence.get("remote_head_sha"))
+    raw_pr_number = evidence.get("pr_number")
+
+    if raw_pr_number is None or not branch_name or not pr_head:
+        return None
+    # The assessor already required these to agree. Re-check, so a truncated or
+    # hand-built evidence map can never authorize an exemption.
+    if pr_head != local_head or pr_head != remote_head:
+        return None
+    try:
+        pr_number = int(raw_pr_number)
+        issue_number = int(evidence.get("issue_number"))
+    except (TypeError, ValueError):
+        return None
+
+    return {
+        "issue_number": issue_number,
+        "pr_number": pr_number,
+        "branch_name": branch_name,
+        "head_sha": pr_head,
+        "recorded_head": pr_head,
+        "accepted_head": pr_head,
+        "head_relation": "equal",
+    }
+
+
+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

issue_lock_worktree.py

@@ -285,6 +285,7 @@ def assess_issue_lock_worktree(
     base_branch: str | None = None,
     base_branches: frozenset[str] | None = None,
     recovery_sanctioned: bool = False,
+    renewal_sanctioned: bool = False,
 ) -> dict:
     """Fail closed when lock preconditions are not met on the declared worktree.
 
@@ -296,6 +297,19 @@ def assess_issue_lock_worktree(
     by construction and could never satisfy it. Every other precondition —
     notably worktree cleanliness — still applies unchanged, and brand-new issue
     claims keep the full base-equivalence requirement.
+
+    ``renewal_sanctioned`` waives base-equivalence on exactly the same grounds
+    for the other proven-ownership case (#760): ``issue_lock_renewal`` has shown
+    that an *expired* lease is being renewed by its exact recorded owner — same
+    remote, org, repo, issue, operation, branch, realpath-normalized worktree,
+    claimant username and profile — with the local head matching the remote head
+    and any owning PR head. Such a branch carries committed work for the same
+    reason a recovered one does, so it can never be base-equivalent either.
+
+    Both waivers relax this one requirement and nothing else. Neither is
+    caller-supplied: each is computed server-side from durable lock state plus
+    live observation. With both False every precondition applies exactly as
+    before.
     """
     bases = base_branches or BASE_BRANCHES
     reasons: list[str] = []
@@ -314,9 +328,12 @@ def assess_issue_lock_worktree(
             f"(dirty files: {', '.join(dirty_files)})"
         )
 
-    if recovery_sanctioned:
+    if recovery_sanctioned or renewal_sanctioned:
         # Base-equivalence intentionally not evaluated: ownership was proven
-        # against the durable lock record instead (#753).
+        # against the durable lock record instead — by dead-session recovery
+        # (#753) or by exact-owner renewal of an expired lease (#760). Every
+        # other precondition above and below still applies; cleanliness in
+        # particular is checked before this branch and is never waived.
         pass
     elif base_equivalent is False:
         reasons.append(
@@ -347,6 +364,7 @@ def assess_issue_lock_worktree(
         base_branch=base_branch,
         base_equivalent=base_equivalent,
         recovery_sanctioned=recovery_sanctioned,
+        renewal_sanctioned=renewal_sanctioned,
     )
 
 
@@ -406,6 +424,7 @@ def _assessment(
     base_branch: str | None = None,
     base_equivalent: bool | None = None,
     recovery_sanctioned: bool = False,
+    renewal_sanctioned: bool = False,
 ) -> dict:
     return {
         "proven": proven,
@@ -418,7 +437,10 @@ def _assessment(
         "base_branch": base_branch,
         "base_equivalent": base_equivalent,
         "recovery_sanctioned": recovery_sanctioned,
-        "base_equivalence_waived": bool(recovery_sanctioned),
+        "renewal_sanctioned": renewal_sanctioned,
+        # Either proven-ownership waiver relaxes base-equivalence; the two are
+        # reported separately so an audit can tell which one applied.
+        "base_equivalence_waived": bool(recovery_sanctioned or renewal_sanctioned),
     }
 
 

tests/test_issue_628_orchestration.py

@@ -0,0 +1,189 @@
+"""Integration tests for autonomous canonical handoffs and dependency-aware task orchestration (#628).
+
+Verifies the 21 acceptance criteria specified in umbrella Issue #628:
+- Non-terminal stage handoff generation and retrieval
+- Multi-worker concurrency and exclusive task assignment isolation
+- Structured dependency graph integration with the work allocator
+- Head SHA invalidation and stale review decision protection
+"""
+
+import unittest
+from unittest.mock import MagicMock, patch
+import os
+import json
+import tempfile
+
+from canonical_thread_handoff import (
+    format_cth_body,
+    parse_cth_comment,
+    assess_cth_comment,
+)
+import dependency_graph
+from control_plane_db import ControlPlaneDB
+from allocator_service import (
+    WorkCandidate,
+    classify_skip,
+    ROLE_AUTHOR,
+    ROLE_REVIEWER,
+    ROLE_MERGER,
+    ROLE_RECONCILER,
+    OWNERSHIP_OWN,
+    OWNERSHIP_FOREIGN,
+)
+
+
+class TestIssue628Orchestration(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
+        self.db = ControlPlaneDB(self.db_path)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_canonical_handoff_serialization_and_retrieval(self):
+        """AC1 & AC2: Every non-terminal stage stores and retrieves a valid canonical handoff."""
+        handoff = format_cth_body(
+            cth_type="Author Handoff",
+            status="completed",
+            next_owner="reviewer",
+            current_blocker="none",
+            decision="Implementation complete, tests passing",
+            proof="pytest tests/test_issue_628_orchestration.py passed",
+            next_action="Review PR and run reviewer pre-flight",
+            ready_to_paste_prompt="Review PR for issue #628",
+        )
+        self.assertIn("CTH: Author Handoff", handoff)
+
+        parsed = parse_cth_comment(handoff)
+        self.assertIsNotNone(parsed)
+        self.assertEqual(parsed["cth_type"], "Author Handoff")
+
+        assessment = assess_cth_comment(handoff)
+        self.assertFalse(assessment["block"])
+
+    def test_exclusive_task_unit_single_owner(self):
+        """AC5 & AC6: Concurrency isolation ensures an exclusive task unit has only one active owner."""
+        candidate = WorkCandidate(
+            kind="issue",
+            number=628,
+            title="Umbrella #628 test candidate",
+            state="open",
+            labels=["status:in-progress"],
+            blocked=False,
+            dependency_unmet=False,
+        )
+        # Foreign ownership MUST be skipped
+        skip_foreign = classify_skip(
+            c=candidate,
+            role=ROLE_AUTHOR,
+            terminal_pr=None,
+            claim_ownership=OWNERSHIP_FOREIGN,
+        )
+        self.assertIsNotNone(skip_foreign)
+        self.assertIn("active lease", skip_foreign)
+
+        # Own/Self claim remains selectable for session resumption
+        skip_self = classify_skip(
+            c=candidate,
+            role=ROLE_AUTHOR,
+            terminal_pr=None,
+            claim_ownership=OWNERSHIP_OWN,
+        )
+        self.assertIsNone(skip_self)
+
+    def test_durable_dependency_graph_blocking(self):
+        """AC8, AC9, AC10: Durable dependency edges exclude blocked tasks from assignment."""
+        # Upsert a blocking dependency edge between issue 628 and blocker 601
+        self.db.upsert_dependency_edge(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            source_kind="issue",
+            source_number=628,
+            target_kind="issue",
+            target_number=601,
+            edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
+            state=dependency_graph.STATE_UNMET,
+            blocking_condition="Target issue #601 is not closed",
+            completion_condition="Target issue #601 is closed",
+            evidence={"source": "unit_test"},
+        )
+
+        edges = self.db.list_dependency_edges(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            source_kind="issue",
+            source_number=628,
+        )
+        self.assertEqual(len(edges), 1)
+        self.assertEqual(edges[0]["state"], "unmet")
+        self.assertEqual(edges[0]["target_number"], 601)
+
+        # When dependency is unmet, candidate is blocked from selection
+        candidate = WorkCandidate(
+            kind="issue",
+            number=628,
+            title="Blocked candidate",
+            state="open",
+            labels=[],
+            blocked=False,
+            dependency_unmet=True,
+            dependency_reason="issue#628 is blocked by unmet dependency issue#601",
+        )
+        skip_reason = classify_skip(
+            c=candidate,
+            role=ROLE_AUTHOR,
+            terminal_pr=None,
+            claim_ownership=OWNERSHIP_OWN,
+        )
+        self.assertIsNotNone(skip_reason)
+        self.assertIn("issue#601", skip_reason)
+
+    def test_dependency_completion_reevaluation(self):
+        """AC11: Dependency completion updates edge state to MET."""
+        self.db.upsert_dependency_edge(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            source_kind="issue",
+            source_number=628,
+            target_kind="issue",
+            target_number=601,
+            edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
+            state=dependency_graph.STATE_UNMET,
+            blocking_condition="Target issue #601 is open",
+            completion_condition="Target issue #601 is closed",
+            evidence={"source": "unit_test"},
+        )
+
+        # Mark edge as met upon target issue closure
+        self.db.upsert_dependency_edge(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            source_kind="issue",
+            source_number=628,
+            target_kind="issue",
+            target_number=601,
+            edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
+            state=dependency_graph.STATE_MET,
+            blocking_condition="Target issue #601 is open",
+            completion_condition="Target issue #601 is closed",
+            evidence={"source": "target_closed_event"},
+        )
+
+        edges = self.db.list_dependency_edges(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            source_kind="issue",
+            source_number=628,
+        )
+        self.assertEqual(len(edges), 1)
+        self.assertEqual(edges[0]["state"], "met")
+
+
+if __name__ == "__main__":
+    unittest.main()

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_760_mcp_renewal_path.py

@@ -0,0 +1,342 @@
+"""MCP-level exact-owner lease renewal through ``gitea_lock_issue`` (#760).
+
+The unit suite in ``test_issue_760_exact_owner_lease_renewal`` proves the
+renewal *disposition*. It cannot prove the disposition survives the rest of the
+tool, and it did not: the waiver was computed and then discarded before
+``assess_issue_lock_worktree``, so every real renewal still failed on
+base-equivalence. A branch being renewed always carries committed work, so it is
+never base-equivalent by construction — exactly the argument #753 already makes
+for recovery.
+
+These tests drive the public tool end to end against a real git repository and a
+real durable lock file, composing every gate in the production order.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from mutation_profile_fixture import shared_mutation_env  # noqa: E402
+
+import issue_lock_provenance  # noqa: E402
+import issue_lock_store  # noqa: E402
+import mcp_server  # noqa: E402
+
+ISSUE = 9760
+BRANCH = f"fix/issue-{ISSUE}-renewal-mcp"
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+
+
+def _past_ts(hours: int = 1) -> str:
+    return (
+        (datetime.now(timezone.utc) - timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+class _RenewalMcpBase(unittest.TestCase):
+    """Real git repo + durable expired lock owned by a live PID.
+
+    The recorded PID is ``os.getpid()`` — unambiguously alive. That is the whole
+    point of #760: the PID belongs to the long-lived MCP daemon, so its liveness
+    says nothing about whether the authoring task still holds the work.
+    """
+
+    def setUp(self):
+        self.lock_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.lock_dir.cleanup)
+        self.repo = tempfile.mkdtemp(prefix="issue760-mcp-")
+        self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.repo], check=False))
+        self._init_worktree()
+        self.remotes = patch.dict(
+            mcp_server.REMOTES,
+            {"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
+        )
+        self.remotes.start()
+        self.addCleanup(patch.stopall)
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def _git(self, *args):
+        return subprocess.run(
+            ["git", "-C", self.repo, *args],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+
+    def _init_worktree(self):
+        self._git("init", "-q", "-b", "master")
+        self._git("config", "user.email", "[email protected]")
+        self._git("config", "user.name", "Test")
+        with open(os.path.join(self.repo, "seed.txt"), "w") as fh:
+            fh.write("seed\n")
+        self._git("add", "seed.txt")
+        self._git("commit", "-q", "-m", "seed")
+        self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
+        # The branch carries committed work, so it is NOT base-equivalent.
+        self._git("checkout", "-q", "-b", BRANCH)
+        with open(os.path.join(self.repo, "work.txt"), "w") as fh:
+            fh.write("author work\n")
+        self._git("add", "work.txt")
+        self._git("commit", "-q", "-m", "author work")
+        self.head_sha = self._git("rev-parse", "HEAD").stdout.strip()
+        self.worktree = os.path.realpath(self.repo)
+
+    def write_expired_lock(self, **overrides):
+        path = issue_lock_store.lock_file_path(
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+        claimant = {"username": IDENTITY, "profile": PROFILE}
+        pid = overrides.pop("session_pid", os.getpid())
+        overrides.pop("pid", None)
+        lease_overrides = overrides.pop("work_lease", {})
+        data = {
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "remote": "prgs",
+            "org": ORG,
+            "repo": REPO,
+            "worktree_path": self.worktree,
+            "session_pid": pid,
+            "pid": pid,
+            "lock_generation": 3,
+            "work_lease": {
+                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+                "issue_number": ISSUE,
+                "pr_number": None,
+                "branch": BRANCH,
+                "worktree_path": self.worktree,
+                "claimant": claimant,
+                "created_at": _past_ts(5),
+                "last_heartbeat_at": _past_ts(5),
+                "expires_at": _past_ts(),          # already expired
+            },
+            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                tool="gitea_lock_issue",
+                claimant=claimant,
+            ),
+        }
+        data["work_lease"].update(lease_overrides)
+        data.update(overrides)
+        data["session_pid"] = pid
+        data["pid"] = pid
+        data["lock_file_path"] = path
+        issue_lock_store.save_lock_file(path, data)
+        return path
+
+    def _tool_env(self):
+        env = shared_mutation_env(
+            PROFILE,
+            include_example_repo=True,
+            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
+        )
+        env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+        return env
+
+    def _git_state(self, *, porcelain="", branch=BRANCH, head=None):
+        return {
+            "current_branch": branch,
+            "porcelain_status": porcelain,
+            # The decisive fact: a branch carrying work is never base-equivalent.
+            "base_equivalent": False,
+            "head_sha": head or self.head_sha,
+            "inspected_git_root": self.worktree,
+            "base_branch": "master",
+        }
+
+    def run_lock_issue(
+        self,
+        *,
+        branch_entries=None,
+        open_prs=None,
+        git_state=None,
+        identity=IDENTITY,
+        profile=PROFILE,
+    ):
+        """Drive the public tool for the published exact-owner renewal shape."""
+        if branch_entries is None:
+            branch_entries = [{"name": BRANCH, "commit": {"id": self.head_sha}}]
+        if open_prs is None:
+            open_prs = [{"number": 4242, "head": {"ref": BRANCH, "sha": self.head_sha}}]
+        if git_state is None:
+            git_state = self._git_state()
+        env = self._tool_env()
+        with patch(
+            "mcp_server.api_get_all", return_value=list(branch_entries)
+        ), patch(
+            "mcp_server._list_open_pulls", return_value=list(open_prs)
+        ), patch(
+            "mcp_server.get_auth_header", return_value="token x"
+        ), patch(
+            "mcp_server._work_lease_claimant",
+            return_value={"username": identity, "profile": profile},
+        ), patch(
+            "mcp_server.issue_lock_worktree.read_worktree_git_state",
+            return_value=git_state,
+        ), patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            side_effect=lambda h, o, r, auth, issue_number: (
+                list(open_prs),
+                [b.get("name") for b in branch_entries if isinstance(b, dict)],
+                {"status": "not_claimed"},
+            ),
+        ), patch.dict(os.environ, env, clear=True):
+            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+            return mcp_server.gitea_lock_issue(
+                issue_number=ISSUE,
+                branch_name=BRANCH,
+                remote="prgs",
+                worktree_path=self.worktree,
+            )
+
+
+class TestRenewalReachableThroughTool(_RenewalMcpBase):
+    """F1: the sanctioned renewal must survive every downstream gate."""
+
+    def test_expired_lease_live_pid_exact_owner_renews_through_the_tool(self):
+        prior = issue_lock_store.read_lock_file(self.write_expired_lock())
+        self.assertTrue(issue_lock_store.is_lease_expired(prior))
+        self.assertTrue(issue_lock_store.is_process_alive(prior["session_pid"]))
+
+        result = self.run_lock_issue()
+
+        self.assertTrue(result["success"], result)
+        self.assertEqual(result["issue_number"], ISSUE)
+        self.assertEqual(result["branch_name"], BRANCH)
+        # The renewal is reported natively, so no lock-file inspection is needed.
+        self.assertIn("lease_renewal", result)
+        self.assertTrue(result["lease_renewal"]["renewed"])
+        self.assertIn("Renewed the expired", result["message"])
+
+    def test_renewed_lock_records_prior_and_replacement_evidence(self):
+        prior = issue_lock_store.read_lock_file(self.write_expired_lock())
+        prior_expiry = prior["work_lease"]["expires_at"]
+        prior_generation = issue_lock_store.lock_generation(prior)
+
+        result = self.run_lock_issue()
+        written = issue_lock_store.read_lock_file(result["lock_file_path"])
+
+        renewal = written["lease_renewal"]
+        self.assertTrue(renewal["renewed"])
+        self.assertEqual(renewal["prior_pid"], prior["session_pid"])
+        self.assertTrue(renewal["prior_pid_alive"])
+        self.assertEqual(renewal["prior_expires_at"], prior_expiry)
+        self.assertEqual(renewal["identity"], IDENTITY)
+        self.assertEqual(renewal["profile"], PROFILE)
+        self.assertEqual(renewal["head_sha"], self.head_sha)
+        self.assertTrue(renewal["proof"])
+        # New expiry is a fresh absolute stamp, later than the one it replaced.
+        self.assertEqual(renewal["new_expires_at"], written["work_lease"]["expires_at"])
+        self.assertGreater(renewal["new_expires_at"], prior_expiry)
+        # Compare-and-swap advanced the generation exactly once.
+        self.assertEqual(
+            issue_lock_store.lock_generation(written), prior_generation + 1
+        )
+
+    def test_renewed_lock_is_live_and_satisfies_mutation_ownership(self):
+        self.write_expired_lock()
+        result = self.run_lock_issue()
+        written = issue_lock_store.read_lock_file(result["lock_file_path"])
+
+        self.assertTrue(issue_lock_store.assess_lock_freshness(written)["live"])
+        verdict = issue_lock_store.verify_lock_for_mutation(
+            written,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.worktree,
+        )
+        self.assertTrue(verdict["proven"], verdict)
+        self.assertFalse(verdict["block"])
+
+    def test_recovery_record_is_not_written_for_a_live_owner_renewal(self):
+        """#753 recovery must not be claimed when the recorded PID is alive."""
+        self.write_expired_lock()
+        result = self.run_lock_issue()
+        written = issue_lock_store.read_lock_file(result["lock_file_path"])
+        self.assertNotIn("dead_session_recovery", written)
+
+
+class TestRenewalWaiverIsNarrow(_RenewalMcpBase):
+    """The waiver relaxes base-equivalence and nothing else."""
+
+    def test_dirty_worktree_still_blocks_a_would_be_renewal(self):
+        """Cleanliness is never waived; the renewal assessor refuses first.
+
+        A dirty worktree makes the renewal refuse, so no waiver is issued and
+        the lease-conflict gate fails closed ahead of the worktree gate. The
+        refusal names the uncommitted files, so the owner still learns why.
+        """
+        self.write_expired_lock()
+        with self.assertRaises(Exception) as ctx:
+            self.run_lock_issue(
+                git_state=self._git_state(porcelain=" M gitea_mcp_server.py\n")
+            )
+        message = str(ctx.exception)
+        self.assertIn("Recovery review is required before takeover", message)
+        self.assertIn("worktree has uncommitted tracked changes", message)
+        self.assertIn("gitea_mcp_server.py", message)
+
+    def test_foreign_claimant_cannot_use_the_waiver(self):
+        """A near-match owner gets no renewal and no base-equivalence waiver."""
+        self.write_expired_lock()
+        with self.assertRaises(Exception) as ctx:
+            self.run_lock_issue(identity="someone-else")
+        message = str(ctx.exception)
+        self.assertIn("Recovery review is required before takeover", message)
+        # The refusal names the missing ownership evidence (#760 diagnostics).
+        self.assertIn("does not match active identity", message)
+
+    def test_foreign_profile_cannot_use_the_waiver(self):
+        self.write_expired_lock()
+        with self.assertRaises(Exception) as ctx:
+            self.run_lock_issue(profile="other-author")
+        self.assertIn(
+            "Recovery review is required before takeover", str(ctx.exception)
+        )
+
+    def test_unpublished_branch_cannot_use_the_waiver(self):
+        """No remote head to agree with, so exact-owner renewal is refused."""
+        self.write_expired_lock()
+        with self.assertRaises(Exception) as ctx:
+            self.run_lock_issue(branch_entries=[], open_prs=[])
+        self.assertIn(
+            "Recovery review is required before takeover", str(ctx.exception)
+        )
+
+    def test_pr_head_mismatch_cannot_use_the_waiver(self):
+        self.write_expired_lock()
+        other = "9" * 40
+        with self.assertRaises(Exception) as ctx:
+            self.run_lock_issue(
+                open_prs=[{"number": 4242, "head": {"ref": BRANCH, "sha": other}}]
+            )
+        self.assertIn(
+            "Recovery review is required before takeover", str(ctx.exception)
+        )
+
+    def test_non_base_equivalent_branch_still_blocks_without_any_waiver(self):
+        """No durable lock at all: the ordinary base-equivalence rule applies."""
+        with self.assertRaises(Exception) as ctx:
+            self.run_lock_issue()
+        self.assertIn("must be base-equivalent", str(ctx.exception))
+
+
+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.