"""Per-PR reviewer leases for safe parallel review sessions (#407)."""

from __future__ import annotations

import os
import re
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any

MARKER = "<!-- mcp-review-lease:v1 -->"

_FIELD_RE = re.compile(
    r"^\s*([a-z_]+)\s*:\s*(.+?)\s*$",
    re.IGNORECASE | re.MULTILINE,
)
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)

# "abandoned" is the owner-session merger finalization outcome (#742). Without
# it here, an abandoned marker would fall through to the generic non-empty
# phase branch and keep re-arming the lease as active.
_TERMINAL_PHASES = frozenset({"done", "released", "blocked", "abandoned"})
_ACTIVE_PHASES = frozenset({
    "claimed",
    "validating",
    "approved",
    "request-changes",
    "merging",
    "adopted",
})

# Reviewer and merger PR leases use a short *sliding* window (#747): a lease
# expires 10 minutes after its last heartbeat, and every heartbeat slides the
# expiry forward. An actively heartbeating session is never evicted, while a
# dead session releases its hold in at most one TTL instead of the two hours
# the previous fixed 120-minute expiry allowed.
LEASE_TTL_MINUTES = 10
# Renewal is named separately from acquisition so the slide amount is tunable
# without silently re-defining how long a fresh lease lives.
LEASE_RENEWAL_MINUTES = 10
# Retained for callers that imported the pre-#747 name.
DEFAULT_LEASE_TTL_MINUTES = LEASE_TTL_MINUTES
# Warn at half the window, while the owner can still heartbeat and recover.
STALE_WARNING_MINUTES = 5

_SESSION_LEASE: dict[str, Any] | None = None


def _parse_timestamp(value: str | None) -> datetime | None:
    if not value:
        return None
    text = value.strip()
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


def _normalize_sha(value: str | None) -> str | None:
    text = (value or "").strip().lower()
    return text if text and _FULL_SHA.match(text) else None


def _parse_pr_ref(value: str | None) -> int | None:
    digits = re.sub(r"[^\d]", "", value or "")
    return int(digits) if digits.isdigit() else None


def new_session_id() -> str:
    return f"{os.getpid()}-{uuid.uuid4().hex[:12]}"


def format_lease_body(
    *,
    repo: str,
    pr_number: int,
    issue_number: int | None,
    reviewer_identity: str,
    profile: str,
    session_id: str,
    worktree: str,
    phase: str,
    candidate_head: str | None,
    target_branch: str,
    target_branch_sha: str | None,
    last_activity: datetime | None = None,
    expires_at: datetime | None = None,
    ttl_minutes: int = LEASE_TTL_MINUTES,
    blocker: str = "none",
) -> str:
    """Serialize a lease marker.

    Every write of this marker — acquisition, heartbeat, adoption — re-derives
    ``expires_at`` from the moment of the write, which is what makes the TTL
    slide (#747). Callers renewing an existing lease pass
    ``ttl_minutes=LEASE_RENEWAL_MINUTES``; an explicit ``expires_at`` still
    wins so a lease can be minted with a deliberate window.
    """
    now = last_activity or datetime.now(timezone.utc)
    expires = expires_at or (now + timedelta(minutes=ttl_minutes))
    last_text = now.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
        "+00:00", "Z"
    )
    expires_text = expires.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
        "+00:00", "Z"
    )
    issue_text = f"#{issue_number}" if issue_number else "none"
    lines = [
        MARKER,
        f"repo: {repo}",
        f"pr: #{pr_number}",
        f"issue: {issue_text}",
        f"reviewer_identity: {reviewer_identity}",
        f"profile: {profile}",
        f"session_id: {session_id}",
        f"worktree: {worktree}",
        f"phase: {phase}",
        f"candidate_head: {candidate_head or 'none'}",
        f"target_branch: {target_branch}",
        f"target_branch_sha: {target_branch_sha or 'none'}",
        f"last_activity: {last_text}",
        f"expires_at: {expires_text}",
        f"blocker: {blocker}",
    ]
    return "\n".join(lines)


def parse_lease_comment(body: str) -> dict[str, Any] | None:
    text = body or ""
    if MARKER not in text:
        return None
    fields: dict[str, str] = {}
    for match in _FIELD_RE.finditer(text):
        fields[match.group(1).strip().lower()] = match.group(2).strip()
    if not fields:
        return None
    return {
        "repo": fields.get("repo"),
        "pr_number": _parse_pr_ref(fields.get("pr")),
        "issue_number": _parse_pr_ref(fields.get("issue")),
        "reviewer_identity": fields.get("reviewer_identity"),
        "profile": fields.get("profile"),
        "session_id": fields.get("session_id"),
        "worktree": fields.get("worktree"),
        "phase": (fields.get("phase") or "").strip().lower() or None,
        "candidate_head": _normalize_sha(fields.get("candidate_head")),
        "target_branch": fields.get("target_branch"),
        "target_branch_sha": _normalize_sha(fields.get("target_branch_sha")),
        "last_activity": fields.get("last_activity"),
        "expires_at": fields.get("expires_at"),
        "blocker": fields.get("blocker"),
        "raw_fields": fields,
    }


def _lease_entries(comments: list[dict], *, pr_number: int) -> list[dict]:
    entries: list[dict] = []
    for comment in comments or []:
        parsed = parse_lease_comment(comment.get("body") or "")
        if not parsed:
            continue
        if parsed.get("pr_number") not in (None, pr_number):
            continue
        entries.append({
            **parsed,
            "comment_id": comment.get("id"),
            "author": (comment.get("user") or {}).get("login") or comment.get("author"),
            "created_at": comment.get("created_at"),
            "updated_at": comment.get("updated_at"),
        })
    return entries


def _lease_expired(lease: dict, *, now: datetime) -> bool:
    expires_at = _parse_timestamp(lease.get("expires_at"))
    return bool(expires_at and expires_at <= now)


def _minutes_since_activity(lease: dict, *, now: datetime) -> float | None:
    last = _parse_timestamp(lease.get("last_activity"))
    if not last:
        return None
    return (now - last).total_seconds() / 60.0


def lease_seconds_remaining(lease: dict, *, now: datetime | None = None) -> int | None:
    """Seconds until *lease* expires, clamped at 0; ``None`` if unparsable.

    Lets diagnostics distinguish "held and live" from "held and dying" (#747)
    rather than only reporting that a lease exists.
    """
    expires_at = _parse_timestamp(lease.get("expires_at"))
    if not expires_at:
        return None
    now = now or datetime.now(timezone.utc)
    return max(0, int((expires_at - now).total_seconds()))


def classify_lease_freshness(lease: dict, *, now: datetime | None = None) -> str:
    """Return active, stale_warning, expired, or terminal.

    Expiry is the only takeover gate (#747). The pre-#747 ``reclaimable`` band
    sat between "stale" and "expired" and made a dead lease wait out a second
    timer before anyone could reclaim it. Under a sliding TTL that band is also
    unreachable: a heartbeat stamps ``last_activity`` and ``expires_at``
    together, so a lease idle for a full TTL is already expired. Foreign
    expired leases are handled by the ``foreign_expired`` classification, which
    carries the same sanctioned release next-action the old tier did.
    """
    now = now or datetime.now(timezone.utc)
    phase = (lease.get("phase") or "").strip().lower()
    if phase in _TERMINAL_PHASES:
        return "terminal"
    if _lease_expired(lease, now=now):
        return "expired"
    minutes = _minutes_since_activity(lease, now=now)
    if minutes is None:
        return "active"
    if minutes >= STALE_WARNING_MINUTES:
        return "stale_warning"
    return "active"


def find_active_reviewer_lease(
    comments: list[dict],
    *,
    pr_number: int,
    now: datetime | None = None,
) -> dict[str, Any] | None:
    """Return the newest unexpired non-terminal lease for *pr_number*.

    Append-only lease markers form a ledger: the **newest** marker is
    authoritative. A later terminal phase (``released`` / ``done`` /
    ``blocked``) ends the lease even when older ``claimed`` markers remain
    on the thread (#577). Skipping only terminal markers and walking older
    claims incorrectly re-arms a lease after a successful release.
    """
    now = now or datetime.now(timezone.utc)
    entries = list(reversed(_lease_entries(comments, pr_number=pr_number)))
    if not entries:
        return None
    newest = entries[0]
    phase = (newest.get("phase") or "").strip().lower()
    if phase in _TERMINAL_PHASES:
        return None
    if _lease_expired(newest, now=now):
        return None
    if phase in _ACTIVE_PHASES or phase:
        lease = dict(newest)
        lease["freshness"] = classify_lease_freshness(lease, now=now)
        return lease
    return None


def assess_acquire_lease(
    comments: list[dict],
    *,
    pr_number: int,
    reviewer_identity: str,
    profile: str,
    session_id: str,
    repo: str,
    issue_number: int | None,
    worktree: str,
    candidate_head: str | None,
    target_branch: str,
    target_branch_sha: str | None,
    pr_merged_or_closed: bool = False,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Fail closed when another session holds an active lease.

    When *pr_merged_or_closed* is true the PR has already merged/closed, so any
    reviewer-lease acquisition or adoption for merge work is moot: fail closed
    with a ``post_merge_moot`` reason and never mint a lease body (#515).
    """
    now = now or datetime.now(timezone.utc)
    reasons: list[str] = []
    existing = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
    post_merge_moot = bool(pr_merged_or_closed)
    if post_merge_moot:
        reasons.append(
            f"post_merge_moot: PR #{pr_number} is already merged/closed; reviewer "
            "lease adoption for merge is moot (fail closed)"
        )
    if existing:
        owner_session = (existing.get("session_id") or "").strip()
        freshness = existing.get("freshness") or classify_lease_freshness(existing, now=now)
        if owner_session and owner_session != session_id and freshness in {
            "active", "stale_warning"
        }:
            reasons.append(
                f"PR #{pr_number} already has active reviewer lease "
                f"(session_id={owner_session}, phase={existing.get('phase')})"
            )
        elif owner_session and owner_session != session_id and freshness == "reclaimable":
            reasons.append(
                f"PR #{pr_number} lease is reclaimable but still held by "
                f"session_id={owner_session}; explicit reclaim not implemented "
                "(fail closed)"
            )

    if not (reviewer_identity or "").strip():
        reasons.append("reviewer identity required for lease acquisition")
    if not (session_id or "").strip():
        reasons.append("session_id required for lease acquisition")
    if not (worktree or "").strip():
        reasons.append("worktree path required for lease acquisition")

    allowed = not reasons
    body = None
    if allowed:
        body = format_lease_body(
            repo=repo,
            pr_number=pr_number,
            issue_number=issue_number,
            reviewer_identity=reviewer_identity,
            profile=profile,
            session_id=session_id,
            worktree=worktree,
            phase="claimed",
            candidate_head=candidate_head,
            target_branch=target_branch,
            target_branch_sha=target_branch_sha,
            last_activity=now,
        )
    return {
        "acquire_allowed": allowed,
        "reasons": reasons,
        "existing_lease": existing,
        "lease_body": body,
        "session_id": session_id,
        "post_merge_moot": post_merge_moot,
    }


def assess_post_merge_moot_lease(
    comments: list[dict],
    *,
    pr_number: int,
    pr_merged: bool = False,
    pr_state: str | None = None,
    merge_commit_sha: str | None = None,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Assess a reviewer lease left lingering on an already-merged/closed PR (#515).

    Read-first and fail-safe:

    - Only treats a lease as moot when the live PR state is merged/closed.
    - Never proposes touching an *active* lease while the PR is still open
      (``cleanup_allowed`` stays false and a refusal reason is returned).
    - When the PR is merged/closed and a lease is still active, ``cleanup_allowed``
      is true and a terminal ``phase: released`` lease body (``blocker:
      post-merge-moot``) is provided so the moot lease can be neutralised by an
      append-only comment — never by deleting a foreign session's comment, and
      never by adopting or merging.

    Posting the released body makes that lease terminal, so a subsequent call
    finds no active lease and reports nothing left to clean (idempotent).
    """
    now = now or datetime.now(timezone.utc)
    merged_or_closed = bool(pr_merged) or (
        str(pr_state or "").strip().lower() == "closed"
    )
    active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
    # The newest lease comment is authoritative: once a terminal marker
    # (released/done/blocked) is the latest entry, the lease is resolved even if
    # an earlier non-terminal comment from the same session still lingers. This
    # keeps post-merge cleanup idempotent.
    entries = _lease_entries(comments, pr_number=pr_number)
    newest = entries[-1] if entries else None
    newest_terminal = bool(newest) and (
        (newest.get("phase") or "").strip().lower() in _TERMINAL_PHASES
    )
    reasons: list[str] = []
    cleanup_allowed = False
    release_body: str | None = None
    is_moot = bool(active) and merged_or_closed and not newest_terminal

    if not merged_or_closed:
        if active:
            reasons.append(
                f"PR #{pr_number} is still open; refusing to touch active reviewer "
                "lease (fail closed)"
            )
        else:
            reasons.append(
                f"PR #{pr_number} is still open; no post-merge lease cleanup applicable"
            )
    elif newest_terminal:
        reasons.append(
            f"PR #{pr_number} reviewer lease already released/terminal; nothing to clean"
        )
    elif active:
        cleanup_allowed = True
        release_body = format_lease_body(
            repo=active.get("repo") or "",
            pr_number=pr_number,
            issue_number=active.get("issue_number"),
            reviewer_identity=active.get("reviewer_identity") or "",
            profile=active.get("profile") or "unknown",
            session_id=active.get("session_id") or "",
            worktree=active.get("worktree") or "",
            phase="released",
            candidate_head=active.get("candidate_head"),
            target_branch=active.get("target_branch") or "master",
            target_branch_sha=active.get("target_branch_sha"),
            last_activity=now,
            blocker="post-merge-moot",
        )
    else:
        reasons.append(
            f"PR #{pr_number} is merged/closed but no active reviewer lease remains; "
            "nothing to clean"
        )

    return {
        "pr_number": pr_number,
        "pr_state": pr_state,
        "pr_merged_or_closed": merged_or_closed,
        "merge_commit_sha": merge_commit_sha,
        "active_lease": active,
        "is_moot": is_moot,
        "cleanup_allowed": cleanup_allowed,
        "release_body": release_body,
        "reasons": reasons,
    }


def record_session_lease(
    lease: dict[str, Any],
    *,
    lease_provenance: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Record the in-session lease mirror for mutation gates.

    *lease_provenance* must be supplied by sanctioned MCP tools (#536). Bare
    manual seeding without provenance cannot satisfy merger/reviewer mutation
    gates.
    """
    global _SESSION_LEASE
    stored = dict(lease)
    if lease_provenance:
        stored["lease_provenance"] = dict(lease_provenance)
    _SESSION_LEASE = stored
    _persist_session_lease_shadow(stored)
    return dict(_SESSION_LEASE)


def clear_session_lease() -> None:
    global _SESSION_LEASE
    prior = _SESSION_LEASE
    _SESSION_LEASE = None
    _persist_session_lease_shadow(None, prior=prior)


def get_session_lease() -> dict[str, Any] | None:
    return dict(_SESSION_LEASE) if _SESSION_LEASE else None


def _persist_session_lease_shadow(
    lease: dict[str, Any] | None,
    *,
    prior: dict[str, Any] | None = None,
) -> None:
    """Best-effort durable shadow of the in-session lease (#702).

    A daemon that dies without teardown leaves this record behind, giving a
    later process provable orphan evidence (owner pid + session id) for the
    guarded cleanup path. Observability only — mutation gates never read it,
    and failures here must never block lease operations.

    Shadows are keyed by lease session id (#702 F5): concurrent same-profile
    sessions each own a distinct record, so one session's heartbeat can never
    overwrite or misattribute another's crash evidence. A sanctioned clear is
    the terminal reconciliation of that record's lifecycle.
    """
    try:
        import mcp_session_state as mss

        if lease is None:
            sid = ((prior or {}).get("session_id") or "").strip() or None
            if sid:
                mss.clear_state(
                    kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=sid
                )
            # Legacy single-slot record from pre-F5 code: clearing it is safe
            # because collision-safe writes never target that key again.
            mss.clear_state(kind=mss.KIND_REVIEWER_SESSION_LEASE)
            return
        mss.save_state(
            kind=mss.KIND_REVIEWER_SESSION_LEASE,
            instance_id=((lease.get("session_id") or "").strip() or None),
            payload={
                "pr_number": lease.get("pr_number"),
                "session_id": lease.get("session_id"),
                "comment_id": lease.get("comment_id"),
                "candidate_head": lease.get("candidate_head"),
                "worktree": lease.get("worktree"),
                "phase": lease.get("phase"),
                "reviewer_identity": lease.get("reviewer_identity"),
                "profile": lease.get("profile"),
                "lease_repo": lease.get("repo"),
                "owner_pid": os.getpid(),
            },
        )
    except Exception:
        pass


def load_session_lease_shadow(
    session_id: str | None = None,
) -> dict[str, Any] | None:
    """Load the durable session-lease shadow left by this profile identity.

    With *session_id*, load that session's collision-safe record (#702 F5),
    falling back to the legacy single-slot record only when its payload names
    the same session. Without *session_id*, return the legacy record or the
    sole surviving instance record — ambiguity (multiple candidates) returns
    ``None`` because a shadow that cannot be attributed proves nothing.
    """
    try:
        import mcp_session_state as mss

        sid = (session_id or "").strip()
        if sid:
            record = mss.load_state(
                kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=sid
            )
            if record is not None:
                return record
            legacy = mss.load_state(kind=mss.KIND_REVIEWER_SESSION_LEASE)
            if legacy and (legacy.get("session_id") or "").strip() == sid:
                return legacy
            return None
        legacy = mss.load_state(kind=mss.KIND_REVIEWER_SESSION_LEASE)
        if legacy is not None:
            return legacy
        records = mss.list_states(kind=mss.KIND_REVIEWER_SESSION_LEASE)
        if len(records) == 1:
            return records[0]
        return None
    except Exception:
        return None


def _pid_alive(pid: int) -> bool:
    """Probe process existence; fail closed toward 'alive' when unsure."""
    try:
        os.kill(int(pid), 0)
        return True
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    except Exception:
        return True


def assess_crashed_session_lease_orphan(
    shadow: dict[str, Any] | None,
    *,
    lease: dict[str, Any] | None,
    current_pid: int | None = None,
    pid_alive: bool | None = None,
) -> dict[str, Any]:
    """Derive owner-process evidence for a durable lease from the shadow (#702).

    Returns ``owner_process_alive`` tri-state. Only a provably dead recorded
    owner pid yields ``False`` (a dead pid cannot still be running the owning
    daemon); an alive pid is *not* ownership proof and stays ``None`` unless
    it is this very process. Session ids must match exactly — the shadow of a
    different lease proves nothing.
    """
    reasons: list[str] = []
    if not shadow or not lease:
        return {
            "orphan_evidence": False,
            "owner_process_alive": None,
            "reasons": ["no durable session-lease shadow evidence available"],
        }
    shadow_sid = (shadow.get("session_id") or "").strip()
    lease_sid = (lease.get("session_id") or "").strip()
    if not shadow_sid or not lease_sid or shadow_sid != lease_sid:
        return {
            "orphan_evidence": False,
            "owner_process_alive": None,
            "reasons": [
                "session-lease shadow session_id does not match the durable "
                f"lease (shadow={shadow_sid or None}, lease={lease_sid or None})"
            ],
        }
    owner_pid = shadow.get("owner_pid") or shadow.get("writer_pid")
    try:
        owner_pid = int(owner_pid)
    except (TypeError, ValueError):
        return {
            "orphan_evidence": False,
            "owner_process_alive": None,
            "reasons": ["session-lease shadow has no usable owner pid (fail closed)"],
        }
    this_pid = current_pid if current_pid is not None else os.getpid()
    if owner_pid == this_pid:
        return {
            "orphan_evidence": False,
            "owner_process_alive": True,
            "owner_pid": owner_pid,
            "reasons": ["shadow owner pid is the current process"],
        }
    alive = pid_alive if pid_alive is not None else _pid_alive(owner_pid)
    if alive:
        reasons.append(
            f"shadow owner pid {owner_pid} observed alive; PID liveness is not "
            "ownership proof — owner evidence stays unknown (fail closed)"
        )
        return {
            "orphan_evidence": False,
            "owner_process_alive": None,
            "owner_pid": owner_pid,
            "reasons": reasons,
        }
    reasons.append(
        f"shadow owner pid {owner_pid} is dead: the daemon that recorded the "
        "matching session lease exited without teardown (crash orphan, #702)"
    )
    return {
        "orphan_evidence": True,
        "owner_process_alive": False,
        "owner_pid": owner_pid,
        "reasons": reasons,
    }


def assess_mutation_lease_gate(
    *,
    pr_number: int,
    comments: list[dict],
    reviewer_identity: str,
    session_id: str | None,
    mutation: str,
    live_head_sha: str | None,
    pinned_head_sha: str | None,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Reviewer mutations require an owned, current PR lease."""
    now = now or datetime.now(timezone.utc)
    reasons: list[str] = []
    session = get_session_lease()
    active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)

    if not session:
        reasons.append(
            f"no in-session reviewer lease recorded; acquire via "
            f"gitea_acquire_reviewer_pr_lease or adopt via "
            f"gitea_adopt_merger_pr_lease before {mutation}"
        )
    else:
        import merger_lease_adoption as mla

        if not mla.is_sanctioned_session_lease(session):
            reasons.append(
                "in-session lease lacks sanctioned provenance; manual "
                "_SESSION_LEASE seeding is not canonical proof — use "
                "gitea_acquire_reviewer_pr_lease or gitea_adopt_merger_pr_lease"
            )
    if session and session.get("pr_number") != pr_number:
        reasons.append(
            f"session lease is for PR #{session.get('pr_number')}, not #{pr_number}"
        )
    elif session and (session.get("session_id") or "") != (
        session_id or session.get("session_id")
    ):
        reasons.append("session lease session_id mismatch (fail closed)")

    if active:
        owner = (active.get("session_id") or "").strip()
        if owner and session_id and owner != session_id:
            reasons.append(
                f"active PR lease owned by session_id={owner}; current session "
                f"cannot {mutation}"
            )
        pinned = _normalize_sha(pinned_head_sha)
        live = _normalize_sha(live_head_sha)
        lease_head = active.get("candidate_head")
        if pinned and live and pinned != live:
            reasons.append(
                "PR head changed during lease; stop and re-validate before "
                f"reviewer {mutation}"
            )
        if lease_head and live and lease_head != live:
            reasons.append(
                "live PR head differs from lease candidate_head; refresh lease "
                f"before {mutation}"
            )
        freshness = active.get("freshness") or classify_lease_freshness(active, now=now)
        if freshness in {"expired", "reclaimable"}:
            reasons.append(f"reviewer lease freshness is '{freshness}' (fail closed)")
    else:
        reasons.append(f"no active reviewer lease found on PR #{pr_number}")

    allowed = not reasons
    return {
        "mutation_allowed": allowed,
        "block": not allowed,
        "reasons": reasons,
        "active_lease": active,
        "session_lease": session,
    }


def assess_lease_inventory(
    comments_by_pr: dict[int, list[dict]],
    *,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Summarize lease states across PR comment threads."""
    now = now or datetime.now(timezone.utc)
    active: list[dict] = []
    stale: list[dict] = []
    reclaimable: list[dict] = []
    for pr_number, comments in (comments_by_pr or {}).items():
        lease = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
        if not lease:
            continue
        freshness = lease.get("freshness") or classify_lease_freshness(lease, now=now)
        entry = {"pr_number": pr_number, "session_id": lease.get("session_id"), "freshness": freshness}
        if freshness == "stale_warning":
            stale.append(entry)
        elif freshness == "reclaimable":
            reclaimable.append(entry)
        else:
            active.append(entry)
    return {
        "active_review_leases": active,
        "stale_review_leases": stale,
        "reclaimable_review_leases": reclaimable,
    }

# Canonical next-action vocabulary for reviewer lease handoff (#599, #691).
NEXT_ACTION_ACQUIRE = "acquire"
NEXT_ACTION_WAIT = "wait"
NEXT_ACTION_RESUME_EXACT_OWNER_SESSION = "resume_exact_owner_session"
NEXT_ACTION_RELEASE_EXPIRED_LEASE = "release_expired_lease"
NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP = "operator_authorized_cleanup"
NEXT_ACTION_REPAIR_WORKTREE_BINDING = "repair_worktree_binding"
NEXT_ACTION_CLEANUP_OBSOLETE_LEASE = "cleanup_obsolete_reviewer_comment_lease"

CLEANUP_OBSOLETE_LEASE_TOOL = "gitea_cleanup_obsolete_reviewer_comment_lease"
CLEANUP_OBSOLETE_LEASE_CONFIRMATION_PREFIX = "CLEANUP OBSOLETE REVIEWER LEASE "

# Formal terminal review verdicts that complete a leased-head workflow (#691).
_TERMINAL_REVIEW_VERDICTS = frozenset({
    "APPROVED",
    "REQUEST_CHANGES",
    "approved",
    "request_changes",
    "REQUESTCHANGES",
})

_HANDOFF_CLASSIFICATIONS = frozenset({
    "no_lease",
    "own_active",
    "own_expired",
    "foreign_active",
    "foreign_reclaimable",
    "foreign_expired",
    "foreign_active_current_head",
    "foreign_expired_current_head",
    "foreign_completed_superseded_head",
    "foreign_expired_superseded_head",
    "orphaned_owner_missing",
    "orphaned_expired_superseded_head",
    "ambiguous_conflicting_evidence",
    "instructed_lease_missing_with_replacement",
    "worktree_binding_mismatch",
})


def _norm_path(value: str | None) -> str:
    text = (value or "").strip()
    if not text:
        return ""
    return os.path.normpath(text.rstrip("/"))


def _normalize_verdict(value: str | None) -> str:
    text = (value or "").strip().upper().replace("-", "_").replace(" ", "_")
    if text in {"REQUESTCHANGES", "REQUEST_CHANGES", "CHANGES_REQUESTED"}:
        return "REQUEST_CHANGES"
    if text in {"APPROVED", "APPROVE"}:
        return "APPROVED"
    return text


def formal_terminal_review_for_head(
    formal_reviews: list[dict] | None,
    *,
    leased_head: str | None,
) -> dict[str, Any] | None:
    """Return the latest undismissed terminal formal review for *leased_head*."""
    head = _normalize_sha(leased_head)
    if not head:
        return None
    matches: list[dict[str, Any]] = []
    for review in formal_reviews or []:
        if review.get("dismissed"):
            continue
        verdict = _normalize_verdict(
            review.get("verdict") or review.get("state") or review.get("body_state")
        )
        if verdict not in {"APPROVED", "REQUEST_CHANGES"}:
            continue
        rhead = _normalize_sha(
            review.get("reviewed_head_sha")
            or review.get("commit_id")
            or review.get("head_sha")
        )
        if rhead != head:
            continue
        matches.append({**review, "verdict": verdict, "reviewed_head_sha": rhead})
    if not matches:
        return None
    return matches[-1]


def find_newest_nonterminal_lease(
    comments: list[dict],
    *,
    pr_number: int,
    now: datetime | None = None,
    include_expired: bool = True,
) -> dict[str, Any] | None:
    """Newest non-terminal lease marker, optionally including expired ones (#691).

    Unlike ``find_active_reviewer_lease``, this retains expired non-terminal
    markers so diagnosis/cleanup can still name the obsolete lease after
    ``expires_at`` (comment-backed ledger; no control-plane ``lease_id``).
    """
    now = now or datetime.now(timezone.utc)
    entries = list(reversed(_lease_entries(comments, pr_number=pr_number)))
    for entry in entries:
        phase = (entry.get("phase") or "").strip().lower()
        if phase in _TERMINAL_PHASES:
            # Newest terminal ends the ledger chain for active acquisition, but
            # an older non-terminal is not "active". Stop at newest marker.
            return None
        expired = _lease_expired(entry, now=now)
        if expired and not include_expired:
            return None
        lease = dict(entry)
        lease["freshness"] = classify_lease_freshness(lease, now=now)
        lease["expired"] = expired
        return lease
    return None


def cleanup_confirmation_for_pr(pr_number: int) -> str:
    return f"{CLEANUP_OBSOLETE_LEASE_CONFIRMATION_PREFIX}{int(pr_number)}"


def assess_obsolete_reviewer_comment_lease_cleanup(
    comments: list[dict],
    *,
    pr_number: int,
    current_head_sha: str | None,
    formal_reviews: list[dict] | None = None,
    repo: str | None = None,
    expected_repo: str | None = None,
    requesting_session_id: str | None = None,
    controller_recovery_authorized: bool = False,
    worktree_exists: bool | None = None,
    worktree_clean: bool | None = None,
    worktree_has_unpreserved_work: bool | None = None,
    owner_process_alive: bool | None = None,
    owner_pid_observed: int | None = None,
    requesting_pid: int | None = None,
    current_head_review_in_progress: bool = False,
    expected_lease_comment_id: int | None = None,
    expected_session_id: str | None = None,
    expected_leased_head: str | None = None,
    confirmation: str | None = None,
    apply: bool = False,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Guarded cleanup eligibility for obsolete comment-backed reviewer leases (#691).

    Cleanup is never transfer/adoption/repoint and never uses PID equality as
    ownership. Eligible only when evidence proves the lease is past expiry and/or
    pinned to a superseded head with a completed formal terminal review, and
    every safety boundary holds.
    """
    now = now or datetime.now(timezone.utc)
    reasons: list[str] = []
    fail_closed_reasons: list[str] = []
    current_head = _normalize_sha(current_head_sha)
    req_session = (requesting_session_id or "").strip()

    lease = find_newest_nonterminal_lease(
        comments, pr_number=pr_number, now=now, include_expired=True
    )
    if not lease:
        # Fall back to active finder (unexpired only) for report consistency.
        lease = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)

    classification = "no_lease"
    cleanup_allowed = False
    release_body: str | None = None
    audit_comment_body: str | None = None
    terminal_review = None
    leased_head = None
    head_superseded = False
    past_expiry = False
    expires_parseable = True

    if not lease:
        fail_closed_reasons.append(
            f"PR #{pr_number}: no non-terminal comment-backed reviewer lease to clean"
        )
        classification = "no_lease"
    else:
        leased_head = _normalize_sha(lease.get("candidate_head"))
        expires_raw = lease.get("expires_at")
        expires_at = _parse_timestamp(expires_raw)
        if expires_raw and expires_at is None:
            expires_parseable = False
            fail_closed_reasons.append(
                "lease expires_at cannot be parsed or trusted (fail closed)"
            )
        past_expiry = bool(expires_at and expires_at <= now)
        freshness = lease.get("freshness") or classify_lease_freshness(lease, now=now)
        owner_session = (lease.get("session_id") or "").strip()
        is_requesting_owner = bool(
            req_session and owner_session and req_session == owner_session
        )

        # Identity pins (repo / PR / session / head / comment) must match when
        # the caller supplies expected evidence.
        lease_repo = (lease.get("repo") or "").strip()
        exp_repo = (expected_repo or repo or "").strip()
        if exp_repo and lease_repo and exp_repo != lease_repo:
            fail_closed_reasons.append(
                f"repository identity mismatch: lease repo={lease_repo!r} "
                f"expected={exp_repo!r}"
            )
        if lease.get("pr_number") not in (None, pr_number):
            fail_closed_reasons.append(
                f"PR identity mismatch: lease pr={lease.get('pr_number')} "
                f"requested={pr_number}"
            )
        if expected_lease_comment_id is not None:
            cid = lease.get("comment_id")
            if cid is None or int(cid) != int(expected_lease_comment_id):
                fail_closed_reasons.append(
                    "lease comment_id does not match expected evidence "
                    f"(lease={cid}, expected={expected_lease_comment_id})"
                )
        if expected_session_id:
            if owner_session != expected_session_id.strip():
                fail_closed_reasons.append(
                    "lease session_id does not match expected evidence "
                    f"(lease={owner_session}, expected={expected_session_id})"
                )
        if expected_leased_head:
            exp_head = _normalize_sha(expected_leased_head)
            if not exp_head or exp_head != leased_head:
                fail_closed_reasons.append(
                    "leased head does not match expected evidence "
                    f"(lease={leased_head}, expected={exp_head})"
                )

        # PID equality is never ownership proof (#691).
        if (
            owner_pid_observed is not None
            and requesting_pid is not None
            and int(owner_pid_observed) == int(requesting_pid)
        ):
            reasons.append(
                "observed PID equals requesting PID but PID equality is NOT "
                "ownership proof; ignored for authorization"
            )

        if is_requesting_owner:
            fail_closed_reasons.append(
                "requesting session owns the lease; use "
                "gitea_release_reviewer_pr_lease (owner path), not non-owner cleanup"
            )

        if current_head_review_in_progress:
            fail_closed_reasons.append(
                "a current-head review submission is in progress; cleanup denied"
            )

        if not current_head:
            fail_closed_reasons.append(
                "current PR head SHA missing or unparseable (fail closed)"
            )
        if not leased_head:
            fail_closed_reasons.append(
                "lease candidate_head missing or unparseable (fail closed)"
            )

        head_superseded = bool(
            current_head and leased_head and current_head != leased_head
        )
        head_matches_current = bool(
            current_head and leased_head and current_head == leased_head
        )

        terminal_review = formal_terminal_review_for_head(
            formal_reviews, leased_head=leased_head
        )
        has_terminal = terminal_review is not None

        # Worktree safety.
        if worktree_has_unpreserved_work is True or worktree_clean is False:
            fail_closed_reasons.append(
                "lease worktree is dirty or has unpreserved work; cleanup denied"
            )
        if (
            worktree_exists is True
            and worktree_clean is None
            and worktree_has_unpreserved_work is None
        ):
            # Unknown cleanliness when path exists — fail closed.
            fail_closed_reasons.append(
                "lease worktree exists but cleanliness is unknown (fail closed)"
            )

        # Classification matrix (#691).
        if head_matches_current and freshness in {"active", "stale_warning"}:
            classification = "foreign_active_current_head"
            fail_closed_reasons.append(
                "genuinely active foreign lease on the current PR head; "
                "cleanup denied (fail closed)"
            )
        elif head_matches_current and past_expiry:
            classification = "foreign_expired_current_head"
            # Expired on current head: owner-missing / orphan path may apply.
            if owner_process_alive is True:
                fail_closed_reasons.append(
                    "lease expired on current head but owner process still alive "
                    "with plausible activity; cleanup denied"
                )
            elif worktree_clean is False:
                fail_closed_reasons.append(
                    "owner process absent but worktree dirty; cleanup denied"
                )
            elif owner_process_alive is False and (
                worktree_clean is True or worktree_exists is False
            ):
                if controller_recovery_authorized:
                    classification = "orphaned_owner_missing"
                else:
                    fail_closed_reasons.append(
                        "controller/recovery capability not authorized for orphan cleanup"
                    )
            else:
                fail_closed_reasons.append(
                    "insufficient orphan evidence for expired current-head lease "
                    "(need owner_process_alive=false and clean/absent worktree)"
                )
        elif head_superseded and has_terminal and past_expiry:
            classification = "foreign_expired_superseded_head"
        elif head_superseded and has_terminal and not past_expiry:
            classification = "foreign_completed_superseded_head"
        elif head_superseded and not has_terminal and past_expiry:
            # Crash-orphan shape (#702): the lease outlived its head with no
            # formal terminal review and has now passed canonical expiry, so
            # it no longer gates fresh acquisition. Guarded cleanup of the
            # ledger marker still demands provable owner-process-exit
            # evidence and a safe worktree — never inferred.
            classification = "orphaned_expired_superseded_head"
            if owner_process_alive is True:
                fail_closed_reasons.append(
                    "lease past expiry on superseded head but owner process "
                    "still alive; cleanup denied"
                )
            elif owner_process_alive is None:
                fail_closed_reasons.append(
                    "expired superseded-head lease with no terminal review: "
                    "cleanup requires provable owner-process-exit evidence "
                    "(owner_process_alive=false); the expired lease no longer "
                    "gates fresh acquisition"
                )
            elif worktree_clean is False:
                fail_closed_reasons.append(
                    "owner process absent but worktree dirty; cleanup denied"
                )
            elif not (worktree_clean is True or worktree_exists is False):
                # #702 F3: unknown worktree evidence must fail closed, exactly
                # like the sibling orphaned_owner_missing gate and the
                # diagnosis path — cleanup needs affirmative safe evidence.
                fail_closed_reasons.append(
                    "owner process absent but worktree evidence is unknown; "
                    "cleanup requires affirmative safe evidence "
                    "(worktree_clean=true or worktree_exists=false)"
                )
        elif head_superseded and not has_terminal:
            classification = "ambiguous_conflicting_evidence"
            fail_closed_reasons.append(
                "lease head is superseded but no formal terminal review exists "
                "for the leased head (fail closed)"
            )
        elif not head_superseded and not past_expiry and freshness in {
            "active", "stale_warning"
        }:
            classification = "foreign_active_current_head"
            fail_closed_reasons.append(
                "foreign lease still active on current head; wait or use owner release"
            )
        elif past_expiry and not head_superseded:
            classification = "foreign_expired_current_head"
        else:
            classification = "ambiguous_conflicting_evidence"
            fail_closed_reasons.append(
                "lease evidence is ambiguous; cleanup denied (fail closed)"
            )

        # Recent owner progress on a non-superseded active lease blocks cleanup.
        minutes = _minutes_since_activity(lease, now=now)
        if (
            head_matches_current
            and freshness == "active"
            and minutes is not None
            and minutes < STALE_WARNING_MINUTES
        ):
            fail_closed_reasons.append(
                "owner session has recent authenticated progress on current head; "
                "cleanup denied"
            )

        # Authority + confirmation for apply.
        if not controller_recovery_authorized:
            if classification in {
                "foreign_completed_superseded_head",
                "foreign_expired_superseded_head",
                "foreign_expired_current_head",
                "orphaned_owner_missing",
                "orphaned_expired_superseded_head",
            }:
                fail_closed_reasons.append(
                    "controller/recovery capability required "
                    "(controller_recovery_authorized=true)"
                )

        expected_conf = cleanup_confirmation_for_pr(pr_number)
        if apply:
            if (confirmation or "").strip() != expected_conf:
                fail_closed_reasons.append(
                    f"confirmation must equal exactly {expected_conf!r}"
                )

        # Superseded + terminal path does not require past_expiry (AC1).
        eligible_class = classification in {
            "foreign_completed_superseded_head",
            "foreign_expired_superseded_head",
            "orphaned_owner_missing",
            "orphaned_expired_superseded_head",
            "foreign_expired_current_head",
        }
        # foreign_expired_current_head needs orphan/clean worktree + authority.
        if classification == "foreign_expired_current_head":
            if worktree_clean is not True and worktree_exists is not False:
                if "worktree" not in " ".join(fail_closed_reasons):
                    fail_closed_reasons.append(
                        "expired current-head cleanup requires proven-clean "
                        "worktree or absent worktree"
                    )
            if owner_process_alive is True:
                fail_closed_reasons.append(
                    "owner process still alive on expired current-head lease"
                )

        cleanup_allowed = (
            eligible_class
            and expires_parseable
            and not fail_closed_reasons
            and not is_requesting_owner
        )

        if cleanup_allowed:
            release_body = format_lease_body(
                repo=lease.get("repo") or exp_repo or "",
                pr_number=pr_number,
                issue_number=lease.get("issue_number"),
                reviewer_identity=lease.get("reviewer_identity") or "",
                profile=lease.get("profile") or "unknown",
                session_id=owner_session or "unknown",
                worktree=lease.get("worktree") or "",
                phase="released",
                candidate_head=leased_head,
                target_branch=lease.get("target_branch") or "master",
                target_branch_sha=lease.get("target_branch_sha"),
                last_activity=now,
                blocker="obsolete-superseded-or-expired-lease",
            )
            audit_comment_body = (
                "## Canonical obsolete reviewer lease cleanup (#691)\n\n"
                f"- pr: #{pr_number}\n"
                f"- lease_comment_id: {lease.get('comment_id')}\n"
                f"- session_id: {owner_session}\n"
                f"- leased_head: {leased_head}\n"
                f"- current_head: {current_head}\n"
                f"- expires_at: {lease.get('expires_at')}\n"
                f"- classification: {classification}\n"
                f"- terminal_review_verdict: "
                f"{(terminal_review or {}).get('verdict')}\n"
                f"- tool: {CLEANUP_OBSOLETE_LEASE_TOOL}\n"
                "- action: posted terminal phase=released lease marker; "
                "did not transfer validation, decision, or workflow proof; "
                "did not repoint lease to the new head; did not delete history\n"
            )
            reasons.append(
                f"cleanup eligible ({classification}); post terminal released "
                "marker via sanctioned tool"
            )
        else:
            reasons.extend(fail_closed_reasons)

    return {
        "pr_number": pr_number,
        "classification": classification,
        "blocker_kind": classification if not cleanup_allowed else "none",
        "cleanup_allowed": cleanup_allowed,
        "mutation_allowed": False,  # never grants review mutation
        "mutation_eligibility": "prohibited" if not cleanup_allowed else "cleanup_only",
        "exact_next_action": (
            NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
            if cleanup_allowed
            else (
                NEXT_ACTION_WAIT
                if classification
                in {
                    "foreign_active_current_head",
                    "ambiguous_conflicting_evidence",
                }
                else NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP
            )
        ),
        "cleanup_tool": CLEANUP_OBSOLETE_LEASE_TOOL,
        "required_confirmation": cleanup_confirmation_for_pr(pr_number),
        "leased_head": leased_head,
        "current_head": current_head,
        "head_superseded": head_superseded,
        "past_expiry": past_expiry,
        "expires_at": (lease or {}).get("expires_at") if lease else None,
        "expires_parseable": expires_parseable,
        "terminal_review": terminal_review,
        "terminal_review_present": terminal_review is not None,
        "worktree_state": {
            "path": (lease or {}).get("worktree") if lease else None,
            "exists": worktree_exists,
            "clean": worktree_clean,
            "has_unpreserved_work": worktree_has_unpreserved_work,
        },
        "owner_session_evidence": {
            "session_id": (lease or {}).get("session_id") if lease else None,
            "reviewer_identity": (
                (lease or {}).get("reviewer_identity") if lease else None
            ),
            "process_alive": owner_process_alive,
            "pid_observed": owner_pid_observed,
            "pid_is_not_ownership_proof": True,
            "requesting_session_is_owner": bool(
                lease
                and req_session
                and (lease.get("session_id") or "").strip() == req_session
            ),
        },
        "active_lease": lease,
        "release_body": release_body,
        "audit_comment_body": audit_comment_body,
        "controller_recovery_authorized": controller_recovery_authorized,
        "reasons": reasons if reasons else fail_closed_reasons,
        "fail_closed_reasons": fail_closed_reasons,
        "forbidden": [
            "manual comment deletion",
            "database edits",
            "mtime manipulation",
            "PID-based ownership",
            "direct session-state seeding",
            "lease stealing or adoption",
            "repointing old lease to new head",
            "transfer of validation or decision state",
            "worktree reuse by replacement reviewer",
        ],
    }


def diagnose_reviewer_pr_lease_handoff(
    comments: list[dict],
    *,
    pr_number: int,
    current_session_id: str | None,
    current_reviewer_identity: str | None,
    proposed_worktree: str | None = None,
    env_bound_worktree: str | None = None,
    instructed_session_id: str | None = None,
    instructed_comment_id: int | None = None,
    current_head_sha: str | None = None,
    formal_reviews: list[dict] | None = None,
    worktree_exists: bool | None = None,
    worktree_clean: bool | None = None,
    owner_process_alive: bool | None = None,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Classify open-PR reviewer lease handoff and emit a canonical next action (#599, #691).

    Read-only diagnosis. Never steals, releases, or adopts a foreign lease.
    Fail-closed acquisition rules for active foreign leases remain intact.

    Returns a structured diagnosis with:
    - classification (including #691 superseded/expired distinctions)
    - next_action (including cleanup_obsolete_reviewer_comment_lease)
    - active_lease identity fields when present
    - worktree_binding match result
    - instructed-lease mismatch flags
    - cleanup tool/confirmation when cleanup-eligible
    """
    now = now or datetime.now(timezone.utc)
    session_id = (current_session_id or "").strip()
    identity = (current_reviewer_identity or "").strip()
    instructed_sid = (instructed_session_id or "").strip()
    reasons: list[str] = []
    current_head = _normalize_sha(current_head_sha)

    active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
    # Also surface expired non-terminal newest marker for #691 diagnosis.
    newest_nt = find_newest_nonterminal_lease(
        comments, pr_number=pr_number, now=now, include_expired=True
    )
    lease_for_class = active or newest_nt
    session_lease = get_session_lease()

    # Worktree binding: env-bound vs proposed vs active lease worktree.
    env_wt = _norm_path(env_bound_worktree)
    prop_wt = _norm_path(proposed_worktree)
    lease_wt = _norm_path((lease_for_class or {}).get("worktree") if lease_for_class else None)
    binding_mismatch = False
    binding_details: dict[str, Any] = {
        "env_bound_worktree": env_bound_worktree or None,
        "proposed_worktree": proposed_worktree or None,
        "lease_worktree": (lease_for_class or {}).get("worktree") if lease_for_class else None,
        "match": True,
    }
    paths = [p for p in (env_wt, prop_wt, lease_wt) if p]
    if len(paths) >= 2 and len(set(paths)) > 1:
        binding_mismatch = True
        binding_details["match"] = False
        reasons.append(
            "worktree binding mismatch: env/proposed/lease worktree paths disagree"
        )

    # Instructed lease gone while a different lease is active (PR #592-style).
    instructed_missing_with_replacement = False
    if instructed_sid or instructed_comment_id is not None:
        if not lease_for_class:
            reasons.append(
                "instructed lease is gone and no active replacement lease remains"
            )
        else:
            owner = (lease_for_class.get("session_id") or "").strip()
            cid = lease_for_class.get("comment_id")
            sid_mismatch = bool(instructed_sid and owner and owner != instructed_sid)
            cid_mismatch = (
                instructed_comment_id is not None
                and cid is not None
                and int(cid) != int(instructed_comment_id)
            )
            if sid_mismatch or cid_mismatch:
                instructed_missing_with_replacement = True
                reasons.append(
                    "instructed lease is gone; a different active lease replaced it "
                    f"(active session_id={owner}, comment_id={cid})"
                )

    # Classification + next_action.
    classification = "no_lease"
    next_action = NEXT_ACTION_ACQUIRE
    cleanup_hint: dict[str, Any] | None = None

    if lease_for_class:
        owner = (lease_for_class.get("session_id") or "").strip()
        freshness = lease_for_class.get("freshness") or classify_lease_freshness(
            lease_for_class, now=now
        )
        owner_identity = (lease_for_class.get("reviewer_identity") or "").strip()
        is_own = bool(session_id and owner and owner == session_id)
        # Same identity alone is NOT ownership for resume; session_id must match.
        same_identity = bool(
            identity and owner_identity and identity == owner_identity
        )
        leased_head = _normalize_sha(lease_for_class.get("candidate_head"))
        head_superseded = bool(
            current_head and leased_head and current_head != leased_head
        )
        head_current = bool(
            current_head and leased_head and current_head == leased_head
        )
        past_expiry = bool(lease_for_class.get("expired")) or freshness == "expired"
        if not past_expiry:
            past_expiry = _lease_expired(lease_for_class, now=now)
        terminal = formal_terminal_review_for_head(
            formal_reviews, leased_head=leased_head
        )

        if is_own and freshness in {"active", "stale_warning"} and not past_expiry:
            classification = "own_active"
            next_action = NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
        elif is_own and (freshness in {"reclaimable", "expired"} or past_expiry):
            classification = "own_expired"
            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
            reasons.append(
                f"own lease freshness is '{freshness}'; release via "
                "gitea_release_reviewer_pr_lease then re-acquire"
            )
        elif not is_own and head_superseded and terminal and past_expiry:
            classification = "foreign_expired_superseded_head"
            next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
            reasons.append(
                "foreign lease expired and pinned to superseded head with "
                "formal terminal review; use guarded obsolete-lease cleanup"
            )
        elif not is_own and head_superseded and terminal and not past_expiry:
            classification = "foreign_completed_superseded_head"
            next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
            reasons.append(
                "foreign lease pinned to superseded head with completed formal "
                "review; use guarded obsolete-lease cleanup (do not wait indefinitely)"
            )
        elif not is_own and head_superseded and not terminal and past_expiry:
            # Crash-orphan shape after canonical expiry (#702): the expired
            # marker no longer gates acquisition, so a fresh reviewer may
            # acquire at the current head. Guarded ledger cleanup becomes
            # available only with provable owner-exit evidence.
            classification = "orphaned_expired_superseded_head"
            if owner_process_alive is False and (
                worktree_clean is True or worktree_exists is False
            ):
                next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
                reasons.append(
                    "expired superseded-head lease with no terminal review and "
                    "provable owner-process exit (crash orphan, #702); guarded "
                    "obsolete-lease cleanup eligible"
                )
            else:
                next_action = NEXT_ACTION_ACQUIRE
                reasons.append(
                    "lease head superseded and lease past canonical expiry with "
                    "no formal terminal review (crash orphan, #702); expired "
                    "marker no longer gates acquisition — acquire a fresh lease "
                    "at the current head; guarded cleanup needs "
                    "owner_process_alive=false and a clean/absent worktree"
                )
        elif not is_own and head_superseded and not terminal:
            classification = "ambiguous_conflicting_evidence"
            next_action = NEXT_ACTION_WAIT
            reasons.append(
                "lease head superseded but no formal terminal review for leased "
                "head; fail closed / wait (no indefinite steal)"
            )
        elif not is_own and head_current and past_expiry:
            classification = "foreign_expired_current_head"
            if owner_process_alive is False and worktree_clean is True:
                classification = "orphaned_owner_missing"
                next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
                reasons.append(
                    "foreign expired lease on current head; owner process absent "
                    "and worktree clean — guarded orphan cleanup eligible"
                )
            elif owner_process_alive is False and worktree_clean is False:
                classification = "ambiguous_conflicting_evidence"
                next_action = NEXT_ACTION_WAIT
                reasons.append(
                    "owner process absent but worktree dirty; cleanup denied"
                )
            else:
                next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
                reasons.append(
                    "foreign expired lease on current head; use guarded cleanup "
                    "when worktree/process evidence permits"
                )
        elif not is_own and freshness in {"active", "stale_warning"} and not past_expiry:
            classification = (
                "foreign_active_current_head" if head_current or not current_head
                else "foreign_active"
            )
            next_action = NEXT_ACTION_WAIT
            reasons.append(
                f"foreign active reviewer lease (session_id={owner}, "
                f"phase={lease_for_class.get('phase')}, freshness={freshness}); "
                "do not submit; do not steal"
            )
            if same_identity:
                reasons.append(
                    "lease identity matches current reviewer but session_id differs; "
                    "resume only from the exact owner session_id or wait"
                )
        elif not is_own and freshness == "reclaimable":
            classification = "foreign_reclaimable"
            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
            reasons.append(
                f"foreign reclaimable lease (session_id={owner}); clear only via "
                "sanctioned gitea_release_reviewer_pr_lease when reclaimable"
            )
        elif not is_own and (freshness == "expired" or past_expiry):
            classification = "foreign_expired"
            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
            reasons.append(
                f"foreign expired lease (session_id={owner}); use sanctioned release "
                f"or {CLEANUP_OBSOLETE_LEASE_TOOL}"
            )
        else:
            classification = "foreign_active"
            next_action = NEXT_ACTION_WAIT
            reasons.append(
                f"unclassified active lease state (session_id={owner}, "
                f"freshness={freshness}); wait fail-closed"
            )

        if instructed_missing_with_replacement and classification.startswith(
            "foreign"
        ):
            # Preserve #691 cleanup path when superseded/expired; otherwise
            # keep the PR #592 instructed-missing label.
            if next_action != NEXT_ACTION_CLEANUP_OBSOLETE_LEASE:
                classification = "instructed_lease_missing_with_replacement"
                if next_action == NEXT_ACTION_WAIT:
                    reasons.append(
                        "replacement foreign lease is active — wait; "
                        "operator_authorized_cleanup only with explicit operator authority"
                    )

        if next_action == NEXT_ACTION_CLEANUP_OBSOLETE_LEASE:
            cleanup_hint = {
                "cleanup_tool": CLEANUP_OBSOLETE_LEASE_TOOL,
                "required_confirmation": cleanup_confirmation_for_pr(pr_number),
                "controller_recovery_authorized_required": True,
                "leased_head": leased_head,
                "current_head": current_head,
                "expires_at": lease_for_class.get("expires_at"),
                "terminal_review_verdict": (terminal or {}).get("verdict"),
            }
    else:
        classification = "no_lease"
        next_action = NEXT_ACTION_ACQUIRE
        reasons.append("no active reviewer lease; acquire via gitea_acquire_reviewer_pr_lease")

    # Binding mismatch is a first-class blocker before submit, but does not
    # erase foreign-lease wait/release guidance. Override next_action only when
    # the session would otherwise be free to acquire or resume (submit path).
    if binding_mismatch:
        if next_action in {
            NEXT_ACTION_ACQUIRE,
            NEXT_ACTION_RESUME_EXACT_OWNER_SESSION,
        }:
            classification = "worktree_binding_mismatch"
            next_action = NEXT_ACTION_REPAIR_WORKTREE_BINDING
        else:
            reasons.append(
                "also repair worktree binding before submit "
                f"(next_action remains {next_action})"
            )

    lease_summary = None
    if lease_for_class:
        lease_summary = {
            "comment_id": lease_for_class.get("comment_id"),
            "session_id": lease_for_class.get("session_id"),
            "phase": lease_for_class.get("phase"),
            "candidate_head": lease_for_class.get("candidate_head"),
            "expires_at": lease_for_class.get("expires_at"),
            "last_activity": lease_for_class.get("last_activity"),
            "freshness": lease_for_class.get("freshness")
            or classify_lease_freshness(lease_for_class, now=now),
            "reviewer_identity": lease_for_class.get("reviewer_identity"),
            "profile": lease_for_class.get("profile"),
            "worktree": lease_for_class.get("worktree"),
            "blocker": lease_for_class.get("blocker"),
        }

    return {
        "pr_number": pr_number,
        "classification": classification,
        "blocker_kind": classification,
        "next_action": next_action,
        "exact_next_action": next_action,
        "active_lease": lease_summary,
        "session_lease": session_lease,
        "worktree_binding": binding_details,
        "leased_head": (lease_summary or {}).get("candidate_head"),
        "current_head": current_head,
        "expires_at": (lease_summary or {}).get("expires_at"),
        "terminal_review_state": (
            formal_terminal_review_for_head(
                formal_reviews,
                leased_head=(lease_summary or {}).get("candidate_head"),
            )
            if lease_summary
            else None
        ),
        "worktree_state": {
            "exists": worktree_exists,
            "clean": worktree_clean,
            "path": (lease_summary or {}).get("worktree"),
        },
        "owner_session_evidence": {
            "session_id": (lease_summary or {}).get("session_id"),
            "process_alive": owner_process_alive,
            "pid_is_not_ownership_proof": True,
        },
        "cleanup": cleanup_hint,
        "cleanup_tool": (cleanup_hint or {}).get("cleanup_tool"),
        "required_confirmation": (cleanup_hint or {}).get("required_confirmation"),
        "instructed_session_id": instructed_session_id,
        "instructed_comment_id": instructed_comment_id,
        "instructed_lease_missing_with_replacement": instructed_missing_with_replacement,
        "mutation_allowed": (
            next_action == NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
            and not binding_mismatch
            and bool(session_lease)
        ),
        "mutation_eligibility": (
            "allowed"
            if (
                next_action == NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
                and not binding_mismatch
                and bool(session_lease)
            )
            else (
                "cleanup_only"
                if next_action == NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
                else "prohibited"
            )
        ),
        "reasons": reasons,
        "forbidden": [
            "manual lock deletion",
            "raw API bypass",
            "mtime manipulation",
            "direct _SESSION_LEASE seeding",
            "silent foreign lease steal",
            "PID-based ownership",
            "repointing old lease to new head",
            "transfer of validation or decision state",
        ],
    }
