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) <[email protected]>
Claude-Session: https://claude.ai/code/session_01Q8RUznLXEA4JoK48sTZiSK
This commit is contained in:
2026-07-27 18:22:07 -04:00
co-authored by Claude Opus 5
parent 79334d4840
commit b1fcf15937
4 changed files with 910 additions and 10 deletions
@@ -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 ──────────────