Gitea-Tools/issue_lock_recovery.py

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

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

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

This module is the pure evidence assessor for that one narrow case. It grants
recovery only when every element of durable ownership still matches exactly and
the recorded process is demonstrably dead. It never trusts caller assertions:
every field is compared against durable lock state or live observation supplied
by the caller. It performs no mutation and no network I/O.

Recovery deliberately does **not** relax base-equivalence for brand-new issue
claims — only for a lock whose own prior record already proves the branch,
worktree, head, and author.
"""

from __future__ import annotations

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

from issue_lock_store import is_process_alive
from reviewer_worktree import parse_dirty_tracked_files

# Outcome values
RECOVERY_SANCTIONED = "RECOVERY_SANCTIONED"
NO_CANDIDATE = "NO_CANDIDATE"
REFUSED = "REFUSED"

# Durable fields a lock must carry before it can be considered at all.
REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path")


def _same_realpath(left: str | None, right: str | None) -> bool:
    if not left or not right:
        return False
    try:
        return os.path.realpath(left) == os.path.realpath(right)
    except OSError:
        return left == right


def _text(value: Any) -> str:
    return str(value or "").strip()


def _lock_claimant(lock: Mapping[str, Any]) -> dict[str, Any]:
    claimant = lock.get("claimant")
    if not isinstance(claimant, Mapping):
        lease = lock.get("work_lease")
        claimant = lease.get("claimant") if isinstance(lease, Mapping) else None
    return dict(claimant) if isinstance(claimant, Mapping) else {}


def _recorded_pid(lock: Mapping[str, Any]) -> Any:
    pid = lock.get("session_pid")
    if pid is None:
        pid = lock.get("pid")
    return pid


def _malformed_reasons(lock: Mapping[str, Any]) -> list[str]:
    """Names of durable fields that are missing or unusable."""
    missing: list[str] = []
    for field in REQUIRED_LOCK_FIELDS:
        if not _text(lock.get(field)):
            missing.append(field)
    pid = _recorded_pid(lock)
    if pid is None or _text(pid) == "":
        missing.append("session_pid/pid")
    else:
        try:
            if int(pid) <= 0:
                missing.append("session_pid/pid")
        except (TypeError, ValueError):
            missing.append("session_pid/pid")
    return missing


def assess_dead_session_lock_recovery(
    existing_lock: Mapping[str, Any] | None,
    *,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    remote: str,
    org: str,
    repo: str,
    identity: str | None,
    profile: str | None,
    current_branch: str | None,
    porcelain_status: str,
    head_sha: str | None,
    remote_head_sha: str | None,
    pr_head_sha: str | None = None,
    pr_number: int | None = None,
    competing_live_locks: Sequence[Mapping[str, Any]] | None = None,
    candidate_branches: Iterable[str] | None = None,
    current_pid: int | None = None,
) -> dict[str, Any]:
    """Decide whether a dead-session author lock may be natively recovered.

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

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

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

    lock = dict(existing_lock)

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

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

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

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

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

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

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

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

    # ── Head agreement: local == remote == PR ───────────────────────────────
    local_head = _text(head_sha)
    remote_head = _text(remote_head_sha)
    if not local_head:
        reasons.append("local head SHA could not be determined")
    if not remote_head:
        reasons.append(
            f"remote head for branch '{locked_branch}' could not be determined"
        )
    if local_head and remote_head and local_head != remote_head:
        reasons.append(
            f"local head {local_head} does not match remote branch head {remote_head}"
        )
    evidence["local_head"] = local_head or None
    evidence["remote_head"] = remote_head or None

    pr_head = _text(pr_head_sha)
    if pr_head:
        evidence["pr_head"] = pr_head
        evidence["pr_number"] = pr_number
        if local_head and pr_head != local_head:
            reasons.append(
                f"open PR #{pr_number} head {pr_head} does not match local head "
                f"{local_head}"
            )

    # ── Author identity ─────────────────────────────────────────────────────
    claimant = _lock_claimant(lock)
    locked_identity = _text(claimant.get("username"))
    locked_profile = _text(claimant.get("profile"))
    evidence["locked_identity"] = locked_identity or None
    evidence["locked_profile"] = locked_profile or None
    if not locked_identity or not locked_profile:
        reasons.append(
            "durable lock does not record a claimant identity/profile; "
            "author ownership could not be proven"
        )
    if not _text(identity) or not _text(profile):
        reasons.append(
            "active session identity/profile is unknown; author ownership "
            "could not be proven"
        )
    if locked_identity and _text(identity) and locked_identity != _text(identity):
        reasons.append(
            f"lock claimant '{locked_identity}' does not match active identity "
            f"'{_text(identity)}'"
        )
    if locked_profile and _text(profile) and locked_profile != _text(profile):
        reasons.append(
            f"lock profile '{locked_profile}' does not match active profile "
            f"'{_text(profile)}'"
        )

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

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

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

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

    return _result(
        RECOVERY_SANCTIONED,
        True,
        [
            f"durable lock for issue #{issue_number} matches branch "
            f"'{locked_branch}', worktree '{locked_worktree}', head {local_head}, "
            f"and claimant '{locked_identity}'; recorded pid {recorded_pid} is dead"
        ],
        evidence,
    )


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


def build_recovery_record(
    assessment: Mapping[str, Any],
    *,
    recovered_at: str,
) -> dict[str, Any]:
    """Durable, secret-free provenance for a completed recovery (#753 AC2/AC6)."""
    evidence = dict(assessment.get("evidence") or {})
    return {
        "recovered": True,
        "reason": "owning MCP session exited; durable ownership evidence matched",
        "recovered_at": recovered_at,
        "prior_session_pid": evidence.get("prior_session_pid"),
        "replacement_session_pid": evidence.get("replacement_session_pid"),
        "prior_pid_alive": evidence.get("prior_pid_alive"),
        "branch_name": evidence.get("locked_branch"),
        "worktree_path": evidence.get("locked_worktree_path"),
        "local_head": evidence.get("local_head"),
        "remote_head": evidence.get("remote_head"),
        "pr_head": evidence.get("pr_head"),
        "pr_number": evidence.get("pr_number"),
        "identity": evidence.get("locked_identity"),
        "profile": evidence.get("locked_profile"),
        "proof": list(assessment.get("reasons") or []),
    }


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