From c31df2130c742ef73e2ce156cc4a7181a82e655c Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 20 Jul 2026 16:13:53 -0400 Subject: [PATCH] test(mcp): AC9 assessor/mutator parity + AC6 unpublished recovery MCP regressions (PR #774 F1/F2) Local worktree commit only (publication blocked by dangling GITEA_AUTHOR_WORKTREE). Tests only; no production code change. Co-Authored-By: Grok 4.5 --- ...st_issue_772_unpublished_claim_recovery.py | 559 ++++++++++++++++++ 1 file changed, 559 insertions(+) diff --git a/tests/test_issue_772_unpublished_claim_recovery.py b/tests/test_issue_772_unpublished_claim_recovery.py index 027c271..e7a3c89 100644 --- a/tests/test_issue_772_unpublished_claim_recovery.py +++ b/tests/test_issue_772_unpublished_claim_recovery.py @@ -441,5 +441,564 @@ class LockGenerationCas(unittest.TestCase): issue_lock_store.bind_session_lock(self.record(), self.dir) +# ──────────────────── F1 AC9 + F2 AC6 MCP regressions (review #487) ─────────── +# +# Reviewer findings on PR #774: the unit suite covered the pure assessor and the +# store CAS, but not (F1) the shared-evaluator projection consumed by the +# diagnostic tool nor (F2) the public gitea_lock_issue entry point for the +# unpublished shape — including the recovery_sanctioned → expected_generation +# wiring that arms AC5. Both gaps are closed below. + + +from datetime import datetime, timedelta, timezone # noqa: E402 +from pathlib import Path # noqa: E402 +from unittest.mock import patch # noqa: E402 + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 + +import issue_lock_provenance # noqa: E402 +import mcp_server # noqa: E402 + +MCP_ISSUE = 9901 +MCP_BRANCH = f"fix/issue-{MCP_ISSUE}-unpublished-mcp" +MCP_IDENTITY = "example-user" +MCP_PROFILE = "test-author-prgs" +MCP_ORG = "Scaled-Tech-Consulting" +MCP_REPO = "Gitea-Tools" + + +def _dead_pid() -> int: + proc = subprocess.Popen([sys.executable, "-c", "pass"]) + proc.wait() + return proc.pid + + +def _future_ts(hours: int = 4) -> str: + return ( + (datetime.now(timezone.utc) + timedelta(hours=hours)) + .isoformat() + .replace("+00:00", "Z") + ) + + +def _past_ts(hours: int = 1) -> str: + return ( + (datetime.now(timezone.utc) - timedelta(hours=hours)) + .isoformat() + .replace("+00:00", "Z") + ) + + +class _UnpublishedMcpBase(unittest.TestCase): + """Shared harness: real git worktree + durable dead lock + MCP tool stubs. + + Gitea inventory is stubbed empty for the issue branch so the production + path selects ``unpublished_claim``. Git ancestry is observed for real + against a temporary repository. + """ + + def setUp(self): + self.lock_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.lock_dir.cleanup) + self.repo = tempfile.mkdtemp(prefix="issue772-mcp-") + self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.repo], check=False)) + self._init_unpublished_worktree() + self.remotes = patch.dict( + mcp_server.REMOTES, + { + "prgs": { + "host": "gitea.prgs.cc", + "org": MCP_ORG, + "repo": MCP_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_unpublished_worktree(self): + self._git("init", "-q", "-b", "master") + self._git("config", "user.email", "test@example.com") + self._git("config", "user.name", "Test") + seed = os.path.join(self.repo, "seed.txt") + with open(seed, "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() + self._git("checkout", "-q", "-b", MCP_BRANCH) + work = os.path.join(self.repo, "work.txt") + with open(work, "w") as fh: + fh.write("unpublished work\n") + self._git("add", "work.txt") + self._git("commit", "-q", "-m", "unpublished claim") + self.head_sha = self._git("rev-parse", "HEAD").stdout.strip() + self.worktree = os.path.realpath(self.repo) + + def write_dead_lock(self, **overrides): + path = issue_lock_store.lock_file_path( + remote="prgs", + org=MCP_ORG, + repo=MCP_REPO, + issue_number=MCP_ISSUE, + lock_dir=self.lock_dir.name, + ) + claimant = {"username": MCP_IDENTITY, "profile": MCP_PROFILE} + # Prefer caller-supplied dead/live pid so live-owner tests can pass one. + pid = overrides.pop("session_pid", None) + if pid is None: + pid = _dead_pid() + overrides.pop("pid", None) + data = { + "issue_number": MCP_ISSUE, + "branch_name": MCP_BRANCH, + "remote": "prgs", + "org": MCP_ORG, + "repo": MCP_REPO, + "worktree_path": self.worktree, + "session_pid": pid, + "pid": pid, + "work_lease": { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": MCP_ISSUE, + "pr_number": None, + "branch": MCP_BRANCH, + "worktree_path": self.worktree, + "claimant": claimant, + "created_at": _past_ts(), + "last_heartbeat_at": _past_ts(), + "expires_at": _future_ts(), + }, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", + claimant=claimant, + ), + } + data.update(overrides) + data["session_pid"] = pid + data["pid"] = pid + data["lock_file_path"] = path + issue_lock_store.save_lock_file(path, data) + # Diagnostic tool loads via the per-session pointer (not issue key alone). + pointer = { + "pid": os.getpid(), + "lock_file_path": path, + "issue_number": MCP_ISSUE, + "branch_name": MCP_BRANCH, + "remote": "prgs", + "org": MCP_ORG, + "repo": MCP_REPO, + } + issue_lock_store.save_lock_file( + issue_lock_store.session_pointer_path(self.lock_dir.name), pointer + ) + return path + def _tool_env(self): + env = shared_mutation_env( + "test-author-prgs", + 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=MCP_BRANCH, head=None): + return { + "current_branch": branch, + "porcelain_status": porcelain, + "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): + """Drive the public ``gitea_lock_issue`` tool for the unpublished shape.""" + # No remote branch for this claim — positive absence observation. + if branch_entries is None: + branch_entries = [] + if open_prs is None: + open_prs = [] + 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": MCP_IDENTITY, "profile": MCP_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=MCP_ISSUE, + branch_name=MCP_BRANCH, + remote="prgs", + worktree_path=self.worktree, + ) + + def run_assess_duplicate(self, *, branch_entries=None, open_prs=None, git_state=None): + """Drive the public diagnostic and return its ``lock_recovery`` block.""" + if branch_entries is None: + branch_entries = [] + if open_prs is None: + open_prs = [] + 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": MCP_IDENTITY, "profile": MCP_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_assess_work_issue_duplicate( + issue_number=MCP_ISSUE, + branch_name=MCP_BRANCH, + remote="prgs", + ) + + +class TestAc9AssessorMutatorParity(_UnpublishedMcpBase): + """F1 — diagnostic and mutator share one decision for one durable state.""" + + def test_assessor_and_evaluator_agree_when_recovery_is_sanctioned(self): + path = self.write_dead_lock() + lock = issue_lock_store.read_lock_file(path) + git_state = self._git_state() + + # Shared evaluator (same function the mutator calls). + env = self._tool_env() + with patch( + "mcp_server.api_get_all", return_value=[] + ), patch( + "mcp_server._list_open_pulls", return_value=[] + ), patch( + "mcp_server.get_auth_header", return_value="token x" + ), patch( + "mcp_server._work_lease_claimant", + return_value={"username": MCP_IDENTITY, "profile": MCP_PROFILE}, + ), patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + evaluation = mcp_server._evaluate_issue_lock_recovery( + lock, + issue_number=MCP_ISSUE, + branch_name=MCP_BRANCH, + worktree_path=self.worktree, + remote="prgs", + h="gitea.prgs.cc", + o=MCP_ORG, + r=MCP_REPO, + git_state=git_state, + ) + + diagnostic = self.run_assess_duplicate(git_state=git_state) + projected = diagnostic.get("lock_recovery") or {} + + self.assertTrue(evaluation.get("recovery_sanctioned"), evaluation) + self.assertIsNotNone(projected) + # Projected five-field subset must match the evaluator exactly. + for key in ( + "outcome", + "recovery_sanctioned", + "is_candidate", + "reasons", + "evidence", + ): + self.assertIn(key, projected, f"projection missing {key}") + self.assertEqual( + projected[key], + evaluation.get(key), + f"AC9 divergence on field {key}", + ) + self.assertEqual( + projected["evidence"].get("recovery_mode"), + issue_lock_recovery.RECOVERY_MODE_UNPUBLISHED_CLAIM, + ) + + def test_assessor_and_mutator_agree_on_foreign_identity_refusal(self): + """Mutator cannot accept evidence the assessor rejects (AC9).""" + self.write_dead_lock() + # Foreign identity on the active session. + env = self._tool_env() + git_state = self._git_state() + + with patch( + "mcp_server.api_get_all", return_value=[] + ), patch( + "mcp_server._list_open_pulls", return_value=[] + ), patch( + "mcp_server.get_auth_header", return_value="token x" + ), patch( + "mcp_server._work_lease_claimant", + return_value={"username": "not-the-owner", "profile": MCP_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 *a, **k: ([], [], {"status": "not_claimed"}), + ), patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + diagnostic = mcp_server.gitea_assess_work_issue_duplicate( + issue_number=MCP_ISSUE, + branch_name=MCP_BRANCH, + remote="prgs", + ) + projected = diagnostic.get("lock_recovery") or {} + self.assertFalse(projected.get("recovery_sanctioned")) + self.assertEqual(projected.get("outcome"), issue_lock_recovery.REFUSED) + with self.assertRaises((ValueError, RuntimeError)) as caught: + mcp_server.gitea_lock_issue( + issue_number=MCP_ISSUE, + branch_name=MCP_BRANCH, + remote="prgs", + worktree_path=self.worktree, + ) + # Mutator must not have written a live recovered lock. + lock = issue_lock_store.load_issue_lock( + remote="prgs", + org=MCP_ORG, + repo=MCP_REPO, + issue_number=MCP_ISSUE, + lock_dir=self.lock_dir.name, + ) + self.assertFalse(issue_lock_store.is_lease_live(lock)) + self.assertIn("does not match active identity", " ".join(projected.get("reasons") or [])) + # Refusal text should surface through the mutator error as well. + self.assertTrue(str(caught.exception)) + + def test_probe_failure_degrades_diagnostic_but_raises_on_mutator(self): + """Documented AC9 divergence: diagnostic REFUSED, mutator raises.""" + self.write_dead_lock() + env = self._tool_env() + git_state = self._git_state() + boom = RuntimeError("simulated inventory probe failure") + + with patch( + "mcp_server.api_get_all", side_effect=boom + ), patch( + "mcp_server._list_open_pulls", return_value=[] + ), patch( + "mcp_server.get_auth_header", return_value="token x" + ), patch( + "mcp_server._work_lease_claimant", + return_value={"username": MCP_IDENTITY, "profile": MCP_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 *a, **k: ([], [], {"status": "not_claimed"}), + ), patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + diagnostic = mcp_server.gitea_assess_work_issue_duplicate( + issue_number=MCP_ISSUE, + branch_name=MCP_BRANCH, + remote="prgs", + ) + projected = diagnostic.get("lock_recovery") or {} + self.assertEqual(projected.get("outcome"), issue_lock_recovery.REFUSED) + self.assertFalse(projected.get("recovery_sanctioned")) + self.assertTrue(projected.get("is_candidate")) + self.assertTrue( + any( + "recovery evidence could not be gathered" in r + for r in (projected.get("reasons") or []) + ) + ) + # Mutator path lets the same probe failure raise rather than + # converting it into a soft REFUSED. + with self.assertRaises(RuntimeError) as caught: + mcp_server.gitea_lock_issue( + issue_number=MCP_ISSUE, + branch_name=MCP_BRANCH, + remote="prgs", + worktree_path=self.worktree, + ) + self.assertIn("Could not list branches", str(caught.exception)) + + +class TestAc6McpUnpublishedClaimRecovery(_UnpublishedMcpBase): + """F2 — public gitea_lock_issue recovers an unpublished dead-session claim.""" + + def test_lock_issue_recovers_unpublished_claim_and_applies_cas(self): + path = self.write_dead_lock() + prior = issue_lock_store.read_lock_file(path) + expected_gen = issue_lock_store.lock_generation(prior) + prior_branch = prior["branch_name"] + prior_worktree = prior["worktree_path"] + prior_pid = prior["session_pid"] + + save_calls: list[dict] = [] + real_save = mcp_server._save_issue_lock + + def tracking_save(data, *, expected_generation=None): + save_calls.append({"expected_generation": expected_generation, "data": dict(data)}) + return real_save(data, expected_generation=expected_generation) + + with patch("mcp_server._save_issue_lock", side_effect=tracking_save): + result = self.run_lock_issue() + + self.assertTrue(result["success"], result) + self.assertEqual(result["issue_number"], MCP_ISSUE) + self.assertEqual(result["branch_name"], prior_branch) + self.assertEqual(result["worktree_path"], prior_worktree) + self.assertTrue(result["lock_freshness"]["live"]) + + recovery = result.get("dead_session_recovery") or {} + self.assertTrue(recovery.get("recovered")) + self.assertEqual(recovery.get("prior_session_pid"), prior_pid) + self.assertEqual(recovery.get("replacement_session_pid"), os.getpid()) + self.assertFalse(recovery.get("prior_pid_alive")) + self.assertEqual( + recovery.get("recovery_mode"), + issue_lock_recovery.RECOVERY_MODE_UNPUBLISHED_CLAIM, + ) + self.assertEqual(recovery.get("branch_name"), prior_branch) + self.assertEqual(recovery.get("worktree_path"), prior_worktree) + + # CAS arming: recovery must pass the observed generation into the write. + self.assertTrue(save_calls, "expected _save_issue_lock to be invoked") + self.assertEqual(save_calls[0]["expected_generation"], expected_gen) + self.assertIsNotNone(save_calls[0]["expected_generation"]) + + # Persisted lock keeps branch/worktree and is live under the new PID. + locked = issue_lock_store.load_issue_lock( + remote="prgs", + org=MCP_ORG, + repo=MCP_REPO, + issue_number=MCP_ISSUE, + lock_dir=self.lock_dir.name, + ) + self.assertEqual(locked["branch_name"], prior_branch) + self.assertEqual(locked["worktree_path"], prior_worktree) + self.assertTrue(issue_lock_store.is_lease_live(locked)) + self.assertEqual( + issue_lock_store.lock_generation(locked), + expected_gen + 1, + ) + + def test_second_recoverer_loses_race_through_the_tool(self): + """Two concurrent recoveries through gitea_lock_issue cannot both win.""" + path = self.write_dead_lock() + prior = issue_lock_store.read_lock_file(path) + observed_gen = issue_lock_store.lock_generation(prior) + + real_bind = issue_lock_store.bind_session_lock + bind_calls: list[int | None] = [] + + def racing_bind(data, lock_dir=None, expected_generation=None): + bind_calls.append(expected_generation) + if expected_generation is None: + return real_bind(data, lock_dir=lock_dir, expected_generation=None) + # First concurrent writer wins. + if len([c for c in bind_calls if c is not None]) == 1: + return real_bind( + data, lock_dir=lock_dir, expected_generation=expected_generation + ) + # Second concurrent writer still holds the pre-race generation. + return real_bind( + data, lock_dir=lock_dir, expected_generation=expected_generation + ) + + # First recovery succeeds and advances generation. + with patch( + "mcp_server.issue_lock_store.bind_session_lock", side_effect=racing_bind + ): + first = self.run_lock_issue() + self.assertTrue(first["success"]) + self.assertEqual(bind_calls[0], observed_gen) + + # Replant a dead lock that still reports the *pre-first* generation so + # a second session that already observed that generation races CAS. + # Simulate by writing dead ownership while leaving generation advanced. + advanced = issue_lock_store.read_lock_file(path) + advanced_gen = issue_lock_store.lock_generation(advanced) + self.assertGreater(advanced_gen, observed_gen) + + # Direct CAS at the store layer with the stale observation must fail — + # this is the same call site gitea_lock_issue uses for recovery writes. + with self.assertRaises(RuntimeError) as caught: + issue_lock_store.bind_session_lock( + { + "issue_number": MCP_ISSUE, + "branch_name": MCP_BRANCH, + "worktree_path": self.worktree, + "remote": "prgs", + "org": MCP_ORG, + "repo": MCP_REPO, + "session_pid": os.getpid(), + "pid": os.getpid(), + }, + self.lock_dir.name, + expected_generation=observed_gen, + ) + self.assertIn("generation changed", str(caught.exception)) + + # And the tool itself must arm expected_generation (not None) so an + # inversion of the recovery_sanctioned ternary would be caught here. + self.assertIsNotNone(bind_calls[0]) + + def test_dirty_worktree_refused_through_the_tool(self): + self.write_dead_lock() + with self.assertRaises((ValueError, RuntimeError)) as caught: + self.run_lock_issue(git_state=self._git_state(porcelain=" M work.txt\n")) + self.assertTrue(str(caught.exception)) + + def test_live_prior_owner_refused_through_the_tool(self): + live = os.getpid() + self.write_dead_lock(session_pid=live, pid=live) + with self.assertRaises((ValueError, RuntimeError)): + self.run_lock_issue() + lock = issue_lock_store.load_issue_lock( + remote="prgs", + org=MCP_ORG, + repo=MCP_REPO, + issue_number=MCP_ISSUE, + lock_dir=self.lock_dir.name, + ) + # Live owner must not be overwritten by a "recovery". + self.assertEqual(lock.get("session_pid"), live) + + if __name__ == "__main__": unittest.main()