"""Unpublished-claim dead-session lock recovery (#772). An author claim whose work exists only as a clean local commit — never pushed, never turned into a PR — was unrecoverable once the owning session exited: every recovery disposition derived ownership from a remote branch head or an owning PR head, and an unpublished claim has neither. These tests exercise the disposition through its evidence, never through any issue number: every case here uses an arbitrary issue number, and the same assertions hold for any other. The #617 *shape* (durable lock, unexpired lease, dead PID, no remote branch, no PR) is what is under test. """ from __future__ import annotations import os import subprocess import sys import tempfile import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import issue_lock_recovery # noqa: E402 import issue_lock_store # noqa: E402 import issue_lock_worktree # noqa: E402 DEAD_PID = 999_999_999 LIVE_PID = os.getpid() WORKTREE = "/tmp/wt/issue-901-example" BRANCH = "fix/issue-901-example" BASE_SHA = "0568f44cb2d87e78fd394a27a670e33c84f7842f" HEAD_SHA = "b46f0f9f138d569edbc73e0d36df4b34239f9934" OTHER_SHA = "1111111111111111111111111111111111111111" def durable_lock(**overrides): """A complete, internally consistent unpublished author lock.""" lock = { "issue_number": 901, "branch_name": BRANCH, "worktree_path": WORKTREE, "remote": "prgs", "org": "Example-Org", "repo": "Example-Repo", "session_pid": DEAD_PID, "pid": DEAD_PID, "work_lease": { "operation_type": "author_issue_work", "issue_number": 901, "pr_number": None, "branch": BRANCH, "worktree_path": WORKTREE, "claimant": {"username": "author-user", "profile": "prgs-author"}, }, "claimant": {"username": "author-user", "profile": "prgs-author"}, } lock.update(overrides) return lock def base_ancestry( *, ancestor=BASE_SHA, descendant=HEAD_SHA, probe_ok=True, ancestor_present=True, strict=True, reasons=None, ): return { "ancestor_sha": ancestor, "descendant_sha": descendant, "probe_ok": probe_ok, "ancestor_present": ancestor_present, "descendant_present": True, "is_ancestor": strict, "is_strict_descendant": strict, "proof": "merge-base --is-ancestor -> exit 0", "reasons": reasons or [], } def assess(lock=None, **overrides): """Assess an unpublished claim; overrides tweak one fact at a time.""" kwargs = { "issue_number": 901, "branch_name": BRANCH, "worktree_path": WORKTREE, "remote": "prgs", "org": "Example-Org", "repo": "Example-Repo", "identity": "author-user", "profile": "prgs-author", "current_branch": BRANCH, "porcelain_status": "", "head_sha": HEAD_SHA, "remote_head_sha": None, "pr_head_sha": None, "pr_number": None, "competing_live_locks": [], "candidate_branches": [BRANCH], "current_pid": LIVE_PID, "remote_branch_exists": False, "recorded_base_sha": BASE_SHA, "base_ancestry": base_ancestry(), } kwargs.update(overrides) return issue_lock_recovery.assess_dead_session_lock_recovery( durable_lock() if lock is None else lock, **kwargs ) class UnpublishedClaimRecoveryGranted(unittest.TestCase): """The #617 shape: every element agrees and the owner is dead.""" def test_recovery_is_sanctioned(self): result = assess() self.assertEqual(result["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED) self.assertTrue(result["recovery_sanctioned"]) def test_mode_and_head_relation_name_the_unpublished_disposition(self): evidence = assess()["evidence"] self.assertEqual( evidence["recovery_mode"], issue_lock_recovery.RECOVERY_MODE_UNPUBLISHED_CLAIM, ) self.assertEqual( evidence["head_relation"], issue_lock_recovery.HEAD_RELATION_DESCENDS_FROM_BASE, ) self.assertEqual(evidence["recorded_base"], BASE_SHA) self.assertEqual(evidence["accepted_head"], HEAD_SHA) self.assertFalse(evidence["remote_branch_exists"]) def test_no_remote_branch_or_pr_is_required(self): """Neither artifact may be demanded for this recovery mode.""" result = assess(remote_head_sha=None, pr_head_sha=None, pr_number=None) self.assertTrue(result["recovery_sanctioned"]) def test_recovery_record_carries_mode_and_base(self): record = issue_lock_recovery.build_recovery_record( assess(), recovered_at="2026-07-20T19:05:27Z" ) self.assertTrue(record["recovered"]) self.assertEqual( record["recovery_mode"], issue_lock_recovery.RECOVERY_MODE_UNPUBLISHED_CLAIM, ) self.assertEqual(record["recorded_base"], BASE_SHA) self.assertEqual(record["accepted_head"], HEAD_SHA) def test_unpublished_recovery_grants_no_owning_pr_exemption(self): """There is no PR, so no duplicate-gate exemption may be produced.""" self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(assess())) class UnpublishedClaimRecoveryRefused(unittest.TestCase): """Every rejection condition must fail closed, independently.""" def refusal(self, **overrides): result = assess(**overrides) self.assertEqual(result["outcome"], issue_lock_recovery.REFUSED) self.assertFalse(result["recovery_sanctioned"]) return " ".join(result["reasons"]) def test_live_prior_owner(self): lock = durable_lock(session_pid=LIVE_PID, pid=LIVE_PID) result = assess(lock=lock, current_pid=LIVE_PID + 1) self.assertFalse(result["recovery_sanctioned"]) self.assertIn("still alive", " ".join(result["reasons"])) def test_foreign_identity(self): self.assertIn( "does not match active identity", self.refusal(identity="someone-else") ) def test_foreign_profile(self): self.assertIn( "does not match active profile", self.refusal(profile="prgs-reviewer") ) def test_dirty_worktree(self): self.assertIn("clean", self.refusal(porcelain_status=" M gitea_mcp_server.py\n")) def test_branch_mismatch(self): self.assertIn("not the locked branch", self.refusal(current_branch="fix/other")) def test_worktree_mismatch(self): self.assertIn( "does not match declared", self.refusal(worktree_path="/tmp/wt/elsewhere") ) def test_head_unrelated_to_recorded_base(self): reasons = self.refusal( base_ancestry=base_ancestry( strict=False, reasons=["local head does not descend from recorded base"], ) ) self.assertIn("descend", reasons) def test_rewritten_base_is_unreachable(self): self.assertIn( "no longer reachable", self.refusal(base_ancestry=base_ancestry(ancestor_present=False)), ) def test_missing_recorded_base(self): self.assertIn( "recorded base", self.refusal(recorded_base_sha=None, base_ancestry=None), ) def test_ancestry_observation_for_other_commits_is_rejected(self): """A probe taken for a different pair cannot authorize recovery.""" self.assertIn( "not the commits under assessment", self.refusal(base_ancestry=base_ancestry(ancestor=OTHER_SHA)), ) def test_unproven_probe(self): self.assertIn( "unproven", self.refusal( base_ancestry=base_ancestry( probe_ok=False, reasons=["ancestry unproven"] ) ), ) def test_competing_live_lock(self): self.assertIn( "competing live lock", self.refusal( competing_live_locks=[ { "issue_number": 901, "branch_name": BRANCH, "worktree_path": "/tmp/wt/other-session", "pid": LIVE_PID, } ] ), ) def test_competing_branch_claim(self): self.assertIn( "ownership is ambiguous", self.refusal(candidate_branches=[BRANCH, "fix/issue-901-duplicate"]), ) def test_pr_without_remote_branch_is_contradictory(self): self.assertIn( "contradictory", self.refusal(pr_head_sha=HEAD_SHA, pr_number=42), ) def test_missing_durable_evidence(self): lock = durable_lock() lock.pop("worktree_path") result = assess(lock=lock) self.assertEqual(result["outcome"], issue_lock_recovery.REFUSED) self.assertIn("incomplete", " ".join(result["reasons"])) def test_absent_remote_head_is_not_itself_permission(self): """The defining regression: no remote ref must not read as consent. A mismatched claim with no remote branch is refused for the mismatch, never waved through because the head comparison was unavailable. """ result = assess(identity="someone-else", profile="prgs-reviewer") self.assertFalse(result["recovery_sanctioned"]) joined = " ".join(result["reasons"]) self.assertIn("does not match active identity", joined) self.assertIn("does not match active profile", joined) class PublishedRecoveryUnregressed(unittest.TestCase): """#753 equality and #768 descendant recovery must still work.""" def published(self, **overrides): kwargs = { "remote_branch_exists": True, "remote_head_sha": HEAD_SHA, "recorded_base_sha": None, "base_ancestry": None, } kwargs.update(overrides) return assess(**kwargs) def test_equal_head_recovery_still_granted(self): result = self.published() self.assertTrue(result["recovery_sanctioned"]) self.assertEqual( result["evidence"]["head_relation"], issue_lock_recovery.HEAD_RELATION_EQUAL, ) self.assertEqual( result["evidence"]["recovery_mode"], issue_lock_recovery.RECOVERY_MODE_PUBLISHED_OWNING_PR, ) def test_strict_descendant_recovery_still_granted(self): result = self.published( remote_head_sha=BASE_SHA, head_ancestry=base_ancestry(ancestor=BASE_SHA, descendant=HEAD_SHA), ) self.assertTrue(result["recovery_sanctioned"]) self.assertEqual( result["evidence"]["head_relation"], issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT, ) def test_owning_pr_exemption_still_produced(self): result = self.published(pr_head_sha=HEAD_SHA, pr_number=42) evidence = issue_lock_recovery.owning_pr_recovery_evidence(result) self.assertIsNotNone(evidence) self.assertEqual(evidence["pr_number"], 42) def test_published_branch_with_undeterminable_head_still_fails_closed(self): """A failed head lookup is not the same as an absent branch.""" result = self.published(remote_head_sha=None) self.assertFalse(result["recovery_sanctioned"]) self.assertIn("could not be determined", " ".join(result["reasons"])) class RecordedBaseObservation(unittest.TestCase): """``read_recorded_base`` observes real git, never a caller's claim.""" def setUp(self): self.tmp = tempfile.mkdtemp() self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.tmp], check=False)) self.git("init", "-q", "-b", "master") self.git("config", "user.email", "test@example.com") self.git("config", "user.name", "Test") with open(os.path.join(self.tmp, "seed.txt"), "w") as fh: fh.write("seed\n") self.git("add", "seed.txt") self.git("commit", "-q", "-m", "seed") self.base = self.rev("HEAD") def git(self, *args): return subprocess.run( ["git", "-C", self.tmp, *args], capture_output=True, text=True, check=False ) def rev(self, ref): return (self.git("rev-parse", ref).stdout or "").strip() def test_base_is_the_merge_base_of_head_and_master(self): self.git("checkout", "-q", "-b", BRANCH) with open(os.path.join(self.tmp, "work.txt"), "w") as fh: fh.write("work\n") self.git("add", "work.txt") self.git("commit", "-q", "-m", "work") head = self.rev("HEAD") observed = issue_lock_worktree.read_recorded_base(self.tmp, head_sha=head) self.assertTrue(observed["probe_ok"]) self.assertEqual(observed["base_sha"], self.base) self.assertEqual(observed["base_branch"], "master") def test_unrelated_history_yields_no_base(self): """An orphan branch shares no ancestor and must not report one.""" self.git("checkout", "-q", "--orphan", "unrelated") self.git("rm", "-rqf", ".") with open(os.path.join(self.tmp, "other.txt"), "w") as fh: fh.write("other\n") self.git("add", "other.txt") self.git("commit", "-q", "-m", "unrelated root") head = self.rev("HEAD") observed = issue_lock_worktree.read_recorded_base(self.tmp, head_sha=head) self.assertIsNone(observed["base_sha"]) self.assertIn("unrelated", " ".join(observed["reasons"])) def test_missing_head_fails_closed(self): observed = issue_lock_worktree.read_recorded_base(self.tmp, head_sha=None) self.assertFalse(observed["probe_ok"]) self.assertIsNone(observed["base_sha"]) class LockGenerationCas(unittest.TestCase): """Two replacement sessions must not both recover one claim.""" def setUp(self): self.dir = tempfile.mkdtemp() self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.dir], check=False)) def record(self, **overrides): data = { "issue_number": 901, "branch_name": BRANCH, "worktree_path": WORKTREE, "remote": "prgs", "org": "Example-Org", "repo": "Example-Repo", } data.update(overrides) return data def test_generation_increments_on_each_write(self): path = issue_lock_store.bind_session_lock(self.record(), self.dir) self.assertEqual( issue_lock_store.lock_generation(issue_lock_store.read_lock_file(path)), 1 ) issue_lock_store.bind_session_lock(self.record(), self.dir) self.assertEqual( issue_lock_store.lock_generation(issue_lock_store.read_lock_file(path)), 2 ) def test_absent_generation_reads_as_zero(self): self.assertEqual(issue_lock_store.lock_generation({}), 0) self.assertEqual(issue_lock_store.lock_generation(None), 0) self.assertEqual(issue_lock_store.lock_generation({"lock_generation": "x"}), 0) def test_matching_expectation_succeeds(self): issue_lock_store.bind_session_lock(self.record(), self.dir) issue_lock_store.bind_session_lock( self.record(), self.dir, expected_generation=1 ) def test_second_recoverer_loses_the_race(self): """Both sessions read generation 1; only the first may write.""" issue_lock_store.bind_session_lock(self.record(), self.dir) observed_by_both = 1 issue_lock_store.bind_session_lock( self.record(), self.dir, expected_generation=observed_by_both ) with self.assertRaises(RuntimeError) as caught: issue_lock_store.bind_session_lock( self.record(), self.dir, expected_generation=observed_by_both ) self.assertIn("generation changed", str(caught.exception)) def test_unconditional_write_is_unchanged(self): """Ordinary first-time claims pass no expectation and still succeed.""" issue_lock_store.bind_session_lock(self.record(), self.dir) issue_lock_store.bind_session_lock(self.record(), self.dir) if __name__ == "__main__": unittest.main()