"""Strict-descendant dead-session recovery (#768).

After a dead author session, a preserved clean remediation commit that strictly
descends from the head recorded at lock time must be recoverable so the author
can publish. Equality alone is still accepted (#753); every other divergence
must keep failing closed.
"""

from __future__ import annotations

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

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

import issue_lock_recovery  # noqa: E402
import issue_lock_store  # noqa: E402
import issue_lock_worktree  # noqa: E402
import issue_work_duplicate_gate  # noqa: E402

ISSUE = 7680
PR_NUMBER = 7681
BRANCH = f"fix/issue-{ISSUE}-descendant-recovery"
WORKTREE = "/scratch/wt-768"
RECORDED = "a" * 40
DESCENDANT = "c" * 40
DIVERGED = "d" * 40
BEHIND = "b" * 40
IDENTITY = "example-user"
PROFILE = "example-author"


def dead_pid() -> int:
    proc = subprocess.Popen([sys.executable, "-c", "pass"])
    proc.wait()
    return proc.pid


def future_ts(hours: int = 4) -> str:
    return (
        (datetime.now(timezone.utc) + timedelta(hours=hours))
        .isoformat()
        .replace("+00:00", "Z")
    )


def make_lock(**overrides):
    lock = {
        "issue_number": ISSUE,
        "branch_name": BRANCH,
        "worktree_path": WORKTREE,
        "remote": "prgs",
        "org": "ExampleOrg",
        "repo": "ExampleRepo",
        "session_pid": dead_pid(),
        "work_lease": {
            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
            "issue_number": ISSUE,
            "branch": BRANCH,
            "worktree_path": WORKTREE,
            "claimant": {"username": IDENTITY, "profile": PROFILE},
            "expires_at": future_ts(),
        },
    }
    lock.update(overrides)
    return lock


def ancestry_ok(
    *,
    ancestor: str = RECORDED,
    descendant: str = DESCENDANT,
    is_strict: bool = True,
    probe_ok: bool = True,
    ancestor_present: bool = True,
    reasons: list[str] | None = None,
) -> dict:
    return {
        "ancestor_sha": ancestor,
        "descendant_sha": descendant,
        "probe_ok": probe_ok,
        "ancestor_present": ancestor_present,
        "descendant_present": True,
        "is_ancestor": is_strict or ancestor == descendant,
        "is_strict_descendant": is_strict,
        "proof": f"git merge-base --is-ancestor {ancestor} {descendant} -> exit 0",
        "reasons": list(reasons or []),
    }


def assess(**overrides):
    kwargs = {
        "issue_number": ISSUE,
        "branch_name": BRANCH,
        "worktree_path": WORKTREE,
        "remote": "prgs",
        "org": "ExampleOrg",
        "repo": "ExampleRepo",
        "identity": IDENTITY,
        "profile": PROFILE,
        "current_branch": BRANCH,
        "porcelain_status": "",
        "head_sha": RECORDED,
        "remote_head_sha": RECORDED,
        "pr_head_sha": RECORDED,
        "pr_number": PR_NUMBER,
        "competing_live_locks": [],
        "candidate_branches": [BRANCH],
        "current_pid": os.getpid(),
        "head_ancestry": None,
    }
    kwargs.update(overrides)
    lock = kwargs.pop("lock", None)
    return issue_lock_recovery.assess_dead_session_lock_recovery(
        make_lock() if lock is None else lock, **kwargs
    )


class TestExactHeadRecoveryStillSucceeds(unittest.TestCase):
    def test_equal_heads_still_sanctioned(self):
        result = assess()
        self.assertTrue(result["recovery_sanctioned"], result["reasons"])
        self.assertEqual(
            result["evidence"]["head_relation"],
            issue_lock_recovery.HEAD_RELATION_EQUAL,
        )
        self.assertEqual(result["evidence"]["recorded_head"], RECORDED)
        self.assertEqual(result["evidence"]["accepted_head"], RECORDED)

    def test_exact_match_record_carries_relation(self):
        record = issue_lock_recovery.build_recovery_record(
            assess(), recovered_at="2026-07-20T00:00:00Z"
        )
        self.assertEqual(record["head_relation"], issue_lock_recovery.HEAD_RELATION_EQUAL)
        self.assertEqual(record["recorded_head"], RECORDED)
        self.assertEqual(record["accepted_head"], RECORDED)


class TestStrictDescendantRecoverySucceeds(unittest.TestCase):
    def test_clean_strict_descendant_recovers(self):
        result = assess(
            head_sha=DESCENDANT,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=ancestry_ok(),
        )
        self.assertTrue(result["recovery_sanctioned"], result["reasons"])
        self.assertEqual(
            result["evidence"]["head_relation"],
            issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
        )
        self.assertEqual(result["evidence"]["recorded_head"], RECORDED)
        self.assertEqual(result["evidence"]["accepted_head"], DESCENDANT)
        self.assertIsNotNone(result["evidence"]["ancestry_proof"])
        self.assertTrue(
            any("strictly descends" in r for r in result["reasons"]),
            result["reasons"],
        )

    def test_pr_still_at_recorded_head_is_ok_for_descendant(self):
        # Remediation is local only; open PR still points at the recorded head.
        result = assess(
            head_sha=DESCENDANT,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=ancestry_ok(),
        )
        self.assertTrue(result["recovery_sanctioned"], result["reasons"])

    def test_recovery_record_names_both_heads_and_proof(self):
        result = assess(
            head_sha=DESCENDANT,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=ancestry_ok(),
        )
        record = issue_lock_recovery.build_recovery_record(
            result, recovered_at="2026-07-20T00:00:00Z"
        )
        self.assertEqual(record["recorded_head"], RECORDED)
        self.assertEqual(record["accepted_head"], DESCENDANT)
        self.assertEqual(
            record["head_relation"],
            issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
        )
        self.assertIn("strictly descends", record["ancestry_proof"] or "")
        self.assertEqual(record["prior_session_pid"], result["evidence"]["prior_session_pid"])
        self.assertEqual(record["replacement_session_pid"], os.getpid())


class TestDescendantEvidenceReachesPublicationGates(unittest.TestCase):
    def _descendant_assessment(self):
        return assess(
            head_sha=DESCENDANT,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=ancestry_ok(),
        )

    def test_owning_pr_evidence_carries_accepted_head(self):
        token = issue_lock_recovery.owning_pr_recovery_evidence(
            self._descendant_assessment()
        )
        self.assertIsNotNone(token)
        assert token is not None
        self.assertEqual(token["head_sha"], RECORDED)
        self.assertEqual(token["accepted_head"], DESCENDANT)
        self.assertEqual(token["recorded_head"], RECORDED)
        self.assertEqual(
            token["head_relation"],
            issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
        )

    def test_persisted_lock_rebuilds_owning_pr_evidence(self):
        assessment = self._descendant_assessment()
        record = issue_lock_recovery.build_recovery_record(
            assessment, recovered_at="2026-07-20T00:00:00Z"
        )
        lock = make_lock(dead_session_recovery=record)
        token = issue_lock_recovery.recovered_owning_pr_from_lock(lock)
        self.assertIsNotNone(token)
        assert token is not None
        self.assertEqual(token["pr_number"], PR_NUMBER)
        self.assertEqual(token["head_sha"], RECORDED)
        self.assertEqual(token["accepted_head"], DESCENDANT)

    def test_duplicate_gate_accepts_pr_at_recorded_or_accepted_head(self):
        token = issue_lock_recovery.owning_pr_recovery_evidence(
            self._descendant_assessment()
        )
        for live_sha in (RECORDED, DESCENDANT):
            with self.subTest(live_sha=live_sha):
                gate = issue_work_duplicate_gate.assess_work_issue_duplicate_gate(
                    ISSUE,
                    open_prs=[
                        {
                            "number": PR_NUMBER,
                            "title": f"Closes #{ISSUE}",
                            "body": f"Closes #{ISSUE}",
                            "head": {"ref": BRANCH, "sha": live_sha},
                        }
                    ],
                    branch_names=[BRANCH],
                    claim_entry={"status": "unclaimed"},
                    locked_branch=BRANCH,
                    phase=issue_work_duplicate_gate.PHASE_COMMIT,
                    recovered_owning_pr=token,
                )
                self.assertFalse(gate["block"], gate)
                self.assertTrue(gate["owning_pr_recovery_exempted"])

    def test_duplicate_gate_still_rejects_foreign_head(self):
        token = issue_lock_recovery.owning_pr_recovery_evidence(
            self._descendant_assessment()
        )
        gate = issue_work_duplicate_gate.assess_work_issue_duplicate_gate(
            ISSUE,
            open_prs=[
                {
                    "number": PR_NUMBER,
                    "title": f"Closes #{ISSUE}",
                    "body": f"Closes #{ISSUE}",
                    "head": {"ref": BRANCH, "sha": DIVERGED},
                }
            ],
            branch_names=[BRANCH],
            claim_entry={"status": "unclaimed"},
            locked_branch=BRANCH,
            phase=issue_work_duplicate_gate.PHASE_COMMIT,
            recovered_owning_pr=token,
        )
        self.assertTrue(gate["block"])
        self.assertFalse(gate["owning_pr_recovery_exempted"])


class TestDirtyDescendantRejected(unittest.TestCase):
    def test_dirty_descendant_refused(self):
        result = assess(
            head_sha=DESCENDANT,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=ancestry_ok(),
            porcelain_status=" M issue_lock_recovery.py\n",
        )
        self.assertFalse(result["recovery_sanctioned"])
        self.assertTrue(any("dirty" in r.lower() for r in result["reasons"]))


class TestDivergedAndBehindRejected(unittest.TestCase):
    def test_diverged_head_refused(self):
        result = assess(
            head_sha=DIVERGED,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=ancestry_ok(
                ancestor=RECORDED,
                descendant=DIVERGED,
                is_strict=False,
                reasons=[
                    f"local head {DIVERGED} does not descend from recorded head "
                    f"{RECORDED}"
                ],
            ),
        )
        self.assertFalse(result["recovery_sanctioned"])
        self.assertIsNone(result["evidence"].get("head_relation"))
        self.assertTrue(
            any("does not match remote" in r for r in result["reasons"]),
            result["reasons"],
        )

    def test_local_behind_recorded_refused(self):
        # merge-base --is-ancestor RECORDED BEHIND is false when BEHIND is ancestor.
        result = assess(
            head_sha=BEHIND,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=ancestry_ok(
                ancestor=RECORDED,
                descendant=BEHIND,
                is_strict=False,
                reasons=[
                    f"local head {BEHIND} does not descend from recorded head "
                    f"{RECORDED}"
                ],
            ),
        )
        self.assertFalse(result["recovery_sanctioned"])
        self.assertTrue(
            any("does not match remote" in r or "not a strict descendant" in r
                for r in result["reasons"]),
            result["reasons"],
        )


class TestUnrelatedAndMalformedAncestryRejected(unittest.TestCase):
    def test_missing_ancestry_observation_fails_closed(self):
        result = assess(
            head_sha=DESCENDANT,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=None,
        )
        self.assertFalse(result["recovery_sanctioned"])
        self.assertTrue(
            any("ancestry" in r.lower() for r in result["reasons"]),
            result["reasons"],
        )

    def test_mismatched_probe_pair_fails_closed(self):
        # Observation for a different commit pair must not authorize this pair.
        result = assess(
            head_sha=DESCENDANT,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=ancestry_ok(ancestor=DIVERGED, descendant=DESCENDANT),
        )
        self.assertFalse(result["recovery_sanctioned"])
        self.assertTrue(
            any("not the heads under assessment" in r for r in result["reasons"]),
            result["reasons"],
        )

    def test_rewritten_recorded_head_fails_closed(self):
        result = assess(
            head_sha=DESCENDANT,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=ancestry_ok(ancestor_present=False, is_strict=False),
        )
        self.assertFalse(result["recovery_sanctioned"])
        self.assertTrue(
            any("no longer reachable" in r or "rewritten" in r
                for r in result["reasons"]),
            result["reasons"],
        )

    def test_failed_probe_fails_closed(self):
        result = assess(
            head_sha=DESCENDANT,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=ancestry_ok(
                probe_ok=False,
                is_strict=False,
                reasons=["ancestry probe failed with exit 128; ancestry unproven"],
            ),
        )
        self.assertFalse(result["recovery_sanctioned"])

    def test_pr_head_not_equal_to_recorded_blocks_descendant(self):
        result = assess(
            head_sha=DESCENDANT,
            remote_head_sha=RECORDED,
            pr_head_sha=DIVERGED,
            head_ancestry=ancestry_ok(),
        )
        self.assertFalse(result["recovery_sanctioned"])
        self.assertTrue(
            any("open PR" in r and "does not match" in r for r in result["reasons"]),
            result["reasons"],
        )


class TestLiveOwnerStillRejected(unittest.TestCase):
    def test_live_prior_pid_refused_even_with_descendant_proof(self):
        result = assess(
            lock=make_lock(session_pid=os.getpid()),
            head_sha=DESCENDANT,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=ancestry_ok(),
        )
        self.assertFalse(result["recovery_sanctioned"])
        self.assertTrue(
            any("still alive" in r or "live" in r.lower() for r in result["reasons"]),
            result["reasons"],
        )


class TestDiagnosticsIdentifyDisposition(unittest.TestCase):
    def test_equal_disposition_named(self):
        result = assess()
        self.assertEqual(
            result["evidence"]["head_relation"],
            issue_lock_recovery.HEAD_RELATION_EQUAL,
        )

    def test_descendant_disposition_named(self):
        result = assess(
            head_sha=DESCENDANT,
            remote_head_sha=RECORDED,
            pr_head_sha=RECORDED,
            head_ancestry=ancestry_ok(),
        )
        self.assertEqual(
            result["evidence"]["head_relation"],
            issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
        )

    def test_rejected_divergence_has_no_accepted_relation(self):
        result = assess(
            head_sha=DIVERGED,
            remote_head_sha=RECORDED,
            head_ancestry=None,
        )
        self.assertIsNone(result["evidence"].get("head_relation"))
        message = issue_lock_recovery.format_recovery_refusal(result)
        self.assertIn("fail closed", message)
        self.assertIn("does not match remote", message)


class TestReadHeadAncestryRealGit(unittest.TestCase):
    def _git(self, repo: str, *args: str) -> subprocess.CompletedProcess[str]:
        return subprocess.run(
            ["git", "-C", repo, *args],
            capture_output=True,
            text=True,
            check=True,
        )

    def _init_repo_with_chain(self) -> tuple[str, str, str, str]:
        """Return (repo, parent_sha, child_sha, sibling_sha)."""
        repo = tempfile.mkdtemp(prefix="issue-768-ancestry-")
        self._git(repo, "init")
        self._git(repo, "config", "user.email", "test@example.com")
        self._git(repo, "config", "user.name", "Test")
        path = Path(repo) / "f.txt"
        path.write_text("one\n")
        self._git(repo, "add", "f.txt")
        self._git(repo, "commit", "-m", "parent")
        parent = self._git(repo, "rev-parse", "HEAD").stdout.strip()
        path.write_text("two\n")
        self._git(repo, "add", "f.txt")
        self._git(repo, "commit", "-m", "child")
        child = self._git(repo, "rev-parse", "HEAD").stdout.strip()
        # Divergent sibling: branch from parent, then unique commit.
        self._git(repo, "checkout", "-B", "side", parent)
        path.write_text("side\n")
        self._git(repo, "add", "f.txt")
        self._git(repo, "commit", "-m", "sibling")
        sibling = self._git(repo, "rev-parse", "HEAD").stdout.strip()
        self._git(repo, "checkout", "-B", "main", child)
        return repo, parent, child, sibling

    def test_strict_descendant_observation(self):
        repo, parent, child, _sibling = self._init_repo_with_chain()
        obs = issue_lock_worktree.read_head_ancestry(
            repo, ancestor_sha=parent, descendant_sha=child
        )
        self.assertTrue(obs["probe_ok"])
        self.assertTrue(obs["ancestor_present"])
        self.assertTrue(obs["is_ancestor"])
        self.assertTrue(obs["is_strict_descendant"])
        self.assertEqual(obs["ancestor_sha"], parent)
        self.assertEqual(obs["descendant_sha"], child)

    def test_equal_heads_not_strict_descendant(self):
        repo, parent, _child, _sibling = self._init_repo_with_chain()
        obs = issue_lock_worktree.read_head_ancestry(
            repo, ancestor_sha=parent, descendant_sha=parent
        )
        self.assertTrue(obs["probe_ok"])
        self.assertTrue(obs["is_ancestor"])
        self.assertFalse(obs["is_strict_descendant"])

    def test_diverged_not_ancestor(self):
        repo, _parent, child, sibling = self._init_repo_with_chain()
        # child and sibling share a parent but neither descends from the other.
        obs = issue_lock_worktree.read_head_ancestry(
            repo, ancestor_sha=child, descendant_sha=sibling
        )
        self.assertTrue(obs["probe_ok"])
        self.assertFalse(obs["is_ancestor"])
        self.assertFalse(obs["is_strict_descendant"])

    def test_missing_sha_fails_closed(self):
        repo, _parent, child, _ = self._init_repo_with_chain()
        obs = issue_lock_worktree.read_head_ancestry(
            repo, ancestor_sha="0" * 40, descendant_sha=child
        )
        self.assertFalse(obs["probe_ok"])
        self.assertFalse(obs["ancestor_present"])

    def test_end_to_end_real_git_descendant_recovery(self):
        repo, parent, child, _sibling = self._init_repo_with_chain()
        obs = issue_lock_worktree.read_head_ancestry(
            repo, ancestor_sha=parent, descendant_sha=child
        )
        result = assess(
            worktree_path=repo,
            head_sha=child,
            remote_head_sha=parent,
            pr_head_sha=parent,
            head_ancestry=obs,
            lock=make_lock(worktree_path=repo),
        )
        self.assertTrue(result["recovery_sanctioned"], result["reasons"])
        self.assertEqual(
            result["evidence"]["head_relation"],
            issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
        )


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