"""Dead-session author issue-lock recovery (#753).

A durable author issue lock records the PID of the MCP session that took it.
When that process exits, ``issue_lock_store.assess_lock_freshness`` classifies
the lock as ``stale`` (``live=False``) even while its lease is still within TTL,
so every ownership check that requires a *live* lock fails closed.

Re-taking the lock through ``gitea_lock_issue`` is unreachable for real work:
``issue_lock_worktree.assess_issue_lock_worktree`` demands the worktree be
base-equivalent to ``master``/``main``/``dev``, and a branch that already
carries commits is ahead of its base by construction. The existing
``assess_expired_lock_reclaim`` affordance does not apply either, because
``assess_same_issue_lease_conflict`` only consults it once the lease has
*expired* — a dead PID under an unexpired lease never reaches it.

This module is the pure evidence assessor for that one narrow case. It grants
recovery only when every element of durable ownership still matches exactly and
the recorded process is demonstrably dead. It never trusts caller assertions:
every field is compared against durable lock state or live observation supplied
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.

#772 adds the remaining uncovered quadrant: a claim that was never published at
all. Two recovery modes now exist, and they require different evidence because
they are answering the same question against different available facts:

``published_owning_pr``
    The branch exists on the remote. Ownership is proven by comparing the local
    head against the remote/PR head — equal (#753) or a strict descendant
    (#768). This is the pre-existing behavior and is unchanged.

``unpublished_claim``
    The branch is absent from the remote and no PR claims it, so there is no
    head to compare against; that absence is the defining fact, not a degraded
    published case. Ownership is instead proven by the durable lock record
    (issue, branch, worktree, claimant, profile, dead PID) plus the local HEAD
    strictly descending from the base the branch was cut from, observed
    server-side by ``issue_lock_worktree.read_recorded_base`` and re-checked
    through ``base_ancestry``.

They cannot share one head-comparison implementation: the published path's
comparison target does not exist in the unpublished case, and inventing one
(defaulting to the base, say) would silently weaken the published path from
"matches what was actually pushed" to "descends from some base". The modes are
therefore selected by observed publication state and never by a caller — and
critically, the absence of a remote head is never itself treated as permission:
every identity, profile, branch, worktree, cleanliness, liveness, and competing
-claim check still applies in full.
"""

from __future__ import annotations

import os
from typing import Any, Iterable, Mapping, Sequence

from issue_lock_store import is_process_alive
from reviewer_worktree import parse_dirty_tracked_files

# Outcome values
RECOVERY_SANCTIONED = "RECOVERY_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")

# How the clean local head relates to the head recorded at lock time (#768).
HEAD_RELATION_EQUAL = "equal"
HEAD_RELATION_STRICT_DESCENDANT = "strict_descendant"
# #772: an unpublished claim has no recorded head to compare against at all, so
# its head is measured against the base the branch was cut from instead.
HEAD_RELATION_DESCENDS_FROM_BASE = "descends_from_recorded_base"

# Which body of evidence a recovery was decided on (#772 AC10). These are not
# interchangeable: a published claim proves ownership against a remote/PR head,
# an unpublished one against the recorded base plus durable lock state. They
# cannot share a single head-comparison implementation because the unpublished
# case has no head to compare — that absence is the defining fact, not a
# degraded version of the published case.
RECOVERY_MODE_PUBLISHED_OWNING_PR = "published_owning_pr"
RECOVERY_MODE_UNPUBLISHED_CLAIM = "unpublished_claim"


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 _text(value: Any) -> str:
    return str(value or "").strip()


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 _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 _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_base_descendancy(
    base_ancestry: Mapping[str, Any] | None,
    *,
    recorded_base: str,
    local_head: str,
) -> tuple[bool, list[str]]:
    """Is ``local_head`` a proven strict descendant of ``recorded_base`` (#772)?

    The unpublished-claim analogue of ``_assess_strict_descendant``. The
    comparison target is the base the branch was cut from — observed server-side
    by ``issue_lock_worktree.read_recorded_base`` — rather than a remote or PR
    head, because an unpublished claim has neither.

    The probe's own endpoints are re-checked against the values under
    assessment, so an observation taken for some other pair of commits cannot
    authorize recovery. Equality is refused: a HEAD that merely equals its base
    carries no committed work, and that is the ordinary base-equivalent case the
    normal lock path already handles.
    """
    if not isinstance(base_ancestry, Mapping):
        return False, [
            "no server-derived ancestry observation was available; an "
            "unpublished claim cannot be recovered without proving its HEAD "
            "descends from the recorded base"
        ]

    probe_ancestor = _text(base_ancestry.get("ancestor_sha"))
    probe_descendant = _text(base_ancestry.get("descendant_sha"))
    if probe_ancestor != recorded_base or probe_descendant != local_head:
        return False, [
            f"ancestry observation covers {probe_ancestor or 'unknown'} -> "
            f"{probe_descendant or 'unknown'}, not the commits under assessment "
            f"({recorded_base} -> {local_head})"
        ]
    if not base_ancestry.get("probe_ok"):
        return False, (
            list(base_ancestry.get("reasons") or [])
            or ["ancestry probe did not complete; ancestry unproven"]
        )
    if not base_ancestry.get("ancestor_present"):
        return False, [
            f"recorded base {recorded_base} is no longer reachable; a rewritten "
            "or force-moved base cannot be recovered"
        ]
    if not base_ancestry.get("is_strict_descendant"):
        return False, (
            list(base_ancestry.get("reasons") or [])
            or [
                f"local head {local_head} is not a strict descendant of the "
                f"recorded base {recorded_base}"
            ]
        )

    proof = _text(base_ancestry.get("proof")) or (
        f"{recorded_base} is an ancestor of {local_head}"
    )
    return True, [
        f"local head {local_head} strictly descends from recorded base "
        f"{recorded_base} ({proof})"
    ]


def assess_dead_session_lock_recovery(
    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,
    current_branch: str | None,
    porcelain_status: str,
    head_sha: str | None,
    remote_head_sha: str | None,
    pr_head_sha: str | None = None,
    pr_number: int | None = None,
    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,
    remote_branch_exists: bool | None = None,
    recorded_base_sha: str | None = None,
    base_ancestry: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Decide whether a dead-session author lock may be natively recovered.

    Returns a dict with ``recovery_sanctioned`` (bool), ``outcome``, ``reasons``
    (why it was refused, or the positive proof when sanctioned), and
    ``evidence`` (a redaction-safe record for auditing).

    ``NO_CANDIDATE`` means no recovery was attempted at all — there is no
    existing lock, or the lock does not describe this issue. The caller must
    treat that exactly as it treated the pre-#753 world. ``REFUSED`` means a
    candidate existed but the evidence did not agree; the caller fails closed.
    """
    reasons: list[str] = []
    evidence: dict[str, Any] = {
        "issue_number": issue_number,
        "branch_name": branch_name,
        "worktree_path": worktree_path,
        "remote": remote,
        "org": org,
        "repo": repo,
    }

    if not existing_lock:
        return _result(
            NO_CANDIDATE, False, ["no existing durable lock for this issue"], evidence
        )

    lock = dict(existing_lock)

    # ── Candidate identification ────────────────────────────────────────────
    # Recovery only ever applies to a lock that already claims THIS issue.
    # Anything else is not a recovery candidate and must not be reinterpreted.
    if lock.get("issue_number") != issue_number:
        return _result(
            NO_CANDIDATE,
            False,
            [
                f"existing lock targets issue #{lock.get('issue_number')}, "
                f"not #{issue_number}; not a recovery candidate"
            ],
            evidence,
        )

    # A malformed/incomplete durable record can never prove ownership.
    missing = _malformed_reasons(lock)
    if missing:
        return _result(
            REFUSED,
            False,
            [
                "durable lock record is incomplete and cannot prove ownership "
                f"(missing/unusable: {', '.join(missing)})"
            ],
            evidence,
        )

    recorded_pid = _recorded_pid(lock)
    evidence["prior_session_pid"] = recorded_pid
    evidence["replacement_session_pid"] = (
        current_pid if current_pid is not None else os.getpid()
    )

    # ── Repository scope ────────────────────────────────────────────────────
    for field, expected in (("remote", remote), ("org", org), ("repo", repo)):
        actual = _text(lock.get(field))
        if actual != _text(expected):
            reasons.append(
                f"lock {field} '{actual}' does not match requested '{_text(expected)}'"
            )

    # ── Branch identity ─────────────────────────────────────────────────────
    locked_branch = _text(lock.get("branch_name"))
    if locked_branch != _text(branch_name):
        reasons.append(
            f"lock branch '{locked_branch}' does not match requested "
            f"'{_text(branch_name)}'"
        )
    evidence["locked_branch"] = locked_branch

    # The worktree must actually be sitting on the locked branch. Without this
    # a clean worktree parked elsewhere could stand in for the real work.
    checked_out = _text(current_branch)
    if not checked_out:
        reasons.append(
            "worktree is not on a named branch (detached HEAD); locked-branch "
            "occupancy could not be proven"
        )
    elif checked_out != locked_branch:
        reasons.append(
            f"worktree is on branch '{checked_out}', not the locked branch "
            f"'{locked_branch}'"
        )

    # ── Worktree identity ───────────────────────────────────────────────────
    locked_worktree = _text(lock.get("worktree_path"))
    if not _same_realpath(locked_worktree, worktree_path):
        reasons.append(
            f"lock worktree '{locked_worktree}' does not match declared "
            f"'{_text(worktree_path)}'"
        )
    evidence["locked_worktree_path"] = locked_worktree

    # ── Cleanliness (never waived) ──────────────────────────────────────────
    dirty_files = parse_dirty_tracked_files(porcelain_status)
    if dirty_files:
        reasons.append(
            "worktree has tracked local edits; recovery requires a clean "
            f"worktree (dirty files: {', '.join(dirty_files)})"
        )
    evidence["dirty_files"] = dirty_files

    # ── 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)
    recorded_base = _text(recorded_base_sha)
    head_relation: str | None = None
    ancestry_proof: str | None = None
    if not local_head:
        reasons.append("local head SHA could not be determined")

    # #772: which body of evidence applies is decided by observed publication
    # state, never by a caller. ``remote_branch_exists is False`` is a positive
    # server-side observation that the branch is absent from the remote — it is
    # not the same as "the head lookup failed", which must still fail closed.
    unpublished = remote_branch_exists is False and not remote_head
    recovery_mode = (
        RECOVERY_MODE_UNPUBLISHED_CLAIM if unpublished
        else RECOVERY_MODE_PUBLISHED_OWNING_PR
    )
    evidence["recovery_mode"] = recovery_mode
    evidence["remote_branch_exists"] = remote_branch_exists

    if unpublished:
        # No remote branch: ownership is measured against the recorded base.
        # An open PR here is contradictory — a PR cannot exist without a remote
        # branch — so it is a mismatch, never a thing to reconcile.
        if _text(pr_head_sha) or pr_number is not None:
            reasons.append(
                f"branch '{locked_branch}' is absent from the remote yet PR "
                f"#{pr_number} claims it; publication state is contradictory"
            )
        if not recorded_base:
            reasons.append(
                f"recorded base for branch '{locked_branch}' could not be "
                "determined; an unpublished claim cannot be recovered without it"
            )
        if local_head and recorded_base:
            descends, notes = _assess_base_descendancy(
                base_ancestry,
                recorded_base=recorded_base,
                local_head=local_head,
            )
            if descends:
                head_relation = HEAD_RELATION_DESCENDS_FROM_BASE
                ancestry_proof = notes[0] if notes else None
            else:
                reasons.extend(notes)
    else:
        if not remote_head:
            reasons.append(
                f"remote head for branch '{locked_branch}' could not be determined"
            )
        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["recorded_base"] = recorded_base or None
    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
        # In unpublished mode the presence of any PR was already refused above as
        # contradictory; re-stating it as a head mismatch would only obscure why.
        if not unpublished and local_head and pr_head != 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)
    locked_identity = _text(claimant.get("username"))
    locked_profile = _text(claimant.get("profile"))
    evidence["locked_identity"] = locked_identity or None
    evidence["locked_profile"] = locked_profile or None
    if not locked_identity or not locked_profile:
        reasons.append(
            "durable lock does not record a claimant identity/profile; "
            "author ownership could not be proven"
        )
    if not _text(identity) or not _text(profile):
        reasons.append(
            "active session identity/profile is unknown; author ownership "
            "could not be proven"
        )
    if locked_identity and _text(identity) and locked_identity != _text(identity):
        reasons.append(
            f"lock claimant '{locked_identity}' does not match active identity "
            f"'{_text(identity)}'"
        )
    if locked_profile and _text(profile) and locked_profile != _text(profile):
        reasons.append(
            f"lock profile '{locked_profile}' does not match active profile "
            f"'{_text(profile)}'"
        )

    # ── The defining condition: the recorded owner must be dead ─────────────
    prior_alive = is_process_alive(recorded_pid)
    evidence["prior_pid_alive"] = prior_alive
    if prior_alive:
        reasons.append(
            f"prior owner pid {recorded_pid} is still alive; this is not a "
            "dead-session recovery"
        )
    if current_pid is not None and recorded_pid is not None:
        try:
            if int(recorded_pid) == int(current_pid):
                reasons.append(
                    "recorded pid is the current session; nothing to recover"
                )
        except (TypeError, ValueError):
            pass

    # ── Competing ownership ─────────────────────────────────────────────────
    competing: list[dict[str, Any]] = []
    for entry in competing_live_locks or ():
        if not isinstance(entry, Mapping):
            continue
        same_issue = entry.get("issue_number") == issue_number
        same_branch = _text(entry.get("branch_name")) == locked_branch
        if not (same_issue or same_branch):
            continue
        # The lock we are recovering is not competition with itself.
        if (
            same_issue
            and same_branch
            and _same_realpath(_text(entry.get("worktree_path")), worktree_path)
        ):
            continue
        competing.append(
            {
                "issue_number": entry.get("issue_number"),
                "branch_name": entry.get("branch_name"),
                "worktree_path": entry.get("worktree_path"),
                "pid": entry.get("pid"),
            }
        )
    if competing:
        described = ", ".join(
            f"issue #{c['issue_number']} branch '{c['branch_name']}'" for c in competing
        )
        reasons.append(f"competing live lock or lease exists ({described})")
    evidence["competing_live_locks"] = competing

    # ── Ambiguous branch claims ─────────────────────────────────────────────
    others = [
        name
        for name in (candidate_branches or ())
        if _text(name) and _text(name) != locked_branch
    ]
    if others:
        reasons.append(
            "multiple branches claim this issue "
            f"({', '.join(sorted(set(others)))}); ownership is ambiguous"
        )
    evidence["other_candidate_branches"] = sorted(set(others))

    if reasons:
        return _result(REFUSED, False, reasons, evidence)

    # No disposition may be granted without a proven head relation. Every path
    # above that leaves it unset also records a reason, so this is a belt-and-
    # braces guard against a future path forgetting one (#772 AC4).
    if head_relation is None:
        return _result(
            REFUSED,
            False,
            ["head relation to the recorded head or base was never proven"],
            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 recovery_mode == RECOVERY_MODE_UNPUBLISHED_CLAIM:
        proof.append(
            f"branch '{locked_branch}' has no remote head and no open PR; "
            f"ownership proven against recorded base {recorded_base}"
        )
    if (
        head_relation
        in (HEAD_RELATION_STRICT_DESCENDANT, HEAD_RELATION_DESCENDS_FROM_BASE)
        and ancestry_proof
    ):
        proof.append(ancestry_proof)
    return _result(RECOVERY_SANCTIONED, True, proof, evidence)


def _result(
    outcome: str,
    sanctioned: bool,
    reasons: list[str],
    evidence: dict[str, Any],
) -> dict[str, Any]:
    return {
        "outcome": outcome,
        "recovery_sanctioned": sanctioned,
        "is_candidate": outcome != NO_CANDIDATE,
        "reasons": reasons,
        "evidence": evidence,
    }


def owning_pr_recovery_evidence(
    assessment: Mapping[str, Any] | None,
) -> dict[str, Any] | None:
    """Server-derived proof of the open PR a sanctioned recovery already owns (#755).

    A dead-session recovery is, by construction, recovery of work that already
    has an open PR — so the duplicate-work gate's linked-open-PR blocker would
    otherwise discard every sanctioned recovery. This distils the completed
    assessment into the minimum evidence that gate needs to tell "the PR this
    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 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
    if assessment.get("outcome") != RECOVERY_SANCTIONED:
        return None
    if not assessment.get("recovery_sanctioned"):
        return None

    evidence = assessment.get("evidence") or {}
    branch_name = _text(evidence.get("locked_branch"))
    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 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)
        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": 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,
    }


def build_recovery_record(
    assessment: Mapping[str, Any],
    *,
    recovered_at: str,
) -> dict[str, Any]:
    """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,
        "reason": "owning MCP session exited; durable ownership evidence matched",
        "recovered_at": recovered_at,
        "prior_session_pid": evidence.get("prior_session_pid"),
        "replacement_session_pid": evidence.get("replacement_session_pid"),
        "prior_pid_alive": evidence.get("prior_pid_alive"),
        "branch_name": evidence.get("locked_branch"),
        "worktree_path": evidence.get("locked_worktree_path"),
        "recovery_mode": evidence.get("recovery_mode"),
        "remote_branch_exists": evidence.get("remote_branch_exists"),
        "recorded_base": evidence.get("recorded_base"),
        "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"),
        "profile": evidence.get("locked_profile"),
        "proof": list(assessment.get("reasons") or []),
    }


def format_recovery_refusal(assessment: Mapping[str, Any]) -> str:
    """Single fail-closed message for a refused recovery attempt."""
    reasons = list(assessment.get("reasons") or []) or [
        "dead-session lock recovery evidence did not agree"
    ]
    return (
        "Dead-session issue-lock recovery refused: "
        + "; ".join(reasons)
        + " (fail closed)"
    )
