vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-12 18:46:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(agents): fail fast with attributable reason after MCP stdio session dies mid-run (#98738)

Browse Source 
* fix(agents): fail fast with attributable reason after MCP stdio session dies mid-run

Wires MCP Client onclose/onerror during bundle-mcp session creation so a
crashed/exited server flips session.connected instead of staying stale.
Next tool/resource/prompt call throws a domain-specific 'is disconnected'
error immediately instead of surfacing the SDK's generic 'Not connected'.
A disconnected reused session is retired and rebuilt fresh on the next
catalog pass rather than reused, since the SDK chains onclose/onerror
cumulatively on repeat connect() and the stdio transport never clears its
read buffer on an unexpected exit.

* fix(agents): retire a reused MCP session that dies mid-refresh, not just pre-refresh

codex review found: the catch-path retirement in getCatalog()'s per-server
task only covered two cases (fresh session that never connected this pass,
and a non-reused session that failed for any reason) - a reused session
that was healthy when this pass started but disconnects mid-refresh (child
process dies between ensureSessionConnected() returning and
listAllToolsBestEffort() finishing) hit neither branch, so onclose flipped
connected=false but the dead session object stayed in the map until the
next catalog rebuild happened to notice it. Added the missing branch plus
a regression test that kills the child process mid tools/list on a reused
session and asserts the session is purged from the map within the same
refresh (not just fail-fast on tool calls, which already worked).

* fix(agents): retire a dead reused MCP session even across overlapping catalog generations

codex round-2 review found: the mid-refresh retirement branch still
skipped when sharedWithNewerGeneration was true, which protects a
still-alive session another generation is actively using - but once
onclose flips session.connected=false, the transport is dead for every
generation sharing that object, so that guard no longer applies.
Simplified to a single !session.connected branch that always retires
(retireSessionIfCurrent already no-ops safely if a newer generation
replaced the map entry), and dropped the now-write-only
connectedForCatalog local it replaced.

* test(scripts): allow MCP SDK callback suppressions

* fix(agents): retire closed MCP sessions

---------

Co-authored-by: Vincent Koc <vincentkoc@ieee.org>
This commit is contained in:
Masato Hoshino
2026-07-02 14:55:15 +09:00
committed by GitHub
parent ebb43e2dca
commit e845a26fd6
3 changed files with 171 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
src/agents/agent-bundle-mcp-runtime.test.ts
View File

@@ -46,7 +46,10 @@ async function writeListToolsMcpServer(params: {
inputSchema?: unknown;
tools?: Array<{ name: string; description?: string; inputSchema?: unknown }>;
capabilities?: Record<string, unknown>;
pidPath?: string;
notifyListChangedOnInitialized?: boolean;
notifyListChangedAfterFirstList?: boolean;
exitOnListCall?: number;
listToolsMethodNotFound?: boolean;
callToolIsError?: boolean;
callToolJsonRpcError?: boolean;
@@ -62,7 +65,10 @@ const delayMs = ${params.delayMs ?? 0};
const initializeDelayMs = ${params.initializeDelayMs ?? 0};
const hang = ${params.hang === true};
const capabilities = ${JSON.stringify(params.capabilities ?? { tools: {} })};
const pidPath = ${JSON.stringify(params.pidPath)};
const notifyListChangedOnInitialized = ${params.notifyListChangedOnInitialized === true};
const notifyListChangedAfterFirstList = ${params.notifyListChangedAfterFirstList === true};
const exitOnListCall = ${params.exitOnListCall ?? 0};
const listToolsMethodNotFound = ${params.listToolsMethodNotFound === true};
const tools = ${JSON.stringify(
params.tools ?? [
@@ -78,8 +84,12 @@ const callToolJsonRpcError = ${params.callToolJsonRpcError === true};
const resourceListJsonRpcError = ${params.resourceListJsonRpcError === true};
let buffer = "";
let listCount = 0;
let pendingTimer;
let keepAlive;
if (pidPath) {
await fs.writeFile(pidPath, String(process.pid), "utf8");
}
function log(line) {
void fs.appendFile(logPath, line + "\\n", "utf8").catch(() => {});
}
@@ -116,6 +126,11 @@ function handle(message) {
return;
}
if (message.method === "tools/list") {
listCount += 1;
if (listCount === exitOnListCall) {
log("exit tools/list " + listCount);
process.exit(1);
}
if (listToolsMethodNotFound) {
log("reject tools/list method not found");
send({
@@ -130,6 +145,7 @@ function handle(message) {
keepAlive = setInterval(() => {}, 1000);
return;
}
const currentListCount = listCount;
log("delay tools/list " + delayMs);
pendingTimer = setTimeout(() => {
send({
@@ -139,6 +155,10 @@ function handle(message) {
tools,
},
});
if (notifyListChangedAfterFirstList && currentListCount === 1) {
log("notify tools/list_changed");
send({ jsonrpc: "2.0", method: "notifications/tools/list_changed" });
}
}, delayMs);
}
if (message.method === "tools/call") {
@@ -247,6 +267,29 @@ async function waitForPredicate(
throw new Error(`Timed out waiting for ${description}`);
}
async function waitForErrorMessage(
action: () => Promise<unknown>,
expectedText: string,
timeoutMs: number,
): Promise<string> {
const deadline = Date.now() + timeoutMs;
let lastMessage = "";
while (Date.now() < deadline) {
try {
await action();
} catch (error) {
lastMessage = error instanceof Error ? error.message : String(error);
if (lastMessage.includes(expectedText)) {
return lastMessage;
}
}
await new Promise((resolve) => {
setTimeout(resolve, 10);
});
}
throw new Error(`Timed out waiting for ${expectedText}; saw ${JSON.stringify(lastMessage)}`);
}
function makeRuntime(
tools: Array<{ toolName: string; description: string }>,
serverName = "bundleProbe",
@@ -1036,6 +1079,92 @@ process.on("SIGINT", shutdown);`,
}
});
it("fails fast with an attributable error after an MCP child process exits", async () => {
const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "bundle-mcp-child-exit-"));
const serverPath = path.join(tempDir, "server.mjs");
const logPath = path.join(tempDir, "server.log");
const pidPath = path.join(tempDir, "server.pid");
await writeListToolsMcpServer({ filePath: serverPath, logPath, pidPath });
const runtime = await getOrCreateSessionMcpRuntime({
sessionId: "session-child-exit",
sessionKey: "agent:test:session-child-exit",
workspaceDir: "/workspace",
cfg: {
mcp: {
servers: {
child: { command: process.execPath, args: [serverPath] },
},
},
},
});
try {
await expect(runtime.callTool("child", "slow_tool", {})).resolves.toMatchObject({
isError: false,
});
await waitForFileText(pidPath, "", LIST_TOOLS_SERVER_LOG_TIMEOUT_MS);
const pid = Number.parseInt((await fs.readFile(pidPath, "utf8")).trim(), 10);
process.kill(pid);
const message = await waitForErrorMessage(
() => runtime.callTool("child", "slow_tool", {}),
"is disconnected",
LIST_TOOLS_SERVER_LOG_TIMEOUT_MS,
);
expect(message).toBe('bundle-mcp server "child" is disconnected: mcp transport closed');
} finally {
await runtime.dispose();
await fs.rm(tempDir, { recursive: true, force: true });
}
});
it("retires a reused MCP session that exits during catalog refresh", async () => {
const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "bundle-mcp-refresh-exit-"));
const serverPath = path.join(tempDir, "server.mjs");
const logPath = path.join(tempDir, "server.log");
await writeListToolsMcpServer({
filePath: serverPath,
logPath,
capabilities: { tools: { listChanged: true } },
notifyListChangedAfterFirstList: true,
exitOnListCall: 2,
});
const runtime = await getOrCreateSessionMcpRuntime({
sessionId: "session-refresh-exit",
sessionKey: "agent:test:session-refresh-exit",
workspaceDir: "/workspace",
cfg: {
mcp: {
servers: {
child: { command: process.execPath, args: [serverPath] },
},
},
},
});
try {
expect((await runtime.getCatalog()).tools).toHaveLength(1);
await waitForFileText(logPath, "notify tools/list_changed", LIST_TOOLS_SERVER_LOG_TIMEOUT_MS);
await waitForPredicate(
() => runtime.peekCatalog() === null,
"list_changed to invalidate the catalog",
LIST_TOOLS_SERVER_LOG_TIMEOUT_MS,
);
const refreshedCatalog = await runtime.getCatalog();
expect(refreshedCatalog.tools).toEqual([]);
expect(refreshedCatalog.diagnostics?.[0]?.serverName).toBe("child");
await expect(runtime.callTool("child", "slow_tool", {})).rejects.toThrow(
'bundle-mcp server "child" is not connected',
);
} finally {
await runtime.dispose();
await fs.rm(tempDir, { recursive: true, force: true });
}
});
it("does not cache a catalog invalidated while discovery is in flight", async () => {
const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "bundle-mcp-inflight-invalidated-"));
const serverPath = path.join(tempDir, "inflight-invalidated.mjs");

67
src/agents/agent-bundle-mcp-runtime.ts
View File

@@ -45,6 +45,7 @@ type BundleMcpSession = {
requestTimeoutMs: number;
supportsParallelToolCalls: boolean;
connected: boolean;
disconnectReason?: string;
retiring: boolean;
catalogUseCount: number;
sharedAcrossCatalogGenerations: boolean;
@@ -542,6 +543,17 @@ export function createSessionMcpRuntime(params: {
throw createDisposedError(params.sessionId);
}
};
const requireConnectedSession = (serverName: string): BundleMcpSession => {
const session = sessions.get(serverName);
if (!session || !session.connected) {
throw new Error(
session?.disconnectReason
? `bundle-mcp server "${serverName}" is disconnected: ${session.disconnectReason}`
: `bundle-mcp server "${serverName}" is not connected`,
);
}
return session;
};
const ensureSessionConnected = async (
session: BundleMcpSession,
connectionTimeoutMs: number,
@@ -640,6 +652,18 @@ export function createSessionMcpRuntime(params: {
failIfDisposed();
let session = sessions.get(serverName);
while (
session &&
!session.retiring &&
!session.connected &&
!session.connectPromise
) {
// A closed SDK client cannot reconnect cleanly on the same transport.
await retireSessionIfCurrent(serverName, session);
// Retirement yields while closing. Preserve any replacement that a
// newer catalog generation installed during that await.
session = sessions.get(serverName);
}
if (session?.retiring) {
session = undefined;
}
@@ -670,7 +694,7 @@ export function createSessionMcpRuntime(params: {
},
},
);
session = {
const createdSession: BundleMcpSession = {
serverName,
client,
transport: resolved.transport,
@@ -683,6 +707,14 @@ export function createSessionMcpRuntime(params: {
sharedAcrossCatalogGenerations: false,
detachStderr: resolved.detachStderr,
};
// The SDK exposes lifecycle hooks as callback properties. A close is
// terminal for this client/transport pair.
// oxlint-disable-next-line unicorn/prefer-add-event-listener -- MCP Client is not an EventTarget.
client.onclose = () => {
createdSession.connected = false;
createdSession.disconnectReason = "mcp transport closed";
};
session = createdSession;
sessions.set(serverName, session);
}
@@ -693,11 +725,9 @@ export function createSessionMcpRuntime(params: {
session.sharedAcrossCatalogGenerations = true;
}
session.catalogUseCount += 1;
let connectedForCatalog = false;
try {
failIfDisposed();
await ensureSessionConnected(session, resolved.connectionTimeoutMs);
connectedForCatalog = true;
failIfDisposed();
const capabilities = summarizeServerCapabilities(
session.client.getServerCapabilities(),
@@ -782,9 +812,9 @@ export function createSessionMcpRuntime(params: {
];
const sharedWithNewerGeneration =
session.sharedAcrossCatalogGenerations || session.catalogUseCount > 1;
if (!connectedForCatalog && !session.connected) {
// Timed-out connects can still leave the SDK client bound to a
// transport. Delete before async close so future catalogs start fresh.
if (!session.connected) {
// A close is terminal for every catalog generation sharing this
// session. The identity guard preserves any newer replacement.
await retireSessionIfCurrent(serverName, session);
} else if (!reusedSession && !sharedWithNewerGeneration) {
// Catalog invalidation can overlap generations; an older failed
@@ -894,10 +924,7 @@ export function createSessionMcpRuntime(params: {
async callTool(serverName, toolName, input) {
failIfDisposed();
await getCatalog();
const session = sessions.get(serverName);
if (!session) {
throw new Error(`bundle-mcp server "${serverName}" is not connected`);
}
const session = requireConnectedSession(serverName);
return await runGuardedServerRequest(
serverName,
async () =>
@@ -914,10 +941,7 @@ export function createSessionMcpRuntime(params: {
async listResources(serverName) {
failIfDisposed();
await getCatalog();
const session = sessions.get(serverName);
if (!session) {
throw new Error(`bundle-mcp server "${serverName}" is not connected`);
}
const session = requireConnectedSession(serverName);
return await runGuardedServerRequest(serverName, async () =>
listAllResources(session.client, session.requestTimeoutMs),
);
@@ -925,10 +949,7 @@ export function createSessionMcpRuntime(params: {
async readResource(serverName, uri) {
failIfDisposed();
await getCatalog();
const session = sessions.get(serverName);
if (!session) {
throw new Error(`bundle-mcp server "${serverName}" is not connected`);
}
const session = requireConnectedSession(serverName);
return await runGuardedServerRequest(
serverName,
async () =>
@@ -938,10 +959,7 @@ export function createSessionMcpRuntime(params: {
async listPrompts(serverName) {
failIfDisposed();
await getCatalog();
const session = sessions.get(serverName);
if (!session) {
throw new Error(`bundle-mcp server "${serverName}" is not connected`);
}
const session = requireConnectedSession(serverName);
return await runGuardedServerRequest(serverName, async () =>
listAllPrompts(session.client, session.requestTimeoutMs),
);
@@ -949,10 +967,7 @@ export function createSessionMcpRuntime(params: {
async getPrompt(serverName, name, args) {
failIfDisposed();
await getCatalog();
const session = sessions.get(serverName);
if (!session) {
throw new Error(`bundle-mcp server "${serverName}" is not connected`);
}
const session = requireConnectedSession(serverName);
return await runGuardedServerRequest(
serverName,
async () =>

1
test/scripts/lint-suppressions.test.ts
View File

@@ -193,6 +193,7 @@ describe("production lint suppressions", () => {
"extensions/feishu/src/bitable.ts|typescript/no-unnecessary-type-parameters|1",
"extensions/matrix/src/onboarding.test-harness.ts|typescript/no-unnecessary-type-parameters|1",
"extensions/slack/src/monitor/provider-support.ts|typescript/no-unnecessary-type-parameters|1",
"src/agents/agent-bundle-mcp-runtime.ts|unicorn/prefer-add-event-listener|1",
"src/channels/plugins/channel-runtime-surface.types.ts|typescript/no-unnecessary-type-parameters|1",
"src/channels/plugins/contracts/test-helpers.ts|typescript/no-unnecessary-type-parameters|1",
"src/channels/plugins/types.plugin.ts|typescript/no-explicit-any|1",