mirror of
https://github.com/andrewyng/openworker.git
synced 2026-09-11 14:50:14 +00:00
Surface MCP server startup failures
Capture stdio stderr tail; record generic connect errors for /v1/mcp (status=error); persistent session notice via mcp_error; tests for client, manager, and GUI mapping.
This commit is contained in:
+44
-2
@@ -13,8 +13,9 @@ Tool execution from the (sync) ToolRegistry bridges back here via
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import tempfile
|
||||||
from contextlib import AsyncExitStack
|
from contextlib import AsyncExitStack
|
||||||
from typing import Any, Optional
|
from typing import Any, IO, Optional
|
||||||
|
|
||||||
from mcp import ClientSession, StdioServerParameters
|
from mcp import ClientSession, StdioServerParameters
|
||||||
from mcp.client.stdio import stdio_client
|
from mcp.client.stdio import stdio_client
|
||||||
@@ -23,6 +24,25 @@ from mcp.client.streamable_http import streamablehttp_client
|
|||||||
from .config import MCPServerDef
|
from .config import MCPServerDef
|
||||||
|
|
||||||
|
|
||||||
|
_STDERR_TAIL_LINES = 20
|
||||||
|
_STDERR_TAIL_CHARS = 1500
|
||||||
|
|
||||||
|
|
||||||
|
def _read_tail(errfile: Optional[IO[str]]) -> Optional[str]:
|
||||||
|
"""Last few lines of a captured stderr file — the crash evidence, not the log."""
|
||||||
|
if errfile is None:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
errfile.seek(0)
|
||||||
|
text = errfile.read()
|
||||||
|
except (OSError, ValueError):
|
||||||
|
return None
|
||||||
|
lines = [ln for ln in text.strip().splitlines() if ln.strip()]
|
||||||
|
if not lines:
|
||||||
|
return None
|
||||||
|
return "\n".join(lines[-_STDERR_TAIL_LINES:])[-_STDERR_TAIL_CHARS:]
|
||||||
|
|
||||||
|
|
||||||
class _Conn:
|
class _Conn:
|
||||||
def __init__(self, session: ClientSession, tools: list[Any]) -> None:
|
def __init__(self, session: ClientSession, tools: list[Any]) -> None:
|
||||||
self.session = session
|
self.session = session
|
||||||
@@ -36,6 +56,7 @@ class MCPManager:
|
|||||||
def __init__(self, secrets: Any = None) -> None:
|
def __init__(self, secrets: Any = None) -> None:
|
||||||
self._conns: dict[str, _Conn] = {}
|
self._conns: dict[str, _Conn] = {}
|
||||||
self._tasks: dict[str, asyncio.Task] = {}
|
self._tasks: dict[str, asyncio.Task] = {}
|
||||||
|
self._stderr_tails: dict[str, str] = {}
|
||||||
self._lock = asyncio.Lock()
|
self._lock = asyncio.Lock()
|
||||||
# SecretStore for OAuth servers' token persistence (mcp/oauth.py); lazy default
|
# SecretStore for OAuth servers' token persistence (mcp/oauth.py); lazy default
|
||||||
# so library/CLI construction without secrets keeps working.
|
# so library/CLI construction without secrets keeps working.
|
||||||
@@ -64,6 +85,10 @@ class MCPManager:
|
|||||||
async def tools(self, server: MCPServerDef) -> list[Any]:
|
async def tools(self, server: MCPServerDef) -> list[Any]:
|
||||||
return (await self.ensure(server)).tools
|
return (await self.ensure(server)).tools
|
||||||
|
|
||||||
|
def last_stderr(self, name: str) -> Optional[str]:
|
||||||
|
"""Stderr tail from the most recent failed startup of `name`, if any."""
|
||||||
|
return self._stderr_tails.get(name)
|
||||||
|
|
||||||
async def call(
|
async def call(
|
||||||
self, name: str, tool: str, arguments: Optional[dict[str, Any]]
|
self, name: str, tool: str, arguments: Optional[dict[str, Any]]
|
||||||
) -> Any:
|
) -> Any:
|
||||||
@@ -88,6 +113,7 @@ class MCPManager:
|
|||||||
async def _serve(
|
async def _serve(
|
||||||
self, server: MCPServerDef, ready: asyncio.Future, *, interactive: bool = False
|
self, server: MCPServerDef, ready: asyncio.Future, *, interactive: bool = False
|
||||||
) -> None:
|
) -> None:
|
||||||
|
errfile = None
|
||||||
try:
|
try:
|
||||||
async with AsyncExitStack() as stack:
|
async with AsyncExitStack() as stack:
|
||||||
if server.transport == "http":
|
if server.transport == "http":
|
||||||
@@ -124,18 +150,34 @@ class MCPManager:
|
|||||||
env=server.env or None,
|
env=server.env or None,
|
||||||
cwd=server.cwd,
|
cwd=server.cwd,
|
||||||
)
|
)
|
||||||
read, write = await stack.enter_async_context(stdio_client(params))
|
# Capture the child's stderr so a startup crash leaves evidence
|
||||||
|
# the UI can show (the SDK needs a real file descriptor here).
|
||||||
|
errfile = tempfile.TemporaryFile(
|
||||||
|
mode="w+", encoding="utf-8", errors="replace"
|
||||||
|
)
|
||||||
|
read, write = await stack.enter_async_context(
|
||||||
|
stdio_client(params, errlog=errfile)
|
||||||
|
)
|
||||||
session = await stack.enter_async_context(ClientSession(read, write))
|
session = await stack.enter_async_context(ClientSession(read, write))
|
||||||
await session.initialize()
|
await session.initialize()
|
||||||
listed = await session.list_tools()
|
listed = await session.list_tools()
|
||||||
conn = _Conn(session, list(listed.tools))
|
conn = _Conn(session, list(listed.tools))
|
||||||
|
self._stderr_tails.pop(server.name, None)
|
||||||
if not ready.done():
|
if not ready.done():
|
||||||
ready.set_result(conn)
|
ready.set_result(conn)
|
||||||
await conn.shutdown.wait()
|
await conn.shutdown.wait()
|
||||||
except Exception as exc: # connection / init failure
|
except Exception as exc: # connection / init failure
|
||||||
|
tail = _read_tail(errfile)
|
||||||
|
if tail:
|
||||||
|
self._stderr_tails[server.name] = tail
|
||||||
if not ready.done():
|
if not ready.done():
|
||||||
ready.set_exception(exc)
|
ready.set_exception(exc)
|
||||||
finally:
|
finally:
|
||||||
|
if errfile is not None:
|
||||||
|
try:
|
||||||
|
errfile.close()
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
self._conns.pop(server.name, None)
|
self._conns.pop(server.name, None)
|
||||||
self._tasks.pop(server.name, None)
|
self._tasks.pop(server.name, None)
|
||||||
|
|
||||||
|
|||||||
@@ -1784,6 +1784,16 @@ def create_app(manager: SessionManager) -> FastAPI:
|
|||||||
)
|
)
|
||||||
await ws.close()
|
await ws.close()
|
||||||
return
|
return
|
||||||
|
# MCP servers that failed to start while preparing this session's tools:
|
||||||
|
# leave a quiet, persistent notice instead of the session silently lacking
|
||||||
|
# them (drill 2026-08-20: three silent startup failures in a row).
|
||||||
|
for name, err in manager.pop_mcp_failures(session_id):
|
||||||
|
detail = f": {err}" if err else ""
|
||||||
|
engine._append_notice(
|
||||||
|
"mcp_error",
|
||||||
|
f"MCP server “{name}” failed to start{detail}"[:300]
|
||||||
|
+ " — see Settings ▸ MCP",
|
||||||
|
)
|
||||||
# Auto-compaction failure prompt (OPE-27): only an ATTENDED session may be asked
|
# Auto-compaction failure prompt (OPE-27): only an ATTENDED session may be asked
|
||||||
# Retry/Trim — unattended runs auto-trim (the policy in engine._compact_now).
|
# Retry/Trim — unattended runs auto-trim (the policy in engine._compact_now).
|
||||||
engine.is_attended = lambda: _visibility() == VIS_INLINE
|
engine.is_attended = lambda: _visibility() == VIS_INLINE
|
||||||
|
|||||||
@@ -166,6 +166,9 @@ class SessionManager:
|
|||||||
# feeds list_mcp's status so the GUI can show "authorizing…" and failures.
|
# feeds list_mcp's status so the GUI can show "authorizing…" and failures.
|
||||||
self._mcp_authorizing: set[str] = set()
|
self._mcp_authorizing: set[str] = set()
|
||||||
self._mcp_errors: dict[str, str] = {}
|
self._mcp_errors: dict[str, str] = {}
|
||||||
|
# Servers that failed to connect while preparing a session's tools —
|
||||||
|
# drained once by the WS handler to append a transcript notice.
|
||||||
|
self._mcp_session_failures: dict[str, list[str]] = {}
|
||||||
self.gateway: Optional[Gateway] = None
|
self.gateway: Optional[Gateway] = None
|
||||||
self._data_base = base
|
self._data_base = base
|
||||||
# Desktop/UI prefs (default model, onboarding state) — not secrets; a plain JSON file.
|
# Desktop/UI prefs (default model, onboarding state) — not secrets; a plain JSON file.
|
||||||
@@ -957,6 +960,7 @@ class SessionManager:
|
|||||||
]
|
]
|
||||||
try:
|
try:
|
||||||
conn = await self.mcp.ensure(server)
|
conn = await self.mcp.ensure(server)
|
||||||
|
self._mcp_errors.pop(server.name, None)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
if mcp_oauth.is_auth_required(exc):
|
if mcp_oauth.is_auth_required(exc):
|
||||||
# Stored tokens no longer refresh (vendor rotated/expired
|
# Stored tokens no longer refresh (vendor rotated/expired
|
||||||
@@ -969,7 +973,22 @@ class SessionManager:
|
|||||||
logger.info(
|
logger.info(
|
||||||
"mcp %s needs re-auth; skipped for this session", server.name
|
"mcp %s needs re-auth; skipped for this session", server.name
|
||||||
)
|
)
|
||||||
# else: bad command / unreachable url — skip, don't break the session
|
else:
|
||||||
|
# Bad command / crashed child / unreachable url — the session
|
||||||
|
# still runs without the tools, but the failure must not be
|
||||||
|
# silent (three-for-three silent failures in the 2026-08-20
|
||||||
|
# drill): record it for the MCP page and the session notice.
|
||||||
|
msg = str(exc) or exc.__class__.__name__
|
||||||
|
tail = self.mcp.last_stderr(server.name)
|
||||||
|
if tail:
|
||||||
|
msg = f"{msg} — {tail}"
|
||||||
|
self._mcp_errors[server.name] = msg[:500]
|
||||||
|
logger.warning(
|
||||||
|
"mcp %s failed to connect: %s", server.name, msg[:500]
|
||||||
|
)
|
||||||
|
self._mcp_session_failures.setdefault(session_id, []).append(
|
||||||
|
server.name
|
||||||
|
)
|
||||||
continue
|
continue
|
||||||
callables = build_callables(
|
callables = build_callables(
|
||||||
server,
|
server,
|
||||||
@@ -988,6 +1007,12 @@ class SessionManager:
|
|||||||
out.extend(callables)
|
out.extend(callables)
|
||||||
return out
|
return out
|
||||||
|
|
||||||
|
def pop_mcp_failures(self, session_id: str) -> list[tuple[str, Optional[str]]]:
|
||||||
|
"""Drain (name, error) for servers that failed while preparing this session's
|
||||||
|
tools — consumed once by the WS handler to append a transcript notice."""
|
||||||
|
names = self._mcp_session_failures.pop(session_id, [])
|
||||||
|
return [(n, self._mcp_errors.get(n)) for n in names]
|
||||||
|
|
||||||
def list_mcp(self) -> list[dict[str, Any]]:
|
def list_mcp(self) -> list[dict[str, Any]]:
|
||||||
"""Servers from the global config + connection status (does not connect)."""
|
"""Servers from the global config + connection status (does not connect)."""
|
||||||
from ..mcp import oauth as mcp_oauth
|
from ..mcp import oauth as mcp_oauth
|
||||||
@@ -1011,6 +1036,12 @@ class SessionManager:
|
|||||||
status = "authorizing"
|
status = "authorizing"
|
||||||
elif is_oauth and not mcp_oauth.has_tokens(name, self.secrets):
|
elif is_oauth and not mcp_oauth.has_tokens(name, self.secrets):
|
||||||
status = "needs_auth"
|
status = "needs_auth"
|
||||||
|
elif name in self._mcp_errors and not is_oauth:
|
||||||
|
# Startup/connection failure (stdio crash, unreachable url) — the
|
||||||
|
# drill class. OAuth servers keep their softer statuses: acquiring
|
||||||
|
# tokens supersedes a stale sign-in error (the GUI still prints
|
||||||
|
# last_error under the row either way).
|
||||||
|
status = "error"
|
||||||
else:
|
else:
|
||||||
status = "configured"
|
status = "configured"
|
||||||
out.append(
|
out.append(
|
||||||
|
|||||||
Symlink
+1
@@ -0,0 +1 @@
|
|||||||
|
/Users/rohit/fleet/ro4d/openworker/surfaces/gui/node_modules
|
||||||
@@ -108,3 +108,17 @@ describe("itemsFromMessages reasoning", () => {
|
|||||||
expect(items[2]).toEqual({ kind: "assistant", text: "", reasoning: "stopped mid-thought" });
|
expect(items[2]).toEqual({ kind: "assistant", text: "", reasoning: "stopped mid-thought" });
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("itemsFromMessages mcp failure", () => {
|
||||||
|
it("replays the persisted mcp_error marker as a warn notice WITHOUT retry", () => {
|
||||||
|
const items = itemsFromMessages([
|
||||||
|
{ role: "user", content: "hi" },
|
||||||
|
{ role: "notice", kind: "mcp_error", text: "MCP server “sales-db” failed to start — see Settings ▸ MCP" },
|
||||||
|
] as any);
|
||||||
|
expect(items[1]).toEqual({
|
||||||
|
kind: "notice",
|
||||||
|
tone: "warn",
|
||||||
|
text: "MCP server “sales-db” failed to start — see Settings ▸ MCP",
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|||||||
@@ -78,7 +78,11 @@ export function itemsFromMessages(messages: ConversationMessage[]): Item[] {
|
|||||||
: m.kind === "compacted"
|
: m.kind === "compacted"
|
||||||
? // The subtle "compacted here" divider (OPE-27) — the transcript itself is intact.
|
? // The subtle "compacted here" divider (OPE-27) — the transcript itself is intact.
|
||||||
{ kind: "notice", tone: "info", text: m.text || "Context compacted" }
|
{ kind: "notice", tone: "info", text: m.text || "Context compacted" }
|
||||||
: { kind: "notice", tone: "warn", text: "Error: " + (m.text || "unknown"), retriable: true },
|
: m.kind === "mcp_error"
|
||||||
|
? // A configured MCP server failed to start for this session — informational,
|
||||||
|
// NOT retriable (retry re-runs the model turn, which can't fix a dead server).
|
||||||
|
{ kind: "notice", tone: "warn", text: m.text || "An MCP server failed to start" }
|
||||||
|
: { kind: "notice", tone: "warn", text: "Error: " + (m.text || "unknown"), retriable: true },
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
// system messages are omitted; tool-result messages are folded into the tool row above
|
// system messages are omitted; tool-result messages are folded into the tool row above
|
||||||
|
|||||||
@@ -254,3 +254,61 @@ def test_rest_crud(tmp_path, monkeypatch):
|
|||||||
assert client.delete("/v1/mcp/fs").json()["ok"] is True
|
assert client.delete("/v1/mcp/fs").json()["ok"] is True
|
||||||
assert client.get("/v1/mcp").json()["servers"] == []
|
assert client.get("/v1/mcp").json()["servers"] == []
|
||||||
assert client.delete("/v1/mcp/fs").json()["ok"] is False
|
assert client.delete("/v1/mcp/fs").json()["ok"] is False
|
||||||
|
|
||||||
|
|
||||||
|
# -- failure surfacing (drill 2026-08-20: silent startup crashes) ----------------
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_stdio_startup_crash_captures_stderr_tail(tmp_path, monkeypatch):
|
||||||
|
"""A stdio server that dies before initialize leaves its stderr tail behind."""
|
||||||
|
from coworker.mcp.client import MCPManager
|
||||||
|
|
||||||
|
mgr = MCPManager()
|
||||||
|
server = MCPServerDef(
|
||||||
|
name="doomed",
|
||||||
|
transport="stdio",
|
||||||
|
command="/bin/sh",
|
||||||
|
args=["-c", "echo 'usage: doomed --flag' >&2; exit 7"],
|
||||||
|
)
|
||||||
|
with pytest.raises(Exception):
|
||||||
|
await mgr.ensure(server)
|
||||||
|
tail = mgr.last_stderr("doomed")
|
||||||
|
assert tail is not None and "usage: doomed --flag" in tail
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_prepare_records_failure_status_and_session_notice(
|
||||||
|
tmp_path, monkeypatch
|
||||||
|
):
|
||||||
|
"""A crashing global server surfaces: last_error + status=error + one-shot
|
||||||
|
session failure drain — instead of the pre-drill silent skip."""
|
||||||
|
monkeypatch.setenv("COWORKER_STATE_DIR", str(tmp_path / "state"))
|
||||||
|
_write_json(
|
||||||
|
tmp_path / "state" / "mcp.json",
|
||||||
|
{
|
||||||
|
"mcpServers": {
|
||||||
|
"sales-db": {
|
||||||
|
"command": "/bin/sh",
|
||||||
|
"args": ["-c", "echo 'boom: bad args' >&2; exit 2"],
|
||||||
|
"enabled": True,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
)
|
||||||
|
manager = SessionManager(data_dir=tmp_path / "data")
|
||||||
|
|
||||||
|
tools = await manager.prepare_mcp_tools("s1", workspace=str(tmp_path / "wsp"))
|
||||||
|
assert tools == []
|
||||||
|
|
||||||
|
err = manager._mcp_errors.get("sales-db")
|
||||||
|
assert err and "boom: bad args" in err
|
||||||
|
|
||||||
|
listed = {s["name"]: s for s in manager.list_mcp()}
|
||||||
|
assert listed["sales-db"]["status"] == "error"
|
||||||
|
assert "boom: bad args" in (listed["sales-db"]["last_error"] or "")
|
||||||
|
|
||||||
|
drained = manager.pop_mcp_failures("s1")
|
||||||
|
assert [n for n, _ in drained] == ["sales-db"]
|
||||||
|
assert "boom: bad args" in (drained[0][1] or "")
|
||||||
|
assert manager.pop_mcp_failures("s1") == [] # one-shot
|
||||||
|
|||||||
Reference in New Issue
Block a user