Gitea-Tools/issue_lock_renewal.py

"""Exact-owner renewal of an expired author issue lease (#760).

An author issue lease carries an absolute wall-clock expiry stamped once at
lock time. The PID recorded alongside it is the long-lived MCP daemon, not the
authoring task, so a lease that expires while its daemon is still up is the
ordinary case for any author task that outlives the TTL — not an anomaly.

Before this module, that case was unreachable.
``issue_lock_store.assess_same_issue_lease_conflict`` computed same-owner
evidence and then returned on the expired branch before consulting it, and
``assess_expired_lock_reclaim`` only permits takeover on a dead PID or a
missing worktree. An exact owner whose daemon is alive and whose worktree is
present satisfied neither, so its own lock became permanently unmodifiable
through sanctioned tools.

This module is the pure evidence assessor for that one narrow case. It answers
a single question: may *this* session renew a lease it can prove it already
owns? It performs no mutation and no network I/O, and it never trusts a caller
assertion — every field is compared against durable lock state or a live
observation supplied by the caller and gathered server-side.

Deliberate boundaries:

* **Renewal is not takeover.** A refusal here never widens what
  ``assess_expired_lock_reclaim`` already allows; foreign expired locks keep
  requiring a dead PID or missing worktree (#760 AC11), and a *live* foreign
  lease stays non-recoverable by construction because only an expired lease is
  ever a candidate (AC12).
* **PID liveness is never authorization.** A live recorded PID proves the
  daemon is up, nothing more. It is recorded as evidence and is neither
  necessary nor sufficient for renewal (AC16).
* **Absolute expiry is preserved.** Renewal issues a new absolute expiry from
  the moment of the write. It does not introduce sliding heartbeat renewal,
  lease generations as fencing tokens, or a shared cross-role lifecycle — that
  is #790's scope and is deliberately not implemented here.
"""

from __future__ import annotations

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

from issue_lock_store import AUTHOR_ISSUE_WORK_LEASE, is_lease_expired, is_process_alive
from reviewer_worktree import parse_dirty_tracked_files

# Outcome values.
RENEWAL_SANCTIONED = "RENEWAL_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 _text(value: Any) -> str:
    return str(value or "").strip()


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 _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 _lock_lease(lock: Mapping[str, Any]) -> dict[str, Any]:
    lease = lock.get("work_lease")
    return dict(lease) if isinstance(lease, Mapping) else {}


def _lock_operation_type(lock: Mapping[str, Any]) -> str:
    lease = _lock_lease(lock)
    return _text(lease.get("operation_type")) or AUTHOR_ISSUE_WORK_LEASE


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 _competing_lock_reasons(
    competing_live_locks: Iterable[Mapping[str, Any]] | None,
    *,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
) -> list[str]:
    """Live locks that would contend with this renewal (#760 AC7).

    A live lock on the *same* issue cannot coexist with this expired lease, so
    any live entry naming this issue, branch, or worktree belongs to somebody
    else and refuses the renewal.
    """
    reasons: list[str] = []
    for entry in competing_live_locks or ():
        if not isinstance(entry, Mapping):
            continue
        entry_issue = entry.get("issue_number")
        entry_branch = _text(entry.get("branch_name"))
        entry_worktree = _text(entry.get("worktree_path"))
        if entry_issue == issue_number:
            reasons.append(
                f"a live lock already exists for issue #{issue_number} "
                f"(pid {entry.get('pid')}); renewal would contend with it"
            )
            continue
        if entry_branch and entry_branch == _text(branch_name):
            reasons.append(
                f"live lock for issue #{entry_issue} already holds branch "
                f"'{branch_name}'"
            )
        if entry_worktree and _same_realpath(entry_worktree, worktree_path):
            reasons.append(
                f"live lock for issue #{entry_issue} already holds worktree "
                f"'{worktree_path}'"
            )
    return reasons


def assess_exact_owner_lease_renewal(
    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,
    operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
    current_branch: str | None = None,
    porcelain_status: str = "",
    worktree_exists: bool = False,
    head_sha: str | None = None,
    remote_head_sha: str | None = None,
    pr_head_sha: str | None = None,
    pr_number: int | None = None,
    competing_live_locks: Sequence[Mapping[str, Any]] | None = None,
    candidate_branches: Sequence[str] | None = None,
    current_pid: int | None = None,
    now: Any = None,
) -> dict[str, Any]:
    """Decide whether an expired lease may be renewed by its exact owner.

    Returns a disposition dict; it never raises and never mutates. A refusal
    withholds permission, leaving every pre-existing guard to fail closed
    exactly as before — this assessment can only ever *add* permission.

    ``NO_CANDIDATE`` means the situation is not an exact-owner renewal at all
    (no lock, different issue, different operation, or an unexpired lease) and
    the caller should carry on with its normal path. ``REFUSED`` means it looked
    like one but the evidence did not hold, and ``reasons`` names exactly what
    was missing.
    """
    evidence: dict[str, Any] = {
        "issue_number": issue_number,
        "branch_name": branch_name,
        "worktree_path": worktree_path,
        "remote": remote,
        "org": org,
        "repo": repo,
        "operation_type": operation_type,
        "identity": identity,
        "profile": profile,
    }

    def _result(outcome: str, reasons: list[str], **extra: Any) -> dict[str, Any]:
        return {
            "outcome": outcome,
            "renewal_sanctioned": outcome == RENEWAL_SANCTIONED,
            "is_candidate": outcome in (RENEWAL_SANCTIONED, REFUSED),
            "reasons": reasons,
            "evidence": {**evidence, **extra},
        }

    if not isinstance(existing_lock, Mapping) or not existing_lock:
        return _result(NO_CANDIDATE, ["no existing lock to renew"])

    if existing_lock.get("issue_number") != issue_number:
        return _result(
            NO_CANDIDATE,
            [
                f"existing lock is for issue #{existing_lock.get('issue_number')}, "
                f"not #{issue_number}"
            ],
        )

    existing_operation = _lock_operation_type(existing_lock)
    if existing_operation != operation_type:
        return _result(
            NO_CANDIDATE,
            [
                f"existing lease operation '{existing_operation}' is not "
                f"'{operation_type}'"
            ],
        )

    # Only an *expired* lease is ever a renewal candidate. An unexpired lease —
    # live, or stale by dead PID — is somebody else's problem: the first needs no
    # renewal, and the second is #753's dead-session recovery. This is also what
    # makes a live foreign lease non-recoverable here (#760 AC12).
    if not is_lease_expired(existing_lock, now=now):
        return _result(
            NO_CANDIDATE,
            ["lease has not expired; renewal does not apply"],
        )

    malformed = _malformed_reasons(existing_lock)
    if malformed:
        return _result(
            REFUSED,
            ["durable lock is missing or has unusable fields: " + ", ".join(malformed)],
        )

    lease = _lock_lease(existing_lock)
    claimant = _lock_claimant(existing_lock)
    recorded_pid = _recorded_pid(existing_lock)
    prior_expires_at = _text(lease.get("expires_at"))

    # #760 AC16: recorded purely as evidence. A live daemon PID is neither
    # necessary nor sufficient for renewal, and nothing below branches on it.
    recorded_pid_alive = is_process_alive(recorded_pid)

    extra: dict[str, Any] = {
        "prior_pid": recorded_pid,
        "prior_pid_alive": recorded_pid_alive,
        "prior_expires_at": prior_expires_at,
        "replacement_pid": current_pid,
        "recorded_claimant": claimant,
        "head_sha": head_sha,
        "remote_head_sha": remote_head_sha,
        "pr_head_sha": pr_head_sha,
        "pr_number": pr_number,
    }

    reasons: list[str] = []

    # ── AC3: exact ownership identity ──
    if _text(existing_lock.get("remote")) != _text(remote):
        reasons.append(
            f"recorded remote '{existing_lock.get('remote')}' does not match "
            f"'{remote}'"
        )
    if _text(existing_lock.get("org")) != _text(org):
        reasons.append(
            f"recorded org '{existing_lock.get('org')}' does not match '{org}'"
        )
    if _text(existing_lock.get("repo")) != _text(repo):
        reasons.append(
            f"recorded repo '{existing_lock.get('repo')}' does not match '{repo}'"
        )
    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}'"
        )
    if not _same_realpath(_text(existing_lock.get("worktree_path")), worktree_path):
        reasons.append(
            f"recorded worktree '{existing_lock.get('worktree_path')}' does not "
            f"match '{worktree_path}'"
        )

    recorded_identity = _text(claimant.get("username"))
    recorded_profile = _text(claimant.get("profile"))
    if not recorded_identity or not recorded_profile:
        reasons.append(
            "durable lock does not record both a claimant username and profile"
        )
    if recorded_identity and recorded_identity != _text(identity):
        reasons.append(
            f"recorded claimant '{recorded_identity}' does not match active "
            f"identity '{_text(identity) or 'unknown'}'"
        )
    if recorded_profile and recorded_profile != _text(profile):
        reasons.append(
            f"recorded profile '{recorded_profile}' does not match active profile "
            f"'{_text(profile) or 'unknown'}'"
        )

    # ── AC4: the registered worktree still exists, is on the branch, and is clean ──
    if not worktree_exists:
        reasons.append(f"declared worktree '{worktree_path}' does not exist")
    if _text(current_branch) != _text(branch_name):
        reasons.append(
            f"worktree is on branch '{_text(current_branch) or 'unknown'}', not "
            f"'{branch_name}'"
        )
    dirty = parse_dirty_tracked_files(porcelain_status or "")
    if dirty:
        reasons.append(
            "worktree has uncommitted tracked changes: " + ", ".join(sorted(dirty))
        )

    # ── AC5/AC6: published heads must agree ──
    if not _text(head_sha):
        reasons.append("local head could not be observed")
    if not _text(remote_head_sha):
        reasons.append(
            "remote branch head could not be observed; an unpublished branch "
            "cannot prove exact-owner renewal"
        )
    if _text(head_sha) and _text(remote_head_sha) and head_sha != remote_head_sha:
        reasons.append(
            f"local head {head_sha} does not equal remote head {remote_head_sha}"
        )
    if pr_number is not None:
        if not _text(pr_head_sha):
            reasons.append(f"owning PR #{pr_number} head could not be observed")
        elif _text(head_sha) and pr_head_sha != head_sha:
            reasons.append(
                f"owning PR #{pr_number} head {pr_head_sha} does not equal local "
                f"head {head_sha}"
            )

    # ── AC7: nothing else claims this work ──
    reasons.extend(
        _competing_lock_reasons(
            competing_live_locks,
            issue_number=issue_number,
            branch_name=branch_name,
            worktree_path=worktree_path,
        )
    )
    other_branches = [
        name
        for name in (candidate_branches or ())
        if _text(name) and _text(name) != _text(branch_name)
    ]
    if other_branches:
        reasons.append(
            "other branches already carry this issue marker: "
            + ", ".join(sorted(other_branches))
        )

    if reasons:
        return _result(REFUSED, reasons, **extra)

    return _result(
        RENEWAL_SANCTIONED,
        [
            f"exact owner '{recorded_identity}' ({recorded_profile}) proved "
            f"ownership of issue #{issue_number} on branch '{branch_name}' from "
            f"worktree '{worktree_path}'; local, remote"
            + (f", and PR #{pr_number}" if pr_number is not None else "")
            + f" heads all equal {head_sha}; lease expired at "
            f"{prior_expires_at or 'unknown'}"
        ],
        **extra,
    )


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

    The mirror of ``issue_lock_recovery.owning_pr_recovery_evidence`` (#755) for
    the renewal disposition. An exact-owner renewal of a published branch is, by
    construction, renewal of work that already has an open PR — so the
    duplicate-work gate's linked-open-PR blocker would otherwise discard every
    sanctioned renewal, exactly as it once discarded every sanctioned recovery.

    Returns ``None`` unless renewal was actually granted and the evidence names
    one owning PR whose head agrees with both the local and remote heads the
    assessor accepted. Nothing is caller-supplied: every field is copied from
    evidence built out of durable lock state plus live git/Gitea observation.

    Renewal has no descendant case — it requires the local, remote, and PR heads
    to be equal — so there is only one head to report.
    """
    if not isinstance(assessment, Mapping):
        return None
    if assessment.get("outcome") != RENEWAL_SANCTIONED:
        return None
    if not assessment.get("renewal_sanctioned"):
        return None

    evidence = assessment.get("evidence") or {}
    branch_name = _text(evidence.get("branch_name"))
    pr_head = _text(evidence.get("pr_head_sha"))
    local_head = _text(evidence.get("head_sha"))
    remote_head = _text(evidence.get("remote_head_sha"))
    raw_pr_number = evidence.get("pr_number")

    if raw_pr_number is None or not branch_name or not pr_head:
        return None
    # The assessor already required these to agree. Re-check, so a truncated or
    # hand-built evidence map can never authorize an exemption.
    if pr_head != local_head or pr_head != remote_head:
        return None
    try:
        pr_number = int(raw_pr_number)
        issue_number = int(evidence.get("issue_number"))
    except (TypeError, ValueError):
        return None

    return {
        "issue_number": issue_number,
        "pr_number": pr_number,
        "branch_name": branch_name,
        "head_sha": pr_head,
        "recorded_head": pr_head,
        "accepted_head": pr_head,
        "head_relation": "equal",
    }


def build_renewal_record(
    assessment: Mapping[str, Any] | None,
    *,
    renewed_at: str,
    new_expires_at: str,
) -> dict[str, Any]:
    """Durable audit record for a sanctioned renewal (#760 AC9).

    Records both sides of the transition — prior PID and expiry, replacement PID
    and new expiry — so a renewed lock is never mistakable for an original
    claim, and so the evidence the waiver was granted on stays inspectable.
    """
    data = dict(assessment or {})
    evidence = dict(data.get("evidence") or {})
    recorded_claimant = dict(evidence.get("recorded_claimant") or {})
    return {
        "renewed": bool(data.get("renewal_sanctioned")),
        "renewed_at": renewed_at,
        "prior_pid": evidence.get("prior_pid"),
        "prior_pid_alive": evidence.get("prior_pid_alive"),
        "prior_expires_at": evidence.get("prior_expires_at"),
        "replacement_pid": evidence.get("replacement_pid"),
        "new_expires_at": new_expires_at,
        "identity": recorded_claimant.get("username"),
        "profile": recorded_claimant.get("profile"),
        "branch_name": evidence.get("branch_name"),
        "worktree_path": evidence.get("worktree_path"),
        "head_sha": evidence.get("head_sha"),
        "remote_head_sha": evidence.get("remote_head_sha"),
        "pr_head_sha": evidence.get("pr_head_sha"),
        "pr_number": evidence.get("pr_number"),
        "reason": "expired lease renewed by its exact recorded owner",
        "proof": list(data.get("reasons") or []),
    }


def format_renewal_refusal(assessment: Mapping[str, Any] | None) -> str:
    """One-line refusal summary for a blocked caller."""
    data = dict(assessment or {})
    reasons = list(data.get("reasons") or [])
    if not reasons:
        return "exact-owner lease renewal was not available (no evidence recorded)"
    return "exact-owner lease renewal refused: " + "; ".join(reasons)