Gitea-Tools/bootstrap_lock_recovery.py

"""Target-specific recovery for incomplete bootstrap issue locks (#953).

The situation this exists for: ``gitea_bootstrap_author_issue_worktree``
reported success, wrote an incomplete lock, and told the author to implement.
The author did — legitimately, following the tool's own reported next action —
and the branch now carries real committed and pushed work. At that point every
pre-existing recovery path is simultaneously ineligible:

* heartbeat refuses, because the claimant is not where it looks;
* ``gitea_lock_issue`` refuses, because the branch is no longer base-equivalent;
* #760 exact-owner renewal never engages, because a lock with no recorded
  expiry is never *expired*;
* the #447 create-PR guard refuses, because there is no provenance.

Distinct from every neighbouring path: #753 ``issue_lock_recovery`` requires a
dead owner PID, #760 ``issue_lock_renewal`` requires an *expired* lease, and
#442 ``issue_lock_adoption`` decides branch adoption. None of them addresses a
lock that is structurally incomplete and therefore never expires at all.

**What this will not do.** It never moves, resets, or rewinds a branch, and
never requires base-equivalence — the committed work is the thing being
preserved. It never pushes and never opens a pull request. It touches only the
one lock file named by (remote, org, repo, issue). It accepts no caller-supplied
provenance and no caller-supplied authorization flag; both are minted
server-side. It refuses a healthy foreign-owned lock outright, and a matching
username alone is never accepted as proof of ownership — the profile must match
too, and the lock's recorded binding must agree with the observed branch,
worktree, and head.
"""

from __future__ import annotations

import os
from typing import Any, Mapping

import author_lock_contract
import issue_lock_store

#: Refusal codes, so callers can branch on cause rather than parse prose.
REFUSAL_NO_LOCK = "no_durable_lock"
REFUSAL_ALREADY_CANONICAL = "already_canonical"
REFUSAL_FOREIGN_CLAIMANT = "foreign_claimant"
REFUSAL_HEALTHY_FOREIGN = "healthy_foreign_lock"
REFUSAL_IDENTITY_UNRESOLVED = "identity_unresolved"
REFUSAL_BINDING_MISMATCH = "binding_mismatch"
REFUSAL_WORKTREE_INVALID = "worktree_invalid"
REFUSAL_HEAD_MISMATCH = "head_mismatch"


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


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


def assess_bootstrap_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,
    observed_head: str | None,
    declared_head: str | None,
    worktree_exists: bool,
    worktree_registered: bool,
    current_branch: str | None,
    now: Any = None,
) -> dict[str, Any]:
    """Decide whether this exact lock may be upgraded by this exact caller.

    Pure: every input is an observation the caller already made, and nothing
    here reads or writes the filesystem, git, or Gitea. That is what makes the
    same decision testable in isolation and reusable by the read-only
    inspection surface, which must not mutate anything (AC16).

    Returns a dict with ``recovery_sanctioned`` plus the full evidence set. A
    refusal never raises — it reports, so the caller can surface exactly which
    piece of evidence was missing.
    """
    reasons: list[str] = []
    refusal_code: str | None = None

    contract = author_lock_contract.assess_lock_contract(existing_lock)

    if not existing_lock:
        return {
            "recovery_sanctioned": False,
            "refusal_code": REFUSAL_NO_LOCK,
            "reasons": [
                f"no durable issue lock exists for issue #{issue_number}; there is "
                "nothing to recover (fail closed)"
            ],
            "contract": contract,
            "evidence": {},
            "expected_generation": None,
        }

    active_identity = _text(identity)
    active_profile = _text(profile)
    recorded = author_lock_contract.lock_claimant(existing_lock)
    freshness = issue_lock_store.assess_lock_freshness(dict(existing_lock), now=now)
    generation = issue_lock_store.lock_generation(existing_lock)

    evidence: dict[str, Any] = {
        "recorded_claimant": recorded,
        "active_identity": active_identity,
        "active_profile": active_profile,
        "recorded_branch": existing_lock.get("branch_name"),
        "recorded_worktree": existing_lock.get("worktree_path"),
        "recorded_owner_session": existing_lock.get("owner_session"),
        "recorded_generation": generation,
        "recorded_remote": existing_lock.get("remote"),
        "recorded_org": existing_lock.get("org"),
        "recorded_repo": existing_lock.get("repo"),
        "observed_head": _text(observed_head),
        "declared_head": _text(declared_head),
        "current_branch": _text(current_branch),
        "worktree_exists": bool(worktree_exists),
        "worktree_registered": bool(worktree_registered),
        "freshness": freshness,
        "claimant_placement": contract.get("claimant_placement"),
        "expiration_state": contract.get("expiration", {}).get("state"),
    }

    # ── Repository and issue identity (AC10) ──
    if _text(existing_lock.get("remote")) != _text(remote):
        reasons.append(
            f"recorded remote '{existing_lock.get('remote')}' does not match '{remote}'"
        )
        refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH
    if _text(existing_lock.get("org")) != _text(org):
        reasons.append(
            f"recorded org '{existing_lock.get('org')}' does not match '{org}'"
        )
        refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH
    if _text(existing_lock.get("repo")) != _text(repo):
        reasons.append(
            f"recorded repo '{existing_lock.get('repo')}' does not match '{repo}'"
        )
        refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH
    if existing_lock.get("issue_number") != issue_number:
        reasons.append(
            f"lock targets issue #{existing_lock.get('issue_number')}, not "
            f"#{issue_number}"
        )
        refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH

    # ── Branch and worktree binding (AC10) ──
    if _text(existing_lock.get("branch_name")) != _text(branch_name):
        reasons.append(
            f"recorded branch '{existing_lock.get('branch_name')}' does not match "
            f"'{branch_name}'"
        )
        refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH
    if not _same_realpath(existing_lock.get("worktree_path"), worktree_path):
        reasons.append(
            f"recorded worktree '{existing_lock.get('worktree_path')}' does not "
            f"match '{worktree_path}'"
        )
        refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH

    # ── The worktree is real, registered, and on the branch (AC10) ──
    # Deliberately no base-equivalence requirement and no constraint on how far
    # the branch has advanced: the whole point is that it already carries the
    # author's legitimate commits (AC9).
    if not worktree_exists:
        reasons.append(f"declared worktree '{worktree_path}' does not exist")
        refusal_code = refusal_code or REFUSAL_WORKTREE_INVALID
    if not worktree_registered:
        reasons.append(f"worktree '{worktree_path}' is not a registered git worktree")
        refusal_code = refusal_code or REFUSAL_WORKTREE_INVALID
    if _text(current_branch) != _text(branch_name):
        reasons.append(
            f"worktree is on branch '{_text(current_branch) or 'unknown'}', not "
            f"'{branch_name}'"
        )
        refusal_code = refusal_code or REFUSAL_WORKTREE_INVALID

    # ── Current head fencing (AC10) ──
    # The caller names the commit it believes it is recovering. A mismatch means
    # the worktree moved under the caller, so the decision is stale.
    if not _text(observed_head):
        reasons.append("could not observe the worktree head")
        refusal_code = refusal_code or REFUSAL_HEAD_MISMATCH
    elif _text(declared_head) and _text(declared_head) != _text(observed_head):
        reasons.append(
            f"declared head '{_text(declared_head)}' does not match observed head "
            f"'{_text(observed_head)}'"
        )
        refusal_code = refusal_code or REFUSAL_HEAD_MISMATCH

    # ── Ownership (AC10, AC11) ──
    # A matching username alone is never sufficient: the profile must match too,
    # and both are compared against server-resolved values the caller cannot set.
    if not active_identity or not active_profile:
        reasons.append(
            "active identity and profile could not both be resolved; ownership "
            "cannot be proven"
        )
        refusal_code = refusal_code or REFUSAL_IDENTITY_UNRESOLVED
    if not recorded["username"] or not recorded["profile"]:
        reasons.append(
            "durable lock does not record both a claimant username and profile"
        )
        refusal_code = refusal_code or REFUSAL_FOREIGN_CLAIMANT
    elif (
        recorded["username"] != active_identity
        or recorded["profile"] != active_profile
    ):
        # AC11: a foreign-owned lock is never recoverable through this path,
        # healthy or not. The healthy case is reported distinctly so the refusal
        # is legible, but both refuse.
        if freshness.get("live"):
            reasons.append(
                f"lock is owned by a healthy foreign claimant "
                f"'{recorded['username']}/{recorded['profile']}'; takeover is not "
                "a recovery path"
            )
            refusal_code = REFUSAL_HEALTHY_FOREIGN
        else:
            reasons.append(
                f"lock claimant '{recorded['username']}/{recorded['profile']}' "
                f"does not match active '{active_identity}/{active_profile}'"
            )
            refusal_code = refusal_code or REFUSAL_FOREIGN_CLAIMANT

    # ── Nothing to recover ──
    # A lock that is already canonical is left strictly alone. Rewriting it would
    # mint a new task-session identifier and invalidate the heartbeat token the
    # legitimate owner is already using.
    if contract.get("canonical") and not reasons:
        return {
            "recovery_sanctioned": False,
            "refusal_code": REFUSAL_ALREADY_CANONICAL,
            "reasons": [
                "lock already satisfies the canonical contract; no recovery is "
                "required"
            ],
            "contract": contract,
            "evidence": evidence,
            "expected_generation": generation,
        }

    sanctioned = not reasons
    return {
        "recovery_sanctioned": sanctioned,
        "refusal_code": None if sanctioned else refusal_code,
        "reasons": reasons,
        "contract": contract,
        "evidence": evidence,
        "expected_generation": generation,
    }


def build_recovery_record(
    assessment: Mapping[str, Any],
    *,
    recovered_at: str,
    new_task_session_id: str,
) -> dict[str, Any]:
    """Auditable record of the ownership and generation transition (AC10).

    A recovered lock must never read as an original claim, so both sides of the
    transition are preserved: what the incomplete lock recorded, and what
    replaced it.
    """
    evidence = dict(assessment.get("evidence") or {})
    contract = dict(assessment.get("contract") or {})
    return {
        "recovery_kind": "incomplete_bootstrap_lock",
        "recovered_at": recovered_at,
        "prior_contract": contract.get("contract"),
        "prior_missing_fields": list(contract.get("missing_fields") or []),
        "prior_claimant_placement": evidence.get("claimant_placement"),
        "prior_expiration_state": evidence.get("expiration_state"),
        "prior_generation": evidence.get("recorded_generation"),
        "prior_owner_session": evidence.get("recorded_owner_session"),
        "prior_freshness": (evidence.get("freshness") or {}).get("status"),
        "replacement_task_session_id": new_task_session_id,
        "preserved_head": evidence.get("observed_head"),
        "branch_reset": False,
        "base_equivalence_required": False,
    }


def format_recovery_refusal(assessment: Mapping[str, Any]) -> str:
    reasons = "; ".join(
        assessment.get("reasons") or ["unknown bootstrap lock recovery refusal"]
    )
    code = assessment.get("refusal_code") or "refused"
    return f"Bootstrap lock recovery refused ({code}): {reasons} (fail closed)"