gitea_mcp_server.py

@@ -2273,7 +2273,14 @@ def _enforce_locked_issue_duplicate_recheck(
     org: str | None = None,
     repo: str | None = None,
 ) -> dict | None:
-    """Re-check duplicate-work gates for the locked issue (#400)."""
+    """Re-check duplicate-work gates for the locked issue (#400).
+
+    #768 AC2: when the lock being re-checked carries a sanctioned dead-session
+    recovery, carry that server-derived owning-PR evidence into the gate. The
+    commit and create-PR phases run in their own calls, long after the recovery
+    assessment ended, so without this they re-block the very PR the recovery
+    already proved belongs to this author.
+    """
     lock_data = _load_existing_issue_lock()
     if not lock_data:
         return None
@@ -2296,6 +2303,9 @@ def _enforce_locked_issue_duplicate_recheck(
         auth=auth,
         locked_branch=locked_branch,
         phase=phase,
+        recovered_owning_pr=issue_lock_recovery.recovered_owning_pr_from_lock(
+            lock_data
+        ),
     )
     if gate.get("block"):
         return gate
@@ -3402,6 +3412,24 @@ def gitea_lock_issue(
                 recovery_pr_number = pull.get("number")
                 break
         recovery_claimant = _work_lease_claimant(h)
+        # #768: the recovering author's clean worktree is, by construction, one
+        # commit ahead of the head recorded at lock time — committing is the
+        # only sanctioned way to clean it without discarding the work. Observe
+        # the ancestry here, server-side, so the assessor can tell an honest
+        # fast-forward from a rewritten head. Probed only when the heads
+        # actually differ, and never from any caller-supplied value.
+        recovery_local_head = git_state.get("head_sha")
+        recovery_ancestry: dict | None = None
+        if (
+            recovery_remote_head
+            and recovery_local_head
+            and recovery_remote_head != recovery_local_head
+        ):
+            recovery_ancestry = issue_lock_worktree.read_head_ancestry(
+                resolved_worktree,
+                ancestor_sha=recovery_remote_head,
+                descendant_sha=recovery_local_head,
+            )
         recovery_assessment = issue_lock_recovery.assess_dead_session_lock_recovery(
             existing_issue_lock,
             issue_number=issue_number,
@@ -3421,6 +3449,7 @@ def gitea_lock_issue(
             competing_live_locks=issue_lock_store.list_live_locks(),
             candidate_branches=recovery_candidates,
             current_pid=os.getpid(),
+            head_ancestry=recovery_ancestry,
         )
 
     recovery_sanctioned = bool(
@@ -3628,6 +3657,15 @@ def gitea_assess_work_issue_duplicate(
         }
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
+    # #768 AC2: recovery evidence comes only from the durable lock this session
+    # already holds, and only when that lock is for the issue being assessed.
+    # Nothing here is reachable through this tool's parameters.
+    recovered_owning_pr = None
+    lock_data = _load_existing_issue_lock()
+    if lock_data and int(lock_data.get("issue_number") or 0) == int(issue_number):
+        recovered_owning_pr = issue_lock_recovery.recovered_owning_pr_from_lock(
+            lock_data
+        )
     gate = _assess_issue_duplicate_gate(
         issue_number,
         h=h,
@@ -3636,6 +3674,7 @@ def gitea_assess_work_issue_duplicate(
         auth=auth,
         locked_branch=branch_name,
         phase=phase,
+        recovered_owning_pr=recovered_owning_pr,
     )
     return {"success": not gate.get("block"), **gate}
 
@@ -16115,6 +16154,16 @@ def _prove_author_ownership_for_pr(
             "durable lock, or branch lock (fail closed; issue_number need not "
             "equal pr_number when the tracking issue is linked)"
         )
+    # #768 AC2: when the lock that proved ownership was itself a sanctioned
+    # dead-session recovery, carry that server-derived evidence into the push
+    # gate rather than making it re-derive ownership blind. ``recorded_head``
+    # and ``accepted_head`` let the caller see that the PR head it is about to
+    # advance is the one recovery already sanctioned, not a foreign head.
+    recovered_owning_pr = None
+    if proven and lock_record:
+        candidate = issue_lock_recovery.recovered_owning_pr_from_lock(lock_record)
+        if candidate and int(candidate.get("pr_number") or 0) == int(pr_number):
+            recovered_owning_pr = candidate
     return {
         "proven": proven,
         "has_author_lock": proven,
@@ -16123,6 +16172,7 @@ def _prove_author_ownership_for_pr(
         "matched_issue": matched_issue,
         "matched_via": matched_via,
         "lock_record": lock_record,
+        "recovered_owning_pr": recovered_owning_pr,
         "reasons": reasons,
     }
 
@@ -16585,6 +16635,9 @@ def gitea_update_pr_branch_by_merge(
         "linked_issues": ownership.get("linked_issues"),
         "matched_issue": ownership.get("matched_issue"),
         "matched_via": ownership.get("matched_via"),
+        # #768 AC2: present only when a sanctioned dead-session recovery already
+        # proved this exact PR belongs to this author.
+        "recovered_owning_pr": ownership.get("recovered_owning_pr"),
     }
     preflight["host"] = h
     preflight["org"] = o

issue_lock_recovery.py

@@ -22,6 +22,18 @@ by the caller. It performs no mutation and no network I/O.
 Recovery deliberately does **not** relax base-equivalence for brand-new issue
 claims — only for a lock whose own prior record already proves the branch,
 worktree, head, and author.
+
+#768 extends the head requirement from strict equality to "equal, or a strict
+descendant". Equality alone made remediation after a session death unreachable:
+recovery needs a clean worktree, the only sanctioned way to clean one without
+discarding work is to commit, and committing advances the head past the value
+recorded at lock time. A commit that strictly descends from the recorded head,
+on the same branch, in the same worktree, by the same claimant, preserves
+everything equality protected — the recorded head is still reachable, still an
+ancestor, still unmodified — so it is accepted, and nothing else is. The
+descendant fact is observed server-side by
+``issue_lock_worktree.read_head_ancestry`` and handed in as ``head_ancestry``;
+no caller can assert it.
 """
 
 from __future__ import annotations
@@ -40,6 +52,10 @@ REFUSED = "REFUSED"
 # Durable fields a lock must carry before it can be considered at all.
 REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path")
 
+# How the clean local head relates to the head recorded at lock time (#768).
+HEAD_RELATION_EQUAL = "equal"
+HEAD_RELATION_STRICT_DESCENDANT = "strict_descendant"
+
 
 def _same_realpath(left: str | None, right: str | None) -> bool:
     if not left or not right:
@@ -87,6 +103,68 @@ def _malformed_reasons(lock: Mapping[str, Any]) -> list[str]:
     return missing
 
 
+def _assess_strict_descendant(
+    head_ancestry: Mapping[str, Any] | None,
+    *,
+    recorded_head: str,
+    local_head: str,
+) -> tuple[bool, list[str]]:
+    """Is ``local_head`` a proven strict descendant of ``recorded_head`` (#768)?
+
+    ``head_ancestry`` is the server-side git observation from
+    ``issue_lock_worktree.read_head_ancestry``. Its own ``ancestor_sha`` /
+    ``descendant_sha`` are re-checked against the heads this assessment is
+    actually reasoning about, so a probe taken for some other pair of commits —
+    stale, mismatched, or hand-built — can never authorize a waiver.
+
+    Returns ``(proven, notes)``. Notes name the exact missing element so a
+    refused caller sees why, never a bare "unproven".
+    """
+    if not isinstance(head_ancestry, Mapping):
+        return False, [
+            "no server-derived ancestry observation was available; a local head "
+            "that differs from the recorded head cannot be accepted"
+        ]
+
+    notes: list[str] = []
+    probe_ancestor = _text(head_ancestry.get("ancestor_sha"))
+    probe_descendant = _text(head_ancestry.get("descendant_sha"))
+    if probe_ancestor != recorded_head or probe_descendant != local_head:
+        return False, [
+            f"ancestry observation covers {probe_ancestor or 'unknown'} -> "
+            f"{probe_descendant or 'unknown'}, not the heads under assessment "
+            f"({recorded_head} -> {local_head})"
+        ]
+    if not head_ancestry.get("probe_ok"):
+        notes.extend(
+            list(head_ancestry.get("reasons") or [])
+            or ["ancestry probe did not complete; ancestry unproven"]
+        )
+        return False, notes
+    if not head_ancestry.get("ancestor_present"):
+        return False, [
+            f"recorded head {recorded_head} is no longer reachable; a rewritten "
+            "or force-moved head cannot be recovered"
+        ]
+    if not head_ancestry.get("is_strict_descendant"):
+        notes.extend(
+            list(head_ancestry.get("reasons") or [])
+            or [
+                f"local head {local_head} is not a strict descendant of the "
+                f"recorded head {recorded_head}"
+            ]
+        )
+        return False, notes
+
+    proof = _text(head_ancestry.get("proof")) or (
+        f"{recorded_head} is an ancestor of {local_head}"
+    )
+    return True, [
+        f"local head {local_head} strictly descends from recorded head "
+        f"{recorded_head} ({proof})"
+    ]
+
+
 def assess_dead_session_lock_recovery(
     existing_lock: Mapping[str, Any] | None,
     *,
@@ -107,6 +185,7 @@ def assess_dead_session_lock_recovery(
     competing_live_locks: Sequence[Mapping[str, Any]] | None = None,
     candidate_branches: Iterable[str] | None = None,
     current_pid: int | None = None,
+    head_ancestry: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Decide whether a dead-session author lock may be natively recovered.
 
@@ -218,31 +297,66 @@ def assess_dead_session_lock_recovery(
         )
     evidence["dirty_files"] = dirty_files
 
-    # ── Head agreement: local == remote == PR ───────────────────────────────
+    # ── Head agreement: local is the recorded head, or strictly descends it ──
+    # The recorded head is what the remote branch still carries. A local head
+    # equal to it is the #753 case. A local head that strictly descends from it
+    # is the #768 case: the author committed remediation, which is the only way
+    # to reach the clean worktree recovery itself demands.
     local_head = _text(head_sha)
     remote_head = _text(remote_head_sha)
+    head_relation: str | None = None
+    ancestry_proof: str | None = None
     if not local_head:
         reasons.append("local head SHA could not be determined")
     if not remote_head:
         reasons.append(
             f"remote head for branch '{locked_branch}' could not be determined"
         )
-    if local_head and remote_head and local_head != remote_head:
-        reasons.append(
-            f"local head {local_head} does not match remote branch head {remote_head}"
-        )
+    if local_head and remote_head:
+        if local_head == remote_head:
+            head_relation = HEAD_RELATION_EQUAL
+        else:
+            descends, notes = _assess_strict_descendant(
+                head_ancestry,
+                recorded_head=remote_head,
+                local_head=local_head,
+            )
+            if descends:
+                head_relation = HEAD_RELATION_STRICT_DESCENDANT
+                ancestry_proof = notes[0] if notes else None
+            else:
+                reasons.append(
+                    f"local head {local_head} does not match remote branch head "
+                    f"{remote_head}"
+                )
+                reasons.extend(notes)
     evidence["local_head"] = local_head or None
     evidence["remote_head"] = remote_head or None
+    # ``recorded_head`` is the head recovery is being measured against;
+    # ``accepted_head`` is the head this recovery actually adopts. They differ
+    # only in the descendant case, and downstream gates need both (#768 AC2/AC7).
+    evidence["recorded_head"] = remote_head or None
+    evidence["accepted_head"] = local_head or None
+    evidence["head_relation"] = head_relation
+    evidence["ancestry_proof"] = ancestry_proof
 
     pr_head = _text(pr_head_sha)
     if pr_head:
         evidence["pr_head"] = pr_head
         evidence["pr_number"] = pr_number
         if local_head and pr_head != local_head:
-            reasons.append(
-                f"open PR #{pr_number} head {pr_head} does not match local head "
-                f"{local_head}"
-            )
+            # A descendant recovery has not been published yet, so the open PR
+            # legitimately still points at the recorded head. Any other
+            # disagreement is a real mismatch.
+            if not (
+                head_relation == HEAD_RELATION_STRICT_DESCENDANT
+                and remote_head
+                and pr_head == remote_head
+            ):
+                reasons.append(
+                    f"open PR #{pr_number} head {pr_head} does not match local head "
+                    f"{local_head}"
+                )
 
     # ── Author identity ─────────────────────────────────────────────────────
     claimant = _lock_claimant(lock)
@@ -335,16 +449,14 @@ def assess_dead_session_lock_recovery(
     if reasons:
         return _result(REFUSED, False, reasons, evidence)
 
-    return _result(
-        RECOVERY_SANCTIONED,
-        True,
-        [
-            f"durable lock for issue #{issue_number} matches branch "
-            f"'{locked_branch}', worktree '{locked_worktree}', head {local_head}, "
-            f"and claimant '{locked_identity}'; recorded pid {recorded_pid} is dead"
-        ],
-        evidence,
-    )
+    proof = [
+        f"durable lock for issue #{issue_number} matches branch "
+        f"'{locked_branch}', worktree '{locked_worktree}', head {local_head}, "
+        f"and claimant '{locked_identity}'; recorded pid {recorded_pid} is dead"
+    ]
+    if head_relation == HEAD_RELATION_STRICT_DESCENDANT and ancestry_proof:
+        proof.append(ancestry_proof)
+    return _result(RECOVERY_SANCTIONED, True, proof, evidence)
 
 
 def _result(
@@ -374,10 +486,16 @@ def owning_pr_recovery_evidence(
     lock already owns" apart from "a competing duplicate PR".
 
     Returns ``None`` unless recovery was actually granted and the assessment's
-    own evidence names exactly one owning PR whose head agrees with the local
-    and remote heads. Nothing here is caller-supplied: every field is copied
-    from evidence the assessor built out of durable lock state plus live
+    own evidence names exactly one owning PR whose head agrees with the heads
+    the assessor accepted. Nothing here is caller-supplied: every field is
+    copied from evidence the assessor built out of durable lock state plus live
     git/Gitea observation, so a caller cannot manufacture an exemption.
+
+    #768: a descendant recovery carries two heads. ``head_sha`` stays the head
+    the open PR currently shows (the recorded head, since the remediation is not
+    published yet) and ``accepted_head`` is the local descendant that
+    publication will move it to. Downstream gates accept either, so the
+    exemption survives the very push it exists to permit.
     """
     if not isinstance(assessment, Mapping):
         return None
@@ -391,13 +509,28 @@ def owning_pr_recovery_evidence(
     pr_head = _text(evidence.get("pr_head"))
     local_head = _text(evidence.get("local_head"))
     remote_head = _text(evidence.get("remote_head"))
+    recorded_head = _text(evidence.get("recorded_head")) or remote_head
+    accepted_head = _text(evidence.get("accepted_head")) or local_head
+    relation = _text(evidence.get("head_relation")) or HEAD_RELATION_EQUAL
     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:
+    if relation == HEAD_RELATION_EQUAL:
+        if pr_head != local_head or pr_head != remote_head:
+            return None
+    elif relation == HEAD_RELATION_STRICT_DESCENDANT:
+        # The PR must still be at the recorded head, and the accepted head must
+        # actually be a different commit — otherwise this is not a descendant.
+        if not recorded_head or pr_head != recorded_head:
+            return None
+        if not accepted_head or accepted_head == recorded_head:
+            return None
+        if accepted_head != local_head:
+            return None
+    else:
         return None
     try:
         pr_number = int(raw_pr_number)
@@ -410,6 +543,77 @@ def owning_pr_recovery_evidence(
         "pr_number": pr_number,
         "branch_name": branch_name,
         "head_sha": pr_head,
+        "recorded_head": recorded_head or None,
+        "accepted_head": accepted_head or None,
+        "head_relation": relation,
+    }
+
+
+def recovered_owning_pr_from_lock(
+    lock_record: Mapping[str, Any] | None,
+) -> dict[str, Any] | None:
+    """Rebuild owning-PR recovery evidence from a persisted lock (#768 AC2).
+
+    ``gitea_lock_issue`` holds the live assessment only for the duration of the
+    lock call. The commit, push, create-PR, and duplicate-assessment gates run
+    later, in their own calls, and re-derive ownership from scratch — so an open
+    PR that recovery already proved belongs to this author reappears there as
+    competing duplicate work.
+
+    This reads the same proof back out of the durable ``dead_session_recovery``
+    block that only the server writes, on a lock the caller must already own.
+    It is a re-read of server-derived state, not a new assertion: a caller that
+    could forge this could equally forge the lock file itself, which every other
+    ownership gate already treats as authoritative.
+    """
+    if not isinstance(lock_record, Mapping):
+        return None
+    record = lock_record.get("dead_session_recovery")
+    if not isinstance(record, Mapping) or not record.get("recovered"):
+        return None
+
+    branch_name = _text(record.get("branch_name")) or _text(
+        lock_record.get("branch_name")
+    )
+    pr_head = _text(record.get("pr_head"))
+    recorded_head = _text(record.get("recorded_head")) or _text(
+        record.get("remote_head")
+    )
+    accepted_head = _text(record.get("accepted_head")) or _text(
+        record.get("local_head")
+    )
+    relation = _text(record.get("head_relation")) or HEAD_RELATION_EQUAL
+    raw_pr_number = record.get("pr_number")
+    raw_issue_number = lock_record.get("issue_number")
+
+    if raw_pr_number is None or raw_issue_number is None:
+        return None
+    if not branch_name or not pr_head:
+        return None
+    if relation == HEAD_RELATION_EQUAL:
+        if accepted_head and accepted_head != pr_head:
+            return None
+    elif relation == HEAD_RELATION_STRICT_DESCENDANT:
+        if not recorded_head or pr_head != recorded_head:
+            return None
+        if not accepted_head or accepted_head == recorded_head:
+            return None
+    else:
+        return None
+    try:
+        pr_number = int(raw_pr_number)
+        issue_number = int(raw_issue_number)
+    except (TypeError, ValueError):
+        return None
+
+    return {
+        "issue_number": issue_number,
+        "pr_number": pr_number,
+        "branch_name": branch_name,
+        "head_sha": pr_head,
+        "recorded_head": recorded_head or None,
+        "accepted_head": accepted_head or None,
+        "head_relation": relation,
     }
 
 
@@ -418,7 +622,13 @@ def build_recovery_record(
     *,
     recovered_at: str,
 ) -> dict[str, Any]:
-    """Durable, secret-free provenance for a completed recovery (#753 AC2/AC6)."""
+    """Durable, secret-free provenance for a completed recovery (#753 AC2/AC6).
+
+    #768 AC7: a granted recovery records, atomically with the lock itself, both
+    session identities, the head it was measured against, the head it adopted,
+    how those two relate, and the ancestry proof — so a descendant recovery can
+    be audited after the fact without re-running any probe.
+    """
     evidence = dict(assessment.get("evidence") or {})
     return {
         "recovered": True,
@@ -431,6 +641,10 @@ def build_recovery_record(
         "worktree_path": evidence.get("locked_worktree_path"),
         "local_head": evidence.get("local_head"),
         "remote_head": evidence.get("remote_head"),
+        "recorded_head": evidence.get("recorded_head"),
+        "accepted_head": evidence.get("accepted_head"),
+        "head_relation": evidence.get("head_relation"),
+        "ancestry_proof": evidence.get("ancestry_proof"),
         "pr_head": evidence.get("pr_head"),
         "pr_number": evidence.get("pr_number"),
         "identity": evidence.get("locked_identity"),

issue_lock_worktree.py

@@ -30,6 +30,108 @@ def resolve_author_worktree_path(
     return os.path.realpath(os.path.abspath(path))
 
 
+def read_head_ancestry(
+    worktree_path: str,
+    *,
+    ancestor_sha: str | None,
+    descendant_sha: str | None,
+) -> dict:
+    """Observe whether ``descendant_sha`` strictly descends from ``ancestor_sha`` (#768).
+
+    Server-side git observation for dead-session lock recovery. The recovering
+    author's only reachable clean-worktree state is one commit *ahead* of the
+    head recorded at lock time, so recovery needs to know whether that commit
+    extends the recorded head or replaces it.
+
+    Reports facts only; the disposition lives in ``issue_lock_recovery``. Every
+    field is read from git in the declared worktree — nothing here is supplied
+    by, or reachable from, an MCP caller (#768 AC6).
+
+    ``ancestor_present`` proves the recorded head is still reachable, which is
+    what separates an honest fast-forward from a rewritten or force-moved
+    history: a rewritten recorded head leaves the object graph and the probe
+    fails closed.
+    """
+    path = (worktree_path or "").strip()
+    ancestor = (ancestor_sha or "").strip()
+    descendant = (descendant_sha or "").strip()
+    result: dict = {
+        "ancestor_sha": ancestor or None,
+        "descendant_sha": descendant or None,
+        "probe_ok": False,
+        "ancestor_present": False,
+        "descendant_present": False,
+        "is_ancestor": False,
+        "is_strict_descendant": False,
+        "proof": None,
+        "reasons": [],
+    }
+    if not path or not ancestor or not descendant:
+        result["reasons"].append(
+            "ancestry probe requires a worktree path and both commit SHAs"
+        )
+        return result
+
+    def _present(sha: str) -> bool:
+        res = subprocess.run(
+            ["git", "-C", path, "rev-parse", "--verify", "--quiet", f"{sha}^{{commit}}"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        return res.returncode == 0
+
+    try:
+        result["ancestor_present"] = _present(ancestor)
+        result["descendant_present"] = _present(descendant)
+    except OSError as exc:  # git unavailable — fail closed, never assume
+        result["reasons"].append(f"ancestry probe could not run: {exc}")
+        return result
+
+    if not result["ancestor_present"]:
+        result["reasons"].append(
+            f"recorded head {ancestor} is not reachable in '{path}'; history may "
+            "have been rewritten or force-moved"
+        )
+    if not result["descendant_present"]:
+        result["reasons"].append(
+            f"local head {descendant} is not reachable in '{path}'"
+        )
+    if not (result["ancestor_present"] and result["descendant_present"]):
+        return result
+
+    probe = subprocess.run(
+        ["git", "-C", path, "merge-base", "--is-ancestor", ancestor, descendant],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    # 0 = is an ancestor, 1 = is not. Anything else is a failed probe, not a "no".
+    if probe.returncode not in (0, 1):
+        result["reasons"].append(
+            f"ancestry probe failed with exit {probe.returncode}; ancestry unproven"
+        )
+        return result
+
+    result["probe_ok"] = True
+    result["is_ancestor"] = probe.returncode == 0
+    result["is_strict_descendant"] = result["is_ancestor"] and ancestor != descendant
+    result["proof"] = (
+        f"git -C <worktree> merge-base --is-ancestor {ancestor} {descendant} "
+        f"-> exit {probe.returncode}"
+    )
+    if not result["is_ancestor"]:
+        result["reasons"].append(
+            f"local head {descendant} does not descend from recorded head {ancestor}"
+        )
+    elif not result["is_strict_descendant"]:
+        result["reasons"].append(
+            f"local head {descendant} equals the recorded head; no descendant "
+            "recovery is involved"
+        )
+    return result
+
+
 def read_worktree_git_state(
     worktree_path: str,
     extra_bases: tuple[str, ...] | list[str] = (),

issue_work_duplicate_gate.py

@@ -122,18 +122,30 @@ def _assess_owning_pr_exemption(
             f"the recovered branch '{token_branch}' (no owning-PR exemption)"
         )
         return False, notes
-    if not token_head or not only_sha or only_sha != token_head:
+    # #768: a descendant recovery is measured against the head the PR still
+    # shows, then publishes the local descendant — so the live PR head is the
+    # recorded head before that push and the accepted head after it. Both are
+    # server-derived and name the same owned PR, so both are accepted; anything
+    # else still fails closed.
+    token_accepted = str(recovered_owning_pr.get("accepted_head") or "").strip()
+    acceptable_heads = [head for head in (token_head, token_accepted) if head]
+    if not acceptable_heads or not only_sha or only_sha not in acceptable_heads:
         notes.append(
             f"open PR #{only_number} head {only_sha or 'unknown'} does not "
-            f"match the recovered head {token_head or 'unknown'} "
-            "(no owning-PR exemption)"
+            f"match the recovered head {token_head or 'unknown'}"
+            + (
+                f" or the accepted head {token_accepted}"
+                if token_accepted and token_accepted != token_head
+                else ""
+            )
+            + " (no owning-PR exemption)"
         )
         return False, notes
 
     return True, [
         f"open PR #{only_number} is the exact PR already owned by the "
         f"recovering lock for issue #{issue_number} (branch '{token_branch}', "
-        f"head {token_head}); not duplicate work"
+        f"head {only_sha}); not duplicate work"
     ]
 
 

tests/test_issue_755_owning_pr_recovery.py

@@ -73,12 +73,21 @@ def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE):
 def sanctioned_token(
     issue_number=ISSUE, pr_number=OWNING_PR, branch=BRANCH, head=HEAD
 ):
-    """The evidence shape the server derives from a granted recovery."""
+    """The evidence shape the server derives from a granted recovery.
+
+    #768 extends the token with recorded/accepted heads and the head relation
+    so a strict-descendant recovery can still exempt the owning PR after the
+    remediation commit lands. Exact-head recovery (#753/#755) reports equal
+    heads under the same shape.
+    """
     return {
         "issue_number": issue_number,
         "pr_number": pr_number,
         "branch_name": branch,
         "head_sha": head,
+        "recorded_head": head,
+        "accepted_head": head,
+        "head_relation": issue_lock_recovery.HEAD_RELATION_EQUAL,
     }
 
 

tests/test_issue_768_strict_descendant_recovery.py

@@ -0,0 +1,551 @@
+"""Strict-descendant dead-session recovery (#768).
+
+After a dead author session, a preserved clean remediation commit that strictly
+descends from the head recorded at lock time must be recoverable so the author
+can publish. Equality alone is still accepted (#753); every other divergence
+must keep failing closed.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_lock_recovery  # noqa: E402
+import issue_lock_store  # noqa: E402
+import issue_lock_worktree  # noqa: E402
+import issue_work_duplicate_gate  # noqa: E402
+
+ISSUE = 7680
+PR_NUMBER = 7681
+BRANCH = f"fix/issue-{ISSUE}-descendant-recovery"
+WORKTREE = "/scratch/wt-768"
+RECORDED = "a" * 40
+DESCENDANT = "c" * 40
+DIVERGED = "d" * 40
+BEHIND = "b" * 40
+IDENTITY = "example-user"
+PROFILE = "example-author"
+
+
+def dead_pid() -> int:
+    proc = subprocess.Popen([sys.executable, "-c", "pass"])
+    proc.wait()
+    return proc.pid
+
+
+def future_ts(hours: int = 4) -> str:
+    return (
+        (datetime.now(timezone.utc) + timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def make_lock(**overrides):
+    lock = {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "worktree_path": WORKTREE,
+        "remote": "prgs",
+        "org": "ExampleOrg",
+        "repo": "ExampleRepo",
+        "session_pid": dead_pid(),
+        "work_lease": {
+            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+            "issue_number": ISSUE,
+            "branch": BRANCH,
+            "worktree_path": WORKTREE,
+            "claimant": {"username": IDENTITY, "profile": PROFILE},
+            "expires_at": future_ts(),
+        },
+    }
+    lock.update(overrides)
+    return lock
+
+
+def ancestry_ok(
+    *,
+    ancestor: str = RECORDED,
+    descendant: str = DESCENDANT,
+    is_strict: bool = True,
+    probe_ok: bool = True,
+    ancestor_present: bool = True,
+    reasons: list[str] | None = None,
+) -> dict:
+    return {
+        "ancestor_sha": ancestor,
+        "descendant_sha": descendant,
+        "probe_ok": probe_ok,
+        "ancestor_present": ancestor_present,
+        "descendant_present": True,
+        "is_ancestor": is_strict or ancestor == descendant,
+        "is_strict_descendant": is_strict,
+        "proof": f"git merge-base --is-ancestor {ancestor} {descendant} -> exit 0",
+        "reasons": list(reasons or []),
+    }
+
+
+def assess(**overrides):
+    kwargs = {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "worktree_path": WORKTREE,
+        "remote": "prgs",
+        "org": "ExampleOrg",
+        "repo": "ExampleRepo",
+        "identity": IDENTITY,
+        "profile": PROFILE,
+        "current_branch": BRANCH,
+        "porcelain_status": "",
+        "head_sha": RECORDED,
+        "remote_head_sha": RECORDED,
+        "pr_head_sha": RECORDED,
+        "pr_number": PR_NUMBER,
+        "competing_live_locks": [],
+        "candidate_branches": [BRANCH],
+        "current_pid": os.getpid(),
+        "head_ancestry": None,
+    }
+    kwargs.update(overrides)
+    lock = kwargs.pop("lock", None)
+    return issue_lock_recovery.assess_dead_session_lock_recovery(
+        make_lock() if lock is None else lock, **kwargs
+    )
+
+
+class TestExactHeadRecoveryStillSucceeds(unittest.TestCase):
+    def test_equal_heads_still_sanctioned(self):
+        result = assess()
+        self.assertTrue(result["recovery_sanctioned"], result["reasons"])
+        self.assertEqual(
+            result["evidence"]["head_relation"],
+            issue_lock_recovery.HEAD_RELATION_EQUAL,
+        )
+        self.assertEqual(result["evidence"]["recorded_head"], RECORDED)
+        self.assertEqual(result["evidence"]["accepted_head"], RECORDED)
+
+    def test_exact_match_record_carries_relation(self):
+        record = issue_lock_recovery.build_recovery_record(
+            assess(), recovered_at="2026-07-20T00:00:00Z"
+        )
+        self.assertEqual(record["head_relation"], issue_lock_recovery.HEAD_RELATION_EQUAL)
+        self.assertEqual(record["recorded_head"], RECORDED)
+        self.assertEqual(record["accepted_head"], RECORDED)
+
+
+class TestStrictDescendantRecoverySucceeds(unittest.TestCase):
+    def test_clean_strict_descendant_recovers(self):
+        result = assess(
+            head_sha=DESCENDANT,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=ancestry_ok(),
+        )
+        self.assertTrue(result["recovery_sanctioned"], result["reasons"])
+        self.assertEqual(
+            result["evidence"]["head_relation"],
+            issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
+        )
+        self.assertEqual(result["evidence"]["recorded_head"], RECORDED)
+        self.assertEqual(result["evidence"]["accepted_head"], DESCENDANT)
+        self.assertIsNotNone(result["evidence"]["ancestry_proof"])
+        self.assertTrue(
+            any("strictly descends" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_pr_still_at_recorded_head_is_ok_for_descendant(self):
+        # Remediation is local only; open PR still points at the recorded head.
+        result = assess(
+            head_sha=DESCENDANT,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=ancestry_ok(),
+        )
+        self.assertTrue(result["recovery_sanctioned"], result["reasons"])
+
+    def test_recovery_record_names_both_heads_and_proof(self):
+        result = assess(
+            head_sha=DESCENDANT,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=ancestry_ok(),
+        )
+        record = issue_lock_recovery.build_recovery_record(
+            result, recovered_at="2026-07-20T00:00:00Z"
+        )
+        self.assertEqual(record["recorded_head"], RECORDED)
+        self.assertEqual(record["accepted_head"], DESCENDANT)
+        self.assertEqual(
+            record["head_relation"],
+            issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
+        )
+        self.assertIn("strictly descends", record["ancestry_proof"] or "")
+        self.assertEqual(record["prior_session_pid"], result["evidence"]["prior_session_pid"])
+        self.assertEqual(record["replacement_session_pid"], os.getpid())
+
+
+class TestDescendantEvidenceReachesPublicationGates(unittest.TestCase):
+    def _descendant_assessment(self):
+        return assess(
+            head_sha=DESCENDANT,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=ancestry_ok(),
+        )
+
+    def test_owning_pr_evidence_carries_accepted_head(self):
+        token = issue_lock_recovery.owning_pr_recovery_evidence(
+            self._descendant_assessment()
+        )
+        self.assertIsNotNone(token)
+        assert token is not None
+        self.assertEqual(token["head_sha"], RECORDED)
+        self.assertEqual(token["accepted_head"], DESCENDANT)
+        self.assertEqual(token["recorded_head"], RECORDED)
+        self.assertEqual(
+            token["head_relation"],
+            issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
+        )
+
+    def test_persisted_lock_rebuilds_owning_pr_evidence(self):
+        assessment = self._descendant_assessment()
+        record = issue_lock_recovery.build_recovery_record(
+            assessment, recovered_at="2026-07-20T00:00:00Z"
+        )
+        lock = make_lock(dead_session_recovery=record)
+        token = issue_lock_recovery.recovered_owning_pr_from_lock(lock)
+        self.assertIsNotNone(token)
+        assert token is not None
+        self.assertEqual(token["pr_number"], PR_NUMBER)
+        self.assertEqual(token["head_sha"], RECORDED)
+        self.assertEqual(token["accepted_head"], DESCENDANT)
+
+    def test_duplicate_gate_accepts_pr_at_recorded_or_accepted_head(self):
+        token = issue_lock_recovery.owning_pr_recovery_evidence(
+            self._descendant_assessment()
+        )
+        for live_sha in (RECORDED, DESCENDANT):
+            with self.subTest(live_sha=live_sha):
+                gate = issue_work_duplicate_gate.assess_work_issue_duplicate_gate(
+                    ISSUE,
+                    open_prs=[
+                        {
+                            "number": PR_NUMBER,
+                            "title": f"Closes #{ISSUE}",
+                            "body": f"Closes #{ISSUE}",
+                            "head": {"ref": BRANCH, "sha": live_sha},
+                        }
+                    ],
+                    branch_names=[BRANCH],
+                    claim_entry={"status": "unclaimed"},
+                    locked_branch=BRANCH,
+                    phase=issue_work_duplicate_gate.PHASE_COMMIT,
+                    recovered_owning_pr=token,
+                )
+                self.assertFalse(gate["block"], gate)
+                self.assertTrue(gate["owning_pr_recovery_exempted"])
+
+    def test_duplicate_gate_still_rejects_foreign_head(self):
+        token = issue_lock_recovery.owning_pr_recovery_evidence(
+            self._descendant_assessment()
+        )
+        gate = issue_work_duplicate_gate.assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[
+                {
+                    "number": PR_NUMBER,
+                    "title": f"Closes #{ISSUE}",
+                    "body": f"Closes #{ISSUE}",
+                    "head": {"ref": BRANCH, "sha": DIVERGED},
+                }
+            ],
+            branch_names=[BRANCH],
+            claim_entry={"status": "unclaimed"},
+            locked_branch=BRANCH,
+            phase=issue_work_duplicate_gate.PHASE_COMMIT,
+            recovered_owning_pr=token,
+        )
+        self.assertTrue(gate["block"])
+        self.assertFalse(gate["owning_pr_recovery_exempted"])
+
+
+class TestDirtyDescendantRejected(unittest.TestCase):
+    def test_dirty_descendant_refused(self):
+        result = assess(
+            head_sha=DESCENDANT,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=ancestry_ok(),
+            porcelain_status=" M issue_lock_recovery.py\n",
+        )
+        self.assertFalse(result["recovery_sanctioned"])
+        self.assertTrue(any("dirty" in r.lower() for r in result["reasons"]))
+
+
+class TestDivergedAndBehindRejected(unittest.TestCase):
+    def test_diverged_head_refused(self):
+        result = assess(
+            head_sha=DIVERGED,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=ancestry_ok(
+                ancestor=RECORDED,
+                descendant=DIVERGED,
+                is_strict=False,
+                reasons=[
+                    f"local head {DIVERGED} does not descend from recorded head "
+                    f"{RECORDED}"
+                ],
+            ),
+        )
+        self.assertFalse(result["recovery_sanctioned"])
+        self.assertIsNone(result["evidence"].get("head_relation"))
+        self.assertTrue(
+            any("does not match remote" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_local_behind_recorded_refused(self):
+        # merge-base --is-ancestor RECORDED BEHIND is false when BEHIND is ancestor.
+        result = assess(
+            head_sha=BEHIND,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=ancestry_ok(
+                ancestor=RECORDED,
+                descendant=BEHIND,
+                is_strict=False,
+                reasons=[
+                    f"local head {BEHIND} does not descend from recorded head "
+                    f"{RECORDED}"
+                ],
+            ),
+        )
+        self.assertFalse(result["recovery_sanctioned"])
+        self.assertTrue(
+            any("does not match remote" in r or "not a strict descendant" in r
+                for r in result["reasons"]),
+            result["reasons"],
+        )
+
+
+class TestUnrelatedAndMalformedAncestryRejected(unittest.TestCase):
+    def test_missing_ancestry_observation_fails_closed(self):
+        result = assess(
+            head_sha=DESCENDANT,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=None,
+        )
+        self.assertFalse(result["recovery_sanctioned"])
+        self.assertTrue(
+            any("ancestry" in r.lower() for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_mismatched_probe_pair_fails_closed(self):
+        # Observation for a different commit pair must not authorize this pair.
+        result = assess(
+            head_sha=DESCENDANT,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=ancestry_ok(ancestor=DIVERGED, descendant=DESCENDANT),
+        )
+        self.assertFalse(result["recovery_sanctioned"])
+        self.assertTrue(
+            any("not the heads under assessment" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_rewritten_recorded_head_fails_closed(self):
+        result = assess(
+            head_sha=DESCENDANT,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=ancestry_ok(ancestor_present=False, is_strict=False),
+        )
+        self.assertFalse(result["recovery_sanctioned"])
+        self.assertTrue(
+            any("no longer reachable" in r or "rewritten" in r
+                for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_failed_probe_fails_closed(self):
+        result = assess(
+            head_sha=DESCENDANT,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=ancestry_ok(
+                probe_ok=False,
+                is_strict=False,
+                reasons=["ancestry probe failed with exit 128; ancestry unproven"],
+            ),
+        )
+        self.assertFalse(result["recovery_sanctioned"])
+
+    def test_pr_head_not_equal_to_recorded_blocks_descendant(self):
+        result = assess(
+            head_sha=DESCENDANT,
+            remote_head_sha=RECORDED,
+            pr_head_sha=DIVERGED,
+            head_ancestry=ancestry_ok(),
+        )
+        self.assertFalse(result["recovery_sanctioned"])
+        self.assertTrue(
+            any("open PR" in r and "does not match" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+
+class TestLiveOwnerStillRejected(unittest.TestCase):
+    def test_live_prior_pid_refused_even_with_descendant_proof(self):
+        result = assess(
+            lock=make_lock(session_pid=os.getpid()),
+            head_sha=DESCENDANT,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=ancestry_ok(),
+        )
+        self.assertFalse(result["recovery_sanctioned"])
+        self.assertTrue(
+            any("still alive" in r or "live" in r.lower() for r in result["reasons"]),
+            result["reasons"],
+        )
+
+
+class TestDiagnosticsIdentifyDisposition(unittest.TestCase):
+    def test_equal_disposition_named(self):
+        result = assess()
+        self.assertEqual(
+            result["evidence"]["head_relation"],
+            issue_lock_recovery.HEAD_RELATION_EQUAL,
+        )
+
+    def test_descendant_disposition_named(self):
+        result = assess(
+            head_sha=DESCENDANT,
+            remote_head_sha=RECORDED,
+            pr_head_sha=RECORDED,
+            head_ancestry=ancestry_ok(),
+        )
+        self.assertEqual(
+            result["evidence"]["head_relation"],
+            issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
+        )
+
+    def test_rejected_divergence_has_no_accepted_relation(self):
+        result = assess(
+            head_sha=DIVERGED,
+            remote_head_sha=RECORDED,
+            head_ancestry=None,
+        )
+        self.assertIsNone(result["evidence"].get("head_relation"))
+        message = issue_lock_recovery.format_recovery_refusal(result)
+        self.assertIn("fail closed", message)
+        self.assertIn("does not match remote", message)
+
+
+class TestReadHeadAncestryRealGit(unittest.TestCase):
+    def _git(self, repo: str, *args: str) -> subprocess.CompletedProcess[str]:
+        return subprocess.run(
+            ["git", "-C", repo, *args],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+
+    def _init_repo_with_chain(self) -> tuple[str, str, str, str]:
+        """Return (repo, parent_sha, child_sha, sibling_sha)."""
+        repo = tempfile.mkdtemp(prefix="issue-768-ancestry-")
+        self._git(repo, "init")
+        self._git(repo, "config", "user.email", "[email protected]")
+        self._git(repo, "config", "user.name", "Test")
+        path = Path(repo) / "f.txt"
+        path.write_text("one\n")
+        self._git(repo, "add", "f.txt")
+        self._git(repo, "commit", "-m", "parent")
+        parent = self._git(repo, "rev-parse", "HEAD").stdout.strip()
+        path.write_text("two\n")
+        self._git(repo, "add", "f.txt")
+        self._git(repo, "commit", "-m", "child")
+        child = self._git(repo, "rev-parse", "HEAD").stdout.strip()
+        # Divergent sibling: branch from parent, then unique commit.
+        self._git(repo, "checkout", "-B", "side", parent)
+        path.write_text("side\n")
+        self._git(repo, "add", "f.txt")
+        self._git(repo, "commit", "-m", "sibling")
+        sibling = self._git(repo, "rev-parse", "HEAD").stdout.strip()
+        self._git(repo, "checkout", "-B", "main", child)
+        return repo, parent, child, sibling
+
+    def test_strict_descendant_observation(self):
+        repo, parent, child, _sibling = self._init_repo_with_chain()
+        obs = issue_lock_worktree.read_head_ancestry(
+            repo, ancestor_sha=parent, descendant_sha=child
+        )
+        self.assertTrue(obs["probe_ok"])
+        self.assertTrue(obs["ancestor_present"])
+        self.assertTrue(obs["is_ancestor"])
+        self.assertTrue(obs["is_strict_descendant"])
+        self.assertEqual(obs["ancestor_sha"], parent)
+        self.assertEqual(obs["descendant_sha"], child)
+
+    def test_equal_heads_not_strict_descendant(self):
+        repo, parent, _child, _sibling = self._init_repo_with_chain()
+        obs = issue_lock_worktree.read_head_ancestry(
+            repo, ancestor_sha=parent, descendant_sha=parent
+        )
+        self.assertTrue(obs["probe_ok"])
+        self.assertTrue(obs["is_ancestor"])
+        self.assertFalse(obs["is_strict_descendant"])
+
+    def test_diverged_not_ancestor(self):
+        repo, _parent, child, sibling = self._init_repo_with_chain()
+        # child and sibling share a parent but neither descends from the other.
+        obs = issue_lock_worktree.read_head_ancestry(
+            repo, ancestor_sha=child, descendant_sha=sibling
+        )
+        self.assertTrue(obs["probe_ok"])
+        self.assertFalse(obs["is_ancestor"])
+        self.assertFalse(obs["is_strict_descendant"])
+
+    def test_missing_sha_fails_closed(self):
+        repo, _parent, child, _ = self._init_repo_with_chain()
+        obs = issue_lock_worktree.read_head_ancestry(
+            repo, ancestor_sha="0" * 40, descendant_sha=child
+        )
+        self.assertFalse(obs["probe_ok"])
+        self.assertFalse(obs["ancestor_present"])
+
+    def test_end_to_end_real_git_descendant_recovery(self):
+        repo, parent, child, _sibling = self._init_repo_with_chain()
+        obs = issue_lock_worktree.read_head_ancestry(
+            repo, ancestor_sha=parent, descendant_sha=child
+        )
+        result = assess(
+            worktree_path=repo,
+            head_sha=child,
+            remote_head_sha=parent,
+            pr_head_sha=parent,
+            head_ancestry=obs,
+            lock=make_lock(worktree_path=repo),
+        )
+        self.assertTrue(result["recovery_sanctioned"], result["reasons"])
+        self.assertEqual(
+            result["evidence"]["head_relation"],
+            issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()