Merge PR #6437: fix(acp): pick up MCP tools that connect after the startup race (@barisdemirdelen)
This commit is contained in:
@@ -51,6 +51,7 @@
|
||||
- Fixed Plan Review annotations being discarded on dismissal and limited to headings; review notes now persist per plan, feed Refine, and can target the top visible plan line.
|
||||
- Fixed the approved plan reference being permanently suppressed when the first post-approval prompt bailed during setup: `#planReferenceSent` was set at message-construction time (before delivery), so a generation-bail, an `@`-mention read error, or a `before_agent_start` hook throw between build and `agent.prompt` left the flag set with nothing delivered and the retry skipped re-injection. The flag is now committed only when the plan-reference message is handed to `agent.prompt` ([#4094](https://github.com/can1357/oh-my-pi/issues/4094)).
|
||||
- Fixed `/new`, handoffs, branching (`/branch`, `/btw`), and cross-session switches retaining staged preview resolve callbacks and per-session ACP `allow_always`/`reject_always` decisions from the previous session ([#4093](https://github.com/can1357/oh-my-pi/issues/4093)).
|
||||
- Fixed ACP sessions silently dropping tools from MCP servers that finished connecting after `MCPManager`'s 250ms startup race window; the initial post-connect refresh and every later `onToolsChanged` follow-up now run through a single ordered queue, so late-arriving tools are still applied to the session.
|
||||
- Fixed `error.notify` raising a "Stopped with error" toast for provider failures while an auto-retry or async-delivery continuation was pending; the toast now waits for the true terminal settle.
|
||||
- Fixed concurrent MCP config mutations losing updates and racing on a shared temp path: every `mcp.json` read-modify-write (add/update/remove server, disabled/force-enabled lists) is now serialized under a per-file lock, and each atomic write uses a unique temp file so overlapping writers no longer rename each other's `.tmp` out from under them (ENOENT or clobbered config) — reachable in-process via the fire-and-forget extensions-dashboard toggle and across processes on a shared `~/.omp/mcp.json` ([#4104](https://github.com/can1357/oh-my-pi/issues/4104)).
|
||||
- Fixed transient provider stream stalls after tool calls failing to auto-retry even when every call already had a tool result, including synthetic `executed:false` results from OpenAI-completions stalls ([#6414](https://github.com/can1357/oh-my-pi/issues/6414)).
|
||||
|
||||
@@ -2380,6 +2380,28 @@ export class AcpAgent implements Agent {
|
||||
}
|
||||
|
||||
const manager = new MCPManager(record.session.sessionManager.getCwd());
|
||||
// MCP servers connect and reconnect independently, so `onToolsChanged` can fire
|
||||
// several times back to back. Each firing is chained onto `refreshChain` so
|
||||
// refreshes apply in order, and each one re-reads `manager.getTools()` at the
|
||||
// time it actually runs rather than the snapshot from when it was queued — so a
|
||||
// refresh can never apply a stale, smaller tool set after a newer one already landed.
|
||||
let refreshChain: Promise<void> = Promise.resolve();
|
||||
const enqueueMcpToolsRefresh = (): Promise<void> => {
|
||||
refreshChain = refreshChain.then(async () => {
|
||||
if (record.mcpManager !== manager) return;
|
||||
try {
|
||||
await record.session.refreshMCPTools(manager.getTools());
|
||||
} catch (error) {
|
||||
logger.warn("ACP MCP tool refresh failed", {
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
}
|
||||
});
|
||||
return refreshChain;
|
||||
};
|
||||
manager.setOnToolsChanged(() => {
|
||||
void enqueueMcpToolsRefresh();
|
||||
});
|
||||
const configs: MCPConfigMap = {};
|
||||
const sources: MCPSourceMap = {};
|
||||
for (const server of servers) {
|
||||
@@ -2402,7 +2424,7 @@ export class AcpAgent implements Agent {
|
||||
}
|
||||
|
||||
record.mcpManager = manager;
|
||||
await record.session.refreshMCPTools(result.tools);
|
||||
await enqueueMcpToolsRefresh();
|
||||
}
|
||||
|
||||
#toMcpConfig(server: McpServer): MCPServerConfig {
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { afterEach, describe, expect, it } from "bun:test";
|
||||
import { afterEach, describe, expect, it, spyOn } from "bun:test";
|
||||
import * as fs from "node:fs";
|
||||
import * as os from "node:os";
|
||||
import * as path from "node:path";
|
||||
@@ -39,6 +39,7 @@ import {
|
||||
} from "@oh-my-pi/pi-coding-agent/tts/models";
|
||||
import { getConfigRootDir, setAgentDir } from "@oh-my-pi/pi-utils";
|
||||
import type { z } from "zod/v4";
|
||||
import { TOOL_NAME as DELAYED_MCP_TOOL_NAME } from "./fixtures/delayed-tool-mcp";
|
||||
|
||||
/**
|
||||
* Validate an ACP wire payload against the external `@agentclientprotocol/sdk`
|
||||
@@ -2517,3 +2518,56 @@ describe("ACP agent", () => {
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("ACP agent MCP server configuration (late-connecting servers)", () => {
|
||||
const FIXTURE_PATH = path.join(import.meta.dir, "fixtures", "delayed-tool-mcp.ts");
|
||||
const BUN_EXEC = process.execPath;
|
||||
|
||||
// Real polling, not fake timers: the fixture is a genuine child process
|
||||
// racing MCPManager's own `Bun.sleep`-based 250ms startup window, and a
|
||||
// subprocess's timers cannot be advanced from this test's fake-timer clock.
|
||||
async function pollUntil(predicate: () => boolean, timeoutMs = 3_000): Promise<void> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (!predicate()) {
|
||||
if (Date.now() >= deadline) throw new Error("pollUntil timed out");
|
||||
await Bun.sleep(5);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Regression test: an MCP server that finishes connecting after
|
||||
* `MCPManager`'s 250ms startup race window used to have its tools
|
||||
* silently discarded — `#configureMcpServers` only called
|
||||
* `session.refreshMCPTools` once, synchronously, with whatever
|
||||
* `connectServers` returned inside the race window. The background
|
||||
* `onToolsChanged` -> `refreshMCPTools` follow-up now runs through a
|
||||
* `refreshChain` queue so late connections still land in the session.
|
||||
*/
|
||||
it("delivers a late-connecting server's tools via a queued refreshMCPTools call", async () => {
|
||||
const harness = await createHarness();
|
||||
const refreshSpy = spyOn(FakeAgentSession.prototype, "refreshMCPTools");
|
||||
const namesOf = (tools: unknown[]) => (tools as Array<{ name: string }>).map(tool => tool.name);
|
||||
|
||||
try {
|
||||
const created = await harness.agent.newSession({
|
||||
cwd: harness.cwdA,
|
||||
mcpServers: [{ name: "delayed", command: BUN_EXEC, args: [FIXTURE_PATH], env: [] }],
|
||||
});
|
||||
expectAcpStructure(zNewSessionResponse, created);
|
||||
|
||||
// The fixture delays its `initialize` response past the 250ms startup
|
||||
// race, so the first (synchronous) refresh inside `#configureMcpServers`
|
||||
// must see no tools yet.
|
||||
expect(refreshSpy.mock.calls).toHaveLength(1);
|
||||
expect(namesOf(refreshSpy.mock.calls[0]?.[0] ?? [])).toEqual([]);
|
||||
|
||||
// Once the delayed `initialize` response lands, the background
|
||||
// `onToolsChanged` -> queued `refreshMCPTools` call must deliver the
|
||||
// server's tool. Before the fix, this late arrival was dropped.
|
||||
await pollUntil(() => refreshSpy.mock.calls.length > 1);
|
||||
expect(namesOf(refreshSpy.mock.calls.at(-1)?.[0] ?? [])).toEqual([`mcp__delayed_${DELAYED_MCP_TOOL_NAME}`]);
|
||||
} finally {
|
||||
refreshSpy.mockRestore();
|
||||
}
|
||||
}, 15_000);
|
||||
});
|
||||
|
||||
+80
@@ -0,0 +1,80 @@
|
||||
#!/usr/bin/env bun
|
||||
/**
|
||||
* Test fixture: a well-behaved stdio MCP server that answers `initialize`
|
||||
* only after a deliberate delay exceeding `MCPManager`'s `STARTUP_TIMEOUT_MS`
|
||||
* (250 ms), then responds normally to `tools/list`.
|
||||
*
|
||||
* Models a server that eventually connects successfully but not within the
|
||||
* manager's startup race window (`Promise.race([Promise.allSettled(...),
|
||||
* delay(STARTUP_TIMEOUT_MS)])`), so its tools only land via the background
|
||||
* `#onToolsChanged` path instead of the initial `connectServers` result.
|
||||
*
|
||||
* Speaks newline-delimited JSON-RPC 2.0 (the wire format of `StdioTransport`):
|
||||
* one JSON object per line on stdin, one JSON response per line on stdout.
|
||||
* Only requests (objects with an `id`) get a response; notifications
|
||||
* (including `notifications/initialized`) are dropped.
|
||||
*/
|
||||
import * as readline from "node:readline";
|
||||
|
||||
export const TOOL_NAME = "late_tool";
|
||||
export const TOOL_RESULT = "MCP_LATE_CONNECT_OK_9d21";
|
||||
export const INITIALIZE_DELAY_MS = 450;
|
||||
|
||||
type JsonRpcRequest = {
|
||||
jsonrpc: "2.0";
|
||||
id?: string | number;
|
||||
method: string;
|
||||
params?: Record<string, unknown>;
|
||||
};
|
||||
|
||||
function buildResult(method: string): Record<string, unknown> {
|
||||
switch (method) {
|
||||
case "initialize":
|
||||
return {
|
||||
protocolVersion: "2025-03-26",
|
||||
serverInfo: { name: "delayed-tool-fixture", version: "1.0.0" },
|
||||
capabilities: { tools: {} },
|
||||
};
|
||||
case "tools/list":
|
||||
return {
|
||||
tools: [
|
||||
{
|
||||
name: TOOL_NAME,
|
||||
description: "Fixture tool that only arrives after the startup race window.",
|
||||
inputSchema: { type: "object", properties: {}, additionalProperties: false },
|
||||
},
|
||||
],
|
||||
};
|
||||
case "tools/call":
|
||||
return { content: [{ type: "text", text: TOOL_RESULT }], isError: false };
|
||||
default:
|
||||
return {};
|
||||
}
|
||||
}
|
||||
|
||||
function startServer(): void {
|
||||
const rl = readline.createInterface({ input: process.stdin });
|
||||
rl.on("line", line => {
|
||||
void (async () => {
|
||||
const trimmed = line.trim();
|
||||
if (trimmed.length === 0) return;
|
||||
let msg: JsonRpcRequest;
|
||||
try {
|
||||
msg = JSON.parse(trimmed) as JsonRpcRequest;
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
if (msg.id === undefined || msg.id === null) return;
|
||||
if (msg.method === "initialize") {
|
||||
await Bun.sleep(INITIALIZE_DELAY_MS);
|
||||
}
|
||||
const response = { jsonrpc: "2.0" as const, id: msg.id, result: buildResult(msg.method) };
|
||||
process.stdout.write(`${JSON.stringify(response)}\n`);
|
||||
})();
|
||||
});
|
||||
rl.on("close", () => process.exit(0));
|
||||
}
|
||||
|
||||
if (import.meta.main) {
|
||||
startServer();
|
||||
}
|
||||
Reference in New Issue
Block a user