"""Durable linked-issue lock head refresh + merge-sync dead-session recovery (#871). ``gitea_update_pr_branch_by_merge`` advances a PR's *remote* head but historically never advanced the linked durable issue lock's recorded head. After the owning session died the drifted lock became unrecoverable and no further synchronization was possible (PR #866 / issue #855). Two halves are covered: * the write-side refresh (``issue_lock_store.assess/apply_durable_lock_head_refresh``) that records the new synced head under compare-and-swap with read-after-write; and * the read-side recovery relation (``issue_lock_recovery`` + ``issue_lock_worktree.read_merge_sync_provenance``) that lets a dead-session lock whose recorded head is a merge-sync *ancestor* of the live PR head be recovered — and nothing else. """ 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 ISSUE = 8710 PR_NUMBER = 8711 BRANCH = f"fix/issue-{ISSUE}-durable-lock-head-refresh" IDENTITY = "example-user" PROFILE = "example-author" OLD = "a" * 40 NEW1 = "b" * 40 NEW2 = "c" * 40 BASE = "d" * 40 REMOTE = "prgs" ORG = "ExampleOrg" REPO = "ExampleRepo" 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 _git(cwd, *args): return subprocess.run( ["git", "-C", cwd, *args], capture_output=True, text=True, check=True, ) def _rev(cwd, ref="HEAD") -> str: return _git(cwd, "rev-parse", ref).stdout.strip() def build_merge_sync_repo(tmp: str) -> dict: """Build a repo where a feature branch was synced by merging master in. Returns a dict with the prior (branch) head, the synced merge-commit head, the master tip, plus a rebase-style linear descendant and an unrelated head. """ _git(tmp, "init", "-q", "-b", "master") _git(tmp, "config", "user.email", "t@example.com") _git(tmp, "config", "user.name", "T") Path(tmp, "base.txt").write_text("base\n") _git(tmp, "add", "-A") _git(tmp, "commit", "-q", "-m", "root") # Feature branch cut from root, one commit — this is the PRIOR/recorded head. _git(tmp, "checkout", "-q", "-b", BRANCH) Path(tmp, "feature.txt").write_text("feature\n") _git(tmp, "add", "-A") _git(tmp, "commit", "-q", "-m", "feature work") prior = _rev(tmp) # Master advances (the base the sync will merge in). _git(tmp, "checkout", "-q", "master") Path(tmp, "base.txt").write_text("base\nmore\n") _git(tmp, "add", "-A") _git(tmp, "commit", "-q", "-m", "master advance") master_tip = _rev(tmp) # Sync: merge master INTO the feature branch → merge commit, first parent = prior. _git(tmp, "checkout", "-q", BRANCH) _git(tmp, "merge", "-q", "--no-ff", "-m", "Merge master into feature", "master") synced = _rev(tmp) # A plain linear descendant of prior (NOT a merge) — a rebase/extra-commit shape. _git(tmp, "checkout", "-q", "-b", "linear-branch", prior) Path(tmp, "extra.txt").write_text("extra\n") _git(tmp, "add", "-A") _git(tmp, "commit", "-q", "-m", "extra linear commit") linear = _rev(tmp) # An unrelated root (force-push / rewritten history shape). unrelated_dir = tempfile.mkdtemp() _git(unrelated_dir, "init", "-q", "-b", "x") _git(unrelated_dir, "config", "user.email", "t@example.com") _git(unrelated_dir, "config", "user.name", "T") Path(unrelated_dir, "z.txt").write_text("z\n") _git(unrelated_dir, "add", "-A") _git(unrelated_dir, "commit", "-q", "-m", "unrelated") unrelated = _rev(unrelated_dir) # Leave the worktree checked out on the feature branch at the PRIOR head, as # a dead author session that never advanced would have left it. _git(tmp, "checkout", "-q", BRANCH) _git(tmp, "reset", "-q", "--hard", prior) return { "prior": prior, "master_tip": master_tip, "synced": synced, "linear": linear, "unrelated": unrelated, } # ─────────────────────────── write-side refresh ─────────────────────────── class TestDurableLockHeadRefresh(unittest.TestCase): def setUp(self): self.lock_dir = tempfile.mkdtemp() self.wt = tempfile.mkdtemp() lock_data = { "issue_number": ISSUE, "branch_name": BRANCH, "worktree_path": self.wt, "remote": REMOTE, "org": ORG, "repo": REPO, "claimant": {"username": IDENTITY, "profile": PROFILE}, "work_lease": { "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, "issue_number": ISSUE, "branch": BRANCH, "worktree_path": self.wt, "claimant": {"username": IDENTITY, "profile": PROFILE}, "expires_at": future_ts(), }, } issue_lock_store.bind_session_lock(lock_data, lock_dir=self.lock_dir) def _apply(self, **over): kw = dict( remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.wt, pr_number=PR_NUMBER, identity=IDENTITY, profile=PROFILE, current_pid=os.getpid(), expected_old_head=OLD, new_head=NEW1, synced_at=future_ts(0), base_head=BASE, lock_dir=self.lock_dir, ) kw.update(over) return issue_lock_store.apply_durable_lock_head_refresh(**kw) def _load(self): return issue_lock_store.load_issue_lock( remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, lock_dir=self.lock_dir, ) def test_first_sync_updates_recorded_head(self): """AC1: first base sync writes the resulting head to the durable lock.""" res = self._apply() self.assertTrue(res["refreshed"], res["reasons"]) self.assertTrue(res["read_after_write_ok"]) self.assertEqual(self._load().get("synced_pr_head"), NEW1) def test_second_sync_after_master_advance(self): """AC2: a later master advance permits a second sanctioned sync.""" self.assertTrue(self._apply()["refreshed"]) res2 = self._apply(expected_old_head=NEW1, new_head=NEW2) self.assertTrue(res2["refreshed"], res2["reasons"]) self.assertEqual(self._load().get("synced_pr_head"), NEW2) history = self._load().get("branch_sync_history") self.assertEqual(len(history), 2) self.assertEqual(history[0]["last_synced_pr_head"], NEW1) self.assertEqual(history[1]["prior_pr_head"], NEW1) def test_cas_detects_concurrent_head_change(self): """AC6: CAS refuses when the recorded synced head is not the old head.""" self.assertTrue(self._apply()["refreshed"]) # recorded head now NEW1 # A second sync claiming the old head is still OLD must fail closed. res = self._apply(expected_old_head=OLD, new_head=NEW2) self.assertFalse(res["refreshed"]) self.assertTrue(any("CAS" in r or "concurrent" in r for r in res["reasons"])) self.assertEqual(self._load().get("synced_pr_head"), NEW1) def test_wrong_issue_fails_closed(self): res = self._apply(issue_number=999999) self.assertFalse(res["refreshed"]) def test_wrong_branch_fails_closed(self): res = self._apply(branch_name="fix/issue-8710-wrong") self.assertFalse(res["refreshed"]) def test_wrong_repo_fails_closed(self): res = self._apply(repo="OtherRepo") self.assertFalse(res["refreshed"]) def test_wrong_identity_fails_closed(self): res = self._apply(identity="intruder") self.assertFalse(res["refreshed"]) def test_wrong_profile_fails_closed(self): res = self._apply(profile="prgs-reviewer") self.assertFalse(res["refreshed"]) def test_foreign_session_fails_closed(self): """A refresh is not a recovery: the current process must own the lock.""" path = issue_lock_store.lock_file_path( remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, lock_dir=self.lock_dir, ) rec = issue_lock_store.read_lock_file(path) rec["session_pid"] = dead_pid() rec["pid"] = rec["session_pid"] issue_lock_store.save_lock_file(path, rec) res = self._apply() self.assertFalse(res["refreshed"]) self.assertTrue(any("current session" in r or "live owner" in r for r in res["reasons"])) def test_new_equals_old_fails_closed(self): res = self._apply(expected_old_head=OLD, new_head=OLD) self.assertFalse(res["refreshed"]) def test_non_full_sha_fails_closed(self): self.assertFalse(self._apply(new_head="deadbeef")["refreshed"]) self.assertFalse(self._apply(expected_old_head="xyz")["refreshed"]) def test_no_lock_fails_closed(self): assessment = issue_lock_store.assess_durable_lock_head_refresh( None, remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.wt, pr_number=PR_NUMBER, identity=IDENTITY, profile=PROFILE, current_pid=os.getpid(), expected_old_head=OLD, new_head=NEW1, ) self.assertFalse(assessment["allowed"]) # ─────────────────────── merge-sync provenance (real git) ─────────────────── class TestMergeSyncProvenanceObservation(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.shas = build_merge_sync_repo(self.tmp) def test_merge_sync_is_recognized(self): obs = issue_lock_worktree.read_merge_sync_provenance( self.tmp, prior_head_sha=self.shas["prior"], synced_head_sha=self.shas["synced"], ) self.assertTrue(obs["is_merge_sync"], obs["reasons"]) self.assertTrue(obs["prior_is_ancestor"]) self.assertTrue(obs["synced_is_merge"]) self.assertTrue(obs["first_parent_reaches_prior"]) def test_linear_descendant_is_not_a_merge_sync(self): """A plain non-merge descendant (rebase/extra commit) is not a sync.""" obs = issue_lock_worktree.read_merge_sync_provenance( self.tmp, prior_head_sha=self.shas["prior"], synced_head_sha=self.shas["linear"], ) self.assertTrue(obs["probe_ok"]) self.assertFalse(obs["is_merge_sync"]) self.assertFalse(obs["synced_is_merge"]) def test_unrelated_history_fails_closed(self): """A rewritten/force-pushed head where prior is unreachable fails closed.""" obs = issue_lock_worktree.read_merge_sync_provenance( self.tmp, prior_head_sha=self.shas["prior"], synced_head_sha=self.shas["unrelated"], ) self.assertFalse(obs["is_merge_sync"]) def test_missing_args_fail_closed(self): obs = issue_lock_worktree.read_merge_sync_provenance( self.tmp, prior_head_sha=None, synced_head_sha=self.shas["synced"], ) self.assertFalse(obs["is_merge_sync"]) # ──────────────────── merge-sync dead-session recovery ────────────────────── def make_dead_lock(worktree, **over): pid = dead_pid() lock = { "issue_number": ISSUE, "branch_name": BRANCH, "worktree_path": worktree, "remote": REMOTE, "org": ORG, "repo": REPO, "session_pid": pid, "pid": pid, "claimant": {"username": IDENTITY, "profile": PROFILE}, "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(over) return lock def sync_prov(prior, synced, **over): d = { "prior_head_sha": prior, "synced_head_sha": synced, "probe_ok": True, "prior_present": True, "synced_present": True, "prior_is_ancestor": True, "synced_is_merge": True, "first_parent_reaches_prior": True, "is_merge_sync": True, "first_parent_sha": prior, "parent_count": 2, "proof": f"{synced} merged base into branch above {prior}", "reasons": [], } d.update(over) return d class TestMergeSyncRecovery(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.shas = build_merge_sync_repo(self.tmp) self.prior = self.shas["prior"] self.synced = self.shas["synced"] def _assess(self, **over): lock = over.pop("_lock", None) or make_dead_lock(self.tmp) kw = dict( issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.tmp, remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, current_branch=BRANCH, porcelain_status="", head_sha=self.prior, remote_head_sha=self.synced, pr_head_sha=self.synced, pr_number=PR_NUMBER, competing_live_locks=[], candidate_branches=[BRANCH], current_pid=os.getpid(), remote_branch_exists=True, sync_provenance=sync_prov(self.prior, self.synced), ) kw.update(over) return issue_lock_recovery.assess_dead_session_lock_recovery(lock, **kw) def test_merge_sync_drift_is_recoverable(self): """AC3/AC4: dead session, recorded head is a merge-sync ancestor of PR head.""" res = self._assess() self.assertEqual(res["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED, res["reasons"]) self.assertEqual( res["evidence"]["head_relation"], issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED, ) self.assertEqual(res["evidence"]["accepted_head"], self.synced) def test_missing_provenance_fails_closed(self): """No server-derived provenance → cannot accept a remote ahead of local.""" res = self._assess(sync_provenance=None) self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED) def test_non_ancestor_recorded_head_fails_closed(self): """AC7: provenance that does not prove ancestry is rejected.""" res = self._assess( sync_provenance=sync_prov( self.prior, self.synced, prior_is_ancestor=False, is_merge_sync=False, reasons=["prior head is not an ancestor"], ) ) self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED) def test_force_pushed_history_fails_closed(self): """AC8: a rewritten head (not a merge sync) stays protected.""" res = self._assess( sync_provenance=sync_prov( self.prior, self.synced, is_merge_sync=False, synced_is_merge=False, reasons=["not a merge-based sync"], ) ) self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED) def test_provenance_for_other_commits_fails_closed(self): """Provenance whose endpoints differ from the heads under assessment is rejected.""" res = self._assess( sync_provenance=sync_prov("f" * 40, self.synced), ) self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED) def test_dirty_worktree_fails_closed(self): """AC11: dirty worktrees remain protected.""" res = self._assess(porcelain_status=" M feature.txt\n") self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED) def test_live_owner_fails_closed(self): """AC10: a live recorded owner is not a dead-session recovery.""" lock = make_dead_lock(self.tmp, session_pid=os.getpid(), pid=os.getpid()) res = self._assess(_lock=lock) self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED) def test_competing_claimant_fails_closed(self): """AC13: a competing live lock blocks recovery.""" res = self._assess( competing_live_locks=[{ "issue_number": ISSUE, "branch_name": BRANCH, "worktree_path": "/some/other/wt", "pid": os.getpid(), }] ) self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED) def test_wrong_branch_fails_closed(self): """AC9: worktree on a different branch fails closed.""" res = self._assess(current_branch="fix/issue-8710-other") self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED) def test_wrong_identity_fails_closed(self): res = self._assess(identity="intruder") self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED) def test_pr_head_mismatch_fails_closed(self): """The open PR must sit at the synced remote head.""" res = self._assess(pr_head_sha="e" * 40) self.assertEqual(res["outcome"], issue_lock_recovery.REFUSED) def test_owning_pr_evidence_for_merge_sync(self): res = self._assess() ev = issue_lock_recovery.owning_pr_recovery_evidence(res) self.assertIsNotNone(ev) self.assertEqual(ev["pr_number"], PR_NUMBER) self.assertEqual(ev["head_sha"], self.synced) self.assertEqual( ev["head_relation"], issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED, ) def test_recovered_owning_pr_from_persisted_record(self): res = self._assess() record = issue_lock_recovery.build_recovery_record(res, recovered_at=future_ts(0)) lock = {"issue_number": ISSUE, "branch_name": BRANCH, "dead_session_recovery": record} rebuilt = issue_lock_recovery.recovered_owning_pr_from_lock(lock) self.assertIsNotNone(rebuilt) self.assertEqual(rebuilt["head_sha"], self.synced) self.assertEqual( rebuilt["head_relation"], issue_lock_recovery.HEAD_RELATION_REMOTE_MERGE_SYNCED, ) class TestExistingRelationsUnchanged(unittest.TestCase): """AC14/AC15: equal-head recovery still works; merge-sync did not weaken it.""" def setUp(self): self.tmp = tempfile.mkdtemp() self.shas = build_merge_sync_repo(self.tmp) def test_equal_head_recovery_still_sanctioned(self): # Worktree at prior head; remote also at prior head → the #753 equal case. prior = self.shas["prior"] lock = make_dead_lock(self.tmp) res = issue_lock_recovery.assess_dead_session_lock_recovery( lock, issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.tmp, remote=REMOTE, org=ORG, repo=REPO, identity=IDENTITY, profile=PROFILE, current_branch=BRANCH, porcelain_status="", head_sha=prior, remote_head_sha=prior, pr_head_sha=prior, pr_number=PR_NUMBER, competing_live_locks=[], candidate_branches=[BRANCH], current_pid=os.getpid(), remote_branch_exists=True, ) self.assertEqual(res["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED, res["reasons"]) self.assertEqual( res["evidence"]["head_relation"], issue_lock_recovery.HEAD_RELATION_EQUAL, ) class TestUpdatePrWrapperPartialFailure(unittest.TestCase): """AC5/AC16: the tool advances the remote head then refreshes the durable lock. When the durable refresh fails after the remote advance, the tool must report a partial lifecycle failure and NOT a fully successful synchronization. Exact PR- head / base-head pinning is preserved (delegated to the real preflight, stubbed here only to isolate the post-update lifecycle branch). """ def setUp(self): import gitea_mcp_server as gms # noqa: E402 self.gms = gms self._orig = {} def _patch(name, value): self._orig[name] = getattr(gms, name) setattr(gms, name, value) _patch("get_profile", lambda *a, **k: { "allowed_operations": ["gitea.branch.push"], "forbidden_operations": [], "profile_name": "prgs-author", }) _patch("_role_kind", lambda *a, **k: "author") _patch("_profile_operation_gate", lambda *a, **k: None) _patch("_permission_block_report", lambda *a, **k: {}) _patch("_resolve", lambda *a, **k: ("gitea.prgs.cc", ORG, REPO)) _patch("_verify_role_mutation_workspace", lambda *a, **k: None) _patch("_get_workspace_porcelain", lambda *a, **k: "") _patch("_canonical_local_git_root", lambda *a, **k: "/x") _patch("_master_parity_block", lambda *a, **k: None) _patch("_auth", lambda *a, **k: {"token": "x"}) _patch("repo_api_url", lambda *a, **k: "http://api") _patch("_redact", lambda s: s) _patch("_work_lease_claimant", lambda *a, **k: { "username": IDENTITY, "profile": PROFILE, }) _patch("_prove_author_ownership_for_pr", lambda *a, **k: { "has_author_lock": True, "matched_issue": ISSUE, "matched_via": "branch", "linked_issues": [ISSUE], "recovered_owning_pr": None, "reasons": [], }) # Real preflight is unit-tested elsewhere; stub it to isolate the # post-update durable-lock lifecycle branch under test. orig_pf = gms.pr_sync_status.assess_update_pr_branch_preflight self._orig_pf = orig_pf gms.pr_sync_status.assess_update_pr_branch_preflight = ( lambda *a, **k: {"mutation_allowed": True, "reasons": [], "performed": False} ) # Sequence the two GET /pulls calls: OLD before update, NEW after. self._pull_calls = {"n": 0} def fake_api_request(method, url, auth, *a, **k): m = method.upper() if m == "GET" and url.endswith(f"/pulls/{PR_NUMBER}"): self._pull_calls["n"] += 1 head = OLD if self._pull_calls["n"] == 1 else NEW1 return { "state": "open", "head": {"sha": head, "ref": BRANCH}, "base": {"sha": BASE, "ref": "master"}, "mergeable": True, "title": "t", "body": "b", } if m == "GET" and "/branches/" in url: return {"commit": {"id": BASE}} if m == "POST" and "/update" in url: return {} return {} _patch("api_request", fake_api_request) def tearDown(self): for name, value in self._orig.items(): setattr(self.gms, name, value) self.gms.pr_sync_status.assess_update_pr_branch_preflight = self._orig_pf def _run(self): return self.gms.gitea_update_pr_branch_by_merge( pr_number=PR_NUMBER, expected_pr_head_sha=OLD, expected_base_head_sha=BASE, remote=REMOTE, worktree_path="/tmp/branches/wt-871", ) def test_partial_failure_when_refresh_fails(self): self._orig["apply_durable_lock_head_refresh"] = ( self.gms.issue_lock_store.apply_durable_lock_head_refresh ) self.gms.issue_lock_store.apply_durable_lock_head_refresh = ( lambda **k: {"refreshed": False, "reasons": ["forced refresh failure"]} ) try: res = self._run() finally: self.gms.issue_lock_store.apply_durable_lock_head_refresh = ( self._orig["apply_durable_lock_head_refresh"] ) self.assertTrue(res["performed"]) self.assertEqual(res["new_pr_head_sha"], NEW1) self.assertFalse(res["success"]) self.assertTrue(res["partial_lifecycle_failure"]) self.assertFalse(res["durable_lock_refreshed"]) def test_full_success_when_refresh_succeeds(self): self._orig["apply_durable_lock_head_refresh"] = ( self.gms.issue_lock_store.apply_durable_lock_head_refresh ) self.gms.issue_lock_store.apply_durable_lock_head_refresh = ( lambda **k: {"refreshed": True, "read_after_write_ok": True, "new_head": NEW1, "reasons": ["ok"]} ) try: res = self._run() finally: self.gms.issue_lock_store.apply_durable_lock_head_refresh = ( self._orig["apply_durable_lock_head_refresh"] ) self.assertTrue(res["success"]) self.assertTrue(res["performed"]) self.assertTrue(res["durable_lock_refreshed"]) self.assertTrue(res["fully_synchronized"]) self.assertEqual(res["new_pr_head_sha"], NEW1) if __name__ == "__main__": unittest.main()