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 <[email protected]>

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", "[email protected]")
+        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()