"""Regression tests for #702: stale runtime binding + orphaned durable lease.

Reproduces the PR #701 incident (lease 65664-4f248d213d60, comments
11129/11131): the MCP daemon crashed without teardown, destroying the
in-memory session lease while the durable comment lease survived, and the
auto-reconnected daemon inherited a stale ``GITEA_ACTIVE_WORKTREE`` binding
(``branches/review-pr-654``) from its parent environment.

Covers the five mandated scenarios:

1. lost in-session leases  — durable shadow survives a crash and yields
   provable owner-process evidence
2. old-head leases         — expired superseded-head lease with no terminal
   review becomes recoverable only with orphan evidence + controller authority
3. runtime/worktree mismatch — env bindings to deleted worktrees are demoted,
   never silently bound
4. managed reconnect       — boot-inherited uncorroborated bindings are
   surfaced for managed recovery; provably stale ones are clear-eligible
5. safe expiry             — pre-expiry stays fail-closed wait (no steal);
   canonical expiry unlocks acquisition and guarded cleanup
"""

from __future__ import annotations

import os
import sys
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path

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

import namespace_workspace_binding as nwb  # noqa: E402
import reviewer_pr_lease as leases  # noqa: E402
import stale_binding_recovery as sbr  # noqa: E402

# PR #701 incident fixtures.
OLD_HEAD = "b4d0cb22e452a07c8e816745c704ddb0f43edf0b"
NEW_HEAD = "6b675f5c834b41f9d74e8a54294ff44dddf28ae4"
CRASHED_SESSION = "65664-4f248d213d60"
LEASE_COMMENT = 11131
REPO = "Scaled-Tech-Consulting/Gitea-Tools"
PR = 701
ISSUE = 699
LEASE_WORKTREE = "branches/review-fix-issue-699-structured-auth-mcp-errors"


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _lease_comment(
    *,
    phase: str = "validation-start",
    candidate_head: str = OLD_HEAD,
    session_id: str = CRASHED_SESSION,
    comment_id: int = LEASE_COMMENT,
    last_activity: datetime | None = None,
    expires_at: datetime | None = None,
    worktree: str = LEASE_WORKTREE,
) -> dict:
    stamp = last_activity or _now()
    body = leases.format_lease_body(
        repo=REPO,
        pr_number=PR,
        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="2" * 40,
        last_activity=stamp,
        expires_at=expires_at,
    )
    return {
        "id": comment_id,
        "body": body,
        "user": {"login": "sysadmin"},
        "author": "sysadmin",
        "created_at": stamp.isoformat(),
        "updated_at": stamp.isoformat(),
    }


def _expired_superseded_comments() -> list[dict]:
    stale = _now() - timedelta(hours=3)
    return [_lease_comment(
        last_activity=stale,
        expires_at=stale + timedelta(minutes=120),
    )]


def _fresh_superseded_comments() -> list[dict]:
    recent = _now() - timedelta(minutes=10)
    return [_lease_comment(
        last_activity=recent,
        expires_at=recent + timedelta(minutes=120),
    )]


class TestLostInSessionLeaseShadow(unittest.TestCase):
    """Scenario 1: the durable shadow survives a daemon crash (#702 AC3)."""

    def setUp(self):
        leases._SESSION_LEASE = None

    def tearDown(self):
        leases._SESSION_LEASE = None

    def _record(self) -> dict:
        return leases.record_session_lease(
            {
                "pr_number": PR,
                "issue_number": ISSUE,
                "session_id": CRASHED_SESSION,
                "reviewer_identity": "sysadmin",
                "profile": "prgs-reviewer",
                "worktree": LEASE_WORKTREE,
                "phase": "claimed",
                "candidate_head": OLD_HEAD,
                "repo": REPO,
                "comment_id": LEASE_COMMENT,
            },
            lease_provenance={"source": "acquire", "comment_id": LEASE_COMMENT},
        )

    def test_shadow_written_on_record_and_survives_simulated_crash(self):
        self._record()
        # Simulated unhandled crash: the in-memory dict dies with the process
        # but the sanctioned clear path never runs.
        leases._SESSION_LEASE = None
        shadow = leases.load_session_lease_shadow()
        self.assertIsNotNone(shadow)
        self.assertEqual(shadow.get("session_id"), CRASHED_SESSION)
        self.assertEqual(shadow.get("pr_number"), PR)
        self.assertEqual(int(shadow.get("owner_pid")), os.getpid())

    def test_sanctioned_clear_removes_shadow(self):
        self._record()
        leases.clear_session_lease()
        self.assertIsNone(leases.load_session_lease_shadow())

    def test_orphan_assessment_dead_owner_pid_proves_exit(self):
        self._record()
        leases._SESSION_LEASE = None
        shadow = leases.load_session_lease_shadow()
        lease = {"session_id": CRASHED_SESSION}
        verdict = leases.assess_crashed_session_lease_orphan(
            shadow, lease=lease, current_pid=os.getpid() + 1, pid_alive=False
        )
        self.assertTrue(verdict["orphan_evidence"])
        self.assertIs(verdict["owner_process_alive"], False)

    def test_orphan_assessment_alive_pid_is_not_ownership_proof(self):
        self._record()
        shadow = leases.load_session_lease_shadow()
        verdict = leases.assess_crashed_session_lease_orphan(
            shadow,
            lease={"session_id": CRASHED_SESSION},
            current_pid=os.getpid() + 1,
            pid_alive=True,
        )
        self.assertFalse(verdict["orphan_evidence"])
        self.assertIsNone(verdict["owner_process_alive"])

    def test_orphan_assessment_session_mismatch_proves_nothing(self):
        self._record()
        shadow = leases.load_session_lease_shadow()
        verdict = leases.assess_crashed_session_lease_orphan(
            shadow, lease={"session_id": "other-999"}, pid_alive=False
        )
        self.assertFalse(verdict["orphan_evidence"])
        self.assertIsNone(verdict["owner_process_alive"])

    def test_orphan_assessment_without_shadow_is_unknown(self):
        verdict = leases.assess_crashed_session_lease_orphan(
            None, lease={"session_id": CRASHED_SESSION}
        )
        self.assertFalse(verdict["orphan_evidence"])
        self.assertIsNone(verdict["owner_process_alive"])


class TestOldHeadLeaseCleanup(unittest.TestCase):
    """Scenario 2: expired superseded-head lease with no terminal review."""

    def _assess(self, comments, **kwargs):
        defaults = dict(
            pr_number=PR,
            current_head_sha=NEW_HEAD,
            formal_reviews=[],
            repo=REPO,
            expected_repo=REPO,
            requesting_session_id="fresh-controller",
            controller_recovery_authorized=True,
            worktree_exists=True,
            worktree_clean=True,
            owner_process_alive=False,
        )
        defaults.update(kwargs)
        return leases.assess_obsolete_reviewer_comment_lease_cleanup(
            comments, **defaults
        )

    def test_expired_orphan_with_full_evidence_is_cleanup_eligible(self):
        result = self._assess(_expired_superseded_comments())
        self.assertEqual(result["classification"], "orphaned_expired_superseded_head")
        self.assertTrue(result["cleanup_allowed"])
        self.assertIn("phase: released", result["release_body"] or "")
        self.assertEqual(
            result["exact_next_action"], "cleanup_obsolete_reviewer_comment_lease"
        )

    def test_expired_orphan_without_owner_evidence_fails_closed(self):
        result = self._assess(
            _expired_superseded_comments(), owner_process_alive=None
        )
        self.assertFalse(result["cleanup_allowed"])
        self.assertIn(
            "provable owner-process-exit evidence",
            " ".join(result["fail_closed_reasons"]),
        )

    def test_expired_orphan_without_controller_authority_fails_closed(self):
        result = self._assess(
            _expired_superseded_comments(), controller_recovery_authorized=False
        )
        self.assertFalse(result["cleanup_allowed"])

    def test_expired_orphan_with_dirty_worktree_fails_closed(self):
        result = self._assess(
            _expired_superseded_comments(),
            worktree_clean=False,
            worktree_has_unpreserved_work=True,
        )
        self.assertFalse(result["cleanup_allowed"])

    def test_unexpired_superseded_no_terminal_stays_ambiguous_wait(self):
        # Regression guard: pre-expiry there is still no steal path.
        result = self._assess(_fresh_superseded_comments())
        self.assertEqual(result["classification"], "ambiguous_conflicting_evidence")
        self.assertFalse(result["cleanup_allowed"])
        self.assertEqual(result["exact_next_action"], "wait")


class TestRuntimeWorktreeMismatch(unittest.TestCase):
    """Scenario 3: env bindings to deleted worktrees are demoted (#702 AC2)."""

    def test_missing_active_env_binding_is_demoted(self):
        demotions: list[str] = []
        missing = "/nonexistent/branches/review-pr-654"
        path, source = nwb.resolve_namespace_workspace(
            role_kind="reviewer",
            process_project_root=os.getcwd(),
            env={"GITEA_ACTIVE_WORKTREE": missing},
            demotions=demotions,
            verify_paths=True,
        )
        self.assertEqual(source, "MCP server process root (default)")
        self.assertEqual(len(demotions), 1)
        self.assertIn("no longer exists", demotions[0])

    def test_missing_active_falls_through_to_existing_role_env(self):
        existing = os.getcwd()
        path, source = nwb.resolve_namespace_workspace(
            role_kind="reviewer",
            process_project_root="/tmp",
            env={
                "GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-pr-654",
                "GITEA_REVIEWER_WORKTREE": existing,
            },
            verify_paths=True,
        )
        self.assertEqual(path, os.path.realpath(existing))
        self.assertEqual(source, "GITEA_REVIEWER_WORKTREE environment variable")

    def test_existing_active_env_binding_still_wins(self):
        existing = os.getcwd()
        path, source = nwb.resolve_namespace_workspace(
            role_kind="reviewer",
            process_project_root="/tmp",
            env={"GITEA_ACTIVE_WORKTREE": existing},
            verify_paths=True,
        )
        self.assertEqual(path, os.path.realpath(existing))
        self.assertEqual(source, "GITEA_ACTIVE_WORKTREE environment variable")

    def test_explicit_worktree_path_argument_is_never_demoted(self):
        missing = "/nonexistent/branches/explicit"
        path, source = nwb.resolve_namespace_workspace(
            role_kind="reviewer",
            worktree_path=missing,
            process_project_root="/tmp",
            env={},
            verify_paths=True,
        )
        self.assertEqual(source, "worktree_path argument")

    def test_mutation_context_reports_demotion_in_ignored_bindings(self):
        ctx = nwb.resolve_namespace_mutation_context(
            role_kind="reviewer",
            worktree_path=None,
            process_project_root=os.getcwd(),
            env={"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-pr-654"},
        )
        joined = " ".join(ctx["ignored_bindings"])
        self.assertIn("stale binding, #702", joined)


class TestManagedReconnectRecovery(unittest.TestCase):
    """Scenario 4: boot-inherited bindings and the sanctioned recovery plan."""

    def test_uncorroborated_inherited_reviewer_binding_is_unverified(self):
        classification = sbr.classify_active_worktree_binding(
            active_value=os.getcwd(),
            role_env_value=None,
            session_lease_worktree=None,
            boot_inherited=True,
            path_exists=True,
            role_kind="reviewer",
        )
        self.assertEqual(
            classification["classification"], sbr.CLASSIFICATION_UNVERIFIED_INHERITED
        )
        self.assertFalse(classification["clear_eligible"])
        plan = sbr.plan_recovery(classification)
        self.assertFalse(plan["clear_allowed"])
        self.assertIn("managed", plan["exact_next_action"])
        self.assertIn("do not clear or rewrite", plan["exact_next_action"])

    def test_missing_path_binding_is_provably_stale_and_clear_eligible(self):
        classification = sbr.classify_active_worktree_binding(
            active_value="/nonexistent/branches/review-pr-654",
            boot_inherited=True,
            path_exists=False,
            role_kind="reviewer",
        )
        self.assertEqual(
            classification["classification"],
            sbr.CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
        )
        plan = sbr.plan_recovery(classification)
        self.assertTrue(plan["clear_allowed"])

    def test_inherited_binding_superseded_by_session_lease_is_clear_eligible(self):
        classification = sbr.classify_active_worktree_binding(
            active_value="/tmp",
            session_lease_worktree=os.getcwd(),
            boot_inherited=True,
            path_exists=True,
            role_kind="reviewer",
        )
        self.assertEqual(
            classification["classification"],
            sbr.CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
        )
        self.assertTrue(sbr.plan_recovery(classification)["clear_allowed"])

    def test_post_boot_binding_conflict_is_surfaced_not_cleared(self):
        classification = sbr.classify_active_worktree_binding(
            active_value="/tmp",
            session_lease_worktree=os.getcwd(),
            boot_inherited=False,
            path_exists=True,
            role_kind="reviewer",
        )
        self.assertEqual(
            classification["classification"], sbr.CLASSIFICATION_UNVERIFIED_INHERITED
        )
        self.assertFalse(sbr.plan_recovery(classification)["clear_allowed"])

    def test_corroborated_bindings_are_untouched(self):
        for kwargs in (
            dict(role_env_value=os.getcwd()),
            dict(session_lease_worktree=os.getcwd()),
        ):
            classification = sbr.classify_active_worktree_binding(
                active_value=os.getcwd(),
                boot_inherited=True,
                path_exists=True,
                role_kind="reviewer",
                **kwargs,
            )
            self.assertEqual(
                classification["classification"], sbr.CLASSIFICATION_CORROBORATED
            )
            self.assertFalse(sbr.plan_recovery(classification)["clear_allowed"])

    def test_author_inherited_binding_stays_corroborated(self):
        classification = sbr.classify_active_worktree_binding(
            active_value=os.getcwd(),
            boot_inherited=True,
            path_exists=True,
            role_kind="author",
        )
        self.assertEqual(
            classification["classification"], sbr.CLASSIFICATION_CORROBORATED
        )

    def test_apply_recovery_clears_env_only_for_authorized_plan(self):
        env = {"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-pr-654"}
        plan = sbr.plan_recovery(
            sbr.classify_active_worktree_binding(
                active_value=env["GITEA_ACTIVE_WORKTREE"],
                boot_inherited=True,
                path_exists=False,
                role_kind="reviewer",
            )
        )
        applied = sbr.apply_recovery(plan, env=env)
        self.assertTrue(applied["performed"])
        self.assertNotIn("GITEA_ACTIVE_WORKTREE", env)
        self.assertIn("review-pr-654", applied["cleared_value"])

    def test_apply_recovery_refuses_unauthorized_plan(self):
        env = {"GITEA_ACTIVE_WORKTREE": os.getcwd()}
        plan = sbr.plan_recovery(
            sbr.classify_active_worktree_binding(
                active_value=env["GITEA_ACTIVE_WORKTREE"],
                boot_inherited=True,
                path_exists=True,
                role_kind="reviewer",
            )
        )
        applied = sbr.apply_recovery(plan, env=env)
        self.assertFalse(applied["performed"])
        self.assertIn("GITEA_ACTIVE_WORKTREE", env)


class TestSafeExpiryDiagnosis(unittest.TestCase):
    """Scenario 5: canonical expiry flips wait into acquire/guarded cleanup."""

    def _diagnose(self, comments, **kwargs):
        defaults = dict(
            pr_number=PR,
            current_session_id=None,
            current_reviewer_identity="fresh-reviewer",
            current_head_sha=NEW_HEAD,
            formal_reviews=[],
        )
        defaults.update(kwargs)
        return leases.diagnose_reviewer_pr_lease_handoff(comments, **defaults)

    def setUp(self):
        leases._SESSION_LEASE = None

    def test_pre_expiry_superseded_no_terminal_stays_wait(self):
        diagnosis = self._diagnose(_fresh_superseded_comments())
        self.assertEqual(
            diagnosis["classification"], "ambiguous_conflicting_evidence"
        )
        self.assertEqual(diagnosis["next_action"], "wait")

    def test_post_expiry_without_orphan_evidence_unlocks_acquire(self):
        diagnosis = self._diagnose(_expired_superseded_comments())
        self.assertEqual(
            diagnosis["classification"], "orphaned_expired_superseded_head"
        )
        self.assertEqual(diagnosis["next_action"], "acquire")

    def test_post_expiry_with_orphan_evidence_offers_guarded_cleanup(self):
        diagnosis = self._diagnose(
            _expired_superseded_comments(),
            owner_process_alive=False,
            worktree_clean=True,
            worktree_exists=True,
        )
        self.assertEqual(
            diagnosis["classification"], "orphaned_expired_superseded_head"
        )
        self.assertEqual(
            diagnosis["next_action"], "cleanup_obsolete_reviewer_comment_lease"
        )

    def test_expired_lease_no_longer_blocks_fresh_acquisition(self):
        assessment = leases.assess_acquire_lease(
            _expired_superseded_comments(),
            pr_number=PR,
            reviewer_identity="fresh-reviewer",
            profile="prgs-reviewer",
            session_id="99999-fresh",
            repo=REPO,
            issue_number=ISSUE,
            worktree="branches/review-pr-701-fresh",
            candidate_head=NEW_HEAD,
            target_branch="master",
            target_branch_sha="2" * 40,
        )
        self.assertTrue(assessment.get("acquire_allowed"))

    def test_unparseable_expiry_fails_closed_in_cleanup(self):
        comment = _lease_comment(
            last_activity=_now() - timedelta(hours=3), expires_at=None
        )
        comment["body"] = comment["body"].replace(
            "expires_at: ", "expires_at: not-a-timestamp"
        )
        result = leases.assess_obsolete_reviewer_comment_lease_cleanup(
            [comment],
            pr_number=PR,
            current_head_sha=NEW_HEAD,
            formal_reviews=[],
            repo=REPO,
            expected_repo=REPO,
            requesting_session_id="fresh-controller",
            controller_recovery_authorized=True,
            worktree_exists=True,
            worktree_clean=True,
            owner_process_alive=False,
        )
        self.assertFalse(result["cleanup_allowed"])


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