fix(lock): make the exact-owner renewal waiver survive downstream gates (#760)

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): "
@@ -12366,39 +12536,10 @@ def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
     return False
 
 
-def _git_default_remote_name(root: str) -> str:
-    """First configured git remote name for *root*, defaulting to 'origin'.
-
-    Used to resolve the live remote master target for parity (#610). Best
-    effort: any failure falls back to 'origin' so callers never raise.
-    """
-    try:
-        res = subprocess.run(
-            ["git", "-C", root, "remote"],
-            capture_output=True, text=True, check=False,
-        )
-    except Exception:
-        return "origin"
-    if res.returncode != 0:
-        return "origin"
-    names = [n.strip() for n in (res.stdout or "").splitlines() if n.strip()]
-    return names[0] if names else "origin"
-
-
 def _current_master_parity() -> dict:
-    """Assess this process's code against local and live remote master (#420/#610).
+    """Assess this process's code against the on-disk master HEAD (#420)."""
-
-    Compares the daemon's startup commit, the on-disk checkout HEAD, and the
-    live remote master target. A stale daemon relative to live master fails
-    closed for mutations even when the local checkout HEAD still matches the
-    startup commit. The live-remote read is best effort: an unresolved live
-    head leaves read-only diagnostics unblocked but is never mutation-safe.
-    """
     current_head = master_parity_gate.read_git_head(PROJECT_ROOT)
-    live_head = master_parity_gate.read_remote_master_head(
+    return master_parity_gate.assess_master_parity(_STARTUP_PARITY, current_head)
-        PROJECT_ROOT, remote=_git_default_remote_name(PROJECT_ROOT))
-    return master_parity_gate.assess_master_parity(
-        _STARTUP_PARITY, current_head, live_remote_head=live_head)
 
 
 def _current_runtime_mode_report(refresh: bool = False) -> dict:
@@ -16339,34 +16480,15 @@ def gitea_get_runtime_context(
         "restart_required": parity["restart_required"],
         "startup_head": parity["startup_head"],
         "current_head": parity["current_head"],
-        # #610 distinguished mutation-safety signals:
-        "daemon_start_head": parity["daemon_start_head"],
-        "local_head": parity["local_head"],
-        "live_remote_head": parity["live_remote_head"],
-        "live_known": parity["live_known"],
-        "live_stale": parity["live_stale"],
-        "mutation_safe": parity["mutation_safe"],
         "summary": master_parity_gate.format_parity(parity),
         "mutation_gate_enforced": not master_parity_gate.gate_disabled(),
-        # #610: the capability resolver is authoritative for mutation safety;
-        # local parity alone must never authorize a mutation.
-        "resolver_authoritative_for_mutation_safety": True,
     }
-    if parity["restart_required"] and not master_parity_gate.gate_disabled():
+    if parity["stale"] and not master_parity_gate.gate_disabled():
-        if parity["live_stale"]:
+        safe_next_action = (
-            safe_next_action = (
+            "Server code is stale relative to master; restart the Gitea MCP "
-                "Daemon is stale relative to LIVE remote master "
+            "server to load current capability gates before mutating. "
-                f"(started {parity['startup_head'][:12] if parity['startup_head'] else 'unknown'}, "
+            f"({master_parity_gate.format_parity(parity)})"
-                f"live master {parity['live_remote_head'][:12] if parity['live_remote_head'] else 'unknown'}); "
+        )
-                "restart/reconnect the Gitea MCP server before mutating. The "
-                "capability resolver is authoritative for mutation safety."
-            )
-        else:
-            safe_next_action = (
-                "Server code is stale relative to master; restart the Gitea MCP "
-                "server to load current capability gates before mutating. "
-                f"({master_parity_gate.format_parity(parity)})"
-            )
         result["safe_next_action"] = safe_next_action
 
     if reveal and h:
@@ -16415,13 +16537,6 @@ def gitea_assess_master_parity(
         "determinable": parity["determinable"],
         "startup_head": parity["startup_head"],
         "current_head": parity["current_head"],
-        # #610 distinguished mutation-safety signals:
-        "daemon_start_head": parity["daemon_start_head"],
-        "local_head": parity["local_head"],
-        "live_remote_head": parity["live_remote_head"],
-        "live_known": parity["live_known"],
-        "live_stale": parity["live_stale"],
-        "mutation_safe": parity["mutation_safe"],
         "mutation_gate_enforced": enforced,
         "summary": master_parity_gate.format_parity(parity),
         "reasons": parity["reasons"],
@@ -16438,7 +16553,7 @@ def gitea_assess_master_parity(
             source=canonical_source,
         ),
     }
-    if parity["restart_required"] and enforced:
+    if parity["stale"] and enforced:
         out["report"] = master_parity_gate.parity_report(parity)
     return out
 

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),
     }
 
 

master_parity_gate.py

@@ -24,26 +24,12 @@ from __future__ import annotations
 
 import os
 import subprocess
-import time
-
-# Live-remote head cache: the parity gate runs on every mutation and every
-# runtime-context read, so the ``git ls-remote`` result is cached briefly to
-# avoid a network round-trip per call (#610). Keyed by (root, remote, branch).
-_REMOTE_HEAD_CACHE: dict[tuple[str, str, str], tuple[float, str | None]] = {}
-_REMOTE_HEAD_TTL = 60.0
-
-
-def _clear_remote_head_cache() -> None:
-    """Reset the live-remote head cache (test isolation / forced refresh)."""
-    _REMOTE_HEAD_CACHE.clear()
 
 # Environment escape hatches (ops + tests):
 #   GITEA_MCP_DISABLE_PARITY_GATE  -> disable enforcement entirely (fail open).
 #   GITEA_TEST_CURRENT_HEAD        -> force the "current" HEAD read, for tests.
 ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE"
 ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD"
-# GITEA_TEST_LIVE_REMOTE_HEAD -> force the live remote master read, for tests.
-ENV_TEST_LIVE_REMOTE_HEAD = "GITEA_TEST_LIVE_REMOTE_HEAD"
 
 
 def read_git_head(root: str) -> str | None:
@@ -72,57 +58,6 @@ def read_git_head(root: str) -> str | None:
     return (res.stdout or "").strip() or None
 
 
-def read_remote_master_head(
-    root: str,
-    remote: str = "origin",
-    branch: str = "master",
-    ttl: float = _REMOTE_HEAD_TTL,
-) -> str | None:
-    """Return the live remote ``branch`` commit SHA, or ``None`` (#610).
-
-    Resolves the *live* target commit via ``git ls-remote`` so parity can tell
-    a daemon that is behind the live remote master apart from one whose local
-    checkout simply hasn't been pulled. ``None`` means the live head could not
-    be resolved (offline, no such remote, git unavailable, error) -- callers
-    must treat unknown live state as *not mutation-safe* while never blocking
-    read-only diagnostics. A ``GITEA_TEST_LIVE_REMOTE_HEAD`` override takes
-    precedence so the wiring can be exercised deterministically and offline.
-
-    The result is cached for *ttl* seconds per (root, remote, branch) so the
-    gate does not run a network probe on every mutation/read (``ttl=0`` forces
-    a live probe). Both hits and ``None`` misses are cached to bound offline
-    latency; the env override bypasses the cache and the subprocess entirely.
-    """
-    forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD)
-    if forced is not None:
-        return forced.strip() or None
-    if not root:
-        return None
-    key = (root, remote, branch)
-    now = time.monotonic()
-    if ttl > 0:
-        cached = _REMOTE_HEAD_CACHE.get(key)
-        if cached is not None and (now - cached[0]) < ttl:
-            return cached[1]
-    sha: str | None = None
-    try:
-        res = subprocess.run(
-            ["git", "-C", root, "ls-remote", remote, f"refs/heads/{branch}"],
-            capture_output=True,
-            text=True,
-            check=False,
-            timeout=5,
-        )
-        if res.returncode == 0:
-            lines = (res.stdout or "").strip().splitlines()
-            if lines:
-                sha = lines[0].split("\t", 1)[0].split()[0].strip() or None
-    except Exception:
-        sha = None
-    _REMOTE_HEAD_CACHE[key] = (now, sha)
-    return sha
-
-
 def capture_startup_parity(root: str, head: str | None = None) -> dict:
     """Capture the process source-tree baseline once at server startup.
 
@@ -137,38 +72,18 @@ def _short(sha: str | None) -> str:
     return sha[:12] if sha else "unknown"
 
 
-def assess_master_parity(
+def assess_master_parity(startup: dict | None, current_head: str | None) -> dict:
-    startup: dict | None,
-    current_head: str | None,
-    live_remote_head: str | None = None,
-) -> dict:
     """Compare the startup baseline against the current on-disk ``HEAD``.
 
-    Pure: all HEADs are supplied by the caller. Returns a structured result:
+    Pure: both HEADs are supplied by the caller. Returns a structured result:
 
     - ``in_parity``    -- server code matches the on-disk master (or parity
       could not be determined, which is not treated as stale).
     - ``stale``        -- the on-disk master has definitively advanced past the
       running process.
-    - ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
+    - ``restart_required`` -- alias of ``stale``; the recovery action.
-    - ``determinable`` -- whether both local HEADs were known well enough to
+    - ``determinable`` -- whether both HEADs were known well enough to compare.
-      compare.
     - ``startup_head`` / ``current_head`` / ``reasons``.
-
-    #610 adds live-remote awareness so a daemon that is stale relative to the
-    *live* remote master cannot report a mutation-safe result even when the
-    local checkout HEAD still matches the daemon's startup commit:
-
-    - ``daemon_start_head`` -- the commit the running process started at
-      (alias of ``startup_head``, named for clarity in reports).
-    - ``local_head`` -- the on-disk checkout HEAD (alias of ``current_head``).
-    - ``live_remote_head`` -- the live remote target commit, or ``None`` when it
-      could not be fetched.
-    - ``live_known`` -- whether the live remote target was resolved.
-    - ``live_stale`` -- the live remote master has advanced past the running
-      process (daemon is behind live master) even if local parity is green.
-    - ``mutation_safe`` -- the daemon code, local checkout, and live remote
-      target all agree; the only state in which a mutation may rely on parity.
     """
     startup_head = (startup or {}).get("startup_head")
     reasons: list[str] = []
@@ -176,56 +91,32 @@ def assess_master_parity(
     if startup_head is None:
         reasons.append(
             "startup commit was not captured; code parity cannot be enforced")
-        return _result(True, False, False, startup_head, current_head,
+        return _result(True, False, False, startup_head, current_head, reasons)
-                       live_remote_head, False, reasons)
 
     if current_head is None:
         reasons.append(
             "current workspace HEAD could not be read; code parity cannot be "
             "enforced")
-        return _result(True, False, False, startup_head, current_head,
+        return _result(True, False, False, startup_head, current_head, reasons)
-                       live_remote_head, False, reasons)
 
-    local_in_parity = startup_head == current_head
+    if startup_head == current_head:
-    local_stale = not local_in_parity
+        return _result(True, False, True, startup_head, current_head, reasons)
-    if local_stale:
-        reasons.append(
-            f"MCP server started at commit {_short(startup_head)} but the "
-            f"workspace master is now {_short(current_head)}; restart the "
-            f"server to load the current capability gates")
 
-    live_known = live_remote_head is not None
+    reasons.append(
-    live_stale = live_known and live_remote_head != startup_head
+        f"MCP server started at commit {_short(startup_head)} but the workspace "
-    if live_stale:
+        f"master is now {_short(current_head)}; restart the server to load the "
-        reasons.append(
+        f"current capability gates")
-            f"live remote master is {_short(live_remote_head)} but the MCP "
+    return _result(False, True, True, startup_head, current_head, reasons)
-            f"server started at {_short(startup_head)}; the daemon is stale "
-            f"relative to live master -- restart/reconnect before mutating")
-
-    return _result(
-        local_in_parity, local_stale, True, startup_head, current_head,
-        live_remote_head, live_stale, reasons)
 
 
-def _result(in_parity, stale, determinable, startup_head, current_head,
+def _result(in_parity, stale, determinable, startup_head, current_head, reasons):
-            live_remote_head, live_stale, reasons):
-    live_known = live_remote_head is not None
-    mutation_safe = (
-        determinable and in_parity and live_known and not live_stale)
     return {
         "in_parity": in_parity,
         "stale": stale,
-        "restart_required": stale or live_stale,
+        "restart_required": stale,
         "determinable": determinable,
         "startup_head": startup_head,
         "current_head": current_head,
-        # #610 distinguished signals:
-        "daemon_start_head": startup_head,
-        "local_head": current_head,
-        "live_remote_head": live_remote_head,
-        "live_known": live_known,
-        "live_stale": live_stale,
-        "mutation_safe": mutation_safe,
         "reasons": list(reasons),
     }
 
@@ -239,13 +130,11 @@ def parity_block_reasons(assessment: dict) -> list[str]:
     """Block reasons for a mutation gate (empty when the mutation may proceed).
 
     A disabled gate or an in-parity / non-determinable assessment yields no
-    reasons. A definitively stale server blocks, and (#610) a daemon that is
+    reasons; only a definitively stale server blocks.
-    stale relative to the *live* remote master blocks even when the local
-    checkout HEAD still matches the daemon's startup commit.
     """
     if gate_disabled():
         return []
-    if assessment.get("stale") or assessment.get("live_stale"):
+    if assessment.get("stale"):
         return list(assessment.get("reasons") or
                     ["server code is stale relative to master (fail closed)"])
     return []
@@ -258,10 +147,6 @@ def parity_report(assessment: dict) -> dict:
         "restart_required": True,
         "startup_head": assessment.get("startup_head"),
         "current_head": assessment.get("current_head"),
-        # #610: name the live remote target so the report distinguishes a
-        # local-code stale from a daemon-behind-live-master stale.
-        "live_remote_head": assessment.get("live_remote_head"),
-        "live_stale": bool(assessment.get("live_stale")),
         "reasons": list(assessment.get("reasons") or []),
         "recovery": [
             "The running MCP server is executing code older than the current "
@@ -272,45 +157,6 @@ def parity_report(assessment: dict) -> dict:
     }
 
 
-def parity_resolver_disagreement(
-    assessment: dict,
-    resolver_restart_required: bool,
-) -> dict | None:
-    """Typed blocker when the resolver requires restart but parity looks green.
-
-    The capability resolver (``gitea_resolve_task_capability``) detects stale
-    runtime authoritatively for mutation safety (#610). When it requires a
-    restart, local-only parity must never override it: this returns a typed,
-    fail-closed blocker that names the resolver as authoritative. Returns
-    ``None`` when the resolver does not require a restart.
-    """
-    if not resolver_restart_required:
-        return None
-    parity_optimistic = bool(assessment.get("in_parity")) and not (
-        assessment.get("stale") or assessment.get("live_stale"))
-    return {
-        "kind": "parity_resolver_disagreement",
-        "restart_required": True,
-        "resolver_authoritative": True,
-        "parity_optimistic": parity_optimistic,
-        "daemon_start_head": assessment.get("daemon_start_head"),
-        "local_head": assessment.get("local_head"),
-        "live_remote_head": assessment.get("live_remote_head"),
-        "reasons": [
-            "The capability resolver requires a restart/reconnect (stale "
-            "runtime) but master-parity reported local code as in-parity. "
-            "The resolver is authoritative for mutation safety; do not mutate "
-            "on local parity alone. Restart/reconnect the Gitea MCP server "
-            "and re-verify before mutating.",
-        ],
-        "recovery": [
-            "Trust the resolver: treat this session as stale.",
-            "Restart or /mcp reconnect the Gitea MCP namespace so it reloads "
-            "current master and live target state, then re-run preflight.",
-        ],
-    }
-
-
 def format_parity(assessment: dict) -> str:
     """One-line human summary for logs / runtime context."""
     if assessment.get("stale"):

runtime_recovery_guard.py

@@ -49,7 +49,7 @@ from __future__ import annotations
 
 import os
 import re
-from typing import Any, Iterable
+from typing import Any, Iterable, Iterator
 
 # Single source of truth for both the redactor and the gated-mutation set: the
 # #671 guard already owns them, so the two contamination models can never drift
@@ -86,8 +86,26 @@ REMEDIATION = (
 
 # Split a compound command line into simple commands on shell separators so
 # ``ps aux | grep mcp_server`` is analysed segment by segment and its harmless
-# inspection half never reaches the kill classifier.
+# inspection half never reaches the kill classifier. The background separator
-_SEGMENT_SPLIT_RE = re.compile(r"(?:\|\||&&|\||;|\n)")
+# ``&`` is a separator too: without it ``sleep 1 & pkill -f mcp_server.py`` was
+# a single segment whose command position held ``sleep``, so the kill was never
+# classified (#787).
+#
+# Splitting is *quote-aware*, and a regex alternation cannot express that, so
+# the scan below replaces the earlier ``_SEGMENT_SPLIT_RE`` pattern. A separator
+# only separates where it is syntactically active: outside single and double
+# quotes, and not backslash-escaped. Without that, adding ``&`` made every
+# benign mention of the canonical kill string classify as a real kill — a commit
+# message quoting ``sleep 1 & pkill -f mcp_server.py``, an ``echo`` of the same
+# sentence, a ``grep`` for it — and a false contamination marker fails review,
+# merge, close and completion mutations closed until a reconciler clears it (PR
+# #789 review finding F1). Quote-awareness is not specific to ``&``: it also
+# retires the same false-positive class that ``;`` and ``|`` carried before #787.
+_SEPARATOR_CHARS = frozenset("|&;\n")
+
+#: Two-character logical separators, consumed whole so ``&&`` and ``||`` are
+#: never split into single characters leaving a stray operator behind.
+_LOGICAL_SEPARATORS = ("&&", "||")
 
 _KILL_VERBS = frozenset({"kill", "pkill", "killall"})
 
@@ -133,8 +151,128 @@ def _clean(value: str | None) -> str:
     return (value or "").strip()
 
 
+def _iter_active(text: str) -> Iterator[tuple[int, str]]:
+    """Yield ``(index, char)`` for every *syntactically active* character.
+
+    Active means outside single and double quotes and not backslash-escaped —
+    the positions where a shell metacharacter actually carries its meaning.
+    Quoted runs, the quote characters themselves, and escaped characters are
+    skipped, so a separator written inside a commit message or a ``grep``
+    pattern is literal text rather than syntax. A backslash escapes nothing
+    inside single quotes, matching POSIX.
+
+    An unterminated quote swallows the rest of the line, exactly as it does for
+    the shell — which would reject such a command as a syntax error rather than
+    run its tail, so nothing executable hides behind it.
+    """
+    quote: str | None = None
+    index = 0
+    end = len(text)
+    while index < end:
+        char = text[index]
+        if quote == "'":
+            if char == "'":
+                quote = None
+            index += 1
+        elif quote == '"':
+            if char == "\\" and index + 1 < end:
+                index += 2
+            else:
+                if char == '"':
+                    quote = None
+                index += 1
+        elif char == "\\" and index + 1 < end:
+            index += 2
+        elif char in ("'", '"'):
+            quote = char
+            index += 1
+        else:
+            yield index, char
+            index += 1
+
+
+def _is_redirection(command: str, index: int, active: frozenset[int]) -> bool:
+    """Is the ``&``/``|`` at *index* part of a redirection, not a separator?
+
+    ``2>&1`` and ``>&2`` put the character immediately after a redirection
+    operator, and ``&>log`` immediately before one; in neither position does it
+    separate commands. Without this, ``a 2>&1`` split into ``['a 2>', '1']``
+    (PR #789 review finding F3).
+    """
+    previous = command[index - 1] if index else ""
+    if previous in ("<", ">") and (index - 1) in active:
+        return True
+    return (
+        command[index] == "&"
+        and command[index + 1:index + 2] == ">"
+        and (index + 1) in active
+    )
+
+
+def _closes_leading_paren(body: str) -> bool:
+    """Does *body* end with the active ``)`` matching a stripped leading ``(``?"""
+    if not body.endswith(")"):
+        return False
+    depth = 0
+    for index, char in _iter_active(body):
+        if char == "(":
+            depth += 1
+        elif char == ")":
+            if depth == 0:
+                return index == len(body) - 1
+            depth -= 1
+    return False
+
+
+def _strip_subshell(segment: str) -> str:
+    """Remove subshell wrappers so ``(pkill -f mcp_server.py)`` is classified.
+
+    The parentheses are shell syntax, not part of the simple command, so a
+    wrapped kill otherwise put ``(pkill`` in command position and never
+    reached the kill classifier (#787). Nested wrappers are unwrapped too.
+
+    A trailing ``)`` is removed only when it closes a leading ``(`` this call
+    stripped. Removing one unconditionally mangled balanced command
+    substitution — ``kill $(pgrep -f myapp)`` became ``kill $(pgrep -f myapp``
+    (PR #789 review finding F3). An unmatched leading ``(`` is still dropped on
+    its own, because splitting a wrapped compound orphans the opening half.
+    """
+    stripped = segment.strip()
+    while stripped.startswith("("):
+        body = stripped[1:].strip()
+        if _closes_leading_paren(body):
+            body = body[:-1].strip()
+        stripped = body
+    return stripped
+
+
 def _split_segments(command: str) -> list[str]:
-    return [seg for seg in _SEGMENT_SPLIT_RE.split(command) if seg.strip()]
+    """Split *command* into simple commands on syntactically active separators."""
+    active = frozenset(index for index, _ in _iter_active(command))
+    segments: list[str] = []
+    start = 0
+    index = 0
+    end = len(command)
+    while index < end:
+        char = command[index]
+        if (
+            char not in _SEPARATOR_CHARS
+            or index not in active
+            or (char in "&|" and _is_redirection(command, index, active))
+        ):
+            index += 1
+            continue
+        width = (
+            2
+            if command[index:index + 2] in _LOGICAL_SEPARATORS
+            and (index + 1) in active
+            else 1
+        )
+        segments.append(command[start:index])
+        index += width
+        start = index
+    segments.append(command[start:])
+    return [seg for seg in (_strip_subshell(seg) for seg in segments) if seg]
 
 
 def is_sanctioned_recovery(text: str | None) -> bool:

tests/test_issue_630_runtime_recovery_guard.py

@@ -91,6 +91,132 @@ def test_compound_command_detects_the_kill_half():
     assert result["contamination"] is True
 
 
+# ── #787: background separator and subshell forms reach the classifier ───────
+
+def test_background_separator_kill_is_contamination():
+    result = guard.classify_recovery_command("sleep 1 & pkill -f mcp_server.py")
+    assert result["process_kill"] is True
+    assert result["contamination"] is True
+    assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
+    assert result["ambiguous"] is False
+
+
+def test_subshell_wrapped_kill_is_contamination():
+    result = guard.classify_recovery_command("(pkill -f mcp_server.py)")
+    assert result["process_kill"] is True
+    assert result["contamination"] is True
+    assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
+    assert result["ambiguous"] is False
+
+
+def test_further_background_and_subshell_forms_are_contamination():
+    for command in (
+        "pkill -f mcp_server.py &",
+        "( sudo pkill -f mcp_server.py )",
+        "((pkill -f gitea_mcp_server))",
+        "sleep 1 & killall mcp_server",
+        "(ps aux | grep mcp_server) & pkill -f mcp_server.py",
+    ):
+        result = guard.classify_recovery_command(command)
+        assert result["contamination"] is True, command
+        assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL, command
+
+
+def test_logical_operators_are_not_split_into_single_characters():
+    # ``&&``/``||`` must still be consumed whole by the separator scan.
+    assert guard._split_segments("a && b || c") == ["a", "b", "c"]
+    assert guard._split_segments("a & b") == ["a", "b"]
+    assert guard._split_segments("(a)") == ["a"]
+    assert guard._split_segments("a; b\nc | d") == ["a", "b", "c", "d"]
+
+
+# ── #789 F1: separators only separate outside quoted or escaped text ─────────
+
+# The three commands the PR #789 review measured as regressions at head
+# 6b58f04: each merely *mentions* the canonical kill string inside quotes.
+F1_QUOTED_COMMANDS = (
+    'git commit -m "block sleep 1 & pkill -f mcp_server.py as recovery"',
+    'echo "docs: sleep 1 & pkill -f mcp_server.py is now detected"',
+    'grep -rn "sleep 1 & pkill -f mcp_server.py" docs/',
+)
+
+
+def test_quoted_ampersand_examples_from_review_f1_are_not_kills():
+    for command in F1_QUOTED_COMMANDS:
+        result = guard.classify_recovery_command(command)
+        assert result["process_kill"] is False, command
+        assert result["contamination"] is False, command
+        assert result["reason_class"] is None, command
+
+
+def test_ampersand_inside_double_quotes_is_not_a_separator():
+    assert guard._split_segments('echo "a & b"') == ['echo "a & b"']
+    result = guard.classify_recovery_command(
+        'echo "restart it: sleep 1 & pkill -f mcp_server.py"'
+    )
+    assert result["process_kill"] is False
+    assert result["contamination"] is False
+
+
+def test_ampersand_inside_single_quotes_is_not_a_separator():
+    assert guard._split_segments("echo 'a & b'") == ["echo 'a & b'"]
+    result = guard.classify_recovery_command(
+        "git commit -m 'sleep 1 & pkill -f mcp_server.py stays quoted'"
+    )
+    assert result["process_kill"] is False
+    assert result["contamination"] is False
+
+
+def test_backslash_escaped_ampersand_is_not_a_separator():
+    command = r"echo a \& pkill -f mcp_server.py"
+    assert guard._split_segments(command) == [command]
+    result = guard.classify_recovery_command(command)
+    assert result["process_kill"] is False
+    assert result["contamination"] is False
+
+
+def test_backslash_does_not_escape_inside_single_quotes():
+    # POSIX: a backslash is literal inside single quotes, so the closing quote
+    # still closes and the following ``&`` is a genuinely active separator.
+    command = r"echo 'a\' & pkill -f mcp_server.py"
+    assert guard._split_segments(command) == [r"echo 'a\'", "pkill -f mcp_server.py"]
+    assert guard.classify_recovery_command(command)["contamination"] is True
+
+
+def test_quote_awareness_also_retires_the_pre_existing_semicolon_and_pipe_cases():
+    # ``;`` and ``|`` misclassified quoted text before #787 as well. The fix is
+    # the quote-unawareness, not the ``&`` instance the issue happens to name.
+    for command in (
+        'git commit -m "fix; pkill -f mcp_server.py"',
+        'git commit -m "fix | pkill -f mcp_server.py"',
+    ):
+        result = guard.classify_recovery_command(command)
+        assert result["process_kill"] is False, command
+        assert result["contamination"] is False, command
+
+
+# ── #789 F3: subshell stripping and redirection stay syntactically honest ────
+
+def test_command_substitution_is_not_mangled_by_subshell_stripping():
+    # Only a wrapper this call opened may be unwrapped; a ``)`` closing ``$(``
+    # must survive intact.
+    assert guard._strip_subshell("kill $(pgrep -f myapp)") == "kill $(pgrep -f myapp)"
+    result = guard.classify_recovery_command("kill $(pgrep -f myapp)")
+    assert result["contamination"] is False
+    assert result["ambiguous"] is True
+
+
+def test_redirection_is_not_treated_as_a_background_separator():
+    assert guard._split_segments("a 2>&1") == ["a 2>&1"]
+    assert guard._split_segments("a &> log") == ["a &> log"]
+    assert guard._split_segments("pkill -f mcp_server.py 2>&1") == [
+        "pkill -f mcp_server.py 2>&1"
+    ]
+    result = guard.classify_recovery_command("pkill -f mcp_server.py 2>&1")
+    assert result["contamination"] is True
+    assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
+
+
 # ── no false positives ───────────────────────────────────────────────────────
 
 def test_read_only_inspection_is_not_a_kill():
@@ -112,6 +238,22 @@ def test_unrelated_pkill_target_is_not_contamination():
     assert result["ambiguous"] is False
 
 
+def test_user_scoped_pkill_of_unrelated_app_is_not_contamination():
+    # ``-u`` consumes ``mcpuser``; the surviving operand names no daemon (#787).
+    result = guard.classify_recovery_command("pkill -u mcpuser -f myapp")
+    assert result["process_kill"] is True
+    assert result["contamination"] is False
+    assert result["ambiguous"] is False
+
+
+def test_commit_message_quoting_the_kill_string_is_not_a_kill():
+    result = guard.classify_recovery_command(
+        'git commit -m "block pkill -f mcp_server.py as workflow recovery"'
+    )
+    assert result["process_kill"] is False
+    assert result["contamination"] is False
+
+
 def test_bare_kill_of_unknown_pid_is_ambiguous_not_contamination():
     result = guard.classify_recovery_command("kill 31337")
     assert result["contamination"] is False
@@ -333,6 +475,43 @@ def test_record_tool_marks_manual_daemon_kill():
     assert "mcp_server.py" in loaded["command_summary"]
 
 
+def test_record_tool_marks_background_separator_kill():
+    _clear_marker()
+    res = srv.gitea_record_daemon_process_kill_attempt(
+        command="sleep 1 & pkill -f mcp_server.py", remote="prgs"
+    )
+    assert res["contaminated"] is True
+    assert res["marked"] is True
+    assert res["marker"]["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
+    loaded = srv._load_runtime_recovery_marker("prgs")
+    assert loaded is not None
+    assert "mcp_server.py" in loaded["command_summary"]
+
+
+def test_record_tool_marks_subshell_wrapped_kill():
+    _clear_marker()
+    res = srv.gitea_record_daemon_process_kill_attempt(
+        command="(pkill -f mcp_server.py)", remote="prgs"
+    )
+    assert res["contaminated"] is True
+    assert res["marked"] is True
+    assert res["marker"]["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
+    loaded = srv._load_runtime_recovery_marker("prgs")
+    assert loaded is not None
+    assert "mcp_server.py" in loaded["command_summary"]
+
+
+def test_record_tool_does_not_mark_a_quoted_mention_of_the_kill_string():
+    # The marker is what fails review/merge/close closed and only a reconciler
+    # may clear it, so a quoted mention must never create one (PR #789 F1).
+    for command in F1_QUOTED_COMMANDS:
+        _clear_marker()
+        res = srv.gitea_record_daemon_process_kill_attempt(command=command, remote="prgs")
+        assert res["contaminated"] is False, command
+        assert res["marked"] is False, command
+        assert srv._load_runtime_recovery_marker("prgs") is None, command
+
+
 def test_record_tool_marks_broad_sweep():
     _clear_marker()
     res = srv.gitea_record_daemon_process_kill_attempt(

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.

tests/test_master_parity_gate.py

@@ -78,95 +78,6 @@ class TestBlockReasonsAndReport(unittest.TestCase):
         self.assertTrue(report["recovery"])
 
 
-class TestLiveRemoteParity(unittest.TestCase):
-    """#610: parity must account for the live remote master, not just local.
-
-    The daemon can be stale relative to the live remote target while the local
-    checkout HEAD still matches the daemon's startup commit, so local parity
-    reports green even though a mutation would run against outdated code.
-    """
-
-    SHA_C = "c" * 40
-
-    def test_distinguishes_three_shas(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        self.assertEqual(res["daemon_start_head"], SHA_A)
-        self.assertEqual(res["local_head"], SHA_A)
-        self.assertEqual(res["live_remote_head"], SHA_B)
-
-    def test_mutation_safe_only_when_all_three_match(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_A)
-        self.assertTrue(res["mutation_safe"])
-        self.assertTrue(res["live_known"])
-        self.assertFalse(res["live_stale"])
-
-    def test_live_stale_when_remote_advanced_past_daemon(self):
-        # Local checkout still matches the daemon start (local parity green),
-        # but the live remote master has advanced -> daemon is live-stale.
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        self.assertTrue(res["in_parity"])  # local parity still green
-        self.assertTrue(res["live_stale"])
-        self.assertFalse(res["mutation_safe"])
-        self.assertTrue(any("live" in r.lower() for r in res["reasons"]))
-
-    def test_live_unknown_is_not_mutation_safe_but_not_stale(self):
-        # Non-goal: unfetchable live remote must not be treated as stale for
-        # read-only, but a mutation-safe claim fails closed.
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=None)
-        self.assertFalse(res["live_known"])
-        self.assertFalse(res["mutation_safe"])
-        self.assertFalse(res["live_stale"])
-        self.assertTrue(res["in_parity"])
-
-    def test_default_live_remote_preserves_legacy_shape(self):
-        # Callers that do not supply a live head keep the pre-#610 behavior:
-        # in-parity, not live-stale, no live-derived block.
-        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
-        self.assertFalse(res["live_stale"])
-        self.assertEqual(mp.parity_block_reasons(res), [])
-
-
-class TestLiveStaleBlockAndReport(unittest.TestCase):
-    """#610: live-staleness must block mutations and surface a typed blocker."""
-
-    def test_live_stale_produces_block_reasons(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        self.assertTrue(mp.parity_block_reasons(res))
-
-    def test_disable_env_suppresses_live_stale_block(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
-            self.assertEqual(mp.parity_block_reasons(res), [])
-
-    def test_resolver_disagreement_returns_typed_blocker(self):
-        # Parity says local-green, resolver says restart required -> disagreement
-        # is a typed, fail-closed blocker naming the resolver as authoritative.
-        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
-        blocker = mp.parity_resolver_disagreement(res, resolver_restart_required=True)
-        self.assertIsNotNone(blocker)
-        self.assertEqual(blocker["kind"], "parity_resolver_disagreement")
-        self.assertTrue(blocker["restart_required"])
-        self.assertTrue(blocker["resolver_authoritative"])
-
-    def test_no_disagreement_when_resolver_agrees(self):
-        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
-        self.assertIsNone(
-            mp.parity_resolver_disagreement(res, resolver_restart_required=False))
-
-    def test_live_stale_report_names_live_remote(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        report = mp.parity_report(res)
-        self.assertEqual(report["live_remote_head"], SHA_B)
-        self.assertTrue(report["restart_required"])
-
-
 class TestReadGitHead(unittest.TestCase):
     def test_test_override_takes_precedence(self):
         with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
@@ -184,78 +95,6 @@ class TestReadGitHead(unittest.TestCase):
             self.assertIsNone(mp.read_git_head(""))
 
 
-class TestReadRemoteMasterHead(unittest.TestCase):
-    """#610: live remote master head reader (env-overridable, fails to None)."""
-
-    def test_test_override_takes_precedence(self):
-        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
-            self.assertEqual(mp.read_remote_master_head("/nonexistent"), SHA_B)
-
-    def test_blank_override_is_none(self):
-        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: "   "}):
-            self.assertIsNone(mp.read_remote_master_head("/nonexistent"))
-
-    def test_unfetchable_remote_is_none(self):
-        # No override; a bogus root/remote must fail closed to None, never raise.
-        env = {k: v for k, v in os.environ.items()
-               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
-        with patch.dict(os.environ, env, clear=True):
-            self.assertIsNone(
-                mp.read_remote_master_head("/nonexistent", remote="nope"))
-
-
-class TestRemoteHeadCache(unittest.TestCase):
-    """#610: live remote reads are cached with a TTL to stay off the network.
-
-    The parity gate runs on every mutation and every runtime-context read, so an
-    unbounded ``git ls-remote`` per call would be a latency/flakiness regression.
-    """
-
-    def setUp(self):
-        mp._clear_remote_head_cache()
-        env = {k: v for k, v in os.environ.items()
-               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
-        self._env = patch.dict(os.environ, env, clear=True)
-        self._env.start()
-        self.addCleanup(self._env.stop)
-        self.addCleanup(mp._clear_remote_head_cache)
-
-    def _fake_run(self, sha):
-        class _R:
-            returncode = 0
-            stdout = f"{sha}\trefs/heads/master\n"
-        calls = {"n": 0}
-
-        def run(*args, **kwargs):
-            calls["n"] += 1
-            return _R()
-        return run, calls
-
-    def test_second_call_within_ttl_uses_cache(self):
-        run, calls = self._fake_run(SHA_B)
-        with patch.object(mp.subprocess, "run", run):
-            a = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
-            b = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
-        self.assertEqual(a, SHA_B)
-        self.assertEqual(b, SHA_B)
-        self.assertEqual(calls["n"], 1)
-
-    def test_zero_ttl_bypasses_cache(self):
-        run, calls = self._fake_run(SHA_B)
-        with patch.object(mp.subprocess, "run", run):
-            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
-            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
-        self.assertEqual(calls["n"], 2)
-
-    def test_env_override_never_touches_subprocess(self):
-        run, calls = self._fake_run(SHA_B)
-        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
-            with patch.object(mp.subprocess, "run", run):
-                self.assertEqual(
-                    mp.read_remote_master_head("/repo", remote="prgs"), SHA_A)
-        self.assertEqual(calls["n"], 0)
-
-
 class TestServerWiring(unittest.TestCase):
     """Integration with the gate choke point in the server namespace."""
 
@@ -266,13 +105,6 @@ class TestServerWiring(unittest.TestCase):
         self._saved = self.srv._STARTUP_PARITY
         self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
                                     "startup_head": SHA_A}
-        # Keep the live-remote read hermetic (no real ls-remote network call):
-        # default the live master to the daemon start so parity is fully green
-        # unless a test overrides the live head explicitly (#610).
-        self._live_patch = patch.dict(
-            os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A})
-        self._live_patch.start()
-        self.addCleanup(self._live_patch.stop)
 
     def tearDown(self):
         self.srv._STARTUP_PARITY = self._saved
@@ -315,36 +147,6 @@ class TestServerWiring(unittest.TestCase):
             self.assertTrue(out["in_parity"])
             self.assertNotIn("report", out)
 
-    # --- #610: live-remote wiring -------------------------------------------
-
-    def test_live_stale_blocks_mutation_though_local_green(self):
-        # Local checkout matches the daemon start (local parity green) but the
-        # live remote master has advanced -> mutations must fail closed.
-        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
-                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
-            self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
-            self.assertTrue(
-                self.srv._master_parity_block("gitea.pr.create"))
-
-    def test_assess_tool_exposes_three_distinct_shas(self):
-        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
-                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
-            out = self.srv.gitea_assess_master_parity(remote="prgs")
-            self.assertEqual(out["daemon_start_head"], SHA_A)
-            self.assertEqual(out["local_head"], SHA_A)
-            self.assertEqual(out["live_remote_head"], SHA_B)
-            self.assertTrue(out["live_stale"])
-            self.assertFalse(out["mutation_safe"])
-            self.assertIn("report", out)
-
-    def test_assess_tool_mutation_safe_when_all_three_match(self):
-        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
-                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
-            out = self.srv.gitea_assess_master_parity(remote="prgs")
-            self.assertTrue(out["mutation_safe"])
-            self.assertFalse(out["live_stale"])
-            self.assertNotIn("report", out)
-
 
 if __name__ == "__main__":
     unittest.main()