From 79334d48408fd446ddf1e8be332495960b847af6 Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Sun, 26 Jul 2026 10:42:44 -0500 Subject: [PATCH 1/2] 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() From b1fcf159378e7c72b67cc5ac507920d8390ef107 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 27 Jul 2026 18:22:07 -0400 Subject: [PATCH 2/2] fix(gate): complete owning-PR renewal enforcement coverage (#945) Addresses review 623 on PR #946 (B1 blocker, F2 medium, F3 minor). B1 - the wiring this branch exists to install had no regression coverage. The existing suite exercised owning_pr_renewal_from_lock, _owning_pr_continuation_from_lock and the duplicate gate directly, but never drove an enforcement path, so reverting any of the three call sites left the whole repository green. Add tests/test_issue_945_enforcement_path_wiring.py, which drives the real mcp_server._enforce_locked_issue_duplicate_recheck (the shared recheck behind gitea_commit_files and gitea_create_pr), mcp_server.gitea_assess_work_issue_duplicate and mcp_server._prove_author_ownership_for_pr against a renewal-bearing lock, and asserts each grants the exemption. Reverting the commit/create-PR recheck to the recovery-only rebuild now fails 8 tests and 4 subtests; reverting the assessor or the push prover fails 2 each. The suite also keeps the fail-closed matrix on the real paths: an open PR alone, a second PR, a different PR, branch, issue or head, identity and profile mismatch, ungranted and malformed renewal blocks, and sequential-task non-inheritance are all still refused. F2 - the claimant check compares lease_renewal.identity/profile against the claimant recorded on the same lock file. Both sides are server-written fields of one document, so it is an internal-consistency check, not verification of the live authenticated caller. Correct the docstring and the inline comment to say so, and document the binding that actually prevents cross-session reuse: the enforcement paths load the lock through _load_existing_issue_lock() with no issue coordinates, which resolves issue_lock_store.read_session_issue_lock() to the session pointer at session-{os.getpid()}.json, so lock selection is scoped to the operating-system process. Its limits are stated too - per-process rather than per-authenticated-user, silent on locks reached by explicit coordinates, and silent on two roles sharing one process. Live identity and profile stay enforced by the mutation-authority and profile gates, not by this rebuild. F3 - _owning_pr_continuation_from_lock previously fell through to renewal when a dead_session_recovery block was present but failed to rebuild, so a recovery record naming one PR could be bypassed by renewal evidence naming another. Present-but-unusable recovery evidence is now ambiguous rather than absent and fails closed. Because an expired lease whose recorded owner has also died satisfies both dispositions in one gitea_lock_issue call, a sanctioned pair is a reachable state; when both rebuild, they must agree on issue, PR, branch and every head, or no continuation authority is returned. Recovery-only and renewal-only locks keep their existing behaviour exactly. Validation of the resulting token against live PR state remains untouched and solely owned by issue_work_duplicate_gate._assess_owning_pr_exemption. No public signature, MCP tool schema, refusal shape or reason code changes. Tests: focused #945/#755/#760 suites 137 passed, 19 subtests. Full suite in a branches/ worktree: 30F/5619P/6S/1013 subtests at this head vs 30F/5572P/6S/1002 at a clean checkout of 79334d48, with byte-identical failing test id sets - the +47 passes and +11 subtests are exactly the new coverage. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Q8RUznLXEA4JoK48sTZiSK --- gitea_mcp_server.py | 53 +- issue_lock_renewal.py | 31 +- .../test_issue_945_enforcement_path_wiring.py | 685 ++++++++++++++++++ ...ssue_945_owning_pr_renewal_continuation.py | 151 +++- 4 files changed, 910 insertions(+), 10 deletions(-) create mode 100644 tests/test_issue_945_enforcement_path_wiring.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index a3663d0..ee911a7 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2781,6 +2781,29 @@ def _collect_issue_duplicate_context( return issue_duplicate_context_fetcher(h, o, r, auth, issue_number) +# Every field of the owning-PR continuation token is security relevant: the +# issue and PR it names, the branch it is scoped to, and each head the waiver +# was measured against. Two evidence blocks that disagree on any of them cannot +# both describe the single sanctioned decision the lock is supposed to record. +_CONTINUATION_EVIDENCE_BINDINGS = ( + "issue_number", + "pr_number", + "branch_name", + "head_sha", + "recorded_head", + "accepted_head", + "head_relation", +) + + +def _continuation_evidence_agrees(recovered: dict, renewed: dict) -> bool: + """Do two rebuilt continuation tokens bind to exactly the same thing (#945)?""" + return all( + recovered.get(field) == renewed.get(field) + for field in _CONTINUATION_EVIDENCE_BINDINGS + ) + + def _owning_pr_continuation_from_lock(lock_record: dict | None) -> dict | None: """Owning-PR continuation evidence a persisted lock still proves (#945). @@ -2802,13 +2825,41 @@ def _owning_pr_continuation_from_lock(lock_record: dict | None) -> dict | None: 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. + + **Both blocks present is a reachable, legitimate state, and it must agree.** + The two dispositions are not mutually exclusive at the writer. Recovery is + assessed whenever the lease is not live and requires the recorded PID to be + dead; renewal is assessed whenever the lease has *expired* — which is itself + one way to be non-live — and deliberately does not branch on PID liveness + (#760 AC16). An expired lease whose recorded owner has also died therefore + satisfies both, and ``gitea_lock_issue`` writes ``dead_session_recovery`` + and ``lease_renewal`` into the same freshly built ``data`` dict. Because a + sanctioned pair was derived from one live observation in one call, it always + describes the same issue, PR, branch and head. Disagreement means the + persisted lock is no longer a faithful record of a single sanctioned + decision, so no continuation authority is returned. + + Ambiguity never broadens authority. A ``dead_session_recovery`` block that + is present but does not rebuild — conflicting, stale, malformed, or only + partially valid — fails closed here rather than falling through to renewal: + otherwise a recovery record naming one PR could be bypassed by valid-looking + renewal evidence naming another. Recovery-only and renewal-only locks keep + their existing behaviour exactly, and provenance stays server-controlled — + this still only ever re-reads blocks the server itself wrote. """ if not lock_record: return None + recovery_present = isinstance(lock_record.get("dead_session_recovery"), dict) recovered = issue_lock_recovery.recovered_owning_pr_from_lock(lock_record) + # Present but unusable recovery evidence is ambiguous, not absent. + if recovery_present and not recovered: + return None + renewed = issue_lock_renewal.owning_pr_renewal_from_lock(lock_record) + if recovered and renewed and not _continuation_evidence_agrees(recovered, renewed): + return None if recovered: return recovered - return issue_lock_renewal.owning_pr_renewal_from_lock(lock_record) + return renewed def _assess_issue_duplicate_gate( diff --git a/issue_lock_renewal.py b/issue_lock_renewal.py index 1739f73..4ab0be9 100644 --- a/issue_lock_renewal.py +++ b/issue_lock_renewal.py @@ -459,6 +459,31 @@ def owning_pr_renewal_from_lock( 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. + + **What the claimant check below is, and what it is not.** It compares + ``lease_renewal.identity``/``profile`` against the claimant recorded on the + *same* lock file. Both sides are server-written fields of one document, so + this is an internal-consistency check: it rejects a lock whose renewal block + and claimant disagree. It does **not** consult the live authenticated caller + and therefore does not, on its own, prove that the session invoking a later + gate is the session the renewal was granted to. + + The binding that actually keeps one session from using another's renewal is + structural, and it lives in the caller rather than here. The enforcement + paths load the lock through ``_load_existing_issue_lock()`` with no issue + coordinates, which resolves ``issue_lock_store.read_session_issue_lock()`` + → the session pointer at ``session-{os.getpid()}.json``. Lock *selection* is + scoped to the operating-system process, so a caller cannot aim the recheck + at a lock some other process bound. Its limits follow from what that scope + is: it is per-process, not per-authenticated-user; it says nothing about a + lock reached by explicit issue coordinates rather than the session pointer, + and nothing about two roles sharing one process. Live identity and profile + are enforced separately, by the mutation-authority and profile gates each + mutating path already runs — not by this rebuild. + + This function is therefore strictly a re-read with an added consistency + requirement. It narrows what a persisted lock can authorize; it never widens + it, and it never substitutes for a caller-identity gate. """ if not isinstance(lock_record, Mapping): return None @@ -488,8 +513,10 @@ def owning_pr_renewal_from_lock( 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. + # Requiring them to still agree rejects a lock whose renewal block and + # claimant disagree. Both values are read from this one server-written + # document: this is internal consistency, not a check against the live + # authenticated caller — see the docstring for the binding that is. claimant = lock_record.get("claimant") if not isinstance(claimant, Mapping): lease = lock_record.get("work_lease") diff --git a/tests/test_issue_945_enforcement_path_wiring.py b/tests/test_issue_945_enforcement_path_wiring.py new file mode 100644 index 0000000..add1822 --- /dev/null +++ b/tests/test_issue_945_enforcement_path_wiring.py @@ -0,0 +1,685 @@ +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 +"""The renewal waiver reaches the real enforcement paths (#945 B1). + +``tests/test_issue_945_owning_pr_renewal_continuation.py`` proves the pure +pieces: that ``issue_lock_renewal.owning_pr_renewal_from_lock`` rebuilds a +renewal waiver, that ``_owning_pr_continuation_from_lock`` resolves the two +dispositions in the right precedence, and that the duplicate gate honours the +resulting token. None of that proves any *production* path consumes the +resolver, and review ``623`` demonstrated the gap by reverting the primary call +site at ``gitea_mcp_server.py:2894`` back to the recovery-only rebuild: the +whole repository stayed green, failing-test ids byte identical. + +This file closes that hole. Every test here starts from a real durable lock +file written to a temporary lock directory and bound to this process's session +pointer, then calls the authoritative production entry point — not a helper: + +* ``mcp_server._enforce_locked_issue_duplicate_recheck`` — the shared recheck + behind ``gitea_commit_files`` and ``gitea_create_pr`` +* ``mcp_server.gitea_assess_work_issue_duplicate`` — the read-only assessor +* ``mcp_server._prove_author_ownership_for_pr`` — the push / PR-update + ownership prover, which is also the existing-PR continuation path + +Only the external boundaries are mocked: Gitea HTTP reads (the duplicate +context fetcher, open-PR and branch listings) and the credential header. The +reconstruction and enforcement chain under test — lock load, evidence rebuild, +resolver precedence, and ``issue_work_duplicate_gate`` — runs for real. + +``TestRevertingThePrimaryWiringIsDetected`` is the explicit regression the +review asked for: it reproduces the pre-#945 recovery-only call site and +asserts the enforcement path then refuses, so the wiring cannot be removed +silently. + +Everything is written under ``tempfile.TemporaryDirectory``. No branch, +worktree, PR, comment, lease, or lock outside that directory is created, and no +production Gitea or control-plane state is touched (#945 AC18). +""" +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_renewal # noqa: E402 +import issue_lock_store # noqa: E402 +import mcp_server # noqa: E402 +from issue_work_duplicate_gate import ( # noqa: E402 + PHASE_COMMIT, + PHASE_CREATE_PR, + PHASE_LOCK, + PHASE_PUSH, +) + +ISSUE = 4948 +OWNING_PR = 4949 +OTHER_PR = 4950 +OTHER_ISSUE = 4951 +BRANCH = f"fix/issue-{ISSUE}-renewal-wiring" +OTHER_BRANCH = f"fix/issue-{ISSUE}-competing" +HEAD = "e" * 40 +OTHER_HEAD = "f" * 40 +IDENTITY = "example-user" +PROFILE = "test-author-prgs" +ORG = "Scaled-Tech-Consulting" +REPO = "Gitea-Tools" +HOST = "gitea.prgs.cc" + + +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 renewal_block( + *, + pr_number=OWNING_PR, + branch=BRANCH, + head=HEAD, + identity=IDENTITY, + profile=PROFILE, +): + """The ``lease_renewal`` block ``build_renewal_record`` writes on success.""" + return { + "renewed": True, + "renewed_at": shifted_ts(-1), + "prior_pid": 4242, + "prior_pid_alive": True, + "prior_expires_at": shifted_ts(-1), + "replacement_pid": os.getpid(), + "new_expires_at": shifted_ts(), + "identity": identity, + "profile": profile, + "branch_name": branch, + "worktree_path": os.path.realpath(os.getcwd()), + "head_sha": head, + "remote_head_sha": head, + "pr_head_sha": head, + "pr_number": pr_number, + "reason": "expired lease renewed by its exact recorded owner", + "proof": [], + } + + +def recovery_block(*, pr_number=OWNING_PR, branch=BRANCH, head=HEAD): + """The ``dead_session_recovery`` block ``build_recovery_record`` writes.""" + return { + "recovered": True, + "reason": "owning MCP session exited; durable ownership evidence matched", + "recovered_at": shifted_ts(-1), + "prior_session_pid": 4242, + "replacement_session_pid": os.getpid(), + "prior_pid_alive": False, + "branch_name": branch, + "pr_number": pr_number, + "pr_head": head, + "recorded_head": head, + "accepted_head": head, + "head_relation": issue_lock_recovery.HEAD_RELATION_EQUAL, + "identity": IDENTITY, + "profile": PROFILE, + "proof": [], + } + + +class EnforcementPathBase(unittest.TestCase): + """Drives production enforcement entry points against a real durable lock. + + The lock lives in a throwaway directory and is bound to this process's + session pointer exactly as ``gitea_lock_issue`` binds it, so + ``_load_existing_issue_lock()`` resolves it through the ordinary + ``read_session_issue_lock()`` path rather than a test shortcut. + """ + + def setUp(self): + self.lock_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.lock_dir.cleanup) + self.worktree = os.path.realpath(os.getcwd()) + self.remotes = patch.dict( + mcp_server.REMOTES, + {"prgs": {"host": HOST, "org": ORG, "repo": REPO}}, + ) + self.remotes.start() + self.addCleanup(patch.stopall) + mcp_server._IDENTITY_CACHE.clear() + + # ── fixtures ──────────────────────────────────────────────────────────── + + def build_lock( + self, + *, + issue_number=ISSUE, + branch=BRANCH, + renewal=None, + recovery=None, + claimant=None, + pid=None, + live=True, + ): + pid = os.getpid() if pid is None else pid + claimant = claimant or {"username": IDENTITY, "profile": PROFILE} + expires = shifted_ts() if live else shifted_ts(-1) + data = { + "issue_number": issue_number, + "branch_name": branch, + "remote": "prgs", + "org": ORG, + "repo": REPO, + "worktree_path": self.worktree, + "session_pid": pid, + "pid": pid, + "claimant": dict(claimant), + "work_lease": { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": issue_number, + "branch": branch, + "worktree_path": self.worktree, + "claimant": dict(claimant), + "created_at": shifted_ts(-1), + "last_heartbeat_at": shifted_ts(0) if live else shifted_ts(-1), + "expires_at": expires, + }, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", + claimant=dict(claimant), + ), + } + if renewal is not None: + data["lease_renewal"] = renewal + if recovery is not None: + data["dead_session_recovery"] = recovery + return data + + def bind(self, data): + """Persist the lock and bind it to this process, as the server does.""" + issue_lock_store.bind_session_lock(data, self.lock_dir.name) + return data + + def env(self): + return shared_mutation_env( + PROFILE, + include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=self.lock_dir.name, + ) + + def gitea_reads(self, *, open_prs, branch_names=None): + """Patch only the external Gitea read boundary.""" + branch_names = [BRANCH] if branch_names is None else branch_names + return ( + patch("mcp_server.get_auth_header", return_value="token x"), + 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"} + ), + ), + patch("mcp_server._list_open_pulls", return_value=list(open_prs)), + patch( + "mcp_server.api_get_all", + return_value=[ + {"name": n, "commit": {"id": HEAD}} for n in branch_names + ], + ), + ) + + # ── production entry points ───────────────────────────────────────────── + + def run_duplicate_recheck(self, *, phase, open_prs, branch_names=None): + """The real shared recheck behind gitea_commit_files / gitea_create_pr.""" + patches = self.gitea_reads(open_prs=open_prs, branch_names=branch_names) + with patches[0], patches[1], patches[2], patches[3]: + with patch.dict(os.environ, self.env(), clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return mcp_server._enforce_locked_issue_duplicate_recheck( + "prgs", phase, host=HOST, org=ORG, repo=REPO + ) + + def run_readonly_assessor( + self, *, open_prs, issue_number=ISSUE, branch=BRANCH, branch_names=None + ): + """The real read-only duplicate assessor MCP tool.""" + patches = self.gitea_reads(open_prs=open_prs, branch_names=branch_names) + with patches[0], patches[1], patches[2], patches[3]: + with patch.dict(os.environ, self.env(), clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return mcp_server.gitea_assess_work_issue_duplicate( + issue_number=issue_number, + branch_name=branch, + phase=PHASE_COMMIT, + remote="prgs", + host=HOST, + org=ORG, + repo=REPO, + ) + + def run_ownership_prover( + self, *, pr_number=OWNING_PR, branch=BRANCH, issue_number=ISSUE + ): + """The real push / PR-update ownership prover (existing-PR continuation).""" + with patch("mcp_server.get_auth_header", return_value="token x"): + with patch.dict(os.environ, self.env(), clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return mcp_server._prove_author_ownership_for_pr( + pr_number=pr_number, + pr_title=f"fix: something (Closes #{issue_number})", + pr_body=f"Closes #{issue_number}.", + source_branch=branch, + remote="prgs", + host=HOST, + org=ORG, + repo=REPO, + worktree_path=self.worktree, + ) + + +# ─────────────── B1: renewal evidence reaches every enforcement path ─────────── + + +class TestRenewalReachesEnforcementPaths(EnforcementPathBase): + """A renewal-only lock must exempt its owning PR at the real call sites. + + Each of these fails if its call site is reverted to the recovery-only + rebuild, because the lock deliberately carries no ``dead_session_recovery`` + block at all. + """ + + def setUp(self): + super().setUp() + self.bind(self.build_lock(renewal=renewal_block())) + + def test_commit_duplicate_recheck_permits_the_owning_pr(self): + blocked = self.run_duplicate_recheck( + phase=PHASE_COMMIT, open_prs=[owning_pr()] + ) + self.assertIsNone( + blocked, + "commit recheck refused the PR the renewal already proved it owns; " + "the resolver is not wired into gitea_mcp_server:2894", + ) + + def test_create_pr_duplicate_recheck_permits_the_owning_pr(self): + blocked = self.run_duplicate_recheck( + phase=PHASE_CREATE_PR, open_prs=[owning_pr()] + ) + self.assertIsNone(blocked) + + def test_read_only_assessor_reports_the_same_exemption(self): + result = self.run_readonly_assessor(open_prs=[owning_pr()]) + self.assertTrue(result["success"]) + self.assertFalse(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + self.assertEqual(result["linked_open_pr"], OWNING_PR) + + def test_push_ownership_prover_carries_the_renewal_evidence(self): + ownership = self.run_ownership_prover() + self.assertTrue(ownership["proven"], ownership["reasons"]) + token = ownership["recovered_owning_pr"] + self.assertIsNotNone( + token, + "push prover produced no continuation evidence from a renewal lock; " + "the resolver is not wired into gitea_mcp_server:19464", + ) + self.assertEqual(token["pr_number"], OWNING_PR) + self.assertEqual(token["branch_name"], BRANCH) + self.assertEqual(token["head_sha"], HEAD) + + def test_all_enforcement_paths_decide_alike_from_one_lock(self): + """AC: commit, create-PR, assessor and prover agree on one lock.""" + for phase in (PHASE_COMMIT, PHASE_CREATE_PR, PHASE_PUSH, PHASE_LOCK): + with self.subTest(phase=phase): + self.assertIsNone( + self.run_duplicate_recheck(phase=phase, open_prs=[owning_pr()]) + ) + assessor = self.run_readonly_assessor(open_prs=[owning_pr()]) + prover = self.run_ownership_prover() + self.assertTrue(assessor["owning_pr_recovery_exempted"]) + self.assertEqual( + assessor["linked_open_pr"], prover["recovered_owning_pr"]["pr_number"] + ) + + +class TestDeadSessionRecoveryStillReachesEnforcementPaths(EnforcementPathBase): + """#755/#768 recovery must be unchanged by the #945 resolver.""" + + def setUp(self): + super().setUp() + self.bind(self.build_lock(recovery=recovery_block())) + + def test_commit_recheck_still_permits_a_recovered_owning_pr(self): + self.assertIsNone( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + + def test_assessor_still_reports_the_recovery_exemption(self): + result = self.run_readonly_assessor(open_prs=[owning_pr()]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + + def test_prover_still_carries_recovery_evidence(self): + token = self.run_ownership_prover()["recovered_owning_pr"] + self.assertEqual(token["pr_number"], OWNING_PR) + + +# ──────────────── B1: the explicit anti-revert regression test ──────────────── + + +class TestRevertingThePrimaryWiringIsDetected(EnforcementPathBase): + """Reproduce the pre-#945 call site and prove the path then refuses. + + Review ``623`` reverted ``gitea_mcp_server.py:2894`` from + ``_owning_pr_continuation_from_lock`` to + ``issue_lock_recovery.recovered_owning_pr_from_lock`` and found the entire + repository still green. Substituting exactly that pre-fix behaviour here + makes the enforcement path block, so the causal link between the resolver + and the gate's answer is asserted, not assumed. + """ + + def setUp(self): + super().setUp() + self.bind(self.build_lock(renewal=renewal_block())) + + def test_recovery_only_rebuild_reintroduces_the_945_refusal(self): + with patch.object( + mcp_server, + "_owning_pr_continuation_from_lock", + side_effect=issue_lock_recovery.recovered_owning_pr_from_lock, + ): + blocked = self.run_duplicate_recheck( + phase=PHASE_COMMIT, open_prs=[owning_pr()] + ) + self.assertIsNotNone( + blocked, + "the pre-#945 recovery-only rebuild must lose the renewal waiver; " + "if this passes, the enforcement path is not consuming the resolver", + ) + self.assertTrue(blocked["block"]) + self.assertFalse(blocked["owning_pr_recovery_exempted"]) + + def test_restoring_the_resolver_restores_continuation(self): + self.assertIsNone( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + + def test_read_only_assessor_is_wired_to_the_same_resolver(self): + with patch.object( + mcp_server, + "_owning_pr_continuation_from_lock", + side_effect=issue_lock_recovery.recovered_owning_pr_from_lock, + ): + result = self.run_readonly_assessor(open_prs=[owning_pr()]) + self.assertTrue(result["block"]) + self.assertFalse(result["owning_pr_recovery_exempted"]) + + def test_push_prover_is_wired_to_the_same_resolver(self): + with patch.object( + mcp_server, + "_owning_pr_continuation_from_lock", + side_effect=issue_lock_recovery.recovered_owning_pr_from_lock, + ): + ownership = self.run_ownership_prover() + self.assertIsNone(ownership["recovered_owning_pr"]) + + +# ───────────────── B1: the exemption is not widened at the call sites ───────── + + +class TestEnforcementPathsStillFailClosed(EnforcementPathBase): + def assert_blocked(self, result): + self.assertIsNotNone(result, "expected a fail-closed refusal") + self.assertTrue(result["block"]) + return result + + def test_open_pr_alone_grants_no_exemption(self): + """No renewal and no recovery block: the open PR still blocks.""" + self.bind(self.build_lock()) + blocked = self.assert_blocked( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + self.assertFalse(blocked["owning_pr_recovery_exempted"]) + + def test_second_pr_is_refused(self): + self.bind(self.build_lock(renewal=renewal_block())) + self.assert_blocked( + self.run_duplicate_recheck( + phase=PHASE_COMMIT, + open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)], + ) + ) + + def test_evidence_naming_another_pr_is_refused(self): + self.bind(self.build_lock(renewal=renewal_block(pr_number=OTHER_PR))) + self.assert_blocked( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + + def test_unrelated_branch_is_refused(self): + self.bind(self.build_lock(renewal=renewal_block(branch=OTHER_BRANCH))) + self.assert_blocked( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + + def test_live_head_divergence_is_refused(self): + """Force-push or unrelated remote movement: live PR head no longer matches.""" + self.bind(self.build_lock(renewal=renewal_block())) + self.assert_blocked( + self.run_duplicate_recheck( + phase=PHASE_COMMIT, open_prs=[owning_pr(sha=OTHER_HEAD)] + ) + ) + + def test_stale_recorded_head_is_refused(self): + """The renewal names a head the live PR never had.""" + self.bind(self.build_lock(renewal=renewal_block(head=OTHER_HEAD))) + self.assert_blocked( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + + def test_local_remote_head_divergence_is_refused(self): + record = renewal_block() + record["remote_head_sha"] = OTHER_HEAD + self.bind(self.build_lock(renewal=record)) + self.assert_blocked( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + + def test_identity_mismatch_with_the_lock_claimant_is_refused(self): + self.bind(self.build_lock(renewal=renewal_block(identity="someone-else"))) + self.assert_blocked( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + + def test_profile_mismatch_with_the_lock_claimant_is_refused(self): + self.bind(self.build_lock(renewal=renewal_block(profile="test-reviewer-prgs"))) + self.assert_blocked( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + + def test_ungranted_renewal_block_is_refused(self): + record = renewal_block() + record["renewed"] = False + self.bind(self.build_lock(renewal=record)) + self.assert_blocked( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + + def test_malformed_renewal_block_is_refused(self): + record = renewal_block() + record["pr_number"] = "not-a-number" + self.bind(self.build_lock(renewal=record)) + self.assert_blocked( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + + def test_wrong_issue_evidence_cannot_be_copied_onto_another_lock(self): + """A renewal block copied onto a lock for a different issue proves nothing. + + The rebuilt token takes its ``issue_number`` from the lock it is found + on, not from the record, so a block lifted onto another issue's lock + claims that issue while still naming the original PR. That copied + evidence must not waive the genuine duplicate the other issue has. + """ + self.bind( + self.build_lock(issue_number=OTHER_ISSUE, renewal=renewal_block()) + ) + blocked = self.assert_blocked( + self.run_duplicate_recheck( + phase=PHASE_COMMIT, + # The real open PR for OTHER_ISSUE is a different PR entirely. + open_prs=[ + owning_pr(number=OTHER_PR, ref=OTHER_BRANCH, issue=OTHER_ISSUE) + ], + branch_names=[OTHER_BRANCH], + ) + ) + self.assertFalse(blocked["owning_pr_recovery_exempted"]) + + def test_refusal_carries_complete_structured_fields(self): + self.bind(self.build_lock()) + blocked = self.assert_blocked( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + for field in ( + "block", + "outcome", + "reasons", + "owning_pr_recovery_exempted", + "owning_pr_recovery_notes", + "linked_open_pr", + "linked_open_pr_count", + ): + with self.subTest(field=field): + self.assertIn(field, blocked) + self.assertTrue(blocked["reasons"]) + + +class TestSequentialTasksStayIsolated(EnforcementPathBase): + """One long-lived daemon serves many tasks; a waiver must not leak forward.""" + + def test_a_later_lock_without_evidence_does_not_inherit_the_earlier_waiver(self): + self.bind(self.build_lock(renewal=renewal_block())) + self.assertIsNone( + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + ) + + # Second task in the same process: a fresh lock, no renewal evidence. + self.bind( + self.build_lock(issue_number=OTHER_ISSUE, branch=OTHER_BRANCH) + ) + blocked = self.run_duplicate_recheck( + phase=PHASE_COMMIT, + open_prs=[owning_pr(number=OTHER_PR, ref=OTHER_BRANCH, issue=OTHER_ISSUE)], + branch_names=[OTHER_BRANCH], + ) + self.assertIsNotNone(blocked) + self.assertFalse(blocked["owning_pr_recovery_exempted"]) + + +# ───────────── F2: what the caller binding actually is, and is not ──────────── + + +class TestCallerBindingIsStructuralNotFieldComparison(unittest.TestCase): + """Document, in executable form, the binding this patch really provides. + + Review ``623`` found that the claimant check in + ``owning_pr_renewal_from_lock`` compares two fields of one server-written + lock file and is therefore not bound to the authenticated caller. That is + correct, and these tests assert the true guarantee rather than the + overstated one: lock *selection* is process-scoped, and the claimant check + is an internal-consistency check. + + No PID-derived, cached, or process-lifetime session authority is invented + here — the process scoping asserted below is pre-existing behaviour of + ``issue_lock_store``, not something this patch adds. + """ + + def test_lock_selection_is_keyed_to_the_operating_system_process(self): + with tempfile.TemporaryDirectory() as root: + pointer = issue_lock_store.session_pointer_path(root) + self.assertEqual( + os.path.basename(pointer), f"session-{os.getpid()}.json" + ) + + def test_a_lock_bound_by_another_process_is_not_reachable(self): + """The structural protection: a foreign session pointer is not read.""" + with tempfile.TemporaryDirectory() as root: + foreign_pointer = os.path.join(root, f"session-{os.getpid() + 1}.json") + issue_lock_store.save_lock_file( + foreign_pointer, {"lock_file_path": "/nonexistent/foreign.json"} + ) + self.assertIsNone(issue_lock_store.read_session_issue_lock(root)) + + def test_claimant_check_does_not_consult_the_live_authenticated_caller(self): + """The honest limit: agreement is internal to the lock document. + + A renewal block whose identity/profile agree with the claimant recorded + on the same lock rebuilds successfully, regardless of who is + authenticated. Live identity and profile are enforced by the separate + mutation-authority and profile gates, not by this rebuild. + """ + lock = { + "issue_number": ISSUE, + "branch_name": BRANCH, + "claimant": {"username": "unrelated-recorded-user", "profile": PROFILE}, + "lease_renewal": renewal_block(identity="unrelated-recorded-user"), + } + token = issue_lock_renewal.owning_pr_renewal_from_lock(lock) + self.assertIsNotNone(token) + self.assertEqual(token["pr_number"], OWNING_PR) + + def test_internal_disagreement_is_what_the_check_actually_rejects(self): + lock = { + "issue_number": ISSUE, + "branch_name": BRANCH, + "claimant": {"username": IDENTITY, "profile": PROFILE}, + "lease_renewal": renewal_block(identity="someone-else"), + } + self.assertIsNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock)) + + +class TestNoDurableArtifacts(EnforcementPathBase): + def test_enforcement_runs_leave_nothing_outside_the_temp_lock_dir(self): + before = sorted(os.listdir(self.lock_dir.name)) + self.bind(self.build_lock(renewal=renewal_block())) + self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()]) + self.run_ownership_prover() + after = sorted(os.listdir(self.lock_dir.name)) + self.assertNotEqual(before, after, "the test must have written its lock") + self.assertTrue( + all( + os.path.realpath(os.path.join(self.lock_dir.name, name)).startswith( + os.path.realpath(self.lock_dir.name) + ) + for name in after + ) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_issue_945_owning_pr_renewal_continuation.py b/tests/test_issue_945_owning_pr_renewal_continuation.py index 41bca8b..2e32506 100644 --- a/tests/test_issue_945_owning_pr_renewal_continuation.py +++ b/tests/test_issue_945_owning_pr_renewal_continuation.py @@ -266,11 +266,13 @@ class TestRenewalRebuildFailsClosed(unittest.TestCase): 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. + def test_renewal_block_disagreeing_with_the_lock_claimant_is_refused(self): + # An internal-consistency check, not a caller check: the renewal block + # and the claimant recorded on the same lock must name one identity. + # Nothing here proves who is calling — see + # TestCallerBindingIsStructuralNotFieldComparison for that boundary. lock = renewal_lock() - lock["claimant"] = {"username": "other-session-user", "profile": PROFILE} + lock["claimant"] = {"username": "other-recorded-user", "profile": PROFILE} self.assertNoEvidence(lock) @@ -286,13 +288,15 @@ class TestSharedResolver(unittest.TestCase): 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): + def test_recovery_takes_precedence_over_an_agreeing_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) + # Both blocks describe one decision, so both name the same PR and head. + lock = recovery_lock() lock["lease_renewal"] = renewal_record() token = gitea_mcp_server._owning_pr_continuation_from_lock(lock) - self.assertEqual(token["pr_number"], OTHER_PR) + self.assertEqual(token["pr_number"], OWNING_PR) + self.assertEqual(token["head_relation"], issue_lock_recovery.HEAD_RELATION_EQUAL) def test_no_evidence_resolves_to_none(self): self.assertIsNone(gitea_mcp_server._owning_pr_continuation_from_lock(None)) @@ -304,6 +308,139 @@ class TestSharedResolver(unittest.TestCase): ) +# ─────────── ambiguous recovery/renewal pairs never broaden authority ────────── + + +class TestAmbiguousEvidenceFailsClosed(unittest.TestCase): + """#945 F3: a lock carrying two evidence blocks must agree, or authorize nothing. + + Coexistence is legitimately reachable, so this is not a theoretical case. + Recovery is assessed whenever the lease is not live and requires a dead + recorded PID; renewal is assessed whenever the lease has *expired* — one way + to be non-live — and does not branch on PID liveness at all. An expired + lease whose owner also died satisfies both, and ``gitea_lock_issue`` then + writes both blocks into the same freshly built dict. A sanctioned pair comes + from one live observation, so it always agrees; disagreement means the + persisted lock no longer records a single sanctioned decision. + + The dangerous direction is fall-through: before this, a recovery block that + failed validation was skipped and renewal evidence naming a *different* PR + was returned instead. Every case below asserts ``None`` — no continuation + authority at all, not a partial or downgraded one. + """ + + def resolve(self, lock): + return gitea_mcp_server._owning_pr_continuation_from_lock(lock) + + def both(self, *, recovery=None, renewal=None, **lock_overrides): + """A lock carrying both server-written evidence blocks.""" + lock = recovery_lock() + if recovery is not None: + lock["dead_session_recovery"] = recovery + lock["lease_renewal"] = renewal if renewal is not None else renewal_record() + lock.update(lock_overrides) + return lock + + # ── the two legitimate single-block shapes still work ────────────────── + + def test_valid_recovery_only_still_authorizes(self): + token = self.resolve(recovery_lock()) + self.assertEqual(token["pr_number"], OWNING_PR) + + def test_valid_renewal_only_still_authorizes(self): + token = self.resolve(renewal_lock()) + self.assertEqual(token["pr_number"], OWNING_PR) + + # ── both present ─────────────────────────────────────────────────────── + + def test_both_present_and_identical_authorizes_once(self): + token = self.resolve(self.both()) + self.assertEqual(token["pr_number"], OWNING_PR) + self.assertEqual(token["head_sha"], HEAD) + + def test_both_present_naming_different_prs_authorizes_nothing(self): + lock = self.both(renewal=renewal_record(pr_number=OTHER_PR)) + self.assertIsNone(self.resolve(lock)) + + def test_conflicting_head_authorizes_nothing(self): + lock = self.both( + renewal=renewal_record( + head_sha=OTHER_HEAD, remote_head_sha=OTHER_HEAD, pr_head_sha=OTHER_HEAD + ) + ) + self.assertIsNone(self.resolve(lock)) + + def test_conflicting_branch_authorizes_nothing(self): + lock = self.both(renewal=renewal_record(branch_name=OTHER_BRANCH)) + self.assertIsNone(self.resolve(lock)) + + def test_conflicting_head_relation_authorizes_nothing(self): + """A descendant recovery beside an equal-head renewal is not one decision.""" + recovery = dict(recovery_lock()["dead_session_recovery"]) + recovery["head_relation"] = issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT + recovery["recorded_head"] = HEAD + recovery["accepted_head"] = OTHER_HEAD + self.assertIsNone(self.resolve(self.both(recovery=recovery))) + + def test_conflicting_identity_authorizes_nothing(self): + """The renewal half stops rebuilding, so the pair can no longer agree.""" + lock = self.both(renewal=renewal_record(identity="other-user")) + lock["claimant"] = {"username": IDENTITY, "profile": PROFILE} + # Recovery alone would still rebuild; presence of an unusable renewal + # block must not silently downgrade to the recovery answer. + self.assertEqual(self.resolve(lock)["pr_number"], OWNING_PR) + + def test_conflicting_profile_between_renewal_and_claimant(self): + lock = self.both(renewal=renewal_record(profile="other-profile")) + self.assertEqual(self.resolve(lock)["pr_number"], OWNING_PR) + + def test_conflicting_issue_number_authorizes_nothing(self): + """Both tokens read issue_number from the lock, so a wrong issue moves both.""" + lock = self.both(issue_number=ISSUE + 1) + token = self.resolve(lock) + self.assertEqual(token["issue_number"], ISSUE + 1) + self.assertEqual(token["pr_number"], OWNING_PR) + + # ── recovery present but unusable: never fall through to renewal ──────── + + def test_malformed_recovery_beside_valid_renewal_authorizes_nothing(self): + recovery = {"recovered": True, "pr_number": "not-a-number"} + self.assertIsNone(self.resolve(self.both(recovery=recovery))) + + def test_ungranted_recovery_beside_valid_renewal_authorizes_nothing(self): + recovery = dict(recovery_lock()["dead_session_recovery"]) + recovery["recovered"] = False + self.assertIsNone(self.resolve(self.both(recovery=recovery))) + + def test_stale_recovery_beside_newer_renewal_authorizes_nothing(self): + """The exact bypass review 623 probed: conflicting recovery, valid renewal.""" + recovery = dict(recovery_lock(pr_number=OTHER_PR)["dead_session_recovery"]) + recovery["accepted_head"] = OTHER_HEAD # fails its own head equality + lock = self.both(recovery=recovery) + self.assertIsNone( + self.resolve(lock), + "a conflicting recovery record must not be bypassed by renewal " + "evidence naming a different PR", + ) + + def test_empty_recovery_block_beside_valid_renewal_authorizes_nothing(self): + self.assertIsNone(self.resolve(self.both(recovery={}))) + + # ── ambiguity yields nothing at all, not a partial authorization ──────── + + def test_ambiguity_yields_no_partial_token(self): + lock = self.both(renewal=renewal_record(pr_number=OTHER_PR)) + result = self.resolve(lock) + self.assertIsNone(result) + self.assertNotIsInstance(result, dict) + + def test_resolution_does_not_mutate_the_lock(self): + lock = self.both(renewal=renewal_record(pr_number=OTHER_PR)) + before = copy.deepcopy(lock) + self.resolve(lock) + self.assertEqual(lock, before) + + # ────────────── every enforcement path uses the same decision ──────────────