diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 1b16902..60579a1 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2081,6 +2081,7 @@ def _assess_issue_duplicate_gate( auth: str, locked_branch: str | None = None, phase: str, + recovered_owning_pr: dict | None = None, ) -> dict: open_prs, branch_names, claim_entry = _collect_issue_duplicate_context( h, o, r, auth, issue_number @@ -2092,6 +2093,7 @@ def _assess_issue_duplicate_gate( claim_entry=claim_entry, locked_branch=locked_branch, phase=phase, + recovered_owning_pr=recovered_owning_pr, ) @@ -3268,6 +3270,17 @@ def gitea_lock_issue( recovery_sanctioned = bool( recovery_assessment and recovery_assessment.get("recovery_sanctioned") ) + # #755: a sanctioned dead-session recovery always has an owning open PR — + # that is what makes it a recovery rather than a fresh claim. Carry the + # server-derived owning-PR evidence into the duplicate-work gate below so + # the PR this lock already owns is not mistaken for competing duplicate + # work. Withheld (None) unless recovery was granted, so the ordinary + # duplicate blocker is untouched on every other path. + recovered_owning_pr = ( + issue_lock_recovery.owning_pr_recovery_evidence(recovery_assessment) + if recovery_sanctioned + else None + ) lock_assessment = issue_lock_worktree.assess_issue_lock_worktree( worktree_path=resolved_worktree, current_branch=git_state.get("current_branch"), @@ -3300,6 +3313,7 @@ def gitea_lock_issue( auth=auth, locked_branch=branch_name, phase=issue_work_duplicate_gate.PHASE_LOCK, + recovered_owning_pr=recovered_owning_pr, ) if duplicate_gate.get("block"): raise ValueError("; ".join(duplicate_gate.get("reasons") or [ diff --git a/issue_lock_recovery.py b/issue_lock_recovery.py index ab27f38..3c2e5e5 100644 --- a/issue_lock_recovery.py +++ b/issue_lock_recovery.py @@ -362,6 +362,57 @@ def _result( } +def owning_pr_recovery_evidence( + assessment: Mapping[str, Any] | None, +) -> dict[str, Any] | None: + """Server-derived proof of the open PR a sanctioned recovery already owns (#755). + + A dead-session recovery is, by construction, recovery of work that already + has an open PR — so the duplicate-work gate's linked-open-PR blocker would + otherwise discard every sanctioned recovery. This distils the completed + assessment into the minimum evidence that gate needs to tell "the PR this + lock already owns" apart from "a competing duplicate PR". + + Returns ``None`` unless recovery was actually granted and the assessment's + own evidence names exactly one owning PR whose head agrees with the local + and remote heads. Nothing here is caller-supplied: every field is copied + from evidence the assessor built out of durable lock state plus live + git/Gitea observation, so a caller cannot manufacture an exemption. + """ + if not isinstance(assessment, Mapping): + return None + if assessment.get("outcome") != RECOVERY_SANCTIONED: + return None + if not assessment.get("recovery_sanctioned"): + return None + + evidence = assessment.get("evidence") or {} + branch_name = _text(evidence.get("locked_branch")) + pr_head = _text(evidence.get("pr_head")) + local_head = _text(evidence.get("local_head")) + remote_head = _text(evidence.get("remote_head")) + raw_pr_number = evidence.get("pr_number") + + if raw_pr_number is None or not branch_name or not pr_head: + return None + # The assessor already required these to agree. Re-check, so a truncated or + # hand-built evidence map can never authorize an exemption. + if pr_head != local_head or pr_head != remote_head: + return None + try: + pr_number = int(raw_pr_number) + issue_number = int(evidence.get("issue_number")) + except (TypeError, ValueError): + return None + + return { + "issue_number": issue_number, + "pr_number": pr_number, + "branch_name": branch_name, + "head_sha": pr_head, + } + + def build_recovery_record( assessment: Mapping[str, Any], *, diff --git a/issue_work_duplicate_gate.py b/issue_work_duplicate_gate.py index fc70437..81350c7 100644 --- a/issue_work_duplicate_gate.py +++ b/issue_work_duplicate_gate.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Any +from typing import Any, Mapping import issue_claim_heartbeat as claim_hb @@ -27,6 +27,116 @@ def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None: return claim_hb._linked_open_pr(issue_number, open_prs) +def _pr_links_issue(issue_number: int, pr: Mapping[str, Any]) -> bool: + """Same linkage rule ``claim_hb._linked_open_pr`` applies, per PR. + + ``_linked_open_pr`` only yields the *first* match, which cannot answer + "is there exactly one linked PR?" — a question the owning-PR exemption + below must answer before it can trust any of them. + """ + pattern = _issue_pattern(issue_number) + head = (pr.get("head") or {}).get("ref") or "" + text = f"{pr.get('title', '')} {pr.get('body', '')}".lower() + if pattern in head.lower(): + return True + return ( + f"closes #{int(issue_number)}" in text + or f"fixes #{int(issue_number)}" in text + ) + + +def _all_linked_open_prs( + issue_number: int, open_prs: list[dict] +) -> list[Mapping[str, Any]]: + return [pr for pr in (open_prs or []) if _pr_links_issue(issue_number, pr)] + + +def _assess_owning_pr_exemption( + issue_number: int, + *, + linked_open_prs: list[Mapping[str, Any]], + locked_branch: str | None, + recovered_owning_pr: Mapping[str, Any] | None, +) -> tuple[bool, list[str]]: + """Is the linked open PR provably the one a sanctioned recovery owns (#755)? + + ``recovered_owning_pr`` is produced by + ``issue_lock_recovery.owning_pr_recovery_evidence`` from a completed + server-side recovery assessment — it is never a caller-supplied field. + Every element is re-checked here against the live PR list this gate was + given, so a stale or partial token cannot widen the exemption. + + Returns ``(exempt, diagnostic_reasons)``. Diagnostics are only emitted when + a token was offered and rejected, so a blocked caller can see which element + of ownership disagreed. + """ + if not recovered_owning_pr: + return False, [] + + notes: list[str] = [] + token_issue = recovered_owning_pr.get("issue_number") + token_pr = recovered_owning_pr.get("pr_number") + token_branch = str(recovered_owning_pr.get("branch_name") or "").strip() + token_head = str(recovered_owning_pr.get("head_sha") or "").strip() + locked = (locked_branch or "").strip() + + if token_issue is not None and int(token_issue) != int(issue_number): + notes.append( + f"recovery evidence is for issue #{token_issue}, not " + f"#{issue_number} (no owning-PR exemption)" + ) + return False, notes + if not locked or not token_branch or locked != token_branch: + notes.append( + f"recovery evidence branch '{token_branch or 'unknown'}' does not " + f"match the branch being locked '{locked or 'unknown'}' " + "(no owning-PR exemption)" + ) + return False, notes + if len(linked_open_prs) != 1: + numbers = ", ".join( + f"#{pr.get('number')}" for pr in linked_open_prs + ) or "none" + notes.append( + f"{len(linked_open_prs)} open PRs link issue #{issue_number} " + f"({numbers}); recovery may only own exactly one " + "(no owning-PR exemption)" + ) + return False, notes + + only = linked_open_prs[0] + head_obj = only.get("head") or {} + only_number = only.get("number") + only_ref = str(head_obj.get("ref") or "").strip() + only_sha = str(head_obj.get("sha") or "").strip() + + if token_pr is None or only_number is None or int(only_number) != int(token_pr): + notes.append( + f"linked open PR #{only_number} is not the recovered owning PR " + f"#{token_pr} (no owning-PR exemption)" + ) + return False, notes + if only_ref != token_branch: + notes.append( + f"open PR #{only_number} head branch '{only_ref}' does not match " + f"the recovered branch '{token_branch}' (no owning-PR exemption)" + ) + return False, notes + if not token_head or not only_sha or only_sha != token_head: + notes.append( + f"open PR #{only_number} head {only_sha or 'unknown'} does not " + f"match the recovered head {token_head or 'unknown'} " + "(no owning-PR exemption)" + ) + return False, notes + + return True, [ + f"open PR #{only_number} is the exact PR already owned by the " + f"recovering lock for issue #{issue_number} (branch '{token_branch}', " + f"head {token_head}); not duplicate work" + ] + + def _matching_branches( issue_number: int, branch_names: list[str], @@ -52,8 +162,15 @@ def assess_work_issue_duplicate_gate( claim_entry: dict | None = None, locked_branch: str | None = None, phase: str = PHASE_LOCK, + recovered_owning_pr: Mapping[str, Any] | None = None, ) -> dict[str, Any]: - """Fail closed when duplicate work is already in flight for an issue.""" + """Fail closed when duplicate work is already in flight for an issue. + + ``recovered_owning_pr`` (#755) is server-derived evidence that a sanctioned + dead-session lock recovery already owns one specific open PR. It exempts + *only* that exact PR from the linked-open-PR blocker; every other duplicate + signal, and every mismatch, keeps failing closed. + """ reasons: list[str] = [] outcome = OUTCOME_DUPLICATE_WORK_NOT_PREVENTED prs = list(open_prs or []) @@ -61,12 +178,23 @@ def assess_work_issue_duplicate_gate( pattern = _issue_pattern(issue_number) linked = _linked_open_pr(issue_number, prs) + linked_open_prs = _all_linked_open_prs(issue_number, prs) + owning_pr_exempted = False + exemption_notes: list[str] = [] if linked: - reasons.append( - f"open PR #{linked.get('number')} already covers issue " - f"#{issue_number} (fail closed)" + owning_pr_exempted, exemption_notes = _assess_owning_pr_exemption( + issue_number, + linked_open_prs=linked_open_prs, + locked_branch=locked_branch, + recovered_owning_pr=recovered_owning_pr, ) - outcome = OUTCOME_DUPLICATE_PR_PREVENTED + if not owning_pr_exempted: + reasons.append( + f"open PR #{linked.get('number')} already covers issue " + f"#{issue_number} (fail closed)" + ) + reasons.extend(exemption_notes) + outcome = OUTCOME_DUPLICATE_PR_PREVENTED conflicting_branches = _matching_branches( issue_number, branches, locked_branch=locked_branch @@ -122,6 +250,9 @@ def assess_work_issue_duplicate_gate( "phase": phase, "outcome": outcome, "linked_open_pr": linked.get("number") if linked else entry.get("linked_open_pr"), + "linked_open_pr_count": len(linked_open_prs), + "owning_pr_recovery_exempted": owning_pr_exempted, + "owning_pr_recovery_notes": list(exemption_notes), "conflicting_branches": conflicting_branches, "claim_status": status or None, "reasons": reasons, diff --git a/tests/test_issue_755_owning_pr_recovery.py b/tests/test_issue_755_owning_pr_recovery.py new file mode 100644 index 0000000..d9fc617 --- /dev/null +++ b/tests/test_issue_755_owning_pr_recovery.py @@ -0,0 +1,609 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 +"""Dead-session lock recovery when the issue already owns an open PR (#755). + +#753 added the recovery *assessor*, but the production ``gitea_lock_issue`` +path still rejected every sanctioned recovery: a dead-session lock is by +construction a lock for work that already has an open PR, and the #400 +duplicate-work gate blocked unconditionally on any linked open PR. These tests +drive the real MCP handler, not just the pure assessor, so that gap cannot +reopen. +""" +import os +import subprocess +import sys +import tempfile +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 issue_lock_provenance # noqa: E402 +import issue_lock_recovery # noqa: E402 +import issue_lock_store # noqa: E402 +import mcp_server # noqa: E402 +from issue_work_duplicate_gate import ( # noqa: E402 + OUTCOME_DUPLICATE_PR_PREVENTED, + OUTCOME_DUPLICATE_WORK_NOT_PREVENTED, + PHASE_LOCK, + assess_work_issue_duplicate_gate, +) + +ISSUE = 4755 +BRANCH = f"fix/issue-{ISSUE}-owning-pr" +OTHER_BRANCH = f"fix/issue-{ISSUE}-competing" +HEAD = "c" * 40 +OTHER_HEAD = "d" * 40 +OWNING_PR = 4756 +OTHER_PR = 4757 +IDENTITY = "example-user" +PROFILE = "test-author-prgs" +ORG = "Scaled-Tech-Consulting" +REPO = "Gitea-Tools" + + +def dead_pid() -> int: + """A PID that has certainly exited (spawned, then reaped).""" + proc = subprocess.Popen([sys.executable, "-c", "pass"]) + proc.wait() + return proc.pid + + +def shifted_ts(hours: int = 4) -> str: + return ( + (datetime.now(timezone.utc) + timedelta(hours=hours)) + .isoformat() + .replace("+00:00", "Z") + ) + + +def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE): + return { + "number": number, + "title": f"fix: something (Closes #{issue})", + "body": f"Closes #{issue}.", + "head": {"ref": ref, "sha": sha}, + } + + +def sanctioned_token( + issue_number=ISSUE, pr_number=OWNING_PR, branch=BRANCH, head=HEAD +): + """The evidence shape the server derives from a granted recovery.""" + return { + "issue_number": issue_number, + "pr_number": pr_number, + "branch_name": branch, + "head_sha": head, + } + + +# ───────────────────────── duplicate gate: exemption ───────────────────────── + + +class TestOwningPrExemptionGranted(unittest.TestCase): + def test_exact_owning_pr_is_not_duplicate_work(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assertFalse(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED) + self.assertEqual(result["linked_open_pr"], OWNING_PR) + self.assertEqual(result["linked_open_pr_count"], 1) + + def test_unrelated_open_pr_alongside_owning_pr_is_ignored(self): + unrelated = { + "number": 999, + "title": "chore: unrelated", + "body": "no linkage", + "head": {"ref": "chore/unrelated", "sha": OTHER_HEAD}, + } + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[unrelated, owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assertFalse(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + self.assertEqual(result["linked_open_pr_count"], 1) + + +class TestOwningPrExemptionRefused(unittest.TestCase): + def assert_blocked(self, result): + self.assertTrue(result["block"]) + self.assertFalse(result["owning_pr_recovery_exempted"]) + self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED) + + def test_no_recovery_evidence_keeps_ordinary_blocker(self): + self.assert_blocked( + assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + ) + ) + + def test_competing_pr_number_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr(number=OTHER_PR)], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + def test_multiple_linked_open_prs_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + self.assertEqual(result["linked_open_pr_count"], 2) + + def test_different_branch_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr(ref=OTHER_BRANCH)], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + def test_locked_branch_differing_from_evidence_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=OTHER_BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + def test_different_head_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr(sha=OTHER_HEAD)], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + def test_evidence_for_another_issue_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(issue_number=ISSUE + 1), + ) + self.assert_blocked(result) + + def test_missing_head_in_live_pr_refused(self): + pr = owning_pr() + pr["head"] = {"ref": BRANCH} + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[pr], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + +class TestOrdinaryDuplicateBehaviorUnchanged(unittest.TestCase): + def test_clean_issue_still_passes(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[], + branch_names=["feat/other-issue-99"], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + ) + self.assertFalse(result["block"]) + self.assertFalse(result["owning_pr_recovery_exempted"]) + + def test_competing_branch_still_blocks_even_with_owning_pr_evidence(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH, OTHER_BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + # The owning PR is exempt, but the competing branch is not. + self.assertTrue(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + self.assertIn(OTHER_BRANCH, result["conflicting_branches"]) + + +# ─────────────────── server-derived evidence cannot be forged ─────────────────── + + +class TestOwningPrEvidenceDerivation(unittest.TestCase): + def granted(self, **evidence_overrides): + evidence = { + "issue_number": ISSUE, + "locked_branch": BRANCH, + "local_head": HEAD, + "remote_head": HEAD, + "pr_head": HEAD, + "pr_number": OWNING_PR, + } + evidence.update(evidence_overrides) + return { + "outcome": issue_lock_recovery.RECOVERY_SANCTIONED, + "recovery_sanctioned": True, + "is_candidate": True, + "reasons": [], + "evidence": evidence, + } + + def test_granted_recovery_yields_evidence(self): + token = issue_lock_recovery.owning_pr_recovery_evidence(self.granted()) + self.assertEqual(token, sanctioned_token()) + + def test_none_assessment_yields_nothing(self): + self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(None)) + + def test_refused_assessment_yields_nothing(self): + refused = self.granted() + refused["outcome"] = issue_lock_recovery.REFUSED + refused["recovery_sanctioned"] = False + self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(refused)) + + def test_sanctioned_flag_without_outcome_yields_nothing(self): + forged = self.granted() + forged["outcome"] = "SOMETHING_ELSE" + self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(forged)) + + def test_missing_pr_number_yields_nothing(self): + self.assertIsNone( + issue_lock_recovery.owning_pr_recovery_evidence( + self.granted(pr_number=None) + ) + ) + + def test_head_disagreement_in_evidence_yields_nothing(self): + self.assertIsNone( + issue_lock_recovery.owning_pr_recovery_evidence( + self.granted(remote_head=OTHER_HEAD) + ) + ) + self.assertIsNone( + issue_lock_recovery.owning_pr_recovery_evidence( + self.granted(local_head=OTHER_HEAD) + ) + ) + + def test_real_refused_assessment_yields_nothing(self): + """End-to-end against the real assessor, not a hand-built dict.""" + lock = { + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": "/scratch/wt", + "remote": "prgs", + "org": "Example-Org", + "repo": "Example-Repo", + "session_pid": os.getpid(), # alive → must refuse + "work_lease": { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": ISSUE, + "branch": BRANCH, + "worktree_path": "/scratch/wt", + "claimant": {"username": IDENTITY, "profile": PROFILE}, + "expires_at": shifted_ts(), + }, + } + assessment = issue_lock_recovery.assess_dead_session_lock_recovery( + lock, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt", + remote="prgs", + org="Example-Org", + repo="Example-Repo", + identity=IDENTITY, + profile=PROFILE, + current_branch=BRANCH, + porcelain_status="", + head_sha=HEAD, + remote_head_sha=HEAD, + pr_head_sha=HEAD, + pr_number=OWNING_PR, + competing_live_locks=[], + candidate_branches=[BRANCH], + current_pid=os.getpid(), + ) + self.assertFalse(assessment["recovery_sanctioned"]) + self.assertIsNone( + issue_lock_recovery.owning_pr_recovery_evidence(assessment) + ) + + +# ──────────────────── end-to-end: the real gitea_lock_issue ──────────────────── + + +class LockIssueEndToEndBase(unittest.TestCase): + """Drives ``mcp_server.gitea_lock_issue`` with live git/Gitea observation + stubbed at the module boundary — the production gate chain itself runs.""" + + def setUp(self): + self.lock_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.lock_dir.cleanup) + self.worktree = os.path.realpath(os.getcwd()) + # Bind host/org/repo to what the ``test-author-prgs`` fixture profile is + # pinned to, so the session-context gate under test is the real one and + # not a cross-host denial. The issue number and lock dir stay synthetic. + self.remotes = patch.dict(mcp_server.REMOTES, { + "prgs": { + "host": "gitea.prgs.cc", + "org": ORG, + "repo": REPO, + }, + }) + self.remotes.start() + self.addCleanup(patch.stopall) + mcp_server._IDENTITY_CACHE.clear() + + def write_durable_lock(self, *, pid, branch=BRANCH, worktree=None): + path = issue_lock_store.lock_file_path( + remote="prgs", + org=ORG, + repo=REPO, + issue_number=ISSUE, + lock_dir=self.lock_dir.name, + ) + claimant = {"username": IDENTITY, "profile": PROFILE} + data = { + "issue_number": ISSUE, + "branch_name": branch, + "remote": "prgs", + "org": ORG, + "repo": REPO, + "worktree_path": worktree or self.worktree, + "session_pid": pid, + "pid": pid, + "work_lease": { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": ISSUE, + "branch": branch, + "worktree_path": worktree or self.worktree, + "claimant": claimant, + "created_at": shifted_ts(-1), + "last_heartbeat_at": shifted_ts(-1), + "expires_at": shifted_ts(), + }, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", + claimant=claimant, + ), + } + issue_lock_store.save_lock_file(path, data) + return path + + def run_lock( + self, + *, + open_prs, + porcelain="", + current_branch=BRANCH, + base_equivalent=False, + branch_names=None, + head_sha=HEAD, + remote_head=HEAD, + ): + branch_names = branch_names if branch_names is not None else [BRANCH] + branch_entries = [ + {"name": name, "commit": {"id": remote_head}} for name in branch_names + ] + env = shared_mutation_env( + "test-author-prgs", + include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=self.lock_dir.name, + ) + with patch( + "mcp_server.api_get_all", return_value=branch_entries + ), patch( + "mcp_server._list_open_pulls", return_value=list(open_prs) + ), patch( + "mcp_server.get_auth_header", return_value="token x" + ), patch( + "mcp_server._work_lease_claimant", + return_value={"username": IDENTITY, "profile": PROFILE}, + ), patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={ + "current_branch": current_branch, + "porcelain_status": porcelain, + "base_equivalent": base_equivalent, + "head_sha": head_sha, + "inspected_git_root": self.worktree, + "base_branch": "master", + }, + ), patch( + "mcp_server.issue_duplicate_context_fetcher", + side_effect=lambda h, o, r, auth, issue_number: ( + list(open_prs), list(branch_names), {"status": "not_claimed"} + ), + ): + with patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return mcp_server.gitea_lock_issue( + issue_number=ISSUE, + branch_name=BRANCH, + remote="prgs", + worktree_path=self.worktree, + ) + + +class TestRecoveryWithOwningPrSucceeds(LockIssueEndToEndBase): + def test_dead_session_recovery_with_owning_pr_relocks(self): + self.write_durable_lock(pid=dead_pid()) + result = self.run_lock(open_prs=[owning_pr()]) + + self.assertTrue(result["success"]) + self.assertEqual(result["issue_number"], ISSUE) + self.assertEqual(result["branch_name"], BRANCH) + self.assertTrue(result["lock_freshness"]["live"]) + self.assertTrue(result["lock_freshness"]["pid_alive"]) + + def test_recovered_lock_records_truthful_provenance(self): + prior = dead_pid() + self.write_durable_lock(pid=prior) + self.run_lock(open_prs=[owning_pr()]) + + lock = issue_lock_store.load_issue_lock( + remote="prgs", + org=ORG, + repo=REPO, + issue_number=ISSUE, + lock_dir=self.lock_dir.name, + ) + record = lock.get("dead_session_recovery") or {} + self.assertTrue(record.get("recovered")) + self.assertEqual(record.get("prior_session_pid"), prior) + self.assertEqual(record.get("replacement_session_pid"), os.getpid()) + self.assertFalse(record.get("prior_pid_alive")) + self.assertEqual(record.get("pr_number"), OWNING_PR) + self.assertEqual(record.get("branch_name"), BRANCH) + self.assertEqual(record.get("identity"), IDENTITY) + + def test_recovered_lock_is_live_and_proves_pr_ownership(self): + """AC6: the persisted lock satisfies update-by-merge's ownership prover.""" + self.write_durable_lock(pid=dead_pid()) + self.run_lock(open_prs=[owning_pr()]) + + lock = issue_lock_store.load_issue_lock( + remote="prgs", + org=ORG, + repo=REPO, + issue_number=ISSUE, + lock_dir=self.lock_dir.name, + ) + self.assertTrue(issue_lock_store.is_lease_live(lock)) + + env = shared_mutation_env( + "test-author-prgs", + include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=self.lock_dir.name, + ) + with patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + ownership = mcp_server._prove_author_ownership_for_pr( + pr_number=OWNING_PR, + pr_title=f"fix: something (Closes #{ISSUE})", + pr_body=f"Closes #{ISSUE}.", + source_branch=BRANCH, + remote="prgs", + host="gitea.prgs.cc", + org=ORG, + repo=REPO, + worktree_path=self.worktree, + ) + self.assertTrue(ownership["proven"], ownership["reasons"]) + self.assertTrue(ownership["has_author_lock"]) + self.assertEqual(ownership["matched_issue"], ISSUE) + + +class TestRecoveryRejectionsEndToEnd(LockIssueEndToEndBase): + def assert_lock_refused(self, **kwargs): + with self.assertRaises((ValueError, RuntimeError)) as ctx: + self.run_lock(**kwargs) + return str(ctx.exception) + + def test_competing_pr_still_blocked(self): + self.write_durable_lock(pid=dead_pid()) + message = self.assert_lock_refused( + open_prs=[owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)], + branch_names=[BRANCH], + ) + self.assertIn("already covers issue", message) + + def test_multiple_linked_open_prs_blocked(self): + self.write_durable_lock(pid=dead_pid()) + message = self.assert_lock_refused( + open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)], + ) + self.assertIn("already covers issue", message) + + def test_owning_pr_on_a_different_head_blocked(self): + self.write_durable_lock(pid=dead_pid()) + self.assert_lock_refused(open_prs=[owning_pr(sha=OTHER_HEAD)]) + + def test_lock_registered_to_a_different_worktree_blocked(self): + self.write_durable_lock( + pid=dead_pid(), worktree=os.path.join(self.worktree, "elsewhere") + ) + self.assert_lock_refused(open_prs=[owning_pr()]) + + def test_dirty_worktree_blocked(self): + self.write_durable_lock(pid=dead_pid()) + self.assert_lock_refused( + open_prs=[owning_pr()], porcelain=" M gitea_mcp_server.py" + ) + + def test_worktree_parked_on_another_branch_blocked(self): + self.write_durable_lock(pid=dead_pid()) + self.assert_lock_refused( + open_prs=[owning_pr()], current_branch="master" + ) + + def test_local_head_differing_from_remote_blocked(self): + self.write_durable_lock(pid=dead_pid()) + self.assert_lock_refused( + open_prs=[owning_pr()], head_sha=OTHER_HEAD + ) + + def test_live_prior_pid_blocked(self): + self.write_durable_lock(pid=os.getpid()) + self.assert_lock_refused(open_prs=[owning_pr()]) + + def test_new_claim_without_prior_lock_still_requires_base_equivalence(self): + """AC10: no durable lock → no recovery → base-equivalence still rules.""" + self.assert_lock_refused(open_prs=[], branch_names=[]) + + +if __name__ == "__main__": + unittest.main()