Gitea-Tools/merger_lease_adoption.py

"""Guarded merger adoption of an existing reviewer PR lease (#536).

Replaces manual in-process ``_SESSION_LEASE`` seeding with an auditable,
comment-backed adoption path for merger sessions that inherit a reviewer lease
after formal approval at the current PR head.
"""

from __future__ import annotations

import re
from datetime import datetime, timezone
from typing import Any

import reviewer_pr_lease as leases

ADOPTION_MARKER = "<!-- mcp-review-lease-adoption:v1 -->"
DEFAULT_ADOPTION_REASON = "merger-handoff-approved-head"

SOURCE_ADOPT = "gitea_adopt_merger_pr_lease"
SOURCE_ACQUIRE = "gitea_acquire_reviewer_pr_lease"
SOURCE_HEARTBEAT = "gitea_heartbeat_reviewer_pr_lease"

SANCTIONED_PROVENANCE_SOURCES = frozenset({
    SOURCE_ADOPT,
    SOURCE_ACQUIRE,
    SOURCE_HEARTBEAT,
})

_MERGER_ADOPTABLE_FRESHNESS = frozenset({"active", "stale_warning"})


def format_adoption_body(
    *,
    repo: str,
    pr_number: int,
    issue_number: int | None,
    adopter_identity: str,
    adopter_profile: str,
    adopter_session_id: str,
    worktree: str,
    candidate_head: str | None,
    target_branch: str,
    target_branch_sha: str | None,
    adopted_from_session_id: str,
    adopted_from_profile: str,
    adopted_from_reviewer_identity: str,
    adopted_from_comment_id: int | None,
    adoption_reason: str = DEFAULT_ADOPTION_REASON,
    adopted_at: datetime | None = None,
) -> str:
    """Render a durable merger adoption proof comment."""
    adopted_at = adopted_at or datetime.now(timezone.utc)
    adopted_text = adopted_at.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
        "+00:00", "Z"
    )
    lease_body = leases.format_lease_body(
        repo=repo,
        pr_number=pr_number,
        issue_number=issue_number,
        reviewer_identity=adopter_identity,
        profile=adopter_profile,
        session_id=adopter_session_id,
        worktree=worktree,
        phase="adopted",
        candidate_head=candidate_head,
        target_branch=target_branch,
        target_branch_sha=target_branch_sha,
        last_activity=adopted_at,
        blocker="none",
    )
    lines = [
        ADOPTION_MARKER,
        f"adopted_at: {adopted_text}",
        f"adopted_by_identity: {adopter_identity}",
        f"adopted_by_profile: {adopter_profile}",
        f"adopted_from_session_id: {adopted_from_session_id}",
        f"adopted_from_profile: {adopted_from_profile}",
        f"adopted_from_reviewer_identity: {adopted_from_reviewer_identity}",
        f"adopted_from_comment_id: {adopted_from_comment_id or 'none'}",
        f"adoption_reason: {adoption_reason}",
        lease_body,
    ]
    return "\n".join(lines)


def is_adoption_comment(body: str) -> bool:
    return ADOPTION_MARKER in (body or "")


def build_lease_provenance(
    *,
    source: str,
    comment_id: int | None = None,
    adopted_from_session_id: str | None = None,
    adopted_from_profile: str | None = None,
    adopted_from_reviewer_identity: str | None = None,
    adoption_reason: str | None = None,
    recorded_at: datetime | None = None,
) -> dict[str, Any]:
    recorded_at = recorded_at or datetime.now(timezone.utc)
    recorded_text = recorded_at.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
        "+00:00", "Z"
    )
    proof = {
        "source": source,
        "recorded_at": recorded_text,
    }
    if comment_id is not None:
        proof["comment_id"] = comment_id
    if adopted_from_session_id:
        proof["adopted_from_session_id"] = adopted_from_session_id
    if adopted_from_profile:
        proof["adopted_from_profile"] = adopted_from_profile
    if adopted_from_reviewer_identity:
        proof["adopted_from_reviewer_identity"] = adopted_from_reviewer_identity
    if adoption_reason:
        proof["adoption_reason"] = adoption_reason
    return proof


def is_sanctioned_session_lease(session: dict[str, Any] | None) -> bool:
    if not session:
        return False
    provenance = session.get("lease_provenance") or {}
    source = (provenance.get("source") or "").strip()
    if source not in SANCTIONED_PROVENANCE_SOURCES:
        return False
    if source == SOURCE_ADOPT:
        return bool(provenance.get("comment_id")) and bool(
            provenance.get("adopted_from_session_id")
        )
    if source in {SOURCE_ACQUIRE, SOURCE_HEARTBEAT}:
        return bool(session.get("comment_id") or provenance.get("comment_id"))
    return False


def describe_session_lease_proof(
    session: dict[str, Any] | None,
) -> dict[str, Any]:
    """Canonical merge-report lease proof fields (#535).

    Surfaces how the in-session lease was established so merge handoffs can
    distinguish sanctioned MCP acquire/adopt paths from bare manual seeding.
    """
    if not session:
        return {
            "lease_proof_source": None,
            "lease_recovery_reason": None,
            "lease_proof_kind": "none",
            "lease_proof_sanctioned": False,
            "lease_proof_comment_id": None,
            "lease_adopted_from_session_id": None,
        }

    provenance = session.get("lease_provenance") or {}
    if not isinstance(provenance, dict):
        provenance = {}
    source = (provenance.get("source") or "").strip() or None
    sanctioned = is_sanctioned_session_lease(session)
    reason = provenance.get("adoption_reason")
    if isinstance(reason, str):
        reason = reason.strip() or None
    else:
        reason = None

    if source == SOURCE_ADOPT and sanctioned:
        kind = "sanctioned_adoption"
        reason = reason or DEFAULT_ADOPTION_REASON
    elif source == SOURCE_ACQUIRE and sanctioned:
        kind = "sanctioned_acquire"
    elif source == SOURCE_HEARTBEAT and sanctioned:
        kind = "sanctioned_heartbeat"
    elif source:
        kind = "unsanctioned"
    else:
        # Session present without provenance = classic manual _SESSION_LEASE seed.
        kind = "unsanctioned_manual_seed"

    comment_id = provenance.get("comment_id")
    if comment_id is None:
        comment_id = session.get("comment_id")

    return {
        "lease_proof_source": source,
        "lease_recovery_reason": reason,
        "lease_proof_kind": kind,
        "lease_proof_sanctioned": sanctioned,
        "lease_proof_comment_id": comment_id,
        "lease_adopted_from_session_id": provenance.get("adopted_from_session_id"),
    }


# Phrases that claim non-MCP / fabricated lease proof in handoffs (#535).
_UNSAFE_LEASE_PROOF_CLAIM_RE = re.compile(
    r"(?is)\b("
    r"manually\s+seeded\s+lease|"
    r"manual(?:ly)?\s+seed(?:ed)?\s+(?:lease|proof)|"
    r"seeded\s+lease\s+proof|"
    r"injected\s+lease|"
    r"lease\s+proof\s+injected|"
    r"manual\s+_?SESSION_LEASE|"
    r"_SESSION_LEASE\s+seed|"
    r"unsanctioned\s+lease\s+proof"
    r")\b"
)

# Evidence that the report used the sanctioned MCP adoption/acquire path.
_SANCTIONED_LEASE_EVIDENCE_RE = re.compile(
    r"(?is)\b("
    r"gitea_adopt_merger_pr_lease|"
    r"gitea_acquire_reviewer_pr_lease|"
    r"lease_proof_source\s*[:=]\s*gitea_adopt_merger_pr_lease|"
    r"lease_proof_source\s*[:=]\s*gitea_acquire_reviewer_pr_lease|"
    r"lease_proof_kind\s*[:=]\s*sanctioned_adoption|"
    r"lease_proof_kind\s*[:=]\s*sanctioned_acquire|"
    r"adoption_comment_id\s*[:=]|"
    r"sanctioned_adoption|"
    r"mcp-review-lease-adoption"
    r")\b"
)


def assess_manual_lease_proof_handoff(report_text: str) -> dict[str, Any]:
    """Controller audit: block handoffs that claim unsafe lease seeding (#535).

    Reports may discuss manual seeding as a blocked/historical anti-pattern only
    when they also cite sanctioned MCP adoption/acquire evidence. Bare claims of
    manual/seeded/injected lease proof without that evidence fail closed.
    """
    text = report_text or ""
    if not _UNSAFE_LEASE_PROOF_CLAIM_RE.search(text):
        return {"proven": True, "block": False, "reasons": []}
    if _SANCTIONED_LEASE_EVIDENCE_RE.search(text):
        return {"proven": True, "block": False, "reasons": []}
    return {
        "proven": False,
        "block": True,
        "reasons": [
            "report claims manual/seeded/injected/unsanctioned lease proof without "
            "sanctioned MCP adoption evidence (use gitea_adopt_merger_pr_lease / "
            "gitea_acquire_reviewer_pr_lease and cite lease_proof_source; #535)"
        ],
    }


def assess_adopt_merger_lease(
    comments: list[dict],
    *,
    pr_number: int,
    adopter_identity: str,
    adopter_profile: str,
    adopter_session_id: str,
    repo: str,
    issue_number: int | None,
    worktree: str,
    expected_head_sha: str | None,
    live_head_sha: str | None,
    approval_at_head: bool,
    pr_open: bool = True,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Decide whether a merger may adopt the active reviewer lease (#536)."""
    now = now or datetime.now(timezone.utc)
    reasons: list[str] = []
    pinned = leases._normalize_sha(expected_head_sha)
    live = leases._normalize_sha(live_head_sha)

    if not pr_open:
        reasons.append(
            f"PR #{pr_number} is not open; merger lease adoption is only for open PRs"
        )
    if not approval_at_head:
        reasons.append(
            "formal APPROVED review at the current PR head is required before "
            "merger lease adoption (fail closed)"
        )
    if not pinned:
        reasons.append("expected_head_sha is required for merger lease adoption")
    if not live:
        reasons.append("live PR head SHA unavailable (fail closed)")
    elif pinned and live and pinned != live:
        reasons.append(
            "expected_head_sha does not match live PR head; refresh approval before adoption"
        )

    active = leases.find_active_reviewer_lease(
        comments, pr_number=pr_number, now=now
    )
    if not active:
        reasons.append(
            f"no active reviewer lease on PR #{pr_number} to adopt; reviewer "
            "must acquire first"
        )
    else:
        owner_session = (active.get("session_id") or "").strip()
        freshness = active.get("freshness") or leases.classify_lease_freshness(
            active, now=now
        )
        if freshness not in _MERGER_ADOPTABLE_FRESHNESS:
            reasons.append(
                f"active reviewer lease freshness is '{freshness}'; explicit "
                "reclaim is not implemented (fail closed)"
            )
        lease_head = active.get("candidate_head")
        if lease_head and live and lease_head != live:
            reasons.append(
                "active reviewer lease candidate_head differs from live PR head"
            )
        if owner_session and owner_session == adopter_session_id:
            # Same session already holds the thread lease — allow idempotent adopt
            # only when the newest entry is not yet an adoption by this session.
            entries = leases._lease_entries(comments, pr_number=pr_number)
            newest = entries[-1] if entries else None
            if newest and (newest.get("phase") or "") == "adopted":
                reasons.append(
                    "merger session already recorded an adoption lease on this PR"
                )
        elif owner_session and owner_session != adopter_session_id:
            # Cross-session merger handoff: adopt the foreign reviewer lease.
            pass

    if not (adopter_identity or "").strip():
        reasons.append("adopter identity required")
    if not (adopter_session_id or "").strip():
        reasons.append("adopter session_id required")
    if not (worktree or "").strip():
        reasons.append("merger worktree path required")
    if "merger" not in (adopter_profile or "").lower():
        reasons.append(
            f"profile '{adopter_profile}' is not a merger profile; adoption is "
            "merger-only (fail closed)"
        )

    adopt_allowed = not reasons
    adoption_body = None
    if adopt_allowed and active:
        adoption_body = format_adoption_body(
            repo=repo,
            pr_number=pr_number,
            issue_number=issue_number or active.get("issue_number"),
            adopter_identity=adopter_identity,
            adopter_profile=adopter_profile,
            adopter_session_id=adopter_session_id,
            worktree=worktree,
            candidate_head=live,
            target_branch=active.get("target_branch") or "master",
            target_branch_sha=active.get("target_branch_sha"),
            adopted_from_session_id=active.get("session_id") or "",
            adopted_from_profile=active.get("profile") or "unknown",
            adopted_from_reviewer_identity=active.get("reviewer_identity") or "",
            adopted_from_comment_id=active.get("comment_id"),
            adopted_at=now,
        )

    return {
        "adopt_allowed": adopt_allowed,
        "reasons": reasons,
        "active_lease": active,
        "adoption_body": adoption_body,
        "adopter_session_id": adopter_session_id,
        "expected_head_sha": pinned,
        "live_head_sha": live,
    }