From 95a5eb254f7758b694a26b3f5629558cb88db328 Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Wed, 22 Jul 2026 17:27:49 -0500 Subject: [PATCH] fix(mcp): forward worktree_path into publication preflight (Closes #815) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit gitea_publish_unpublished_issue_branch accepted a required worktree_path but resolved it only after verify_preflight_purity had already run, so the #618 branches-only guard and every other workspace-resolution layer behind preflight received None and fell back to the MCP process root. A daemon rooted at the stable control checkout therefore refused a valid registered issue worktree the caller had explicitly supplied, before the publication assessor could use it — the sole verify_preflight_purity call site that accepted a worktree argument and dropped it. Resolve the workspace once, before preflight, and forward it. A blank or absent path forwards None and keeps the ordinary #618 fail-closed fallback, so guard strictness is unchanged for missing, empty, unregistered, foreign, or control-checkout worktrees. Public tool contract, ownership, cleanliness, hash, ancestry, and read-after-write protections from PR #814 are untouched. Adds tests/test_issue_815_preflight_worktree_forwarding.py: a #735-style capture proving the argument reaches verify_preflight_purity, a faithful production reproduction (control-rooted daemon, no session lock, explicit worktree) that clears #618 on the fixed source and is trapped at #618 on the unpatched source, an end-to-end control-rooted publication in the real topology (PROJECT_ROOT is the stable control checkout, the issue worktree is a distinct registered path, production guards forced on), and negative coverage keeping every #618 and assessor refusal intact. The prior #812 suite masked the defect by patching PROJECT_ROOT to equal the issue worktree. Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 20 +- ...issue_815_preflight_worktree_forwarding.py | 622 ++++++++++++++++++ 2 files changed, 640 insertions(+), 2 deletions(-) create mode 100644 tests/test_issue_815_preflight_worktree_forwarding.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index f71360b..acd56ca 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -9173,11 +9173,27 @@ def gitea_publish_unpublished_issue_branch( if blocked: return blocked - verify_preflight_purity(remote, task=task, org=org, repo=repo) + # #815: resolve the caller's worktree *before* preflight and forward it, so + # every workspace-resolution layer behind verify_preflight_purity — including + # the #618 branches-only guard — judges the registered issue worktree this + # publication actually operates on. Resolving it afterwards let preflight + # fall back to the MCP process root, so a daemon rooted at the stable control + # checkout refused a valid explicit worktree before the assessor ever ran. + # A caller supplying nothing usable forwards None and keeps the ordinary + # fail-closed fallback. + explicit_worktree = (worktree_path or "").strip() + workspace = os.path.realpath(os.path.abspath(explicit_worktree or ".")) + + verify_preflight_purity( + remote, + worktree_path=workspace if explicit_worktree else None, + task=task, + org=org, + repo=repo, + ) h, o, r = _resolve(remote, host, org, repo) git_remote = (git_remote_name or remote or "").strip() - workspace = os.path.realpath(os.path.abspath((worktree_path or "").strip() or ".")) existing_lock = issue_lock_store.load_issue_lock( remote=remote, org=o, repo=r, issue_number=int(issue_number) diff --git a/tests/test_issue_815_preflight_worktree_forwarding.py b/tests/test_issue_815_preflight_worktree_forwarding.py new file mode 100644 index 0000000..8cc97b2 --- /dev/null +++ b/tests/test_issue_815_preflight_worktree_forwarding.py @@ -0,0 +1,622 @@ +"""Publication preflight must receive the caller's worktree (#815). + +``gitea_publish_unpublished_issue_branch`` takes a **required** ``worktree_path`` +but resolved it only *after* ``verify_preflight_purity`` had already run. Every +workspace-resolution layer behind that preflight — canonical root, root checkout, +create-issue bootstrap, the #618 branches-only guard, issue scope, and anti-stomp +— therefore received ``None`` and fell back to the MCP process root. A daemon +rooted at the stable control checkout refused a valid registered issue worktree +that the caller had explicitly supplied, before the publication assessor ever ran. + +The #812 suite could not see this. Its fixture sets ``self.worktree = +os.path.realpath(self.repo)`` and patches ``PROJECT_ROOT`` to that same path, so +the fallback resolved to the very worktree the argument named. The production +topology — control checkout on a stable branch, issue worktree somewhere else — +was never constructed, and preflight additionally no-ops under pytest unless +production guards are forced on. + +These tests build that topology honestly: + +* ``PROJECT_ROOT`` is a control checkout sitting on ``master``; +* the registered issue worktree is a genuinely separate path under ``branches/``; +* ``GITEA_TEST_FORCE_PRODUCTION_GUARDS`` is set so the #618 guard really runs; +* no patch makes the issue worktree appear to be ``PROJECT_ROOT``. + +Nothing here reads, writes, or references the protected worktree named in #812 +AC17 and #815 AC9. Every fixture is built from scratch against a local bare +remote, so publication and read-after-write verification genuinely execute. +""" + +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__))) + +import issue_lock_provenance # noqa: E402 +import issue_lock_store # noqa: E402 +import mcp_server # noqa: E402 +from mutation_profile_fixture import shared_mutation_env # noqa: E402 + +ISSUE = 9815 +BRANCH = f"feat/issue-{ISSUE}-forwarding-fixture" +WORKTREE_DIRNAME = BRANCH.replace("/", "-") +IDENTITY = "example-user" +PROFILE = "test-author-prgs" +ORG = "Scaled-Tech-Consulting" +REPO = "Gitea-Tools" +GIT_REMOTE = "prgs" + +AUTHOR_PROFILE = { + "profile_name": "prgs-author", + "role": "author", + "allowed_operations": [ + "gitea.read", "gitea.issue.create", "gitea.issue.comment", + "gitea.pr.create", "gitea.repo.commit", "gitea.branch.push", + ], + "forbidden_operations": [], + "audit_label": "prgs-author", +} + + +def _ts(hours: int) -> str: + return ( + (datetime.now(timezone.utc) + timedelta(hours=hours)) + .isoformat() + .replace("+00:00", "Z") + ) + + +class TestPreflightReceivesTheWorktree(unittest.TestCase): + """AC1 — the supplied path reaches ``verify_preflight_purity`` itself. + + Follows the #735 capture pattern: replace preflight with a recorder that + raises, so the argument can be proven forwarded without performing the + mutation. This is the direct unit-level statement of the defect. + """ + + def _capture_preflight(self, **kwargs): + captured: dict = {} + + def _capture(*a, **kw): + captured.update(kw) + captured["_args"] = a + raise RuntimeError("capture-only") + + with patch.object( + mcp_server, "verify_preflight_purity", side_effect=_capture + ), patch.object( + mcp_server, "get_profile", return_value=AUTHOR_PROFILE + ), patch.object( + mcp_server, "_resolve", + return_value=("gitea.prgs.cc", ORG, REPO), + ), patch.object( + mcp_server, "_auth", return_value="token fake", + ), patch.object( + mcp_server.role_session_router, + "check_author_mutation_after_reviewer_stop", + return_value=(True, []), + ), patch.object( + mcp_server, "_namespace_mutation_block", return_value=None + ), patch.object( + mcp_server, "_profile_permission_block", return_value=None + ): + try: + mcp_server.gitea_publish_unpublished_issue_branch(**kwargs) + except RuntimeError as exc: + if "capture-only" not in str(exc) and not captured: + raise + self.assertTrue( + captured, + "gitea_publish_unpublished_issue_branch never called " + "verify_preflight_purity", + ) + return captured + + def _base_kwargs(self, **overrides): + kwargs = { + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": "/tmp/issue-815-explicit-worktree", + "expected_head": "a" * 40, + "remote": "prgs", + "org": ORG, + "repo": REPO, + "git_remote_name": GIT_REMOTE, + } + kwargs.update(overrides) + return kwargs + + def test_explicit_worktree_path_reaches_preflight(self): + captured = self._capture_preflight(**self._base_kwargs()) + self.assertEqual( + captured.get("worktree_path"), + os.path.realpath(os.path.abspath("/tmp/issue-815-explicit-worktree")), + "the authoritative worktree_path must be forwarded into preflight", + ) + + def test_forwarded_path_is_the_one_publication_uses(self): + """AC4 — preflight and publication must judge the same resolved path.""" + raw = "/tmp/issue-815-explicit-worktree/./" + captured = self._capture_preflight(**self._base_kwargs(worktree_path=raw)) + expected = os.path.realpath(os.path.abspath(raw.strip())) + self.assertEqual(captured.get("worktree_path"), expected) + + def test_blank_worktree_path_forwards_none(self): + """AC5/AC8 — nothing usable supplied keeps the fail-closed fallback.""" + for blank in ("", " "): + with self.subTest(blank=repr(blank)): + captured = self._capture_preflight( + **self._base_kwargs(worktree_path=blank) + ) + self.assertIsNone( + captured.get("worktree_path"), + "a blank worktree must not resolve to the process cwd", + ) + + def test_org_repo_and_task_forwarding_are_not_regressed(self): + """AC6 — #735's org/repo forwarding and the task name still hold.""" + captured = self._capture_preflight(**self._base_kwargs()) + self.assertEqual(captured.get("org"), ORG) + self.assertEqual(captured.get("repo"), REPO) + self.assertEqual(captured.get("task"), "publish_unpublished_branch") + + +class _ProductionTopologyBase(unittest.TestCase): + """Control checkout on master + a distinct registered issue worktree. + + This is the shape the production daemon runs in and the shape the #812 + fixture never built. ``PROJECT_ROOT`` is the control checkout; the issue + worktree is a real registered worktree at a different path; production + guards are forced on so the #618 branches-only guard genuinely evaluates. + """ + + def setUp(self): + self.lock_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.lock_dir.cleanup) + self.origin = tempfile.mkdtemp(prefix="issue815-origin-") + self.control = tempfile.mkdtemp(prefix="issue815-control-") + for path in (self.origin, self.control): + self.addCleanup( + lambda p=path: subprocess.run(["rm", "-rf", p], check=False) + ) + self._init_repos() + 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, cwd=None): + return subprocess.run( + ["git", "-C", cwd or self.control, *args], + capture_output=True, text=True, check=True, + ) + + def _init_repos(self): + subprocess.run( + ["git", "init", "-q", "--bare", "-b", "master", self.origin], check=True + ) + self._git("init", "-q", "-b", "master") + self._git("config", "user.email", "test@example.com") + self._git("config", "user.name", "Test") + self._git("remote", "add", GIT_REMOTE, self.origin) + + with open(os.path.join(self.control, "seed.txt"), "w") as fh: + fh.write("seed\n") + # The real repository gitignores branches/, so a registered worktree + # living there does not dirty the stable control checkout. Mirror that, + # or the #615 dirty-runtime block fires on the worktree we just created. + with open(os.path.join(self.control, ".gitignore"), "w") as fh: + fh.write("branches/\n") + self._git("add", "seed.txt", ".gitignore") + self._git("commit", "-q", "-m", "seed") + self.base_sha = self._git("rev-parse", "HEAD").stdout.strip() + self._git("push", "-q", GIT_REMOTE, "master") + + # The control checkout STAYS on master. This is the whole point: the + # daemon's process root is the stable control checkout, never the + # worktree the publication targets. + self.worktree = os.path.realpath( + os.path.join(self.control, "branches", WORKTREE_DIRNAME) + ) + self._git("worktree", "add", "-q", "-b", BRANCH, self.worktree, "master") + + with open(os.path.join(self.worktree, "work.txt"), "w") as fh: + fh.write("unpublished implementation\n") + self._git("add", "work.txt", cwd=self.worktree) + self._git("commit", "-q", "-m", "unpublished implementation", cwd=self.worktree) + self.head_sha = self._git("rev-parse", "HEAD", cwd=self.worktree).stdout.strip() + + self.control_branch = self._git( + "rev-parse", "--abbrev-ref", "HEAD" + ).stdout.strip() + + # ── durable lock naming the caller and the issue worktree ──────────── + def lock_path(self): + return issue_lock_store.lock_file_path( + remote="prgs", org=ORG, repo=REPO, issue_number=ISSUE, + lock_dir=self.lock_dir.name, + ) + + def write_lock(self, *, bind_session=True, **overrides): + path = self.lock_path() + claimant = overrides.pop( + "claimant", {"username": IDENTITY, "profile": PROFILE} + ) + pid = overrides.pop("session_pid", os.getpid()) + lease = { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": ISSUE, + "pr_number": None, + "branch": overrides.get("branch_name", BRANCH), + "worktree_path": overrides.get("worktree_path", self.worktree), + "claimant": claimant, + "created_at": _ts(-2), + "last_heartbeat_at": _ts(-2), + "expires_at": _ts(-1), + } + lease.update(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": 1, + "work_lease": lease, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", claimant=claimant + ), + } + data.update(overrides) + data["lock_file_path"] = path + issue_lock_store.save_lock_file(path, data) + # Bind the session pointer so the #683 issue-scope guard resolves an + # owning issue for this author session. In real production the publish + # task does not require a session lock — require_author_lock is keyed on + # the test-only production_guards_forced() flag, which this suite must + # set to make preflight run at all — so this pointer is fixture + # scaffolding to clear a guard production would not apply here, never a + # softening of the worktree-forwarding behaviour under test. The + # preflight-negative cases below leave it unbound precisely so the #618 + # guard is reached with no session fallback to rescue a bad worktree. + if bind_session: + pointer = { + "pid": os.getpid(), + "lock_file_path": path, + "issue_number": ISSUE, + "branch_name": data["branch_name"], + "remote": "prgs", + "org": ORG, + "repo": 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( + PROFILE, include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=self.lock_dir.name, + ) + env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + # The defect only exists where preflight actually runs. Under pytest the + # production root/branches/scope guards are skipped unless forced on, so + # force them: this test exists to exercise the #618 guard, not to bypass + # it. Parity is pinned to the server's own startup head so repointing + # PROJECT_ROOT does not read as a stale daemon. + env["GITEA_TEST_FORCE_PRODUCTION_GUARDS"] = "1" + # Production is a promoted stable-control runtime. The pytest process + # itself runs from a branches/ worktree, which the #615 runtime-mode + # gate correctly classifies as dev-test; declaring the sanctioned mode + # models the production daemon rather than defeating the gate. Without + # this, forcing production guards on would trip the *runtime-mode* block + # for a reason unrelated to the #815 worktree-forwarding defect. + env["GITEA_MCP_RUNTIME_MODE"] = "stable-control" + startup_head = mcp_server._STARTUP_PARITY.get("startup_head") or "" + env["GITEA_TEST_CURRENT_HEAD"] = startup_head + env["GITEA_TEST_LIVE_REMOTE_HEAD"] = startup_head + return env + + def run_publish(self, *, open_prs=None, expected_head=None, **kwargs): + """Drive the public tool with PROJECT_ROOT pinned to the CONTROL checkout.""" + env = self._tool_env() + with patch( + "mcp_server._list_open_pulls", return_value=list(open_prs or []) + ), patch( + "mcp_server._auth", return_value="token x" + ), patch( + "mcp_server.get_auth_header", return_value="token x" + ), patch( + "mcp_server._work_lease_claimant", + return_value={"username": IDENTITY, "profile": PROFILE}, + ), patch.object( + # NOTE: the control checkout — deliberately NOT self.worktree. + mcp_server, "PROJECT_ROOT", self.control + ), patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return mcp_server.gitea_publish_unpublished_issue_branch( + issue_number=kwargs.pop("issue_number", ISSUE), + branch_name=kwargs.pop("branch_name", BRANCH), + worktree_path=kwargs.pop("worktree_path", self.worktree), + expected_head=expected_head or self.head_sha, + remote="prgs", + git_remote_name=kwargs.pop("git_remote_name", GIT_REMOTE), + **kwargs, + ) + + def remote_head(self, branch=BRANCH): + res = subprocess.run( + ["git", "-C", self.origin, "rev-parse", "--verify", "--quiet", branch], + capture_output=True, text=True, check=False, + ) + return (res.stdout or "").strip() or None + + +class TestForwardingClearsThe618Guard(_ProductionTopologyBase): + """AC2 — the faithful production reproduction, and the sharpest fix proof. + + The production recovery worker had **no** session issue lock — acquiring one + was the very thing the deadlock prevented — so preflight had nothing but the + explicit ``worktree_path`` argument to resolve the workspace from. This class + reproduces exactly that: no session pointer is bound, so there is no + author-lock fallback to rescue a dropped argument. + + With the argument forwarded (fixed source) the #618 branches-only guard + accepts the registered issue worktree and the call advances to the next + guard. With the argument dropped (the buggy source this issue reports) + preflight falls back to ``PROJECT_ROOT`` — the stable control checkout — and + the #618 guard traps the call there. The two outcomes are told apart by the + guard that fired, on its own error text. + + This test therefore *fails* against the unpatched source (the call is trapped + at #618 instead of clearing it), which is what makes it a regression rather + than a smoke test. + """ + + _CONTROL_CHECKOUT_MARKERS = ("stable control checkout", "#618") + + def test_explicit_worktree_clears_618_without_a_session_lock(self): + # No write_lock(): the session is deliberately unbound, as in production. + with self.assertRaises(RuntimeError) as ctx: + self.run_publish() + message = str(ctx.exception) + # The workspace guard is satisfied — the failure is the *later* scope + # guard (no owning issue), never the control-checkout refusal. If the + # argument were dropped, this call would be trapped at #618 instead. + for marker in self._CONTROL_CHECKOUT_MARKERS: + self.assertNotIn( + marker, message, + f"the explicit worktree must clear #618; got a control-checkout " + f"refusal instead: {message}", + ) + self.assertIn( + "owning issue", message, + f"expected the downstream scope guard to fire, got: {message}", + ) + self.assertIsNone(self.remote_head()) + + def test_dropped_argument_would_be_trapped_at_618(self): + # Simulate the buggy call shape directly: no session lock, and preflight + # given no worktree, exactly as the unpatched source left it. This pins + # the control-checkout refusal that the fix eliminates, so the pair of + # tests brackets the defect from both sides regardless of which source + # version is loaded. + env = self._tool_env() + with patch( + "mcp_server._list_open_pulls", return_value=[] + ), patch( + "mcp_server._auth", return_value="token x" + ), patch( + "mcp_server.get_auth_header", return_value="token x" + ), patch( + "mcp_server._work_lease_claimant", + return_value={"username": IDENTITY, "profile": PROFILE}, + ), patch.object( + mcp_server, "PROJECT_ROOT", self.control + ), patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + with self.assertRaises(RuntimeError) as ctx: + # Drive verify_preflight_purity the way the buggy body did: + # no worktree_path forwarded at all. + mcp_server.verify_preflight_purity( + "prgs", + task="publish_unpublished_branch", + org=ORG, + repo=REPO, + ) + message = str(ctx.exception) + self.assertTrue( + any(m in message for m in self._CONTROL_CHECKOUT_MARKERS), + f"a dropped worktree must trap at the control checkout: {message}", + ) + self.assertIsNone(self.remote_head()) + + +class TestProductionTopologyPublishes(_ProductionTopologyBase): + """AC2/AC4/AC7 — the explicit registered worktree is what preflight validates.""" + + def test_fixture_is_genuinely_the_production_topology(self): + """Guard the guard: if this drifts, the regression stops meaning anything.""" + self.assertNotEqual( + os.path.realpath(self.control), self.worktree, + "the issue worktree must not be PROJECT_ROOT", + ) + self.assertEqual( + self.control_branch, "master", + "the control checkout must sit on a stable branch", + ) + self.assertTrue( + os.path.realpath(self.worktree).startswith( + os.path.realpath(os.path.join(self.control, "branches")) + os.sep + ), + "the issue worktree must live under branches/", + ) + listed = subprocess.run( + ["git", "-C", self.control, "worktree", "list"], + capture_output=True, text=True, check=True, + ).stdout + self.assertIn( + self.worktree, listed, "the issue worktree must be genuinely registered" + ) + + def test_publishes_from_a_control_rooted_daemon(self): + """The exact production failure: this refused with #618 before the fix.""" + self.write_lock() + self.assertIsNone(self.remote_head(), "fixture must start unpublished") + res = self.run_publish() + self.assertTrue(res.get("success"), res) + self.assertTrue(res.get("performed"), res) + self.assertEqual(self.remote_head(), self.head_sha) + + def test_dry_run_uses_the_explicit_worktree(self): + """AC4 — dry-run reaches the same decision without publishing.""" + self.write_lock() + res = self.run_publish(dry_run=True) + self.assertTrue(res.get("success"), res) + self.assertFalse(res.get("performed"), res) + self.assertTrue(res.get("would_publish"), res) + self.assertIsNone(self.remote_head(), "dry-run must not publish") + + def test_dry_run_and_apply_agree_on_the_same_worktree(self): + """AC4 — both paths resolve the same workspace, so both succeed.""" + self.write_lock() + dry = self.run_publish(dry_run=True) + self.assertTrue(dry.get("would_publish"), dry) + applied = self.run_publish() + self.assertTrue(applied.get("performed"), applied) + self.assertEqual(self.remote_head(), self.head_sha) + + def test_read_after_write_verification_still_runs(self): + """AC6 — PR #814's post-publication verification is unchanged.""" + self.write_lock() + res = self.run_publish() + self.assertTrue(res.get("verified"), res) + self.assertEqual(res.get("remote_head_sha"), self.head_sha) + + +class TestProductionTopologyFailsClosed(_ProductionTopologyBase): + """AC3/AC5/AC8 — the fix does not weaken any refusal. + + A refusal reaches the caller by one of two mechanisms, and this class holds + them apart deliberately. A bad *workspace* is caught by the #618 preflight + guard, which raises before the assessor is built. A bad *content/ownership* + fact passes preflight (the worktree itself is fine) and is then refused by + the publication assessor, which returns ``success: False``. Both are + fail-closed; asserting the wrong mechanism would hide a regression. + """ + + # ── #618 preflight refusals: no session lock, so nothing rescues a bad + # workspace and the guard fires exactly as it does in production ────── + def _assert_preflight_raises(self, **kwargs): + with self.assertRaises(RuntimeError) as ctx: + self.run_publish(**kwargs) + self.assertIsNone( + self.remote_head(), "a blocked publication must not reach the remote" + ) + return str(ctx.exception) + + def test_blank_worktree_path_fails_closed_via_618(self): + """AC5 — a blank path forwards None, so preflight sees the control root.""" + for blank in ("", " "): + with self.subTest(blank=repr(blank)): + message = self._assert_preflight_raises(worktree_path=blank) + self.assertIn("618", message) + + def test_control_checkout_as_worktree_fails_closed_via_618(self): + """AC5 — naming the stable control checkout explicitly is still refused.""" + message = self._assert_preflight_raises(worktree_path=self.control) + self.assertIn("618", message) + + def test_unregistered_directory_fails_closed(self): + """AC3 — a plain directory under branches/ is not a registered worktree.""" + bogus = os.path.join(self.control, "branches", "not-a-worktree") + os.makedirs(bogus, exist_ok=True) + self._assert_preflight_raises(worktree_path=bogus) + + def test_missing_worktree_path_fails_closed(self): + """AC3 — a path that does not exist is refused, not silently replaced.""" + missing = os.path.join(self.control, "branches", "absent-worktree") + self._assert_preflight_raises(worktree_path=missing) + + # ── assessor refusals: preflight passes on a valid worktree, then the + # publication assessor refuses on content/ownership evidence ────────── + def _assert_assessor_refuses(self, **kwargs): + res = self.run_publish(**kwargs) + self.assertFalse(res.get("success"), res) + self.assertFalse(res.get("performed"), res) + self.assertIsNone(self.remote_head()) + return res + + def test_changed_local_head_still_refuses(self): + """AC6 — the declared expected_head remains authoritative.""" + self.write_lock() + self._assert_assessor_refuses(expected_head="b" * 40) + + def test_foreign_claimant_still_refuses(self): + """AC6 — ownership still comes from the durable lock record.""" + self.write_lock(claimant={"username": "someone-else", "profile": PROFILE}) + self._assert_assessor_refuses() + + def test_dirty_worktree_still_refuses(self): + """AC6 — cleanliness enforcement survives the forwarding change.""" + self.write_lock() + with open(os.path.join(self.worktree, "work.txt"), "a") as fh: + fh.write("uncommitted drift\n") + self._assert_assessor_refuses() + + def test_competing_open_pr_still_refuses(self): + """AC6 — a rival claim on another branch still blocks.""" + self.write_lock() + self._assert_assessor_refuses( + open_prs=[{"number": 4242, "head": {"ref": f"fix/issue-{ISSUE}-rival"}}] + ) + + def test_issue_lock_record_is_not_mutated_by_a_refusal(self): + """AC6 — record separation (#812 AC23) is unaffected by this change.""" + path = self.write_lock() + with open(path, "rb") as fh: + before = fh.read() + self._assert_assessor_refuses(expected_head="c" * 40) + with open(path, "rb") as fh: + self.assertEqual(before, fh.read()) + + +class TestProtectedFixtureNotReferenced(unittest.TestCase): + """AC9 — this regression never names the protected #635 fixture. + + The forbidden tokens are reconstructed from fragments so this assertion + file does not itself contain them and produce a false positive. + """ + + def test_no_reference_to_the_protected_worktree(self): + forbidden = [ + "issue-635-" + "project-registry-api", + "b2f6e9a6dc40e9651ef8" + "76f322dd0a68bddebfd8", + ] + here = os.path.abspath(__file__) + with open(here, "r", encoding="utf-8") as fh: + text = fh.read() + for token in forbidden: + self.assertNotIn( + token, text, + f"the protected #635 fixture must not be referenced: {token}", + ) + + +if __name__ == "__main__": + unittest.main()