Gitea-Tools/tests/test_issue_691_obsolete_reviewer_lease_cleanup.py

"""Guarded cleanup for obsolete comment-backed reviewer leases (#691).

Reproduces PR #688 lease comment 10749 class of defect: completed
REQUEST_CHANGES on head A, author pushes head B, foreign lease remains
pinned to A (and after expiry still has no non-owner cleanup path that
diagnosis can prescribe beyond indefinite wait).
"""

from __future__ import annotations

import sys
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import merger_lease_adoption as mla  # noqa: E402
import reviewer_pr_lease as leases  # noqa: E402

# PR #688 / comment 10749 reproduction fixtures
HEAD_A = "c7a444eb4b41cf916fdbd20a4999ffd78af496d0"
HEAD_B = "4a6357800364718a27a36ebc73578d4b929ff4aa"
SESSION_FOREIGN = "25883-7d4c6e6ebd53"
COMMENT_10749 = 10749
REPO = "Scaled-Tech-Consulting/Gitea-Tools"
PR = 688
ISSUE = 687
EXPIRES_10749 = "2026-07-13T04:23:00Z"


def _utc(dt: datetime) -> datetime:
    return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)


def _lease_comment(
    *,
    pr_number: int = PR,
    session_id: str = SESSION_FOREIGN,
    phase: str = "claimed",
    candidate_head: str = HEAD_A,
    comment_id: int = COMMENT_10749,
    last_activity: datetime | None = None,
    expires_at: datetime | None = None,
    worktree: str = "branches/review-pr688-feat-issue-687-reconciler-branch-delete",
    repo: str = REPO,
) -> dict:
    now = last_activity or datetime.now(timezone.utc)
    body = leases.format_lease_body(
        repo=repo,
        pr_number=pr_number,
        issue_number=ISSUE,
        reviewer_identity="sysadmin",
        profile="prgs-reviewer",
        session_id=session_id,
        worktree=worktree,
        phase=phase,
        candidate_head=candidate_head,
        target_branch="master",
        target_branch_sha="b" * 40,
        last_activity=now,
        expires_at=expires_at,
    )
    return {
        "id": comment_id,
        "body": body,
        "user": {"login": "sysadmin"},
        "author": "sysadmin",
        "created_at": now.isoformat(),
        "updated_at": now.isoformat(),
    }


def _reviews_for_head(head: str, verdict: str = "REQUEST_CHANGES") -> list[dict]:
    return [
        {
            "verdict": verdict,
            "reviewed_head_sha": head,
            "dismissed": False,
            "reviewer": "sysadmin",
        }
    ]


def _assess(
    comments,
    *,
    current_head=HEAD_B,
    formal_reviews=None,
    controller=True,
    worktree_exists=True,
    worktree_clean=True,
    owner_process_alive=False,
    requesting_session="fresh-controller",
    apply=False,
    confirmation=None,
    **kwargs,
):
    return leases.assess_obsolete_reviewer_comment_lease_cleanup(
        comments,
        pr_number=PR,
        current_head_sha=current_head,
        formal_reviews=formal_reviews if formal_reviews is not None else _reviews_for_head(HEAD_A),
        repo=REPO,
        expected_repo=REPO,
        requesting_session_id=requesting_session,
        controller_recovery_authorized=controller,
        worktree_exists=worktree_exists,
        worktree_clean=worktree_clean,
        worktree_has_unpreserved_work=not worktree_clean if worktree_exists else False,
        owner_process_alive=owner_process_alive,
        owner_pid_observed=kwargs.get("owner_pid_observed"),
        requesting_pid=kwargs.get("requesting_pid", 99999),
        current_head_review_in_progress=kwargs.get(
            "current_head_review_in_progress", False
        ),
        expected_lease_comment_id=kwargs.get("expected_lease_comment_id"),
        expected_session_id=kwargs.get("expected_session_id"),
        expected_leased_head=kwargs.get("expected_leased_head"),
        confirmation=confirmation
        if confirmation is not None
        else (leases.cleanup_confirmation_for_pr(PR) if apply else ""),
        apply=apply,
        now=kwargs.get("now"),
    )


class TestIssue691SupersededHeadCleanup(unittest.TestCase):
    def setUp(self):
        leases.clear_session_lease()

    def test_1_request_changes_then_push_expired_cleanup_succeeds(self):
        """Completed REQUEST_CHANGES on A, push B, lease A expires → cleanup OK."""
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(
            last_activity=expires - timedelta(hours=2),
            expires_at=expires,
        )
        result = _assess(
            [claim],
            formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
            now=now,
            apply=True,
        )
        self.assertTrue(result["cleanup_allowed"], result["reasons"])
        self.assertEqual(result["classification"], "foreign_expired_superseded_head")
        self.assertEqual(
            result["exact_next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
        )
        self.assertIn("phase: released", result["release_body"])
        self.assertIn("obsolete-superseded-or-expired-lease", result["release_body"])
        self.assertIsNotNone(result["audit_comment_body"])
        self.assertEqual(result["cleanup_tool"], leases.CLEANUP_OBSOLETE_LEASE_TOOL)

    def test_2_approve_then_push_cleanup_policy(self):
        """Completed APPROVE on A, push B → cleanup eligible (completed superseded)."""
        now = datetime(2026, 7, 13, 1, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)  # not yet expired
        claim = _lease_comment(
            last_activity=now - timedelta(minutes=10),
            expires_at=expires,
        )
        result = _assess(
            [claim],
            formal_reviews=_reviews_for_head(HEAD_A, "APPROVED"),
            now=now,
        )
        self.assertTrue(result["cleanup_allowed"], result["reasons"])
        self.assertEqual(result["classification"], "foreign_completed_superseded_head")

    def test_3_active_current_head_cleanup_denied(self):
        now = datetime.now(timezone.utc)
        claim = _lease_comment(
            candidate_head=HEAD_B,
            last_activity=now - timedelta(minutes=5),
            expires_at=now + timedelta(hours=2),
        )
        result = _assess(
            [claim],
            current_head=HEAD_B,
            formal_reviews=_reviews_for_head(HEAD_B, "REQUEST_CHANGES"),
            now=now,
        )
        self.assertFalse(result["cleanup_allowed"])
        self.assertEqual(result["classification"], "foreign_active_current_head")

    def test_4_foreign_owner_recent_progress_denied(self):
        now = datetime.now(timezone.utc)
        claim = _lease_comment(
            candidate_head=HEAD_B,
            last_activity=now - timedelta(minutes=2),
            expires_at=now + timedelta(hours=2),
        )
        result = _assess(
            [claim],
            current_head=HEAD_B,
            formal_reviews=[],
            owner_process_alive=True,
            now=now,
        )
        self.assertFalse(result["cleanup_allowed"])
        self.assertTrue(
            any("recent authenticated progress" in r for r in result["reasons"])
            or any("current PR head" in r for r in result["reasons"])
        )

    def test_5_owner_absent_worktree_dirty_denied(self):
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(
            candidate_head=HEAD_B,
            last_activity=expires - timedelta(hours=1),
            expires_at=expires,
        )
        result = _assess(
            [claim],
            current_head=HEAD_B,
            formal_reviews=[],
            worktree_clean=False,
            owner_process_alive=False,
            now=now,
        )
        self.assertFalse(result["cleanup_allowed"])
        self.assertTrue(any("dirty" in r for r in result["reasons"]))

    def test_6_owner_absent_worktree_clean_orphan_ok(self):
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(
            candidate_head=HEAD_B,
            last_activity=expires - timedelta(hours=1),
            expires_at=expires,
        )
        result = _assess(
            [claim],
            current_head=HEAD_B,
            formal_reviews=[],
            worktree_clean=True,
            owner_process_alive=False,
            now=now,
        )
        self.assertTrue(result["cleanup_allowed"], result["reasons"])
        self.assertEqual(result["classification"], "orphaned_owner_missing")

    def test_7_pid_reuse_not_ownership(self):
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
        result = _assess(
            [claim],
            now=now,
            owner_pid_observed=4242,
            requesting_pid=4242,
        )
        self.assertTrue(
            any("PID equality" in r or "NOT ownership" in r for r in result["reasons"])
            or result["owner_session_evidence"]["pid_is_not_ownership_proof"]
        )
        # Still eligible via superseded+terminal+expired despite matching PID.
        self.assertTrue(result["cleanup_allowed"], result["reasons"])

    def test_8_comment_backed_no_db_lease_id(self):
        """Cleanup uses comment ledger only — no control-plane lease_id required."""
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
        # Explicitly no lease_id field anywhere.
        self.assertNotIn("lease_id", claim)
        result = _assess([claim], now=now)
        self.assertTrue(result["cleanup_allowed"], result["reasons"])
        self.assertEqual(result["active_lease"]["comment_id"], COMMENT_10749)
        self.assertIsNone(result["active_lease"].get("lease_id"))

    def test_9_cleanup_posts_durable_audit_bodies(self):
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
        result = _assess([claim], now=now, apply=True)
        self.assertTrue(result["cleanup_allowed"])
        self.assertIn("#691", result["audit_comment_body"])
        self.assertIn(str(COMMENT_10749), result["audit_comment_body"])
        self.assertIn(HEAD_A, result["audit_comment_body"])
        self.assertIn(HEAD_B, result["audit_comment_body"])

    def test_10_fresh_acquire_after_cleanup_marker(self):
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(
            expires_at=expires,
            last_activity=expires - timedelta(hours=1),
            comment_id=COMMENT_10749,
        )
        assessed = _assess([claim], now=now, apply=True)
        self.assertTrue(assessed["cleanup_allowed"])
        release = {
            "id": 99999,
            "body": assessed["release_body"],
            "user": {"login": "sysadmin"},
        }
        # Newest terminal marker ends active lease.
        active = leases.find_active_reviewer_lease(
            [claim, release], pr_number=PR, now=now
        )
        self.assertIsNone(active)
        acq = leases.assess_acquire_lease(
            [claim, release],
            pr_number=PR,
            reviewer_identity="sysadmin",
            profile="prgs-reviewer",
            session_id="fresh-reviewer-1",
            repo=REPO,
            issue_number=ISSUE,
            worktree="branches/review-pr-688-fresh",
            candidate_head=HEAD_B,
            target_branch="master",
            target_branch_sha="b" * 40,
            now=now,
        )
        self.assertTrue(acq["acquire_allowed"], acq["reasons"])

    def test_11_no_validation_state_transfer(self):
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
        result = _assess([claim], now=now, apply=True)
        self.assertTrue(result["cleanup_allowed"])
        # Release body keeps old candidate_head (no repoint).
        self.assertIn(f"candidate_head: {HEAD_A}", result["release_body"])
        self.assertNotIn(HEAD_B, result["release_body"].split("candidate_head:")[1].split("\n")[0])
        self.assertIn("did not transfer validation", result["audit_comment_body"])
        self.assertIn("did not repoint", result["audit_comment_body"])
        # Session lease must remain unset by assessment (tool never seeds).
        self.assertIsNone(leases.get_session_lease())

    def test_12_repeated_cleanup_idempotent(self):
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
        first = _assess([claim], now=now, apply=True)
        self.assertTrue(first["cleanup_allowed"])
        release = {"id": 100001, "body": first["release_body"], "user": {"login": "x"}}
        second = _assess([claim, release], now=now, apply=True)
        self.assertFalse(second["cleanup_allowed"])
        self.assertEqual(second["classification"], "no_lease")

    def test_13_malformed_expiry_fails_closed(self):
        claim = _lease_comment()
        # Corrupt expires_at in the body.
        claim["body"] = claim["body"].replace(
            claim["body"].split("expires_at:")[1].split("\n")[0].strip(),
            "not-a-timestamp",
        )
        result = _assess([claim], formal_reviews=_reviews_for_head(HEAD_A))
        self.assertFalse(result["cleanup_allowed"])
        self.assertFalse(result["expires_parseable"])
        self.assertTrue(any("expires_at" in r for r in result["reasons"]))

    def test_14_wrong_identity_evidence_fails_closed(self):
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
        bad_repo = _assess(
            [claim],
            now=now,
            # force repo mismatch via expected_repo
        )
        # Patch via direct call with wrong expected_repo
        bad = leases.assess_obsolete_reviewer_comment_lease_cleanup(
            [claim],
            pr_number=PR,
            current_head_sha=HEAD_B,
            formal_reviews=_reviews_for_head(HEAD_A),
            expected_repo="Other-Org/Other-Repo",
            requesting_session_id="fresh",
            controller_recovery_authorized=True,
            worktree_exists=True,
            worktree_clean=True,
            owner_process_alive=False,
            now=now,
        )
        self.assertFalse(bad["cleanup_allowed"])
        self.assertTrue(any("repository identity" in r for r in bad["reasons"]))

        bad_pr = leases.assess_obsolete_reviewer_comment_lease_cleanup(
            [claim],
            pr_number=999,
            current_head_sha=HEAD_B,
            formal_reviews=_reviews_for_head(HEAD_A),
            expected_repo=REPO,
            requesting_session_id="fresh",
            controller_recovery_authorized=True,
            worktree_exists=True,
            worktree_clean=True,
            owner_process_alive=False,
            expected_lease_comment_id=COMMENT_10749,
            now=now,
        )
        self.assertFalse(bad_pr["cleanup_allowed"])

        bad_head = _assess(
            [claim],
            now=now,
            expected_leased_head="0" * 40,
        )
        self.assertFalse(bad_head["cleanup_allowed"])

        bad_session = _assess(
            [claim],
            now=now,
            expected_session_id="wrong-session",
        )
        self.assertFalse(bad_session["cleanup_allowed"])

    def test_15_current_head_active_cannot_be_displaced(self):
        now = datetime.now(timezone.utc)
        claim = _lease_comment(
            candidate_head=HEAD_B,
            last_activity=now - timedelta(minutes=1),
            expires_at=now + timedelta(hours=2),
        )
        result = _assess(
            [claim],
            current_head=HEAD_B,
            formal_reviews=_reviews_for_head(HEAD_B),
            now=now,
            current_head_review_in_progress=True,
        )
        self.assertFalse(result["cleanup_allowed"])
        # Acquire also blocked by foreign active lease.
        acq = leases.assess_acquire_lease(
            [claim],
            pr_number=PR,
            reviewer_identity="other",
            profile="prgs-reviewer",
            session_id="thief",
            repo=REPO,
            issue_number=ISSUE,
            worktree="branches/steal",
            candidate_head=HEAD_B,
            target_branch="master",
            target_branch_sha="b" * 40,
            now=now,
        )
        self.assertFalse(acq["acquire_allowed"])

    def test_regression_pr688_comment_10749_after_expiry(self):
        """Exact 10749 reproduction after recorded expires_at 2026-07-13T04:23:00Z."""
        now = datetime(2026, 7, 13, 4, 30, 0, tzinfo=timezone.utc)
        expires = datetime.fromisoformat(EXPIRES_10749.replace("Z", "+00:00"))
        claim = _lease_comment(
            session_id=SESSION_FOREIGN,
            candidate_head=HEAD_A,
            comment_id=COMMENT_10749,
            last_activity=expires - timedelta(hours=2),
            expires_at=expires,
        )
        # Diagnosis must not prescribe indefinite wait-only for this case.
        diag = leases.diagnose_reviewer_pr_lease_handoff(
            [claim],
            pr_number=PR,
            current_session_id="fresh-reviewer",
            current_reviewer_identity="sysadmin",
            current_head_sha=HEAD_B,
            formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
            worktree_exists=True,
            worktree_clean=True,
            owner_process_alive=False,
            instructed_session_id=SESSION_FOREIGN,
            instructed_comment_id=COMMENT_10749,
            now=now,
        )
        self.assertEqual(diag["classification"], "foreign_expired_superseded_head")
        self.assertEqual(
            diag["next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
        )
        self.assertEqual(diag["cleanup_tool"], leases.CLEANUP_OBSOLETE_LEASE_TOOL)
        self.assertIn("CLEANUP OBSOLETE REVIEWER LEASE 688", diag["required_confirmation"])
        self.assertEqual(diag["mutation_eligibility"], "cleanup_only")
        self.assertFalse(diag["mutation_allowed"])

        # Assessment still returns the lease as active/stale class of block for acquire
        # when unexpired path... here it's expired so find_active is None; acquire free
        # only after cleanup if somehow still active. With expired, find_active is None:
        active = leases.find_active_reviewer_lease([claim], pr_number=PR, now=now)
        self.assertIsNone(active)

        # But superseded unexpired still blocks acquire and diagnosis points to cleanup:
        unexpired_now = datetime(2026, 7, 13, 1, 0, 0, tzinfo=timezone.utc)
        claim2 = _lease_comment(
            session_id=SESSION_FOREIGN,
            candidate_head=HEAD_A,
            comment_id=COMMENT_10749,
            last_activity=unexpired_now - timedelta(minutes=45),
            expires_at=expires,
        )
        active2 = leases.find_active_reviewer_lease(
            [claim2], pr_number=PR, now=unexpired_now
        )
        self.assertIsNotNone(active2)
        self.assertEqual(active2["freshness"], "stale_warning")
        diag2 = leases.diagnose_reviewer_pr_lease_handoff(
            [claim2],
            pr_number=PR,
            current_session_id="fresh-reviewer",
            current_reviewer_identity="sysadmin",
            current_head_sha=HEAD_B,
            formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
            worktree_exists=True,
            worktree_clean=True,
            now=unexpired_now,
        )
        self.assertEqual(diag2["classification"], "foreign_completed_superseded_head")
        self.assertEqual(
            diag2["next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
        )
        acq = leases.assess_acquire_lease(
            [claim2],
            pr_number=PR,
            reviewer_identity="sysadmin",
            profile="prgs-reviewer",
            session_id="fresh-reviewer",
            repo=REPO,
            issue_number=ISSUE,
            worktree="branches/review-pr-688-new",
            candidate_head=HEAD_B,
            target_branch="master",
            target_branch_sha="b" * 40,
            now=unexpired_now,
        )
        self.assertFalse(acq["acquire_allowed"])

    def test_missing_controller_auth_denied(self):
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
        result = _assess([claim], controller=False, now=now)
        self.assertFalse(result["cleanup_allowed"])

    def test_wrong_confirmation_denied_on_apply(self):
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
        result = _assess(
            [claim], now=now, apply=True, confirmation="MERGE PR 688"
        )
        self.assertFalse(result["cleanup_allowed"])
        self.assertTrue(any("confirmation" in r for r in result["reasons"]))

    def test_owner_session_must_use_owner_release(self):
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
        result = _assess(
            [claim],
            now=now,
            requesting_session=SESSION_FOREIGN,
        )
        self.assertFalse(result["cleanup_allowed"])
        self.assertTrue(any("owns the lease" in r for r in result["reasons"]))

    def test_superseded_without_terminal_review_denied(self):
        now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
        expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
        claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
        result = _assess([claim], formal_reviews=[], now=now)
        self.assertFalse(result["cleanup_allowed"])
        self.assertEqual(result["classification"], "ambiguous_conflicting_evidence")


class TestIssue691DiagnoseClassifications(unittest.TestCase):
    def setUp(self):
        leases.clear_session_lease()

    def test_foreign_active_current_head_still_wait(self):
        now = datetime.now(timezone.utc)
        claim = _lease_comment(
            candidate_head=HEAD_B,
            last_activity=now - timedelta(minutes=5),
            expires_at=now + timedelta(hours=1),
        )
        claim["id"] = 1
        result = leases.diagnose_reviewer_pr_lease_handoff(
            [claim],
            pr_number=PR,
            current_session_id="other",
            current_reviewer_identity="sysadmin",
            current_head_sha=HEAD_B,
            formal_reviews=[],
            now=now,
        )
        self.assertEqual(result["classification"], "foreign_active_current_head")
        self.assertEqual(result["next_action"], leases.NEXT_ACTION_WAIT)
        self.assertFalse(result["mutation_allowed"])


if __name__ == "__main__":
    unittest.main()