plugins: every base takes a stdio server; pi's adapter and dsh's client spawn it themselves

docs/self-hosting-guide.md

@@ -752,8 +752,9 @@ The record you read back carries what the server derived from the package (`mani
 `mcpServers`, `skills`, and `skipped` for anything it could not load) and a `blob` handle that
 round-trips on the next PUT. The harness's own `mcpServers` and `skills` stay exactly what you wrote
 there; a plugin's components are listed on the plugin. Names must not collide across the harness and
-its enabled plugins, and a package that declares a process (a `stdio` server) is refused on a base
-that cannot run one. `GET /v1/harnesses/{id}/plugin` exports the harness's own tools and Skills as a
+its enabled plugins. A package that declares a process (a `stdio` server) installs on every base: the
+runner writes one launcher per server and each agent's own MCP client runs it (pi's adapter and dsh's
+client included). `GET /v1/harnesses/{id}/plugin` exports the harness's own tools and Skills as a
 package, credentials omitted. The full contract is the
 [Plugins chapter](https://unifiedharnessprotocol.org/spec/2026-09-12/plugins) of the protocol.
 

gateway/app.py

@@ -13484,12 +13484,6 @@ def _skill_key(name: str) -> str:
     """The runner's skill directory name (its _skill_dir_name), so two skills the gateway sees
     as different cannot land in one folder."""
     return re.sub(r"[^A-Za-z0-9_-]+", "-", str(name or ""))
-# Runner backends that can launch a stdio MCP server (each writer emits command/args). pi's MCP
-# adapter and dsh take URLs only; a package that needs a process is refused for those bases rather
-# than accepted and skipped in a turn, which Harnesses §4.1 forbids.
-_STDIO_MCP_BACKENDS = {"claude", "codex", "hermes", "gemini", "qwen", "opencode", "goose", "omp", "cline"}
-
-
 def _plugin_invalid(name: str, path: str, reason: str) -> HTTPException:
     return uhp_error(422, "plugin_invalid", f"Plugin {name or 'package'}: {reason}", "plugins",
                      {"path": path, "reason": reason})
@@ -13822,9 +13816,6 @@ async def _plugins_prepare(body: HarnessBody, org: str, previous: list[dict] | N
     derived object and the blob handle, which is exactly what a client reads back and PUTs again.
     `previous` is what the harness held before an update; packages the new list no longer refers to
     are deleted once the new list is validated."""
-    base = _require_supported_base(body.base) if body.base else ""
-    body.base = base or body.base
-    backend = str(_BASE_CATALOG.get(base, {}).get("backend") or "")
     staged: list[tuple[dict, str | None, str | None]] = []   # (entry, encoded files to store, kept blob)
     seen: dict[str, str] = {}
     for i, item in enumerate(body.plugins or []):
@@ -13859,11 +13850,6 @@ async def _plugins_prepare(body: HarnessBody, org: str, previous: list[dict] | N
                             {"component": "plugin", "name": name, "between": [other, name]})
         seen[name] = key
         enabled = item.get("enabled", True) is not False
-        if enabled and backend and backend not in _STDIO_MCP_BACKENDS \
-                and any(s.get("transport") == "stdio" for s in derived["mcpServers"]):
-            raise uhp_error(422, "unsupported_transport",
-                            f"Plugin {name} declares a stdio MCP server, which the {base} base cannot run.",
-                            "plugins", {"transport": "stdio", "base": base, "plugin": name})
         # A remote server this deployment's network policy would refuse at turn time is refused
         # here instead, and written down: the turn would otherwise drop it with only a log line.
         kept: list[dict] = []

gateway/tests/test_plugins.py

@@ -180,15 +180,18 @@ def test_a_disabled_plugin_stays_installed_and_clears_collisions(api):
     assert pl["name"] == "demo-plugin" and pl["enabled"] is False
 
 
-def test_stdio_on_a_base_that_cannot_run_it_is_refused(api):
-    r = api.post("/v1/harnesses", json={"name": "x", "base": "dsh",
-                                        "plugins": [{"files": _package(manifest=MANIFEST, mcp=MCP)}]})
-    assert r.status_code in (422, 400), r.text
-    e = r.json()["error"]
-    if r.status_code == 422 and e["code"] == "unsupported_transport":
-        assert e["detail"]["transport"] == "stdio" and e["detail"]["base"] == "dsh"
-    else:
-        pytest.skip(f"dsh is not a creatable base here: {e['code']}")
+def test_a_stdio_server_installs_on_every_base(api):
+    """pi's MCP adapter and dsh's MCP client each spawn a stdio server themselves (pi-mcp-adapter's
+    `command`, dsh-mcp-client's `transport: stdio`), so no base refuses a package for needing a
+    process: the runner hands every client the same launcher and the client runs it."""
+    for base in ("pi", "dsh", "claude-code"):
+        r = api.post("/v1/harnesses", json={"name": "x", "base": base,
+                                            "plugins": [{"files": _package(manifest=MANIFEST, mcp=MCP)}]})
+        if r.status_code == 400:
+            pytest.skip(f"{base} is not a creatable base here: {r.json()['error']['code']}")
+        assert r.status_code == 200, r.text
+        (pl,) = r.json()["plugins"]
+        assert {s["name"]: s["transport"] for s in pl["mcpServers"]} == {"remote": "http", "local": "stdio"}
 
 
 def test_export_is_a_package_that_installs_again_without_credentials(api):

runner/dsh_driver.py

@@ -418,9 +418,15 @@ def _compose_patch(home: pathlib.Path, servers: list[dict], llm: dict | None = N
                       {"id": "hr-fs-local", "name": "@deepseek-ai/dsh-fs-local", "config": {"cwd": cwd}}]
     for i, s in enumerate(servers):
         url = (s or {}).get("url")
+        name = "".join(c for c in str((s or {}).get("name") or f"mcp{i}") if c.isalnum() or c in "_-")[:32] or f"mcp{i}"
         if not url:
+            # A plugin's stdio server: dsh-mcp-client's own stdio transport spawns the command
+            # (StdioClientTransport); the launcher the runner wrote carries env and cwd.
+            if (s or {}).get("command"):
+                inserted.append({"id": f"hr-mcp-{i}", "name": "@deepseek-ai/dsh-mcp-client",
+                                 "config": {"transport": "stdio", "serverName": name,
+                                            "command": str(s["command"]), "args": [str(a) for a in (s.get("args") or [])]}})
             continue
-        name = "".join(c for c in str((s or {}).get("name") or f"mcp{i}") if c.isalnum() or c in "_-")[:32] or f"mcp{i}"
         entry = {"id": f"hr-mcp-{i}", "name": "@deepseek-ai/dsh-mcp-client",
                  "config": {"transport": "streamable-http", "serverName": name, "url": url}}
         hdrs: dict = {}

runner/server.py

@@ -2038,15 +2038,20 @@ def _pi_models_json(api: str, base_url: str, api_key: str, model: str,
 
 def _pi_write_mcp(home: pathlib.Path, servers: list[dict] | None) -> bool:
     """Write $HOME/.pi/agent/mcp.json for pi-mcp-adapter (same input contract as the claude/codex
-    writers: url + optional auth/headers). Returns whether any server was written. The agent-dir
+    writers: url + optional auth/headers, or command + args for a plugin's stdio server). Returns
+    whether any server was written. The agent-dir
     location is deliberate: project-local .pi/mcp.json sits behind pi's trust gate; the agent dir
     does not."""
     entries: dict = {}
     for s in servers or []:
+        name = _mcp_name((s or {}).get("name") or (s or {}).get("id") or "mcp")
         url = (s or {}).get("url")
         if not url:
+            # A plugin's stdio server: the adapter spawns `command` itself (its stdio transport,
+            # mutually exclusive with url), and the launcher the runner wrote carries env and cwd.
+            if (s or {}).get("command"):
+                entries[name] = {"command": s["command"], "args": list(s.get("args") or [])}
             continue
-        name = _mcp_name((s or {}).get("name") or (s or {}).get("id") or "mcp")
         entry: dict = {"url": url}
         auth = (s or {}).get("auth")
         if auth:

runner/tests/test_uhp_plugins.py

@@ -132,3 +132,37 @@ def test_a_symlinked_data_directory_is_not_followed(tmp_path):
     assert not data.is_symlink() and data.is_dir(), "the planted link is replaced by a real directory"
     assert pathlib.Path(out["command"]).parent.resolve() == data.resolve()
     assert not list(outside.iterdir()), "nothing was written through the link"
+
+
+def test_pi_adapter_gets_a_stdio_command_entry(tmp_path):
+    """pi-mcp-adapter spawns a stdio server itself (`command`, mutually exclusive with `url`), so
+    a plugin's server reaches pi as the launcher the runner wrote, beside the URL servers. Before
+    this the writer skipped anything without a url and the gateway refused the package for pi."""
+    home = tmp_path / "home"
+    servers = [{"name": "probe", "command": "/ws/.harness/plugin-data/hr_probe/.launch-probe.sh",
+                "args": ["--data", "/ws/.harness/plugin-data/hr_probe"], "plugin": "hr-probe"},
+               {"name": "remote", "url": "https://mcp.example.invalid/mcp", "headers": {"X-Team": "t"}}]
+    assert rn._pi_write_mcp(home, servers) is True
+    cfg = json.loads((home / ".pi" / "agent" / "mcp.json").read_text())["mcpServers"]
+    assert cfg["probe"] == {"command": "/ws/.harness/plugin-data/hr_probe/.launch-probe.sh",
+                            "args": ["--data", "/ws/.harness/plugin-data/hr_probe"]}
+    assert cfg["remote"] == {"url": "https://mcp.example.invalid/mcp", "headers": {"X-Team": "t"}}
+
+
+def test_dsh_client_gets_a_stdio_row(tmp_path):
+    """dsh-mcp-client has a stdio transport of its own (StdioClientTransport on command/args), so a
+    plugin's server is one more inserted client row, discriminated on transport."""
+    import yaml
+    import dsh_driver
+    home = tmp_path / "home"
+    servers = [{"name": "probe", "command": "/ws/.harness/plugin-data/hr_probe/.launch-probe.sh",
+                "args": ["--data", "/ws/.harness/plugin-data/hr_probe"], "plugin": "hr-probe"},
+               {"name": "remote", "url": "https://mcp.example.invalid/mcp"}]
+    out = pathlib.Path(dsh_driver._compose_patch(home, servers, relay_port=9999, model="m", cwd=str(tmp_path / "ws")))
+    rows = [r for e in yaml.safe_load(out.read_text()) if isinstance(e, dict) for r in (e.get("insert") or [])
+            if r.get("name") == "@deepseek-ai/dsh-mcp-client"]
+    by = {r["config"]["serverName"]: r["config"] for r in rows}
+    assert by["probe"] == {"transport": "stdio", "serverName": "probe",
+                           "command": "/ws/.harness/plugin-data/hr_probe/.launch-probe.sh",
+                           "args": ["--data", "/ws/.harness/plugin-data/hr_probe"]}
+    assert by["remote"]["transport"] == "streamable-http" and by["remote"]["url"] == "https://mcp.example.invalid/mcp"