"""MCP-level exact-owner lease renewal through ``gitea_lock_issue`` (#760). The unit suite in ``test_issue_760_exact_owner_lease_renewal`` proves the renewal *disposition*. It cannot prove the disposition survives the rest of the tool, and it did not: the waiver was computed and then discarded before ``assess_issue_lock_worktree``, so every real renewal still failed on base-equivalence. A branch being renewed always carries committed work, so it is never base-equivalent by construction — exactly the argument #753 already makes for recovery. These tests drive the public tool end to end against a real git repository and a real durable lock file, composing every gate in the production order. """ from __future__ import annotations import os import subprocess import sys import tempfile import unittest from datetime import datetime, timedelta, timezone from unittest.mock import patch sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from mutation_profile_fixture import shared_mutation_env # noqa: E402 import issue_lock_provenance # noqa: E402 import issue_lock_store # noqa: E402 import mcp_server # noqa: E402 ISSUE = 9760 BRANCH = f"fix/issue-{ISSUE}-renewal-mcp" IDENTITY = "example-user" PROFILE = "test-author-prgs" ORG = "Scaled-Tech-Consulting" REPO = "Gitea-Tools" def _past_ts(hours: int = 1) -> str: return ( (datetime.now(timezone.utc) - timedelta(hours=hours)) .isoformat() .replace("+00:00", "Z") ) class _RenewalMcpBase(unittest.TestCase): """Real git repo + durable expired lock owned by a live PID. The recorded PID is ``os.getpid()`` — unambiguously alive. That is the whole point of #760: the PID belongs to the long-lived MCP daemon, so its liveness says nothing about whether the authoring task still holds the work. """ def setUp(self): self.lock_dir = tempfile.TemporaryDirectory() self.addCleanup(self.lock_dir.cleanup) self.repo = tempfile.mkdtemp(prefix="issue760-mcp-") self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.repo], check=False)) self._init_worktree() self.remotes = patch.dict( mcp_server.REMOTES, {"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}}, ) self.remotes.start() self.addCleanup(patch.stopall) mcp_server._IDENTITY_CACHE.clear() def _git(self, *args): return subprocess.run( ["git", "-C", self.repo, *args], capture_output=True, text=True, check=True, ) def _init_worktree(self): self._git("init", "-q", "-b", "master") self._git("config", "user.email", "test@example.com") self._git("config", "user.name", "Test") with open(os.path.join(self.repo, "seed.txt"), "w") as fh: fh.write("seed\n") self._git("add", "seed.txt") self._git("commit", "-q", "-m", "seed") self.base_sha = self._git("rev-parse", "HEAD").stdout.strip() # The branch carries committed work, so it is NOT base-equivalent. self._git("checkout", "-q", "-b", BRANCH) with open(os.path.join(self.repo, "work.txt"), "w") as fh: fh.write("author work\n") self._git("add", "work.txt") self._git("commit", "-q", "-m", "author work") self.head_sha = self._git("rev-parse", "HEAD").stdout.strip() self.worktree = os.path.realpath(self.repo) def write_expired_lock(self, **overrides): path = issue_lock_store.lock_file_path( remote="prgs", org=ORG, repo=REPO, issue_number=ISSUE, lock_dir=self.lock_dir.name, ) claimant = {"username": IDENTITY, "profile": PROFILE} pid = overrides.pop("session_pid", os.getpid()) overrides.pop("pid", None) lease_overrides = overrides.pop("work_lease", {}) data = { "issue_number": ISSUE, "branch_name": BRANCH, "remote": "prgs", "org": ORG, "repo": REPO, "worktree_path": self.worktree, "session_pid": pid, "pid": pid, "lock_generation": 3, "work_lease": { "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, "issue_number": ISSUE, "pr_number": None, "branch": BRANCH, "worktree_path": self.worktree, "claimant": claimant, "created_at": _past_ts(5), "last_heartbeat_at": _past_ts(5), "expires_at": _past_ts(), # already expired }, "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( tool="gitea_lock_issue", claimant=claimant, ), } data["work_lease"].update(lease_overrides) data.update(overrides) data["session_pid"] = pid data["pid"] = pid data["lock_file_path"] = path issue_lock_store.save_lock_file(path, data) return path def _tool_env(self): env = shared_mutation_env( PROFILE, include_example_repo=True, GITEA_ISSUE_LOCK_DIR=self.lock_dir.name, ) env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name return env def _git_state(self, *, porcelain="", branch=BRANCH, head=None): return { "current_branch": branch, "porcelain_status": porcelain, # The decisive fact: a branch carrying work is never base-equivalent. "base_equivalent": False, "head_sha": head or self.head_sha, "inspected_git_root": self.worktree, "base_branch": "master", } def run_lock_issue( self, *, branch_entries=None, open_prs=None, git_state=None, identity=IDENTITY, profile=PROFILE, ): """Drive the public tool for the published exact-owner renewal shape.""" if branch_entries is None: branch_entries = [{"name": BRANCH, "commit": {"id": self.head_sha}}] if open_prs is None: open_prs = [{"number": 4242, "head": {"ref": BRANCH, "sha": self.head_sha}}] if git_state is None: git_state = self._git_state() env = self._tool_env() with patch( "mcp_server.api_get_all", return_value=list(branch_entries) ), patch( "mcp_server._list_open_pulls", return_value=list(open_prs) ), patch( "mcp_server.get_auth_header", return_value="token x" ), patch( "mcp_server._work_lease_claimant", return_value={"username": identity, "profile": profile}, ), patch( "mcp_server.issue_lock_worktree.read_worktree_git_state", return_value=git_state, ), patch( "mcp_server.issue_duplicate_context_fetcher", side_effect=lambda h, o, r, auth, issue_number: ( list(open_prs), [b.get("name") for b in branch_entries if isinstance(b, dict)], {"status": "not_claimed"}, ), ), patch.dict(os.environ, env, clear=True): os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name return mcp_server.gitea_lock_issue( issue_number=ISSUE, branch_name=BRANCH, remote="prgs", worktree_path=self.worktree, ) class TestRenewalReachableThroughTool(_RenewalMcpBase): """F1: the sanctioned renewal must survive every downstream gate.""" def test_expired_lease_live_pid_exact_owner_renews_through_the_tool(self): prior = issue_lock_store.read_lock_file(self.write_expired_lock()) self.assertTrue(issue_lock_store.is_lease_expired(prior)) self.assertTrue(issue_lock_store.is_process_alive(prior["session_pid"])) result = self.run_lock_issue() self.assertTrue(result["success"], result) self.assertEqual(result["issue_number"], ISSUE) self.assertEqual(result["branch_name"], BRANCH) # The renewal is reported natively, so no lock-file inspection is needed. self.assertIn("lease_renewal", result) self.assertTrue(result["lease_renewal"]["renewed"]) self.assertIn("Renewed the expired", result["message"]) def test_renewed_lock_records_prior_and_replacement_evidence(self): prior = issue_lock_store.read_lock_file(self.write_expired_lock()) prior_expiry = prior["work_lease"]["expires_at"] prior_generation = issue_lock_store.lock_generation(prior) result = self.run_lock_issue() written = issue_lock_store.read_lock_file(result["lock_file_path"]) renewal = written["lease_renewal"] self.assertTrue(renewal["renewed"]) self.assertEqual(renewal["prior_pid"], prior["session_pid"]) self.assertTrue(renewal["prior_pid_alive"]) self.assertEqual(renewal["prior_expires_at"], prior_expiry) self.assertEqual(renewal["identity"], IDENTITY) self.assertEqual(renewal["profile"], PROFILE) self.assertEqual(renewal["head_sha"], self.head_sha) self.assertTrue(renewal["proof"]) # New expiry is a fresh absolute stamp, later than the one it replaced. self.assertEqual(renewal["new_expires_at"], written["work_lease"]["expires_at"]) self.assertGreater(renewal["new_expires_at"], prior_expiry) # Compare-and-swap advanced the generation exactly once. self.assertEqual( issue_lock_store.lock_generation(written), prior_generation + 1 ) def test_renewed_lock_is_live_and_satisfies_mutation_ownership(self): self.write_expired_lock() result = self.run_lock_issue() written = issue_lock_store.read_lock_file(result["lock_file_path"]) self.assertTrue(issue_lock_store.assess_lock_freshness(written)["live"]) verdict = issue_lock_store.verify_lock_for_mutation( written, issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.worktree, ) self.assertTrue(verdict["proven"], verdict) self.assertFalse(verdict["block"]) def test_recovery_record_is_not_written_for_a_live_owner_renewal(self): """#753 recovery must not be claimed when the recorded PID is alive.""" self.write_expired_lock() result = self.run_lock_issue() written = issue_lock_store.read_lock_file(result["lock_file_path"]) self.assertNotIn("dead_session_recovery", written) class TestRenewalWaiverIsNarrow(_RenewalMcpBase): """The waiver relaxes base-equivalence and nothing else.""" def test_dirty_worktree_still_blocks_a_would_be_renewal(self): """Cleanliness is never waived; the renewal assessor refuses first. A dirty worktree makes the renewal refuse, so no waiver is issued and the lease-conflict gate fails closed ahead of the worktree gate. The refusal names the uncommitted files, so the owner still learns why. """ self.write_expired_lock() with self.assertRaises(Exception) as ctx: self.run_lock_issue( git_state=self._git_state(porcelain=" M gitea_mcp_server.py\n") ) message = str(ctx.exception) self.assertIn("Recovery review is required before takeover", message) self.assertIn("worktree has uncommitted tracked changes", message) self.assertIn("gitea_mcp_server.py", message) def test_foreign_claimant_cannot_use_the_waiver(self): """A near-match owner gets no renewal and no base-equivalence waiver.""" self.write_expired_lock() with self.assertRaises(Exception) as ctx: self.run_lock_issue(identity="someone-else") message = str(ctx.exception) self.assertIn("Recovery review is required before takeover", message) # The refusal names the missing ownership evidence (#760 diagnostics). self.assertIn("does not match active identity", message) def test_foreign_profile_cannot_use_the_waiver(self): self.write_expired_lock() with self.assertRaises(Exception) as ctx: self.run_lock_issue(profile="other-author") self.assertIn( "Recovery review is required before takeover", str(ctx.exception) ) def test_unpublished_branch_cannot_use_the_waiver(self): """No remote head to agree with, so exact-owner renewal is refused.""" self.write_expired_lock() with self.assertRaises(Exception) as ctx: self.run_lock_issue(branch_entries=[], open_prs=[]) self.assertIn( "Recovery review is required before takeover", str(ctx.exception) ) def test_pr_head_mismatch_cannot_use_the_waiver(self): self.write_expired_lock() other = "9" * 40 with self.assertRaises(Exception) as ctx: self.run_lock_issue( open_prs=[{"number": 4242, "head": {"ref": BRANCH, "sha": other}}] ) self.assertIn( "Recovery review is required before takeover", str(ctx.exception) ) def test_non_base_equivalent_branch_still_blocks_without_any_waiver(self): """No durable lock at all: the ordinary base-equivalence rule applies.""" with self.assertRaises(Exception) as ctx: self.run_lock_issue() self.assertIn("must be base-equivalent", str(ctx.exception)) if __name__ == "__main__": unittest.main()