Gitea-Tools/stale_review_decision_lock.py

"""Stale / moot #332 review-decision lock detection and cleanup policy (#594/#620).

#332 correctly hard-stops a reviewer session after a terminal live review
mutation. After #559 those locks are durable on disk, so they can outlive the
work they protect: when the referenced PR is already merged or closed, no
same-PR merge sequence remains, yet the durable ledger still blocks unrelated
new terminal reviews.

#620 scopes the terminal boundary by **reviewed head SHA**: a prior
REQUEST_CHANGES (or APPROVE) on head A must not block a fresh formal decision
on head B of the **same open PR**. Same-head duplicates remain fail-closed.
Open-PR lock *cleanup* remains forbidden unless the PR is truly merged/closed
(#594); new-head re-review is allowed without deleting the durable lock.

This module is pure policy (no Gitea I/O). The MCP tool
``gitea_cleanup_stale_review_decision_lock`` fetches live PR state, calls these
helpers, and only then clears durable state when cleanup is allowed.

Manual deletion of ``~/.cache/gitea-tools/session-state/*`` is never the
canonical path.
"""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any

# Keep in sync with gitea_mcp_server._TERMINAL_REVIEW_ACTIONS.
TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})


def normalize_head_sha(value: Any) -> str | None:
    """Normalize a git SHA for equality comparison; None if empty/invalid."""
    if value is None:
        return None
    text = str(value).strip().lower()
    if not text:
        return None
    return text


def heads_equal(a: Any, b: Any) -> bool:
    na = normalize_head_sha(a)
    nb = normalize_head_sha(b)
    if not na or not nb:
        return False
    return na == nb


def mutation_head_sha(mutation: dict | None, lock: dict | None = None) -> str | None:
    """Head SHA that bounds a live mutation (#620).

    Prefers fields recorded on the mutation itself. For *legacy* mutations
    that predate head-scoping, falls back to the lock's
    ``ready_expected_head_sha`` only when that ready binding is for the same
    PR number (the frozen durable PR #619-style ledger).
    """
    if not isinstance(mutation, dict):
        return None
    for key in ("head_sha", "expected_head_sha", "reviewed_head_sha"):
        head = normalize_head_sha(mutation.get(key))
        if head:
            return head
    if not isinstance(lock, dict):
        return None
    if lock.get("ready_pr_number") != mutation.get("pr_number"):
        return None
    return normalize_head_sha(lock.get("ready_expected_head_sha"))


def last_terminal_mutation(lock: dict | None) -> dict | None:
    """Return the last terminal live mutation on *lock*, or None."""
    if not lock:
        return None
    terminals = [
        m
        for m in (lock.get("live_mutations") or [])
        if isinstance(m, dict) and m.get("action") in TERMINAL_REVIEW_ACTIONS
    ]
    return terminals[-1] if terminals else None


def prior_live_mutations_block_boundary(
    lock: dict | None,
    *,
    pr_number: int,
    expected_head_sha: str | None,
) -> bool:
    """True when prior live mutations block a new decision for PR+head (#620).

    Historical terminals on a **different head of the same PR** do not block.
    Any prior mutation on another PR, the same head, or without a comparable
    head SHA fails closed (blocks).

    Head resolution uses ``mutation_head_sha(m, lock)`` so pre-#620 ledgers
    that only stored ``ready_expected_head_sha`` still compare correctly
    (PR #619-style).
    """
    if not lock:
        return False
    if lock.get("correction_authorized"):
        return False
    prior = list(lock.get("live_mutations") or [])
    if not prior:
        return False
    target = normalize_head_sha(expected_head_sha)
    for m in prior:
        if not isinstance(m, dict):
            return True
        m_pr = m.get("pr_number")
        m_head = mutation_head_sha(m, lock)
        if (
            m_pr == pr_number
            and target
            and m_head
            and not heads_equal(m_head, target)
        ):
            # Same open PR, different reviewed head — historical boundary only.
            continue
        return True
    return False


def terminal_boundary_allows_fresh_decision(
    lock: dict | None,
    *,
    pr_number: int,
    expected_head_sha: str | None,
    operation: str,
) -> bool:
    """Whether #332 hard-stop should yield for a new PR+head boundary (#620).

    Only for ``mark_ready`` / ``review`` / ``resume`` on the **same PR** when
    the requested head differs from the last terminal's head. Merge is never
    reopened by head change (approved head merge path is separate).
    """
    if operation not in ("mark_ready", "review", "resume"):
        return False
    if not lock:
        return False
    if lock.get("correction_authorized"):
        return True
    last = last_terminal_mutation(lock)
    if last is None:
        return True
    if last.get("pr_number") != pr_number:
        return False
    locked_head = mutation_head_sha(last, lock)
    target = normalize_head_sha(expected_head_sha)
    if not locked_head or not target:
        return False
    return not heads_equal(locked_head, target)


def backfill_terminal_heads_from_ready(lock: dict) -> dict:
    """Stamp head_sha onto legacy terminal mutations before overwriting ready_*.

    Called when marking a fresh decision on a new head so historical rows keep
    their original boundary after ``ready_expected_head_sha`` advances (#620).
    """
    if not isinstance(lock, dict):
        return lock
    ready_pr = lock.get("ready_pr_number")
    ready_head = normalize_head_sha(lock.get("ready_expected_head_sha"))
    if ready_pr is None or not ready_head:
        return lock
    mutations = list(lock.get("live_mutations") or [])
    changed = False
    for m in mutations:
        if not isinstance(m, dict):
            continue
        if m.get("action") not in TERMINAL_REVIEW_ACTIONS:
            continue
        if m.get("pr_number") != ready_pr:
            continue
        if mutation_head_sha(m):
            continue
        m["head_sha"] = ready_head
        changed = True
    if changed:
        lock["live_mutations"] = mutations
    return lock


def classify_pr_live_state(pr_live: dict | None) -> dict[str, Any]:
    """Normalize a Gitea PR payload into merge/closed flags."""
    pr = pr_live or {}
    merged = bool(pr.get("merged") or pr.get("merged_at"))
    state = (pr.get("state") or "").strip().lower() or None
    closed = state == "closed"
    head = pr.get("head") if isinstance(pr.get("head"), dict) else {}
    head_sha = normalize_head_sha(
        pr.get("head_sha") or pr.get("head_commit_sha") or head.get("sha")
    )
    return {
        "pr_state": state,
        "pr_merged": merged,
        "pr_merged_or_closed": merged or closed,
        "merge_commit_sha": pr.get("merge_commit_sha")
        or pr.get("merged_commit_sha"),
        "current_pr_head_sha": head_sha,
    }


def lock_summary(lock: dict | None) -> dict[str, Any] | None:
    """LLM-safe summary of a decision lock (no secrets)."""
    if lock is None:
        return None
    last = last_terminal_mutation(lock)
    mutations = list(lock.get("live_mutations") or [])
    last_head = mutation_head_sha(last, lock) if last else None
    return {
        "task": lock.get("task"),
        "remote": lock.get("remote"),
        "org": lock.get("org") or lock.get("ready_org"),
        "repo": lock.get("repo") or lock.get("ready_repo"),
        "profile_identity": lock.get("profile_identity")
        or lock.get("session_profile_lock")
        or lock.get("session_profile"),
        "session_profile": lock.get("session_profile"),
        "ready_pr_number": lock.get("ready_pr_number"),
        "ready_action": lock.get("ready_action"),
        "ready_expected_head_sha": normalize_head_sha(
            lock.get("ready_expected_head_sha")
        ),
        "live_mutations_count": len(mutations),
        "last_terminal": (
            {
                "pr_number": last.get("pr_number"),
                "action": last.get("action"),
                "review_id": last.get("review_id"),
                "head_sha": last_head,
            }
            if last
            else None
        ),
        "locked_head_sha": last_head,
        "correction_authorized": bool(lock.get("correction_authorized")),
        "updated_at": lock.get("updated_at") or lock.get("recorded_at"),
    }


def assess_stale_review_decision_lock(
    lock: dict | None,
    *,
    pr_live: dict | None = None,
    pr_lookup_error: str | None = None,
    active_profile_identity: str | None = None,
    current_pr_head_sha: str | None = None,
) -> dict[str, Any]:
    """Assess whether a durable review-decision lock is stale/moot (#594/#620).

    *pr_live* must be the live Gitea payload for the **last terminal
    mutation's PR**. When no lock / no terminal mutation exists, cleanup is
    not applicable. When the PR is still open (or lookup fails), cleanup is
    forbidden (#594). Open PR + different live head reports
    ``stale_by_head`` / ``fresh_review_on_current_head_allowed`` (#620)
    without enabling cleanup.
    """
    summary = lock_summary(lock)
    result: dict[str, Any] = {
        "has_lock": lock is not None,
        "lock_summary": summary,
        "last_terminal_pr": None,
        "last_terminal_action": None,
        "locked_head_sha": None,
        "current_pr_head_sha": normalize_head_sha(current_pr_head_sha),
        "stale_by_head": False,
        "fresh_review_on_current_head_allowed": False,
        "is_moot": False,
        "cleanup_allowed": False,
        "profile_match": True,
        "pr_state": None,
        "pr_merged": False,
        "pr_merged_or_closed": False,
        "merge_commit_sha": None,
        "reasons": [],
        "recovery_tool": "gitea_cleanup_stale_review_decision_lock",
    }

    if lock is None:
        result["reasons"].append("no review decision lock present")
        return result

    # Profile identity gate (caller may re-check with env; pure assess still
    # reports mismatch when active identity is supplied).
    stored = (
        (lock.get("profile_identity") or lock.get("session_profile_lock") or "")
        .strip()
    )
    active = (active_profile_identity or "").strip()
    if active and stored and active != stored:
        result["profile_match"] = False
        result["reasons"].append(
            "review decision lock profile identity does not match active "
            f"profile '{active}' (fail closed, #594)"
        )
        return result

    last = last_terminal_mutation(lock)
    if last is None:
        result["reasons"].append(
            "review decision lock has no terminal live mutations; nothing to clear"
        )
        return result

    pr_number = last.get("pr_number")
    action = last.get("action")
    locked_head = mutation_head_sha(last, lock)
    result["last_terminal_pr"] = pr_number
    result["last_terminal_action"] = action
    result["locked_head_sha"] = locked_head

    if pr_number is None:
        result["reasons"].append(
            "last terminal mutation missing pr_number; refuse cleanup (fail closed)"
        )
        return result

    if pr_lookup_error:
        result["reasons"].append(
            f"could not load live state for PR #{pr_number}: {pr_lookup_error} "
            "(fail closed, #594)"
        )
        return result

    if pr_live is None:
        result["reasons"].append(
            f"live PR state required for terminal PR #{pr_number} before cleanup "
            "(fail closed, #594)"
        )
        return result

    live = classify_pr_live_state(pr_live)
    current_head = normalize_head_sha(
        current_pr_head_sha or live.get("current_pr_head_sha")
    )
    result.update(
        {
            "pr_state": live["pr_state"],
            "pr_merged": live["pr_merged"],
            "pr_merged_or_closed": live["pr_merged_or_closed"],
            "merge_commit_sha": live["merge_commit_sha"],
            "current_pr_head_sha": current_head,
        }
    )

    if not live["pr_merged_or_closed"]:
        stale_by_head = bool(
            locked_head and current_head and not heads_equal(locked_head, current_head)
        )
        result["stale_by_head"] = stale_by_head
        # Fresh formal decision is allowed on the new head without cleanup (#620).
        result["fresh_review_on_current_head_allowed"] = stale_by_head
        if stale_by_head:
            result["reasons"].append(
                f"terminal {action} on PR #{pr_number} is bound to head "
                f"{locked_head[:12]}…; live head is {current_head[:12]}… — "
                "fresh formal review on current head is allowed without "
                "cleanup (fail closed for cleanup, #620); #332 same-head "
                "duplicate protection remains"
            )
        else:
            result["reasons"].append(
                f"terminal review mutation on open PR #{pr_number} is still active "
                f"({action}); #332 hard-stop remains — cleanup forbidden (#594)"
            )
        return result

    # Merged or closed: lock is moot. Same-PR merge continuation is impossible.
    result["is_moot"] = True
    result["cleanup_allowed"] = True
    result["stale_by_head"] = False
    result["fresh_review_on_current_head_allowed"] = False
    result["reasons"].append(
        f"terminal mutation ({action} on PR #{pr_number}) is moot: PR is "
        f"{'merged' if live['pr_merged'] else 'closed'}; canonical cleanup allowed (#594)"
    )
    return result


def build_cleanup_audit_record(
    *,
    assessment: dict[str, Any],
    actor_username: str | None,
    profile_name: str | None,
    applied: bool,
) -> dict[str, Any]:
    """Structured durable audit payload for a cleanup attempt."""
    last = (assessment.get("lock_summary") or {}).get("last_terminal") or {}
    return {
        "event": "stale_review_decision_lock_cleanup",
        "issue_ref": "#594",
        "applied": bool(applied),
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "actor_username": actor_username,
        "profile_name": profile_name,
        "last_terminal_pr": assessment.get("last_terminal_pr") or last.get("pr_number"),
        "last_terminal_action": assessment.get("last_terminal_action")
        or last.get("action"),
        "pr_state": assessment.get("pr_state"),
        "pr_merged": assessment.get("pr_merged"),
        "pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
        "merge_commit_sha": assessment.get("merge_commit_sha"),
        "prior_live_mutations_count": (
            (assessment.get("lock_summary") or {}).get("live_mutations_count")
        ),
        "prior_profile_identity": (
            (assessment.get("lock_summary") or {}).get("profile_identity")
        ),
        "cleanup_allowed": assessment.get("cleanup_allowed"),
        "is_moot": assessment.get("is_moot"),
        "reasons": list(assessment.get("reasons") or []),
    }


def format_cleanup_audit_comment(audit: dict[str, Any]) -> str:
    """Markdown body for a durable Gitea audit comment after cleanup."""
    status = "APPLIED" if audit.get("applied") else "ASSESSED (not applied)"
    lines = [
        "## Stale #332 review-decision lock cleanup (#594)",
        "",
        f"Status: **{status}**",
        "",
        f"- actor: `{audit.get('actor_username')}`",
        f"- profile: `{audit.get('profile_name')}`",
        f"- timestamp: `{audit.get('timestamp')}`",
        f"- last terminal: `{audit.get('last_terminal_action')}` on "
        f"PR #{audit.get('last_terminal_pr')}",
        f"- PR state: `{audit.get('pr_state')}` "
        f"(merged={audit.get('pr_merged')})",
        f"- merge_commit_sha: `{audit.get('merge_commit_sha')}`",
        f"- prior live_mutations_count: "
        f"`{audit.get('prior_live_mutations_count')}`",
        f"- prior profile_identity: `{audit.get('prior_profile_identity')}`",
        "",
        "Manual deletion of session-state files is **not** the workflow.",
        "This path only clears a lock when the referenced PR is merged/closed.",
    ]
    return "\n".join(lines)