From 79334d48408fd446ddf1e8be332495960b847af6 Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Sun, 26 Jul 2026 10:42:44 -0500 Subject: [PATCH] fix(gate): preserve exact-owner renewal evidence across duplicate rechecks (Closes #945) An exact-owner lease renewal (#760) is granted the owning-PR duplicate-work waiver inside gitea_lock_issue, but every later enforcement path rebuilt that proof from the durable lock through issue_lock_recovery.recovered_owning_pr_from_lock, which reads only the dead_session_recovery block. A plain renewal persists its proof under lease_renewal instead, so the waiver expired with the lock call and the next author mutation was refused duplicate_commit_prevented with owning_pr_recovery_exempted: false on the very PR the renewal had just proved. Add issue_lock_renewal.owning_pr_renewal_from_lock as the renewal mirror of the recovery rebuild, and resolve both halves through one helper, _owning_pr_continuation_from_lock, in the same precedence gitea_lock_issue applies. The three enforcement paths that previously saw only recovery evidence now share that resolver: the commit/create-PR duplicate recheck, the read-only duplicate assessor, and the push-ownership prover. The rebuild re-applies the equality the renewal assessor required (PR head == local head == remote head) and additionally binds the record to the claimant the lock names, so a renewal block cannot be reused under another identity, profile, or workflow session. Validation of the resulting token against live PR state is unchanged and still owned solely by issue_work_duplicate_gate._assess_owning_pr_exemption, so repository, issue, PR, branch and head binding continue to be enforced in exactly one place. No public signature, MCP tool schema, refusal shape, or reason code changes. Dead-session recovery behaviour is untouched. Tests: tests/test_issue_945_owning_pr_renewal_continuation.py - 49 tests, 8 subtests, all in-memory (no durable branch, worktree, lock, lease or PR). Against unmodified aab54d48 the suite is 47 failed / 2 passed; the 2 that pass are the defect-pinning reproductions. Full suite 28F/5574P/6S/1002 subtests at head vs 28F/5525P/6S/994 at base, with identical failing test id sets. Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 51 +- issue_lock_renewal.py | 84 ++++ ...ssue_945_owning_pr_renewal_continuation.py | 465 ++++++++++++++++++ 3 files changed, 593 insertions(+), 7 deletions(-) create mode 100644 tests/test_issue_945_owning_pr_renewal_continuation.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index f3f6edc..a3663d0 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2781,6 +2781,36 @@ def _collect_issue_duplicate_context( return issue_duplicate_context_fetcher(h, o, r, auth, issue_number) +def _owning_pr_continuation_from_lock(lock_record: dict | None) -> dict | None: + """Owning-PR continuation evidence a persisted lock still proves (#945). + + ``gitea_lock_issue`` grants the duplicate-work waiver from either a + sanctioned dead-session recovery (#755) or a sanctioned exact-owner renewal + (#760), in that precedence. Every later enforcement path — commit, + create-PR, push-ownership, and the read-only duplicate assessor — re-derives + ownership from the durable lock instead of that live assessment. + + Until #945 only the recovery half was rebuilt there, so an ordinary + exact-owner renewal lost its waiver the moment ``gitea_lock_issue`` + returned: the author renewed successfully and was then refused + ``duplicate_commit_prevented`` with ``owning_pr_recovery_exempted: false`` + on the very PR the renewal had just proved it owned. + + Resolving both halves here, in the same precedence the lock path applies, + keeps the answer from drifting between the gate that grants the waiver and + the gates that enforce it. This only decides which server-written block the + token is rebuilt from — the token is still re-validated against live PR + state by ``issue_work_duplicate_gate._assess_owning_pr_exemption``, which + remains the single authoritative policy for whether an exemption applies. + """ + if not lock_record: + return None + recovered = issue_lock_recovery.recovered_owning_pr_from_lock(lock_record) + if recovered: + return recovered + return issue_lock_renewal.owning_pr_renewal_from_lock(lock_record) + + def _assess_issue_duplicate_gate( issue_number: int, *, @@ -2833,6 +2863,11 @@ def _enforce_locked_issue_duplicate_recheck( commit and create-PR phases run in their own calls, long after the recovery assessment ended, so without this they re-block the very PR the recovery already proved belongs to this author. + + #945: an exact-owner *renewal* (#760) owns its open PR for exactly the same + reason, and ``gitea_lock_issue`` already waives the blocker for both. Both + halves are resolved together here so the renewal waiver survives past the + lock call instead of expiring with it. """ lock_data = _load_existing_issue_lock() if not lock_data: @@ -2856,9 +2891,7 @@ def _enforce_locked_issue_duplicate_recheck( auth=auth, locked_branch=locked_branch, phase=phase, - recovered_owning_pr=issue_lock_recovery.recovered_owning_pr_from_lock( - lock_data - ), + recovered_owning_pr=_owning_pr_continuation_from_lock(lock_data), ) if gate.get("block"): return gate @@ -5140,9 +5173,10 @@ def gitea_assess_work_issue_duplicate( recovered_owning_pr = None lock_data = _load_existing_issue_lock() if lock_data and int(lock_data.get("issue_number") or 0) == int(issue_number): - recovered_owning_pr = issue_lock_recovery.recovered_owning_pr_from_lock( - lock_data - ) + # #945: rebuilt from a sanctioned recovery *or* a sanctioned exact-owner + # renewal, so this read-only assessor reports the same disposition the + # commit and create-PR gates will enforce. + recovered_owning_pr = _owning_pr_continuation_from_lock(lock_data) gate = _assess_issue_duplicate_gate( issue_number, h=h, @@ -19424,7 +19458,10 @@ def _prove_author_ownership_for_pr( # advance is the one recovery already sanctioned, not a foreign head. recovered_owning_pr = None if proven and lock_record: - candidate = issue_lock_recovery.recovered_owning_pr_from_lock(lock_record) + # #945: a sanctioned exact-owner renewal proves the same ownership of + # the same PR, so the push gate resolves both halves rather than seeing + # only the recovery one. + candidate = _owning_pr_continuation_from_lock(lock_record) if candidate and int(candidate.get("pr_number") or 0) == int(pr_number): recovered_owning_pr = candidate return { diff --git a/issue_lock_renewal.py b/issue_lock_renewal.py index ec8a436..1739f73 100644 --- a/issue_lock_renewal.py +++ b/issue_lock_renewal.py @@ -436,6 +436,90 @@ def owning_pr_renewal_evidence( } +def owning_pr_renewal_from_lock( + lock_record: Mapping[str, Any] | None, +) -> dict[str, Any] | None: + """Rebuild owning-PR renewal evidence from a persisted lock (#945). + + The renewal mirror of ``issue_lock_recovery.recovered_owning_pr_from_lock``. + ``owning_pr_renewal_evidence`` supplies the waiver for the duration of the + ``gitea_lock_issue`` call only. The commit, push, create-PR, and + duplicate-assessment gates run later in their own calls and re-derive + ownership from the durable lock instead — so without this the open PR that + renewal already proved belongs to this author reappears there as competing + duplicate work, and the exact owner is refused with + ``duplicate_commit_prevented`` despite complete matching evidence. + + This reads only the ``lease_renewal`` block that the server itself writes, + on a lock the caller must already own. Like the recovery mirror it is a + re-read of server-derived state, never a fresh assertion: a caller able to + forge it could equally forge the lock file every other ownership gate + already treats as authoritative. + + Renewal has no descendant case — the assessor required the local, remote and + PR heads to be equal — so that equality is re-checked here, and the record + must still name the claimant the lock records. + """ + if not isinstance(lock_record, Mapping): + return None + record = lock_record.get("lease_renewal") + if not isinstance(record, Mapping) or not record.get("renewed"): + return None + + branch_name = _text(record.get("branch_name")) or _text( + lock_record.get("branch_name") + ) + pr_head = _text(record.get("pr_head_sha")) + local_head = _text(record.get("head_sha")) + remote_head = _text(record.get("remote_head_sha")) + raw_pr_number = record.get("pr_number") + raw_issue_number = lock_record.get("issue_number") + + if raw_pr_number is None or raw_issue_number is None: + return None + if not branch_name or not pr_head: + return None + # The assessor required all three heads to agree before it granted renewal. + # Re-check, so a truncated, drifted, or hand-built record cannot widen the + # exemption past the single head the renewal disposition actually proved. + if not local_head or not remote_head: + return None + if pr_head != local_head or pr_head != remote_head: + return None + # Renewal is refused outright unless the durable lock records both a + # claimant username and profile, so a sanctioned record always carries them. + # Requiring them to still agree keeps a renewal block from being reused + # under an identity or profile the lock no longer names. + claimant = lock_record.get("claimant") + if not isinstance(claimant, Mapping): + lease = lock_record.get("work_lease") + claimant = lease.get("claimant") if isinstance(lease, Mapping) else None + if not isinstance(claimant, Mapping): + return None + identity = _text(record.get("identity")) + profile = _text(record.get("profile")) + if not identity or identity != _text(claimant.get("username")): + return None + if not profile or profile != _text(claimant.get("profile")): + return None + + try: + pr_number = int(raw_pr_number) + issue_number = int(raw_issue_number) + except (TypeError, ValueError): + return None + + return { + "issue_number": issue_number, + "pr_number": pr_number, + "branch_name": branch_name, + "head_sha": pr_head, + "recorded_head": pr_head, + "accepted_head": pr_head, + "head_relation": "equal", + } + + def build_renewal_record( assessment: Mapping[str, Any] | None, *, diff --git a/tests/test_issue_945_owning_pr_renewal_continuation.py b/tests/test_issue_945_owning_pr_renewal_continuation.py new file mode 100644 index 0000000..41bca8b --- /dev/null +++ b/tests/test_issue_945_owning_pr_renewal_continuation.py @@ -0,0 +1,465 @@ +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: F401,E402 +"""Exact-owner renewal keeps its owning-PR waiver past lock_issue (#945). + +#755 taught the duplicate-work gate that a sanctioned *dead-session recovery* +owns its open PR, and #768 taught the later gates to rebuild that proof from the +durable lock. #760 added the exact-owner *renewal* disposition and granted it +the same waiver inside ``gitea_lock_issue`` — but never added the matching +rebuild. So an ordinary renewal held the waiver only for the duration of the +lock call: ``_enforce_locked_issue_duplicate_recheck`` asked +``recovered_owning_pr_from_lock``, which reads only ``dead_session_recovery``, +and the very next commit was refused ``duplicate_commit_prevented`` with +``owning_pr_recovery_exempted: false`` on the PR the renewal had just proved. + +``TestPreFixReproduction`` pins that defect directly: the recovery-only rebuild +still returns ``None`` for a renewal lock, which is exactly why the gates lost +the waiver. Everything else proves the renewal half now survives, that recovery +is unchanged, and that no path grants an exemption on weaker evidence. + +Every fixture here is an in-memory mapping. Nothing writes a branch, worktree, +lock file, lease, comment, or PR (#945 AC18). +""" +import copy +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import gitea_mcp_server # noqa: E402 +import issue_lock_recovery # noqa: E402 +import issue_lock_renewal # noqa: E402 +from issue_work_duplicate_gate import ( # noqa: E402 + OUTCOME_DUPLICATE_WORK_NOT_PREVENTED, + PHASE_COMMIT, + PHASE_CREATE_PR, + PHASE_LOCK, + PHASE_PUSH, + assess_work_issue_duplicate_gate, +) + +ISSUE = 4945 +OWNING_PR = 4946 +OTHER_PR = 4947 +BRANCH = f"fix/issue-{ISSUE}-owning-pr-renewal" +OTHER_BRANCH = f"fix/issue-{ISSUE}-competing" +HEAD = "a" * 40 +OTHER_HEAD = "b" * 40 +IDENTITY = "example-user" +PROFILE = "test-author-prgs" + + +def renewal_record(**overrides): + """The ``lease_renewal`` block ``build_renewal_record`` writes on success.""" + record = { + "renewed": True, + "renewed_at": "2026-01-01T00:00:00Z", + "prior_pid": 4242, + "prior_pid_alive": True, + "prior_expires_at": "2026-01-01T00:00:00Z", + "replacement_pid": 4243, + "new_expires_at": "2026-01-01T00:10:00Z", + "identity": IDENTITY, + "profile": PROFILE, + "branch_name": BRANCH, + "worktree_path": f"branches/issue-{ISSUE}-owning-pr-renewal", + "head_sha": HEAD, + "remote_head_sha": HEAD, + "pr_head_sha": HEAD, + "pr_number": OWNING_PR, + "reason": "expired lease renewed by its exact recorded owner", + "proof": [], + } + record.update(overrides) + return record + + +def renewal_lock(record=None, *, issue_number=ISSUE, claimant=True, **lock_overrides): + lock = { + "issue_number": issue_number, + "branch_name": BRANCH, + "lease_renewal": renewal_record() if record is None else record, + } + if claimant: + lock["claimant"] = {"username": IDENTITY, "profile": PROFILE} + lock.update(lock_overrides) + return lock + + +def recovery_lock(pr_number=OWNING_PR, head=HEAD): + """A lock carrying sanctioned dead-session recovery evidence (#755/#768).""" + return { + "issue_number": ISSUE, + "branch_name": BRANCH, + "claimant": {"username": IDENTITY, "profile": PROFILE}, + "dead_session_recovery": { + "recovered": True, + "branch_name": BRANCH, + "pr_number": pr_number, + "pr_head": head, + "recorded_head": head, + "accepted_head": head, + "head_relation": issue_lock_recovery.HEAD_RELATION_EQUAL, + }, + } + + +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 gate(phase, *, token, open_prs=None, branch_names=None, locked_branch=BRANCH): + return assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()] if open_prs is None else open_prs, + branch_names=branch_names or [], + claim_entry={}, + locked_branch=locked_branch, + phase=phase, + recovered_owning_pr=token, + ) + + +# ───────────────────── the defect this issue exists to fix ───────────────────── + + +class TestPreFixReproduction(unittest.TestCase): + """The exact wiring gap: renewal evidence was invisible to later gates.""" + + def test_recovery_only_rebuild_cannot_see_a_renewal_lock(self): + # This is the pre-fix behaviour of every enforcement path. It is correct + # for the recovery rebuild to ignore a renewal block -- the defect was + # that nothing else looked at it. + self.assertIsNone( + issue_lock_recovery.recovered_owning_pr_from_lock(renewal_lock()) + ) + + def test_renewal_lock_produced_no_exemption_before_the_fix(self): + # Feeding the gate what the pre-fix code fed it (recovery rebuild only) + # reproduces the reported refusal at the commit phase. + token = issue_lock_recovery.recovered_owning_pr_from_lock(renewal_lock()) + result = gate(PHASE_COMMIT, token=token) + self.assertTrue(result["block"]) + self.assertEqual(result["outcome"], "duplicate_commit_prevented") + self.assertFalse(result["owning_pr_recovery_exempted"]) + self.assertEqual(result["owning_pr_recovery_notes"], []) + + def test_shared_resolver_now_sees_it(self): + self.assertIsNotNone( + gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock()) + ) + + +# ───────────────────────── rebuild: the granted case ───────────────────────── + + +class TestRenewalRebuildGranted(unittest.TestCase): + def test_sanctioned_renewal_rebuilds_owning_pr_evidence(self): + token = issue_lock_renewal.owning_pr_renewal_from_lock(renewal_lock()) + self.assertEqual( + token, + { + "issue_number": ISSUE, + "pr_number": OWNING_PR, + "branch_name": BRANCH, + "head_sha": HEAD, + "recorded_head": HEAD, + "accepted_head": HEAD, + "head_relation": "equal", + }, + ) + + def test_branch_falls_back_to_the_lock_branch(self): + lock = renewal_lock(renewal_record(branch_name="")) + token = issue_lock_renewal.owning_pr_renewal_from_lock(lock) + self.assertEqual(token["branch_name"], BRANCH) + + def test_claimant_may_live_under_work_lease(self): + lock = renewal_lock(claimant=False) + lock["work_lease"] = {"claimant": {"username": IDENTITY, "profile": PROFILE}} + self.assertIsNotNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock)) + + def test_rebuild_does_not_mutate_the_lock(self): + lock = renewal_lock() + before = copy.deepcopy(lock) + issue_lock_renewal.owning_pr_renewal_from_lock(lock) + self.assertEqual(lock, before) + + +# ───────────────────────── rebuild: fails closed ───────────────────────── + + +class TestRenewalRebuildFailsClosed(unittest.TestCase): + def assertNoEvidence(self, lock): + self.assertIsNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock)) + + def test_no_lock_at_all(self): + self.assertNoEvidence(None) + self.assertNoEvidence({}) + self.assertNoEvidence("not-a-mapping") + + def test_lock_without_renewal_block(self): + # A fresh claim, or a lock whose renewal block was replaced. + self.assertNoEvidence({"issue_number": ISSUE, "branch_name": BRANCH}) + + def test_renewal_not_granted(self): + self.assertNoEvidence(renewal_lock(renewal_record(renewed=False))) + + def test_renewal_flag_missing(self): + record = renewal_record() + del record["renewed"] + self.assertNoEvidence(renewal_lock(record)) + + def test_renewal_block_malformed(self): + self.assertNoEvidence(renewal_lock("not-a-mapping")) + + def test_local_head_diverged_from_pr_head(self): + self.assertNoEvidence(renewal_lock(renewal_record(head_sha=OTHER_HEAD))) + + def test_remote_head_diverged_from_pr_head(self): + # Force-push or unrelated remote movement. + self.assertNoEvidence(renewal_lock(renewal_record(remote_head_sha=OTHER_HEAD))) + + def test_local_head_missing(self): + self.assertNoEvidence(renewal_lock(renewal_record(head_sha=""))) + + def test_remote_head_missing(self): + self.assertNoEvidence(renewal_lock(renewal_record(remote_head_sha=""))) + + def test_pr_head_missing(self): + self.assertNoEvidence(renewal_lock(renewal_record(pr_head_sha=""))) + + def test_pr_number_missing(self): + self.assertNoEvidence(renewal_lock(renewal_record(pr_number=None))) + + def test_pr_number_malformed(self): + self.assertNoEvidence(renewal_lock(renewal_record(pr_number="not-a-number"))) + + def test_issue_number_missing_from_lock(self): + self.assertNoEvidence(renewal_lock(issue_number=None)) + + def test_branch_unknown_everywhere(self): + lock = renewal_lock(renewal_record(branch_name="")) + lock["branch_name"] = "" + self.assertNoEvidence(lock) + + def test_identity_mismatch(self): + self.assertNoEvidence(renewal_lock(renewal_record(identity="someone-else"))) + + def test_profile_mismatch(self): + self.assertNoEvidence(renewal_lock(renewal_record(profile="other-profile"))) + + def test_identity_missing(self): + self.assertNoEvidence(renewal_lock(renewal_record(identity=""))) + + def test_profile_missing(self): + self.assertNoEvidence(renewal_lock(renewal_record(profile=""))) + + def test_claimant_absent(self): + self.assertNoEvidence(renewal_lock(claimant=False)) + + def test_evidence_from_a_different_session_is_not_reusable(self): + # A renewal block left by another workflow session names another + # claimant, so the lock it is found on cannot inherit its authority. + lock = renewal_lock() + lock["claimant"] = {"username": "other-session-user", "profile": PROFILE} + self.assertNoEvidence(lock) + + +# ───────────────────────── the shared resolver ───────────────────────── + + +class TestSharedResolver(unittest.TestCase): + def test_recovery_lock_resolves_to_recovery_evidence(self): + token = gitea_mcp_server._owning_pr_continuation_from_lock(recovery_lock()) + self.assertEqual(token["pr_number"], OWNING_PR) + + def test_renewal_lock_resolves_to_renewal_evidence(self): + token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock()) + self.assertEqual(token["pr_number"], OWNING_PR) + + def test_recovery_takes_precedence_over_renewal(self): + # Same precedence gitea_lock_issue applies when granting the waiver, so + # the answer cannot differ between the granting and enforcing paths. + lock = recovery_lock(pr_number=OTHER_PR, head=OTHER_HEAD) + lock["lease_renewal"] = renewal_record() + token = gitea_mcp_server._owning_pr_continuation_from_lock(lock) + self.assertEqual(token["pr_number"], OTHER_PR) + + def test_no_evidence_resolves_to_none(self): + self.assertIsNone(gitea_mcp_server._owning_pr_continuation_from_lock(None)) + self.assertIsNone(gitea_mcp_server._owning_pr_continuation_from_lock({})) + self.assertIsNone( + gitea_mcp_server._owning_pr_continuation_from_lock( + {"issue_number": ISSUE, "branch_name": BRANCH} + ) + ) + + +# ────────────── every enforcement path uses the same decision ────────────── + + +class TestEnforcementPathsShareOneDecision(unittest.TestCase): + """AC: commit, push and create-PR gates consume one authoritative token.""" + + def setUp(self): + self.token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock()) + + def test_commit_phase_permits_continuation(self): + result = gate(PHASE_COMMIT, token=self.token) + self.assertFalse(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED) + + def test_create_pr_phase_permits_continuation(self): + result = gate(PHASE_CREATE_PR, token=self.token) + self.assertFalse(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + + def test_push_phase_permits_continuation(self): + result = gate(PHASE_PUSH, token=self.token) + self.assertFalse(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + + def test_lock_phase_permits_continuation(self): + result = gate(PHASE_LOCK, token=self.token) + self.assertFalse(result["block"]) + + def test_all_phases_agree(self): + outcomes = { + phase: gate(phase, token=self.token)["block"] + for phase in (PHASE_LOCK, PHASE_COMMIT, PHASE_PUSH, PHASE_CREATE_PR) + } + self.assertEqual(set(outcomes.values()), {False}, outcomes) + + def test_dead_session_recovery_still_permits_continuation(self): + token = gitea_mcp_server._owning_pr_continuation_from_lock(recovery_lock()) + for phase in (PHASE_COMMIT, PHASE_PUSH, PHASE_CREATE_PR): + with self.subTest(phase=phase): + result = gate(phase, token=token) + self.assertFalse(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + + +# ───────────────── the exemption cannot be widened ───────────────── + + +class TestExemptionCannotBeWidened(unittest.TestCase): + def setUp(self): + self.token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock()) + + def test_an_open_pr_alone_grants_nothing(self): + result = gate(PHASE_COMMIT, token=None) + self.assertTrue(result["block"]) + self.assertFalse(result["owning_pr_recovery_exempted"]) + + def test_a_second_pr_is_refused(self): + result = gate( + PHASE_CREATE_PR, + token=self.token, + open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)], + ) + self.assertTrue(result["block"]) + self.assertFalse(result["owning_pr_recovery_exempted"]) + + def test_a_different_pr_is_refused(self): + result = gate( + PHASE_COMMIT, token=self.token, open_prs=[owning_pr(number=OTHER_PR)] + ) + self.assertTrue(result["block"]) + + def test_a_different_branch_is_refused(self): + result = gate( + PHASE_COMMIT, token=self.token, open_prs=[owning_pr(ref=OTHER_BRANCH)] + ) + self.assertTrue(result["block"]) + + def test_locked_branch_mismatch_is_refused(self): + result = gate(PHASE_COMMIT, token=self.token, locked_branch=OTHER_BRANCH) + self.assertTrue(result["block"]) + + def test_live_pr_head_divergence_is_refused(self): + # Force-push or unrelated remote movement after renewal. + result = gate( + PHASE_COMMIT, token=self.token, open_prs=[owning_pr(sha=OTHER_HEAD)] + ) + self.assertTrue(result["block"]) + + def test_evidence_for_another_issue_is_refused(self): + foreign = gitea_mcp_server._owning_pr_continuation_from_lock( + renewal_lock(issue_number=ISSUE + 1) + ) + result = gate(PHASE_COMMIT, token=foreign) + self.assertTrue(result["block"]) + + def test_sequential_tasks_do_not_inherit_continuation(self): + # One daemon serves many tasks. A renewal proved for issue N must not + # authorize continuation for the next task's issue. + prior_task = gitea_mcp_server._owning_pr_continuation_from_lock( + renewal_lock(issue_number=ISSUE + 7) + ) + self.assertIsNotNone(prior_task) + self.assertTrue(gate(PHASE_COMMIT, token=prior_task)["block"]) + + +# ───────────────── ordinary duplicate prevention is intact ───────────────── + + +class TestDuplicatePreventionRetained(unittest.TestCase): + def test_competing_branch_still_blocks(self): + token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock()) + result = gate( + PHASE_COMMIT, + token=token, + open_prs=[], + branch_names=[BRANCH, OTHER_BRANCH], + ) + self.assertTrue(result["block"]) + + def test_unrelated_work_without_a_lock_still_blocks(self): + token = gitea_mcp_server._owning_pr_continuation_from_lock(None) + self.assertIsNone(token) + self.assertTrue(gate(PHASE_COMMIT, token=token)["block"]) + + +# ───────────────── refusals stay structured and auditable ───────────────── + + +class TestRefusalShapePreserved(unittest.TestCase): + def test_blocked_result_keeps_its_audit_fields(self): + token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock()) + result = gate( + PHASE_COMMIT, token=token, open_prs=[owning_pr(number=OTHER_PR)] + ) + for field in ( + "block", + "outcome", + "reasons", + "owning_pr_recovery_exempted", + "owning_pr_recovery_notes", + ): + with self.subTest(field=field): + self.assertIn(field, result) + self.assertTrue(result["reasons"]) + # A rejected token explains which element of ownership disagreed. + self.assertTrue(result["owning_pr_recovery_notes"]) + + def test_granted_result_records_why(self): + token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock()) + result = gate(PHASE_COMMIT, token=token) + self.assertTrue(result["owning_pr_recovery_notes"]) + self.assertIn( + f"#{OWNING_PR}", " ".join(result["owning_pr_recovery_notes"]) + ) + + +if __name__ == "__main__": + unittest.main()