Gitea-Tools/reviewer_pr_lease.py

"""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)

_TERMINAL_PHASES = frozenset({"done", "released", "blocked"})
_ACTIVE_PHASES = frozenset({
    "claimed",
    "validating",
    "approved",
    "request-changes",
    "merging",
    "adopted",
})

DEFAULT_LEASE_TTL_MINUTES = 120
STALE_WARNING_MINUTES = 30
RECLAIMABLE_MINUTES = 60

_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,
    blocker: str = "none",
) -> str:
    now = last_activity or datetime.now(timezone.utc)
    expires = expires_at or (now + timedelta(minutes=DEFAULT_LEASE_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 classify_lease_freshness(lease: dict, *, now: datetime | None = None) -> str:
    """Return active, stale_warning, reclaimable, expired, or terminal."""
    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 >= RECLAIMABLE_MINUTES:
        return "reclaimable"
    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
    return dict(_SESSION_LEASE)


def clear_session_lease() -> None:
    global _SESSION_LEASE
    _SESSION_LEASE = None


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


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).
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"

_HANDOFF_CLASSIFICATIONS = frozenset({
    "no_lease",
    "own_active",
    "own_expired",
    "foreign_active",
    "foreign_reclaimable",
    "foreign_expired",
    "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 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,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Classify open-PR reviewer lease handoff and emit a canonical next action (#599).

    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
    - next_action (one of wait / resume_exact_owner_session /
      release_expired_lease / operator_authorized_cleanup /
      repair_worktree_binding / acquire)
    - active_lease identity fields when present
    - worktree_binding match result
    - instructed-lease mismatch flags
    """
    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] = []

    active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
    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((active or {}).get("worktree") if active 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": (active or {}).get("worktree") if active 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 active:
            reasons.append(
                "instructed lease is gone and no active replacement lease remains"
            )
        else:
            owner = (active.get("session_id") or "").strip()
            cid = active.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

    if active:
        owner = (active.get("session_id") or "").strip()
        freshness = active.get("freshness") or classify_lease_freshness(
            active, now=now
        )
        owner_identity = (active.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
        )

        if is_own and freshness in {"active", "stale_warning"}:
            classification = "own_active"
            next_action = NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
        elif is_own and freshness in {"reclaimable", "expired"}:
            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 freshness in {"active", "stale_warning"}:
            classification = "foreign_active"
            next_action = NEXT_ACTION_WAIT
            reasons.append(
                f"foreign active reviewer lease (session_id={owner}, "
                f"phase={active.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":
            classification = "foreign_expired"
            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
            reasons.append(
                f"foreign expired lease (session_id={owner}); use sanctioned release"
            )
        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"
        ):
            classification = "instructed_lease_missing_with_replacement"
            # Foreign active still means wait; reclaimable still release.
            if next_action == NEXT_ACTION_WAIT:
                reasons.append(
                    "replacement foreign lease is active — wait; "
                    "operator_authorized_cleanup only with explicit operator authority"
                )
    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 active:
        lease_summary = {
            "comment_id": active.get("comment_id"),
            "session_id": active.get("session_id"),
            "phase": active.get("phase"),
            "candidate_head": active.get("candidate_head"),
            "expires_at": active.get("expires_at"),
            "last_activity": active.get("last_activity"),
            "freshness": active.get("freshness")
            or classify_lease_freshness(active, now=now),
            "reviewer_identity": active.get("reviewer_identity"),
            "profile": active.get("profile"),
            "worktree": active.get("worktree"),
            "blocker": active.get("blocker"),
        }

    return {
        "pr_number": pr_number,
        "classification": classification,
        "next_action": next_action,
        "active_lease": lease_summary,
        "session_lease": session_lease,
        "worktree_binding": binding_details,
        "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)
        ),
        "reasons": reasons,
        "forbidden": [
            "manual lock deletion",
            "raw API bypass",
            "mtime manipulation",
            "direct _SESSION_LEASE seeding",
            "silent foreign lease steal",
        ],
    }