Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 101 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

Merge pull request 'fix(gate): preserve exact-owner renewal evidence across duplicate rechecks (Closes #945)' (#946) from fix/issue-945-owning-pr-renewal-evidence into master

Browse Source
This commit was merged in pull request #946.
This commit is contained in:
sysadmin
2026-07-27 18:27:35 -05:00
parent ed9414ebda b1fcf15937
commit 35ed8a2fcb
4 changed files with 1493 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_issue_945_enforcement_path_wiring.py
+685
View File
@@ -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()
tests/test_issue_945_owning_pr_renewal_continuation.py
+602
View File
@@ -0,0 +1,602 @@
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_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-recorded-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_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.
# 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"], 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))
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}
)
)
# ─────────── 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 ──────────────
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()