fix(mcp): forward worktree_path into publication preflight (Closes #815)

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

docs/webui-local-dev.md

@@ -67,11 +67,6 @@ that govern when a write path may open (#632, epic #631).
 | `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
 | `/leases` | Lease and collision visibility (#433) |
 | `/api/leases` | JSON lease/collision export |
-| `/sessions` | Phase 1 shell stub — session inventory (backed by #636) |
-| `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) |
-| `/timeline` | Phase 1 shell stub — workflow event timeline |
-| `/policy` | Phase 1 shell stub — capability/role policy placeholder |
-| `/insights` | Phase 1 shell stub — operational insights placeholder |
 
 Most routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
 `read-only-mvp`, except `/audit` and `/api/audit` which accept POST for
@@ -152,26 +147,6 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
 checkout is behind merged safety-gate changes. Restart guidance links to #420;
 no tokens or MCP restart actions are exposed.
 
-## Application shell — Phase 1 (#638)
-
-The console shell (`webui/layout.py`) renders a grouped navigation driven by a
-single nav-config module, `webui/nav.py`. Nav groups follow the epic #631
-Phase 1 information architecture: **Health, Traffic, Runtime/Sessions,
-Projects, Inventory, Timeline, Policy** (placeholder), and **Insights**
-(placeholder). Live views and Phase 1 placeholders (`stub`) are declared in one
-place so the layout and the route table cannot drift.
-
-The header carries two read-only status badges — an **environment** badge
-(`local` for loopback binds, `remote` otherwise, derived from `WEBUI_HOST`) and
-a **mode: read-only** badge — plus a **Docs** link to this document. No
-privileged action controls are present in the Phase 1 shell.
-
-Not-yet-implemented surfaces (`/sessions`, `/inventory`, `/timeline`,
-`/policy`, `/insights`) resolve to graceful read-only stub pages instead of
-404s; their backing views land in later child issues of #631 (the inventory
-surfaces are backed by #636). Mutating methods on stub routes still fail closed
-with `read-only-mvp`.
-
 ## Deployment boundary (#435)
 
 MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**

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)

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

tests/test_webui_shell.py

@@ -1,135 +0,0 @@
-"""Tests for the Phase 1 operator console application shell (#638)."""
-import sys
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from starlette.routing import Route
-from starlette.testclient import TestClient
-
-from webui import layout
-from webui.app import create_app
-from webui.nav import NAV_GROUPS, STUB_PAGES, nav_hrefs
-
-
-class TestShellNav(unittest.TestCase):
-    def setUp(self):
-        self.client = TestClient(create_app())
-
-    def test_nav_group_labels_present(self):
-        text = self.client.get("/").text
-        for group in NAV_GROUPS:
-            with self.subTest(group=group.label):
-                self.assertIn(f">{group.label}<", text)
-
-    def test_phase1_group_labels_cover_expected_ia(self):
-        labels = {group.label for group in NAV_GROUPS}
-        for expected in (
-            "Health",
-            "Traffic",
-            "Runtime/Sessions",
-            "Projects",
-            "Inventory",
-            "Timeline",
-            "Policy",
-            "Insights",
-        ):
-            with self.subTest(label=expected):
-                self.assertIn(expected, labels)
-
-    def test_every_nav_href_resolves_to_a_get_route(self):
-        app = create_app()
-        get_paths = {
-            route.path
-            for route in app.routes
-            if isinstance(route, Route) and "GET" in route.methods
-        }
-        for href in nav_hrefs():
-            with self.subTest(href=href):
-                self.assertIn(href, get_paths, f"nav href {href} has no GET route")
-
-    def test_legacy_hrefs_still_navigable(self):
-        text = self.client.get("/").text
-        for href in ("/queue", "/projects", "/prompts", "/runtime",
-                     "/audit", "/worktrees", "/leases", "/actions"):
-            with self.subTest(href=href):
-                self.assertIn(f'href="{href}"', text)
-
-
-class TestShellBadges(unittest.TestCase):
-    def setUp(self):
-        self.client = TestClient(create_app())
-
-    def test_mode_badge_present(self):
-        self.assertIn("mode: read-only", self.client.get("/").text)
-
-    def test_environment_badge_present(self):
-        self.assertIn("env:", self.client.get("/").text)
-
-    def test_default_environment_is_local(self):
-        self.assertEqual(layout.environment_label(), "local")
-
-    def test_remote_bind_reports_remote_environment(self):
-        import os
-
-        prior = os.environ.get("WEBUI_HOST")
-        os.environ["WEBUI_HOST"] = "10.0.0.5"
-        try:
-            self.assertEqual(layout.environment_label(), "remote")
-        finally:
-            if prior is None:
-                os.environ.pop("WEBUI_HOST", None)
-            else:
-                os.environ["WEBUI_HOST"] = prior
-
-    def test_docs_link_present(self):
-        text = self.client.get("/").text
-        self.assertIn(layout.DOCS_URL, text)
-        self.assertIn(">Docs<", text)
-
-
-class TestShellStubs(unittest.TestCase):
-    def setUp(self):
-        self.client = TestClient(create_app())
-
-    def test_stub_routes_render_200(self):
-        for path, (title, _desc) in STUB_PAGES.items():
-            with self.subTest(path=path):
-                response = self.client.get(path)
-                self.assertEqual(response.status_code, 200, path)
-                self.assertIn(title, response.text)
-                self.assertIn("placeholder", response.text)
-
-    def test_stub_routes_are_read_only(self):
-        for path in STUB_PAGES:
-            with self.subTest(path=path):
-                response = self.client.post(path)
-                self.assertEqual(response.status_code, 405)
-                self.assertEqual(response.json()["error"], "read-only-mvp")
-
-    def test_stub_pages_carry_nav_and_badges(self):
-        response = self.client.get("/inventory")
-        self.assertIn("mode: read-only", response.text)
-        self.assertIn('href="/queue"', response.text)
-
-
-class TestShellHome(unittest.TestCase):
-    def setUp(self):
-        self.client = TestClient(create_app())
-
-    def test_home_summarizes_console(self):
-        text = self.client.get("/").text
-        self.assertIn("Operator console", text)
-        self.assertIn("Phase 1", text)
-
-    def test_home_links_legacy_pages(self):
-        text = self.client.get("/").text
-        self.assertIn("MVP legacy pages", text)
-        for href in ("/queue", "/audit", "/leases"):
-            with self.subTest(href=href):
-                self.assertIn(f'href="{href}"', text)
-
-
-if __name__ == "__main__":
-    unittest.main()

webui/app.py

@@ -11,7 +11,6 @@ from starlette.routing import Route
 
 from webui.deployment_boundary import deployment_snapshot
 from webui.layout import render_page
-from webui.nav import NAV_GROUPS, STUB_PAGES
 from webui.project_registry import find_project, load_registry, registry_to_dict
 from webui.project_views import render_project_detail, render_projects_list
 from webui.prompt_library import find_prompt, library_to_dict
@@ -44,62 +43,24 @@ def _stub_page(title: str, description: str) -> HTMLResponse:
     return HTMLResponse(render_page(title=title, body_html=body))
 
 
-_LEGACY_PAGES = (
-    ("/queue", "Queue", "live PR and issue dashboard (#429)"),
-    ("/projects", "Projects", "registry and onboarding (#427)"),
-    ("/prompts", "Prompts", "canonical workflow prompt library (#428)"),
-    ("/runtime", "Runtime", "MCP health and stale-runtime detection (#430)"),
-    ("/audit", "Audit", "final-report paste and validator preview (#431)"),
-    ("/worktrees", "Worktrees", "branch hygiene dashboard (#432)"),
-    ("/leases", "Leases", "collision and lease visibility (#433)"),
-    ("/actions", "Actions", "gated write-action framework (#434)"),
-)
-
-
-def _render_home_nav_groups() -> str:
-    groups = []
-    for group in NAV_GROUPS:
-        items = "".join(
-            f'<li><a href="{item.href}">{item.label}</a>'
-            + ("" if item.status == "live" else " <span class=\"muted\">(stub)</span>")
-            + "</li>"
-            for item in group.items
-        )
-        groups.append(f"<h3>{group.label}</h3><ul>{items}</ul>")
-    return "".join(groups)
-
-
 async def home(_request: Request) -> HTMLResponse:
-    legacy = "".join(
-        f"<li><strong>{label}</strong> — {desc} "
-        f'(<a href="{href}">{href}</a>)</li>'
-        for href, label, desc in _LEGACY_PAGES
-    )
     body = (
         "<h2>Operator console</h2>"
-        "<p>Read-only home for the MCP Control Plane Phase 1 operator console. "
-        "Gitea, MCP capability gates, and canonical workflows remain the source "
-        "of truth; this console never mutates them.</p>"
-        "<h2>Phase 1 surfaces</h2>"
-        + _render_home_nav_groups()
-        + "<h2>MVP legacy pages</h2>"
-        "<ul>" + legacy + "</ul>"
+        "<p>Local entry point for MCP Control Plane operational views.</p>"
+        "<ul>"
+        "<li><strong>Queue</strong> — live PR and issue dashboard (#429)</li>"
+        "<li><strong>Projects</strong> — registry and onboarding (#427)</li>"
+        "<li><strong>Prompts</strong> — canonical workflow prompt library (#428)</li>"
+        "<li><strong>Runtime</strong> — MCP health and stale-runtime detection (#430)</li>"
+        "<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
+        "<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
+        "<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
+        "<li><strong>Actions</strong> — gated write-action framework (#434)</li>"
+        "</ul>"
     )
     return HTMLResponse(render_page(title="Home", body_html=body))
 
 
-async def phase_stub(request: Request) -> HTMLResponse:
-    """Graceful read-only placeholder for a not-yet-implemented Phase 1 surface."""
-    title, description = STUB_PAGES[request.url.path]
-    body = (
-        f"<h2>{title}</h2>"
-        f'<div class="stub"><p>{description}</p>'
-        "<p>Phase 1 shell placeholder — no write actions. Tracked under "
-        "epic #631.</p></div>"
-    )
-    return HTMLResponse(render_page(title=title, body_html=body))
-
-
 async def health(_request: Request) -> JSONResponse:
     bind_host = _request.app.state.webui_bind_host
     return JSONResponse({
@@ -330,10 +291,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                 methods=["POST"],
             ),
             Route("/api/leases", api_leases, methods=["GET"]),
-            *[
-                Route(path, phase_stub, methods=["GET"])
-                for path in STUB_PAGES
-            ],
         ],
         exception_handlers={405: method_not_allowed},
     )

webui/layout.py

@@ -2,66 +2,28 @@
 
 from __future__ import annotations
 
-import os
-
-from webui.nav import NAV_GROUPS
+NAV_ITEMS = (
+    ("/", "Home"),
+    ("/queue", "Queue"),
+    ("/projects", "Projects"),
+    ("/prompts", "Prompts"),
+    ("/runtime", "Runtime"),
+    ("/audit", "Audit"),
+    ("/worktrees", "Worktrees"),
+    ("/leases", "Leases"),
+    ("/actions", "Actions"),
+)
 
 MVP_NOTICE = (
     "Read-only MVP — Gitea, MCP tools, and canonical workflows remain the "
     "source of truth. No mutation endpoints."
 )
 
-# Canonical docs entry point surfaced from the shell header (#638).
-DOCS_URL = (
-    "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/src/branch/"
-    "master/docs/webui-local-dev.md"
-)
-
-_LOCAL_HOSTS = frozenset({"", "127.0.0.1", "localhost", "::1"})
-
-
-def environment_label() -> str:
-    """Classify the serving environment as ``local`` or ``remote`` (#638).
-
-    Derived from the same ``WEBUI_HOST`` default the app binds to; loopback
-    hosts are ``local``, anything else is ``remote``. Read-only signal only.
-    """
-    host = (os.environ.get("WEBUI_HOST", "127.0.0.1") or "").strip().lower()
-    return "local" if host in _LOCAL_HOSTS else "remote"
-
-
-def _render_nav() -> str:
-    groups_html = []
-    for group in NAV_GROUPS:
-        links = "".join(
-            f'<a href="{item.href}"'
-            + (' class="nav-stub"' if item.status == "stub" else "")
-            + f'>{item.label}</a>'
-            for item in group.items
-        )
-        groups_html.append(
-            '<div class="nav-group">'
-            f'<span class="nav-group-label">{group.label}</span>'
-            f'<span class="nav-group-links">{links}</span>'
-            "</div>"
-        )
-    return "".join(groups_html)
-
-
-def _render_badges() -> str:
-    env = environment_label()
-    return (
-        '<div class="header-badges">'
-        f'<span class="badge env-badge env-{env}">env: {env}</span>'
-        '<span class="badge mode-badge">mode: read-only</span>'
-        f'<a class="badge docs-link" href="{DOCS_URL}">Docs</a>'
-        "</div>"
-    )
-
 
 def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
-    nav_links = _render_nav()
-    header_badges = _render_badges()
+    nav_links = "".join(
+        f'<a href="{href}">{label}</a>' for href, label in NAV_ITEMS
+    )
     return f"""<!DOCTYPE html>
 <html lang="en">
 <head>
@@ -91,58 +53,21 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
       padding: 0.75rem 1.25rem;
     }}
     header h1 {{
-      margin: 0;
+      margin: 0 0 0.5rem;
       font-size: 1.1rem;
       font-weight: 600;
     }}
-    .header-top {{
-      display: flex;
-      flex-wrap: wrap;
-      align-items: center;
-      justify-content: space-between;
-      gap: 0.5rem 1rem;
-      margin-bottom: 0.6rem;
-    }}
-    .header-badges {{ display: inline-flex; flex-wrap: wrap; gap: 0.4rem; }}
-    .env-badge.env-local {{ color: #8fd19e; border-color: #3d6b4a; }}
-    .env-badge.env-remote {{ color: #e0c27a; border-color: #6b5730; }}
-    .mode-badge {{ color: #9ec8f0; border-color: #3d5f7a; }}
-    a.docs-link {{
-      color: var(--accent);
-      border-color: var(--accent);
-      text-decoration: none;
-      text-transform: none;
-    }}
-    a.docs-link:hover {{ filter: brightness(1.12); }}
     nav {{
       display: flex;
       flex-wrap: wrap;
-      gap: 0.5rem 1.25rem;
+      gap: 0.75rem 1rem;
     }}
-    .nav-group {{
-      display: flex;
-      flex-direction: column;
-      gap: 0.15rem;
-    }}
-    .nav-group-label {{
-      font-size: 0.68rem;
-      text-transform: uppercase;
-      letter-spacing: 0.04em;
-      color: var(--muted);
-    }}
-    .nav-group-links {{ display: inline-flex; flex-wrap: wrap; gap: 0.6rem; }}
     nav a {{
       color: var(--accent);
       text-decoration: none;
       font-size: 0.9rem;
     }}
     nav a:hover {{ text-decoration: underline; }}
-    nav a.nav-stub {{ color: var(--muted); }}
-    nav a.nav-stub::after {{
-      content: " ·stub";
-      font-size: 0.7rem;
-      color: var(--muted);
-    }}
     main {{
       max-width: 52rem;
       margin: 0 auto;
@@ -241,10 +166,7 @@ def render_page(*, title: str, body_html: str, extra_head: str = "") -> str:
 </head>
 <body>
   <header>
-    <div class="header-top">
-      <h1>MCP Control Plane</h1>
-      {header_badges}
-    </div>
+    <h1>MCP Control Plane</h1>
     <nav>{nav_links}</nav>
   </header>
   <main>

webui/nav.py

@@ -1,111 +0,0 @@
-"""Navigation IA for the Phase 1 operator console shell (#638).
-
-Single source of truth for the console navigation so ``webui/layout.py`` and
-the ``webui/app.py`` route table stay aligned with epic #631. Read-only: every
-destination is a GET view or a Phase 1 placeholder. No mutation links.
-
-Nav groups follow the #631 Phase 1 information architecture: Health, Traffic,
-Runtime/Sessions, Projects, Inventory, Timeline, Policy (placeholder), and
-Insights (placeholder). Later-phase surfaces are declared as ``stub`` items and
-backed by ``STUB_PAGES`` so their nav links resolve to a graceful placeholder
-instead of a 404.
-"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-
-
-@dataclass(frozen=True)
-class NavItem:
-    """A single navigation destination.
-
-    ``status`` is ``"live"`` for implemented views and ``"stub"`` for Phase 1
-    placeholders whose backing view lands in a later child issue.
-    """
-
-    href: str
-    label: str
-    status: str = "live"
-
-
-@dataclass(frozen=True)
-class NavGroup:
-    label: str
-    items: tuple[NavItem, ...]
-
-
-NAV_GROUPS: tuple[NavGroup, ...] = (
-    NavGroup("Health", (
-        NavItem("/health", "Liveness"),
-    )),
-    NavGroup("Traffic", (
-        NavItem("/queue", "Queue"),
-        NavItem("/leases", "Leases"),
-        NavItem("/actions", "Actions"),
-    )),
-    NavGroup("Runtime/Sessions", (
-        NavItem("/runtime", "Runtime health"),
-        NavItem("/sessions", "Sessions", "stub"),
-    )),
-    NavGroup("Projects", (
-        NavItem("/projects", "Projects"),
-    )),
-    NavGroup("Inventory", (
-        NavItem("/inventory", "Inventory", "stub"),
-        NavItem("/worktrees", "Worktrees"),
-    )),
-    NavGroup("Timeline", (
-        NavItem("/timeline", "Timeline", "stub"),
-    )),
-    NavGroup("Policy", (
-        NavItem("/policy", "Policy", "stub"),
-        NavItem("/prompts", "Prompts"),
-    )),
-    NavGroup("Insights", (
-        NavItem("/insights", "Insights", "stub"),
-        NavItem("/audit", "Audit"),
-    )),
-)
-
-
-# Phase 1 placeholder destinations whose backing views land in later child
-# issues of epic #631. Each maps a path to (title, description). Routes are
-# registered so nav links resolve to a graceful, read-only stub page.
-STUB_PAGES: dict[str, tuple[str, str]] = {
-    "/sessions": (
-        "Sessions",
-        "Active session, capability, and role inventory. Backed by the unified "
-        "inventory API (#636) once it lands.",
-    ),
-    "/inventory": (
-        "Inventory",
-        "Unified sessions, leases, locks, namespaces, and worktree inventory. "
-        "Backed by the Phase 1 inventory API (#636).",
-    ),
-    "/timeline": (
-        "Timeline",
-        "Workflow event timeline across issues and PRs. A later Phase 1 surface.",
-    ),
-    "/policy": (
-        "Policy",
-        "Capability and role policy surface. Placeholder until a later phase.",
-    ),
-    "/insights": (
-        "Insights",
-        "Aggregate operational insights and trends. Placeholder until a later "
-        "phase.",
-    ),
-}
-
-
-def iter_nav_items():
-    """Yield every ``NavItem`` across all groups in declared order."""
-    for group in NAV_GROUPS:
-        for item in group.items:
-            yield item
-
-
-def nav_hrefs() -> tuple[str, ...]:
-    """Return every navigation href in declared order."""
-    return tuple(item.href for item in iter_nav_items())