fix: retire uncertain MCP names so late OpenCode adds cannot replace live connections
This commit is contained in:
@@ -2,7 +2,7 @@
|
||||
import { existsSync } from 'node:fs'
|
||||
import { createHash } from 'node:crypto'
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { createServer } from 'node:http'
|
||||
import { createServer, request as httpRequest } from 'node:http'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { createOpencodeClient } from '@opencode-ai/sdk/v2'
|
||||
@@ -91,7 +91,8 @@ it.skipIf(!existsSync(binaryPath))(
|
||||
expect(modelCalls).toBe(1)
|
||||
expect(operations).toEqual(['add-connected', 'disconnect-succeeded', 'add-connected', 'disconnect-succeeded'])
|
||||
expect(registrations).toHaveLength(2)
|
||||
expect(registrations[1]).toEqual(registrations[0])
|
||||
expect(registrations[1]?.name).not.toBe(registrations[0]?.name)
|
||||
expect(registrations[1]?.config).toEqual(registrations[0]?.config)
|
||||
expect(registrations[0]?.config.headers.Authorization).toBe(`Bearer ${token}`)
|
||||
expect(records).toHaveLength(1)
|
||||
expect(records[0]?.mcp).toMatchObject({
|
||||
@@ -115,6 +116,143 @@ it.skipIf(!existsSync(binaryPath))(
|
||||
60_000
|
||||
)
|
||||
|
||||
it.skipIf(!existsSync(binaryPath))(
|
||||
'keeps a later MCP connection when a timed-out OpenCode add settles afterwards',
|
||||
async () => {
|
||||
const workspace = await mkdtemp(join(tmpdir(), 'goodbuddy-opencode-mcp-late-add-'))
|
||||
let releaseModel!: () => void
|
||||
const modelReleased = new Promise<void>(resolve => { releaseModel = resolve })
|
||||
let modelStarted!: () => void
|
||||
const modelPending = new Promise<void>(resolve => { modelStarted = resolve })
|
||||
const model = createServer((_request, response) => {
|
||||
modelStarted()
|
||||
void modelReleased.then(() => {
|
||||
response.writeHead(200, { 'content-type': 'text/event-stream' })
|
||||
response.end(`data: ${JSON.stringify({
|
||||
id: 'late', object: 'chat.completion.chunk', created: 1, model: 'late',
|
||||
choices: [{ index: 0, delta: { role: 'assistant', content: 'LATE_OK' }, finish_reason: 'stop' }]
|
||||
})}\n\ndata: [DONE]\n\n`)
|
||||
})
|
||||
})
|
||||
await new Promise<void>(resolve => model.listen(0, '127.0.0.1', resolve))
|
||||
const modelAddress = model.address()
|
||||
if (!modelAddress || typeof modelAddress === 'string') throw new Error('No model port')
|
||||
const gateway = new KnowledgeMcpGateway({} as never)
|
||||
await gateway.start()
|
||||
const upstream = new URL(gateway.getEndpoint()!)
|
||||
// Holds the first MCP initialize so OpenCode keeps initializing after the client gives up.
|
||||
let releaseInitialize!: () => void
|
||||
const initializeReleased = new Promise<void>(resolve => { releaseInitialize = resolve })
|
||||
let heldInitializeSettled!: () => void
|
||||
const heldInitializeDone = new Promise<void>(resolve => { heldInitializeSettled = resolve })
|
||||
let heldInitialize = false
|
||||
const proxy = createServer((request, response) => {
|
||||
void (async () => {
|
||||
const chunks: Buffer[] = []
|
||||
for await (const chunk of request) chunks.push(chunk as Buffer)
|
||||
const body = Buffer.concat(chunks)
|
||||
const hold = !heldInitialize && request.method === 'POST' && body.toString().includes('"initialize"')
|
||||
if (hold) {
|
||||
heldInitialize = true
|
||||
await initializeReleased
|
||||
}
|
||||
const forwarded = httpRequest({
|
||||
host: upstream.hostname, port: upstream.port, path: request.url,
|
||||
method: request.method, headers: request.headers
|
||||
}, upstreamResponse => {
|
||||
response.writeHead(upstreamResponse.statusCode ?? 502, upstreamResponse.headers)
|
||||
upstreamResponse.pipe(response)
|
||||
if (hold) upstreamResponse.on('end', heldInitializeSettled)
|
||||
})
|
||||
forwarded.on('error', () => response.destroy())
|
||||
response.on('close', () => forwarded.destroy())
|
||||
forwarded.end(body)
|
||||
})().catch(() => response.destroy())
|
||||
})
|
||||
await new Promise<void>(resolve => proxy.listen(0, '127.0.0.1', resolve))
|
||||
const proxyAddress = proxy.address()
|
||||
if (!proxyAddress || typeof proxyAddress === 'string') throw new Error('No proxy port')
|
||||
const proxiedGateway = new Proxy(gateway, {
|
||||
get: (target, key, receiver) => key === 'getEndpoint'
|
||||
? () => `http://127.0.0.1:${proxyAddress.port}/mcp`
|
||||
: Reflect.get(target, key, receiver)
|
||||
})
|
||||
let client: ReturnType<typeof createOpencodeClient> | undefined
|
||||
const added: string[] = []
|
||||
const runtime = new OpenCodeRuntime({
|
||||
embedded: true, binaryPath: '', bundledBinaryPath: binaryPath,
|
||||
configPath: '', defaultWorkspace: workspace, knowledgeGateway: proxiedGateway,
|
||||
modelProfile: {
|
||||
id: crypto.randomUUID(), name: 'Late add fixture', modelName: 'late',
|
||||
baseUrl: `http://127.0.0.1:${modelAddress.port}/v1`,
|
||||
protocol: 'openai-chat-completions', authentication: 'none'
|
||||
}
|
||||
}, {
|
||||
controlRequestTimeoutMs: 3_000,
|
||||
createClient: options => {
|
||||
client = createOpencodeClient({
|
||||
...options,
|
||||
fetch: async (input, init) => {
|
||||
const request = input instanceof Request ? input : new Request(input, init)
|
||||
if (new URL(request.url).pathname === '/mcp' && request.method === 'POST') {
|
||||
added.push((await request.clone().json() as { name: string }).name)
|
||||
}
|
||||
return fetch(request, init)
|
||||
}
|
||||
})
|
||||
return client
|
||||
}
|
||||
})
|
||||
const conversationId = crypto.randomUUID()
|
||||
const run = async (): Promise<string> => {
|
||||
const controller = new AbortController()
|
||||
const token = gateway.grant(crypto.randomUUID(), [crypto.randomUUID()], controller.signal)!
|
||||
const deadline = setTimeout(() => controller.abort(), 30_000)
|
||||
let text = ''
|
||||
try {
|
||||
for await (const event of runtime.run({
|
||||
requestId: crypto.randomUUID(), conversationId, workMode: 'ask',
|
||||
prompt: 'Reply LATE_OK without tools.', knowledgeCapabilityToken: token
|
||||
}, controller.signal)) {
|
||||
if (event.type === 'text') text += event.delta
|
||||
}
|
||||
return text
|
||||
} finally {
|
||||
clearTimeout(deadline)
|
||||
gateway.revoke(token)
|
||||
}
|
||||
}
|
||||
try {
|
||||
await expect(run()).rejects.toThrow('超时')
|
||||
expect(added).toHaveLength(1)
|
||||
const second = run()
|
||||
await modelPending
|
||||
expect(added).toHaveLength(2)
|
||||
expect((await client!.mcp.status({ directory: workspace })).data?.[added[1]!]?.status).toBe('connected')
|
||||
// The late add now finishes with a revoked token while the second request is live.
|
||||
releaseInitialize()
|
||||
await heldInitializeDone
|
||||
await new Promise(resolve => setTimeout(resolve, 500))
|
||||
const status = await client!.mcp.status({ directory: workspace })
|
||||
expect(status.data?.[added[1]!]?.status).toBe('connected')
|
||||
expect(added[1]).not.toBe(added[0])
|
||||
releaseModel()
|
||||
await expect(second).resolves.toBe('LATE_OK')
|
||||
} finally {
|
||||
releaseInitialize()
|
||||
releaseModel()
|
||||
await runtime.dispose()
|
||||
await gateway.dispose()
|
||||
proxy.closeAllConnections()
|
||||
model.closeAllConnections()
|
||||
await new Promise<void>(resolve => proxy.close(() => resolve()))
|
||||
await new Promise<void>(resolve => model.close(() => resolve()))
|
||||
await rm(workspace, { recursive: true, force: true })
|
||||
}
|
||||
},
|
||||
90_000
|
||||
)
|
||||
|
||||
it.skipIf(!existsSync(binaryPath))(
|
||||
'closes real event streams after parallel runs and compaction without cancelling a peer',
|
||||
async () => {
|
||||
|
||||
@@ -240,6 +240,7 @@ function runClient(events: Record<string, unknown>[]) {
|
||||
[input.name]: { status: "connected" },
|
||||
},
|
||||
error: undefined,
|
||||
response: { status: 200 },
|
||||
})),
|
||||
disconnect: vi.fn().mockResolvedValue({ data: true, error: undefined }),
|
||||
},
|
||||
@@ -3248,7 +3249,7 @@ describe("OpenCodeRuntime embedded permission mediation", () => {
|
||||
const fail = async (input?: { name?: string }) => {
|
||||
if (thrown) throw new Error(failure);
|
||||
if (apiError) return { error: { name: "UnknownError", data: { message: failure } }, response: { status: 403 } };
|
||||
return { data: status === "missing" ? {} : {
|
||||
return { response: { status: 200 }, data: status === "missing" ? {} : {
|
||||
[input!.name!]: { status: status ?? "failed", error: failure },
|
||||
} };
|
||||
};
|
||||
@@ -3312,10 +3313,22 @@ describe("OpenCodeRuntime embedded permission mediation", () => {
|
||||
expect((record.error as Error).message).toBe("Local MCP operation failed");
|
||||
}
|
||||
if (attempts === 2) {
|
||||
expect(add.mock.calls[1]![0]).toEqual(add.mock.calls[0]![0]);
|
||||
expect(add.mock.calls[1]![0]?.config).toEqual(add.mock.calls[0]![0]?.config);
|
||||
if (thrown) expect(add.mock.calls[1]![0]?.name).not.toBe(add.mock.calls[0]![0]?.name);
|
||||
else expect(add.mock.calls[1]![0]?.name).toBe(add.mock.calls[0]![0]?.name);
|
||||
expect(disconnect.mock.invocationCallOrder[0]).toBeGreaterThan(add.mock.invocationCallOrder[0]!);
|
||||
expect(disconnect.mock.invocationCallOrder[0]).toBeLessThan(add.mock.invocationCallOrder[1]!);
|
||||
}
|
||||
if (succeeds) {
|
||||
const name = add.mock.calls.at(-1)![0]!.name;
|
||||
const toolId = `${name}_${kind === "custom" ? "custom_tool" : "knowledge_search"}`;
|
||||
expect(setup.session.create).toHaveBeenCalledWith(expect.objectContaining({
|
||||
permission: expect.arrayContaining([{ permission: toolId, pattern: "*", action: "allow" }]),
|
||||
}), expect.anything());
|
||||
expect(setup.session.promptAsync).toHaveBeenCalledWith(expect.objectContaining({
|
||||
tools: expect.objectContaining({ [toolId]: true }),
|
||||
}), expect.anything());
|
||||
}
|
||||
expect(add.mock.calls[0]![0]?.config).toMatchObject({
|
||||
headers: { Authorization: `Bearer ${secret}` }, oauth: false,
|
||||
});
|
||||
@@ -3403,6 +3416,9 @@ describe("OpenCodeRuntime embedded permission mediation", () => {
|
||||
expect(observeFailure.mock.calls[0]![0].mcp).toMatchObject({ category: "transport-failure" });
|
||||
}
|
||||
expect(add).toHaveBeenCalledTimes(scenario === "recovered" || scenario === "exhausted" ? 2 : 1);
|
||||
if (add.mock.calls.length === 2) {
|
||||
expect(add.mock.calls[1]![0]?.name).not.toBe(add.mock.calls[0]![0]?.name);
|
||||
}
|
||||
expect(order).toEqual(scenario === "recovered" || scenario === "exhausted"
|
||||
? ["add", "disconnect", "add", "disconnect"]
|
||||
: scenario.startsWith("cleanup-") ? ["add", "disconnect", "disconnect"] : ["add", "disconnect"]);
|
||||
@@ -3465,12 +3481,13 @@ describe("OpenCodeRuntime embedded permission mediation", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["reject", "error", "false", "timeout"])("does not retry or reuse the slot after MCP cleanup %s", async (failure) => {
|
||||
it.each(["reject", "error", "false", "timeout"])("does not retry or reuse the name after MCP cleanup %s", async (failure) => {
|
||||
const setup = runClient([
|
||||
{ id: "idle", type: "session.idle", properties: { sessionID: "session-1" } },
|
||||
]);
|
||||
const add = vi.mocked(setup.client.mcp.add);
|
||||
add.mockImplementationOnce((async (input: { name: string }) => ({
|
||||
response: { status: 200 },
|
||||
data: { [input.name]: { status: "failed", error: "connect ECONNREFUSED" } },
|
||||
})) as never);
|
||||
const disconnect = vi.mocked(setup.client.mcp.disconnect);
|
||||
@@ -3504,7 +3521,9 @@ describe("OpenCodeRuntime embedded permission mediation", () => {
|
||||
expect(new Set(observeFailure.mock.calls.map(([record]) => record.mcp?.correlationId)).size).toBe(1);
|
||||
disconnect.mockResolvedValue({ data: true } as never);
|
||||
await collect("second");
|
||||
expect(add.mock.calls[1]![0]?.name).toBe(`${add.mock.calls[0]![0]?.name}-2`);
|
||||
expect(add.mock.calls[1]![0]?.name).not.toBe(add.mock.calls[0]![0]?.name);
|
||||
await collect("third");
|
||||
expect(add.mock.calls[2]![0]?.name).toBe(add.mock.calls[1]![0]?.name);
|
||||
} finally {
|
||||
await runtime.dispose();
|
||||
}
|
||||
@@ -3669,12 +3688,10 @@ describe("OpenCodeRuntime embedded permission mediation", () => {
|
||||
);
|
||||
const base = names[0]!;
|
||||
expect(base).toMatch(/^gbd-[a-f0-9]{6}$/u);
|
||||
// request-1 failed to disconnect (rejected) -> slot 1 stays leased.
|
||||
expect(names[1]).toBe(`${base}-2`);
|
||||
// request-2 disconnect returned an error -> slot 2 stays leased.
|
||||
expect(names[2]).toBe(`${base}-3`);
|
||||
// request-3 disconnected cleanly -> slot 3 is reusable.
|
||||
expect(names[3]).toBe(`${base}-3`);
|
||||
expect(new Set(names.slice(0, 3)).size).toBe(3);
|
||||
expect(names.every(name => name.length <= 12)).toBe(true);
|
||||
// Successful cleanup preserves the replacement name for the same slot.
|
||||
expect(names[3]).toBe(names[2]);
|
||||
} finally {
|
||||
await runtime.dispose();
|
||||
}
|
||||
@@ -3701,6 +3718,81 @@ describe("OpenCodeRuntime embedded permission mediation", () => {
|
||||
expect(runtime.acquireTemporaryMcpSlot(setup.client, "abc123")).toBe(4);
|
||||
expect(createTemporaryMcpName("gbd-", "abc123", 1)).toBe("gbd-abc123");
|
||||
expect(createTemporaryMcpName("gbc-", "abc123", 4)).toBe("gbc-abc123-4");
|
||||
for (const generation of [1, 35, 36, 36 ** 8 - 1]) {
|
||||
const name = createTemporaryMcpName("gbc-", "abc123", 9, generation);
|
||||
expect(`${name}_${"x".repeat(51)}`).toHaveLength(64);
|
||||
expect(name).not.toBe(createTemporaryMcpName("gbc-", "abc123", 9));
|
||||
}
|
||||
expect(() => createTemporaryMcpName("gbc-", "abc123", 9, 36 ** 8)).toThrow("generations exhausted");
|
||||
});
|
||||
|
||||
it.each(["throw", "no-response", "timeout", "abort"])("retires an uncertain MCP add after %s despite successful disconnect", async (failure) => {
|
||||
const setup = runClient([
|
||||
{ id: "idle", type: "session.idle", properties: { sessionID: "session-1" } },
|
||||
]);
|
||||
const controller = new AbortController();
|
||||
const add = vi.mocked(setup.client.mcp.add);
|
||||
add.mockImplementationOnce((async () => {
|
||||
if (failure === "throw") throw new Error("unknown initialization failure");
|
||||
if (failure === "no-response") return { error: new Error("unknown initialization failure") };
|
||||
if (failure === "abort") controller.abort(new Error("cancel initialization"));
|
||||
return new Promise(() => {});
|
||||
}) as never);
|
||||
const runtime = embeddedRuntime(setup.client, {
|
||||
knowledgeGateway: {
|
||||
getEndpoint: () => "http://127.0.0.1:4567/mcp",
|
||||
getAvailableToolNames: () => ["knowledge_search"],
|
||||
} as unknown as KnowledgeMcpGateway,
|
||||
}, { controlRequestTimeoutMs: 50 });
|
||||
const run = async (signal: AbortSignal) => {
|
||||
for await (const event of runtime.run({
|
||||
requestId: crypto.randomUUID(), conversationId: "uncertain", prompt: "test", workMode: "ask",
|
||||
knowledgeCapabilityToken: "token",
|
||||
}, signal)) void event;
|
||||
};
|
||||
try {
|
||||
await expect(run(controller.signal)).rejects.toThrow();
|
||||
expect(setup.client.mcp.disconnect).toHaveBeenCalledOnce();
|
||||
await run(new AbortController().signal);
|
||||
await run(new AbortController().signal);
|
||||
const names = add.mock.calls.map(([input]) => input!.name);
|
||||
expect(names[1]).not.toBe(names[0]);
|
||||
expect(names[2]).toBe(names[1]);
|
||||
expect(vi.mocked(setup.client.mcp.disconnect).mock.calls.map(([input]) => input!.name)).toEqual(names);
|
||||
} finally { await runtime.dispose(); }
|
||||
});
|
||||
|
||||
it("releases active slots after more than nine failed cleanups and isolates replacement generations", async () => {
|
||||
const setup = runClient([
|
||||
{ id: "idle", type: "session.idle", properties: { sessionID: "session-1" } },
|
||||
]);
|
||||
const disconnect = vi.mocked(setup.client.mcp.disconnect);
|
||||
disconnect.mockResolvedValue({ data: false } as never);
|
||||
const runtime = embeddedRuntime(setup.client, {
|
||||
knowledgeGateway: {
|
||||
getEndpoint: () => "http://127.0.0.1:4567/mcp",
|
||||
getAvailableToolNames: () => ["knowledge_search"],
|
||||
} as unknown as KnowledgeMcpGateway,
|
||||
});
|
||||
const run = async (conversationId: string) => {
|
||||
for await (const event of runtime.run({
|
||||
requestId: crypto.randomUUID(), conversationId, prompt: "test", workMode: "ask",
|
||||
knowledgeCapabilityToken: "token",
|
||||
}, new AbortController().signal)) void event;
|
||||
};
|
||||
try {
|
||||
for (let i = 0; i < 12; i++) await run("cleanup-failures");
|
||||
await run("other-conversation");
|
||||
disconnect.mockResolvedValue({ data: true } as never);
|
||||
await run("other-conversation");
|
||||
await run("cleanup-failures");
|
||||
await run("cleanup-failures");
|
||||
const names = vi.mocked(setup.client.mcp.add).mock.calls.map(([input]) => input!.name);
|
||||
expect(new Set(names.slice(0, 15)).size).toBe(15);
|
||||
expect(names[15]).toBe(names[14]);
|
||||
expect(names.every(name => name !== undefined && name.length <= 12)).toBe(true);
|
||||
expect(disconnect.mock.calls.map(([input]) => input!.name)).toEqual(names);
|
||||
} finally { await runtime.dispose(); }
|
||||
});
|
||||
|
||||
it("enables the deterministic MCP tool name when tool ids omit dynamic tools", async () => {
|
||||
|
||||
@@ -553,7 +553,13 @@ export function createTemporaryMcpName(
|
||||
prefix: (typeof TEMPORARY_MCP_PREFIXES)[number],
|
||||
conversationKey: string,
|
||||
slot: number,
|
||||
generation = 0,
|
||||
): string {
|
||||
if (generation > 0) {
|
||||
// Eight base-36 digits keep the registration within the 12-character budget.
|
||||
if (generation >= 36 ** 8) throw new Error("OpenCode MCP name generations exhausted");
|
||||
return `${prefix}${generation.toString(36).padStart(8, "0")}`;
|
||||
}
|
||||
return `${prefix}${conversationKey}${slot === 1 ? "" : `-${slot}`}`;
|
||||
}
|
||||
|
||||
@@ -776,14 +782,11 @@ export class OpenCodeRuntime implements AgentRuntime {
|
||||
>();
|
||||
private readonly conversationRunTails = new Map<string, Promise<void>>();
|
||||
private mcpMutationTail: Promise<void> = Promise.resolve();
|
||||
// Temporary MCP name slots per OpenCode client and conversation. A slot is
|
||||
// held while a request owns its MCP connections and is only freed after
|
||||
// every disconnect succeeds, so a request never reuses a name that may still
|
||||
// be bound to another request's capability token. A new client (restarted
|
||||
// server) starts with a clean MCP registry and therefore fresh slots.
|
||||
// Slots bound active requests; generations retire uncertain names without
|
||||
// retaining a slot forever. Replacement names are unique across this client.
|
||||
private readonly temporaryMcpSlots = new WeakMap<
|
||||
OpencodeClient,
|
||||
Map<string, Set<number>>
|
||||
{ generation: number; conversations: Map<string, Map<number, { active: boolean; generation: number }>> }
|
||||
>();
|
||||
private readonly dependencies: OpenCodeRuntimeDependencies;
|
||||
|
||||
@@ -865,19 +868,20 @@ export class OpenCodeRuntime implements AgentRuntime {
|
||||
client: OpencodeClient,
|
||||
conversationKey: string,
|
||||
): number {
|
||||
let byConversation = this.temporaryMcpSlots.get(client);
|
||||
if (!byConversation) {
|
||||
byConversation = new Map();
|
||||
this.temporaryMcpSlots.set(client, byConversation);
|
||||
let state = this.temporaryMcpSlots.get(client);
|
||||
if (!state) {
|
||||
state = { generation: 0, conversations: new Map() };
|
||||
this.temporaryMcpSlots.set(client, state);
|
||||
}
|
||||
let occupied = byConversation.get(conversationKey);
|
||||
let occupied = state.conversations.get(conversationKey);
|
||||
if (!occupied) {
|
||||
occupied = new Set();
|
||||
byConversation.set(conversationKey, occupied);
|
||||
occupied = new Map();
|
||||
state.conversations.set(conversationKey, occupied);
|
||||
}
|
||||
for (let slot = 1; slot <= MAX_TEMPORARY_MCP_SLOTS; slot += 1) {
|
||||
if (!occupied.has(slot)) {
|
||||
occupied.add(slot);
|
||||
const entry = occupied.get(slot);
|
||||
if (!entry?.active) {
|
||||
occupied.set(slot, { active: true, generation: entry?.generation ?? 0 });
|
||||
return slot;
|
||||
}
|
||||
}
|
||||
@@ -891,12 +895,14 @@ export class OpenCodeRuntime implements AgentRuntime {
|
||||
conversationKey: string,
|
||||
slot: number,
|
||||
): void {
|
||||
const byConversation = this.temporaryMcpSlots.get(client);
|
||||
const byConversation = this.temporaryMcpSlots.get(client)?.conversations;
|
||||
const occupied = byConversation?.get(conversationKey);
|
||||
if (!byConversation || !occupied) {
|
||||
return;
|
||||
}
|
||||
occupied.delete(slot);
|
||||
const entry = occupied.get(slot);
|
||||
if (entry?.generation) entry.active = false;
|
||||
else occupied.delete(slot);
|
||||
if (occupied.size === 0) {
|
||||
byConversation.delete(conversationKey);
|
||||
}
|
||||
@@ -2251,7 +2257,17 @@ export class OpenCodeRuntime implements AgentRuntime {
|
||||
prefix: (typeof TEMPORARY_MCP_PREFIXES)[number],
|
||||
): string => {
|
||||
mcpSlot ??= this.acquireTemporaryMcpSlot(client, mcpConversationKey);
|
||||
return createTemporaryMcpName(prefix, mcpConversationKey, mcpSlot);
|
||||
const entry = this.temporaryMcpSlots.get(client)!.conversations.get(mcpConversationKey)!.get(mcpSlot)!;
|
||||
return createTemporaryMcpName(prefix, mcpConversationKey, mcpSlot, entry.generation);
|
||||
};
|
||||
const retireMcpName = (name: string): void => {
|
||||
const state = this.temporaryMcpSlots.get(client)!;
|
||||
const entry = state.conversations.get(mcpConversationKey)!.get(mcpSlot!)!;
|
||||
const prefix = name.startsWith(KNOWLEDGE_MCP_PREFIX) ? KNOWLEDGE_MCP_PREFIX : CUSTOM_MCP_PREFIX;
|
||||
// A late cleanup of an already retired name must not retire its successor.
|
||||
if (createTemporaryMcpName(prefix, mcpConversationKey, mcpSlot!, entry.generation) === name) {
|
||||
entry.generation = ++state.generation;
|
||||
}
|
||||
};
|
||||
const subscriptionController = new AbortController();
|
||||
const imageCapabilityToken = request.imageToolBinding && request.workMode === 'execute'
|
||||
@@ -2284,15 +2300,17 @@ export class OpenCodeRuntime implements AgentRuntime {
|
||||
category = result.error && !result.response ? "transport-failure"
|
||||
: status && status >= 400 ? "http-rejection" : "cleanup-failure";
|
||||
} catch {
|
||||
// The uncertain connection keeps its slot until cleanup succeeds.
|
||||
// The name is retired below; the active request can still release its slot.
|
||||
}
|
||||
retireMcpName(name);
|
||||
observeMcpFailure({ ...metadata, phase: "disconnect", category, status,
|
||||
elapsedMs: Math.round(performance.now() - startedAt) });
|
||||
return false;
|
||||
};
|
||||
const addMcp = (name: string, url: string, token: string, label: string) =>
|
||||
const addMcp = (prefix: (typeof TEMPORARY_MCP_PREFIXES)[number], url: string, token: string, label: string) =>
|
||||
this.mutateMcp(signal, async () => {
|
||||
for (let attempt = 0; attempt < 2; attempt++) {
|
||||
let name = temporaryMcpName(prefix);
|
||||
for (let attempt = 0; ; attempt++) {
|
||||
signal.throwIfAborted();
|
||||
const startedAt = performance.now();
|
||||
const metadata: McpAttempt = {
|
||||
@@ -2304,9 +2322,12 @@ export class OpenCodeRuntime implements AgentRuntime {
|
||||
let retryableStatus = true;
|
||||
let category: DesktopMcpDiagnosticMetadata["category"] = "transport-failure";
|
||||
let httpStatus: number | undefined;
|
||||
let attempted = false;
|
||||
let confirmed = false;
|
||||
try {
|
||||
const added = await this.controlRequest(label, (controlSignal) => {
|
||||
attemptedMcpNames.set(name, metadata);
|
||||
attempted = true;
|
||||
return client.mcp.add({
|
||||
directory, name,
|
||||
config: {
|
||||
@@ -2316,10 +2337,11 @@ export class OpenCodeRuntime implements AgentRuntime {
|
||||
},
|
||||
}, { signal: controlSignal });
|
||||
}, signal);
|
||||
confirmed = !!added.response;
|
||||
const status = added.data?.[name];
|
||||
httpStatus = added.response?.status;
|
||||
const httpRejected = httpStatus !== undefined && httpStatus >= 400;
|
||||
if (!httpRejected && !added.error && status?.status === "connected") return;
|
||||
if (!httpRejected && !added.error && status?.status === "connected") return name;
|
||||
// With throwOnError:false the SDK resolves fetch failures without a response.
|
||||
const transportFailure = !!added.error && !added.response;
|
||||
category = transportFailure ? "transport-failure"
|
||||
@@ -2330,6 +2352,9 @@ export class OpenCodeRuntime implements AgentRuntime {
|
||||
: status?.status ?? "unknown");
|
||||
} catch (error) {
|
||||
failure = error;
|
||||
} finally {
|
||||
// Aborting the HTTP client does not cancel OpenCode's pending add.
|
||||
if (attempted && !confirmed) retireMcpName(name);
|
||||
}
|
||||
observeMcpFailure({ ...metadata, phase: "connect",
|
||||
category: signal.aborted ? "cancelled" : category, status: httpStatus,
|
||||
@@ -2346,6 +2371,7 @@ export class OpenCodeRuntime implements AgentRuntime {
|
||||
if (attempt > 0 || !retryableStatus || !transient) throw error;
|
||||
if (!await disconnectMcp(name, metadata)) throw error;
|
||||
attemptedMcpNames.delete(name);
|
||||
name = temporaryMcpName(prefix);
|
||||
}
|
||||
});
|
||||
try {
|
||||
@@ -2355,8 +2381,7 @@ export class OpenCodeRuntime implements AgentRuntime {
|
||||
this.options.knowledgeGateway?.getEndpoint()
|
||||
) {
|
||||
const knowledgeEndpoint = this.options.knowledgeGateway.getEndpoint()!;
|
||||
knowledgeMcpName = temporaryMcpName(KNOWLEDGE_MCP_PREFIX);
|
||||
await addMcp(knowledgeMcpName, knowledgeEndpoint, scopedCapabilityToken, "连接内置只读工具");
|
||||
knowledgeMcpName = await addMcp(KNOWLEDGE_MCP_PREFIX, knowledgeEndpoint, scopedCapabilityToken, "连接内置只读工具");
|
||||
// OpenCode 1.18.x does not include dynamically added MCP tools in
|
||||
// experimental/tool/ids. Its model tool namespace is deterministic:
|
||||
// "<MCP server name>_<declared tool name>".
|
||||
@@ -2382,8 +2407,7 @@ export class OpenCodeRuntime implements AgentRuntime {
|
||||
customMcpToken,
|
||||
signal,
|
||||
);
|
||||
customMcpName = temporaryMcpName(CUSTOM_MCP_PREFIX);
|
||||
await addMcp(customMcpName, customMcpEndpoint, customMcpToken, "连接自定义 MCP 工具");
|
||||
customMcpName = await addMcp(CUSTOM_MCP_PREFIX, customMcpEndpoint, customMcpToken, "连接自定义 MCP 工具");
|
||||
knowledgeToolIds.push(
|
||||
...tools.map((tool) => `${customMcpName}_${tool.name}`),
|
||||
);
|
||||
@@ -3215,17 +3239,13 @@ export class OpenCodeRuntime implements AgentRuntime {
|
||||
} finally {
|
||||
subscriptionController.abort();
|
||||
if (imageCapabilityToken && imageCapabilityToken !== request.knowledgeCapabilityToken) this.options.knowledgeGateway?.revoke(imageCapabilityToken);
|
||||
let allMcpDisconnected = true;
|
||||
for (const [name, metadata] of attemptedMcpNames) {
|
||||
const disconnected = await this.mutateMcp(
|
||||
await this.mutateMcp(
|
||||
new AbortController().signal,
|
||||
() => disconnectMcp(name, metadata),
|
||||
);
|
||||
allMcpDisconnected &&= disconnected;
|
||||
}
|
||||
// A slot whose connection may still be alive stays occupied so a later
|
||||
// request cannot reuse a name bound to this request's token.
|
||||
if (mcpSlot !== undefined && allMcpDisconnected) {
|
||||
if (mcpSlot !== undefined) {
|
||||
this.releaseTemporaryMcpSlot(client, mcpConversationKey, mcpSlot);
|
||||
}
|
||||
if (customMcpToken) {
|
||||
|
||||
Reference in New Issue
Block a user