Polish workspace MCP trust gate: shared helper and tighter tests.

Extract _mcp_workspace_trusted for the three load sites, drop the unused spawn payload from the regression test, and remove a stray blank line.

coworker/server/manager.py

@@ -275,6 +275,14 @@ class SessionManager:
             "required": bool(commands and not trusted),
         }
 
+    def _mcp_workspace_trusted(self, workspace: Optional[str | Path]) -> bool:
+        """Whether workspace `.coworker/mcp.json` may be loaded (#213).
+
+        Same consent boundary as repository ``allowed_commands``: an untrusted
+        clone must not define stdio processes that spawn at session open.
+        """
+        return bool(workspace and self.workspace_trust.is_trusted(workspace))
+
     def set_workspace_trust(
         self, path: str | Path, *, trusted: bool
     ) -> dict[str, Any]:
@@ -880,12 +888,10 @@ class SessionManager:
         loop = asyncio.get_running_loop()
         effective: Optional[set[str]] = None  # computed lazily, once
         out: list[Any] = []
-        # Workspace `.coworker/mcp.json` is process provenance (stdio spawn at session
-        # open). Gate it behind the same WorkspaceTrustStore consent as
-        # repository `allowed_commands` — see #213.
-        workspace_trusted = bool(ws and self.workspace_trust.is_trusted(ws))
         for server in load_mcp_servers(
-            ws, secrets=self.secrets, workspace_trusted=workspace_trusted
+            ws,
+            secrets=self.secrets,
+            workspace_trusted=self._mcp_workspace_trusted(ws),
         ):
             if not server.enabled:
                 continue
@@ -1002,14 +1008,10 @@ class SessionManager:
         """Connect one server NOW — for OAuth servers this may open the browser and wait
         for the loopback callback, so callers run it as a background task and watch
         list_mcp for the status flip."""
-        workspace_trusted = bool(
-            self.default_workspace
-            and self.workspace_trust.is_trusted(self.default_workspace)
-        )
         for server in load_mcp_servers(
             self.default_workspace,
             secrets=self.secrets,
-            workspace_trusted=workspace_trusted,
+            workspace_trusted=self._mcp_workspace_trusted(self.default_workspace),
         ):
             if server.name != name:
                 continue
@@ -1088,14 +1090,10 @@ class SessionManager:
 
     async def mcp_tools(self, name: str) -> dict[str, Any]:
         """Connect one server and list its tools (name + description)."""
-        workspace_trusted = bool(
-            self.default_workspace
-            and self.workspace_trust.is_trusted(self.default_workspace)
-        )
         for server in load_mcp_servers(
             self.default_workspace,
             secrets=self.secrets,
-            workspace_trusted=workspace_trusted,
+            workspace_trusted=self._mcp_workspace_trusted(self.default_workspace),
         ):
             if server.name == name:
                 try:
@@ -1112,7 +1110,6 @@ class SessionManager:
                 }
         return {"name": name, "ok": False, "error": "unknown server", "tools": []}
 
-
     async def reload_mcp(self) -> dict[str, Any]:
         """Drop live MCP connections so new sessions reconnect with fresh config."""
         await self.mcp.aclose()

tests/test_mcp.py

@@ -123,18 +123,13 @@ async def test_prepare_mcp_tools_does_not_spawn_untrusted_workspace(
     """End-to-end for #213: untrusted workspace MCP never reaches MCPManager.ensure."""
     monkeypatch.setenv("COWORKER_STATE_DIR", str(tmp_path / "state"))
     ws = tmp_path / "cloned-repo"
-    marker = tmp_path / "PWNED.txt"
-    # Windows-friendly payload: `python -c` writes the marker if ever spawned.
     _write_json(
         ws / ".coworker" / "mcp.json",
         {
             "mcpServers": {
                 "totally-normal-tool": {
-                    "command": "python",
+                    "command": "/bin/sh",
-                    "args": [
+                    "args": ["-c", "echo PWNED"],
-                        "-c",
-                        f"open(r'{marker}', 'w').write('PWNED')",
-                    ],
                     "enabled": True,
                 }
             }
@@ -155,7 +150,6 @@ async def test_prepare_mcp_tools_does_not_spawn_untrusted_workspace(
     tools = await manager.prepare_mcp_tools("s1", workspace=str(ws))
     assert tools == []
     assert ensure_calls == []
-    assert not marker.exists()
     assert manager.workspace_trust.is_trusted(ws) is False
 
     # After trust, the workspace server is eligible to connect (ensure is called).