Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
169 changes: 169 additions & 0 deletions packages/acp-bridge/src/bridge.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11876,6 +11876,175 @@ describe('createAcpSessionBridge', () => {
});
});

describe('session-scoped runtime MCP servers', () => {
it('routes add and remove to the selected live session only', async () => {
const calls: Array<{
method: string;
params: Record<string, unknown>;
}> = [];
const factory: ChannelFactory = async () => {
const { clientStream, agentStream } = createInMemoryChannel();
const agent = new FakeAgent({
extMethodImpl: (method, params) => {
calls.push({ method, params });
if (method === SERVE_CONTROL_EXT_METHODS.sessionMcpRuntimeAdd) {
return Promise.resolve({
name: params['name'],
transport: 'sdk',
replaced: false,
shadowedSettings: false,
toolCount: 1,
originatorClientId: params['originatorClientId'],
});
}
if (method === SERVE_CONTROL_EXT_METHODS.sessionMcpRuntimeRemove) {
return Promise.resolve({
name: params['name'],
removed: true,
wasShadowingSettings: false,
originatorClientId: params['originatorClientId'],
});
}
return Promise.resolve({});
},
});
new AgentSideConnection(() => agent as Agent, agentStream);
return {
stream: clientStream,
exited: new Promise<
| { exitCode: number | null; signalCode: NodeJS.Signals | null }
| undefined
>(() => {}),
kill: async () => {},
killSync: () => {},
};
};
const bridge = makeBridge({
sessionScope: 'thread',
channelFactory: factory,
});
await bridge.spawnOrAttach({ workspaceCwd: WS_A });
const target = await bridge.spawnOrAttach({ workspaceCwd: WS_A });

await bridge.addSessionRuntimeMcpServer(
target.sessionId,
'channel-loop',
{ type: 'sdk' },
'daemon-channel',
);
await bridge.removeSessionRuntimeMcpServer(
target.sessionId,
'channel-loop',
'daemon-channel',
);

expect(calls.slice(-2)).toEqual([
{
method: SERVE_CONTROL_EXT_METHODS.sessionMcpRuntimeAdd,
params: {
sessionId: target.sessionId,
name: 'channel-loop',
config: { type: 'sdk' },
originatorClientId: 'daemon-channel',
},
},
{
method: SERVE_CONTROL_EXT_METHODS.sessionMcpRuntimeRemove,
params: {
sessionId: target.sessionId,
name: 'channel-loop',
originatorClientId: 'daemon-channel',
},
},
]);
await bridge.shutdown();
});

it('rejects an unknown session without falling back to a live channel', async () => {
const extMethodImpl = vi.fn().mockResolvedValue({});
const bridge = makeBridge({
channelFactory: async () => makeChannel({ extMethodImpl }).channel,
});
await bridge.spawnOrAttach({ workspaceCwd: WS_A });

await expect(
bridge.addSessionRuntimeMcpServer('missing-session', 'channel-loop', {
type: 'sdk',
}),
).rejects.toBeInstanceOf(SessionNotFoundError);
expect(extMethodImpl).not.toHaveBeenCalledWith(
SERVE_CONTROL_EXT_METHODS.sessionMcpRuntimeAdd,
expect.anything(),
);
await bridge.shutdown();
});

it('allows session MCP mutations to outlive the initialize timeout', async () => {
const add = deferred<Record<string, unknown>>();
const remove = deferred<Record<string, unknown>>();
const bridge = makeBridge({
initializeTimeoutMs: 5,
channelFactory: async () =>
makeChannel({
extMethodImpl: (method) => {
if (method === SERVE_CONTROL_EXT_METHODS.sessionMcpRuntimeAdd) {
return add.promise;
}
if (
method === SERVE_CONTROL_EXT_METHODS.sessionMcpRuntimeRemove
) {
return remove.promise;
}
return Promise.resolve({});
},
}).channel,
});
const target = await bridge.spawnOrAttach({ workspaceCwd: WS_A });

vi.useFakeTimers();
try {
const addResult = bridge.addSessionRuntimeMcpServer(
target.sessionId,
'channel-loop',
{
type: 'sdk',
},
);
await vi.advanceTimersByTimeAsync(6);
add.resolve({
name: 'channel-loop',
transport: 'sdk',
replaced: false,
shadowedSettings: false,
toolCount: 1,
originatorClientId: 'daemon',
});
await expect(addResult).resolves.toMatchObject({
name: 'channel-loop',
});

const removeResult = bridge.removeSessionRuntimeMcpServer(
target.sessionId,
'channel-loop',
);
await vi.advanceTimersByTimeAsync(6);
remove.resolve({
name: 'channel-loop',
removed: true,
wasShadowingSettings: false,
originatorClientId: 'daemon',
});
await expect(removeResult).resolves.toMatchObject({
name: 'channel-loop',
removed: true,
});
} finally {
vi.useRealTimers();
await bridge.shutdown();
}
});
});

describe('removeRuntimeMcpServer (T2.8 #4514)', () => {
/**
* Build a channel factory whose ACP `extMethod` handler returns a
Expand Down
28 changes: 27 additions & 1 deletion packages/acp-bridge/src/bridge.ts
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,8 @@ import type {
BridgeSessionTranscriptPageRequest,
BridgeGenerationStreamEvent,
BridgeWorkspaceGenerationStreamEvent,
RuntimeMcpServerAddResult,
RuntimeMcpServerRemoveResult,
} from './bridgeTypes.js';
import type {
BridgeFreshSessionAdmissionContext,
Expand Down Expand Up @@ -3487,6 +3489,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
sessionId: string,
method: string,
params: Record<string, unknown> = {},
timeoutMs = initTimeoutMs,
): Promise<T> => {
const entry = byId.get(sessionId);
if (!entry) throw new SessionNotFoundError(sessionId);
Expand All @@ -3495,7 +3498,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
const response = await Promise.race([
withTimeout(
entry.connection.extMethod(method, { ...params, sessionId }),
initTimeoutMs,
timeoutMs,
method,
),
getTransportClosedReject(entry),
Expand Down Expand Up @@ -8193,6 +8196,29 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
return response;
},

async addSessionRuntimeMcpServer(
sessionId,
name,
config,
originatorClientId,
) {
return requestSessionStatus<RuntimeMcpServerAddResult>(
sessionId,
SERVE_CONTROL_EXT_METHODS.sessionMcpRuntimeAdd,
{ name, config, originatorClientId },
MCP_RESTART_SERVER_DEADLINE_MS,
);
},
Comment on lines +8202 to +8214

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

已修复:session-scoped MCP add/remove 现在与 workspace 路径一致使用 300s MCP restart deadline,并新增回归测试覆盖操作超过 initialize timeout 后仍可完成。验证证据:cd packages/acp-bridge && npx vitest run src/bridge.test.ts(450/450 passed);cd packages/cli && npx vitest run src/acp-integration/acpAgent.test.ts -t "mutates runtime MCP state on only the selected live session"(1/1 passed);npm run build --workspace=@qwen-code/acp-bridge、npm run typecheck --workspace=@qwen-code/acp-bridge、Prettier 与 git diff --check 均通过。Commit: f57b02307c。


async removeSessionRuntimeMcpServer(sessionId, name, originatorClientId) {
return requestSessionStatus<RuntimeMcpServerRemoveResult>(
sessionId,
SERVE_CONTROL_EXT_METHODS.sessionMcpRuntimeRemove,
{ name, originatorClientId },
MCP_RESTART_SERVER_DEADLINE_MS,
);
},

async killSession(sessionId, opts) {
const entry = byId.get(sessionId);
if (!entry) return false;
Expand Down
84 changes: 84 additions & 0 deletions packages/acp-bridge/src/bridgeClient.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2834,6 +2834,90 @@ describe('BridgeClient — reverse tool channel (qwen/control/client_mcp/message
expect((result as { payload?: unknown }).payload).toBeDefined();
});

it('forwards the originating session id to the client MCP sender', async () => {
const contexts: unknown[] = [];
const sender: ClientMcpMessageSender = () => async (_payload, context) => {
contexts.push(context);
return { jsonrpc: '2.0', id: 1, result: {} };
};
const client = new BridgeClient(
(() => undefined) as never,
(() => undefined) as never,
{ request: thrower } as never,
0,
Infinity,
undefined,
undefined,
undefined,
sender,
);

await client.extMethod('qwen/control/client_mcp/message', {
server: 'channel-loop',
sessionId: 'session-channel-1',
payload: { jsonrpc: '2.0', id: 1, method: 'tools/list' },
});

expect(contexts).toEqual([{ sessionId: 'session-channel-1' }]);
});

it('rejects a malformed optional session id', async () => {
const sender: ClientMcpMessageSender = () => async () => ({
jsonrpc: '2.0',
id: 1,
result: {},
});
const client = new BridgeClient(
(() => undefined) as never,
(() => undefined) as never,
{ request: thrower } as never,
0,
Infinity,
undefined,
undefined,
undefined,
sender,
);

await expect(
client.extMethod('qwen/control/client_mcp/message', {
server: 'channel-loop',
sessionId: '',
payload: { jsonrpc: '2.0', id: 1, method: 'tools/list' },
}),
).rejects.toMatchObject({ code: -32602 });
});

it('rejects a session id owned by another ACP channel', async () => {
const send = vi.fn().mockResolvedValue({
jsonrpc: '2.0',
id: 1,
result: {},
});
const sender: ClientMcpMessageSender = () => send;
const client = new BridgeClient(
(() => undefined) as never,
(() => undefined) as never,
{ request: thrower } as never,
0,
Infinity,
undefined,
undefined,
undefined,
sender,
() => false,
);

await expect(
client.extMethod('qwen/control/client_mcp/message', {
server: 'channel-loop',
sessionId: 'foreign-session',
payload: { jsonrpc: '2.0', id: 1, method: 'tools/list' },
}),
).rejects.toMatchObject({ code: -32602 });
expect(send).not.toHaveBeenCalled();
});

it('rejects (invalidParams) when the named server is not connected', async () => {
const registrar = new ClientMcpRegistrar({ sendFrame: () => {} });
// No registerServer call — the lookup returns undefined.
Expand Down
21 changes: 20 additions & 1 deletion packages/acp-bridge/src/bridgeClient.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1300,7 +1300,26 @@ export class BridgeClient implements Client {
`client-hosted MCP server '${server}' is not currently connected`,
);
}
const response = await send(payload);
const sessionId = params['sessionId'];
if (
sessionId !== undefined &&
(typeof sessionId !== 'string' || sessionId.length === 0)
) {
throw RequestError.invalidParams(
undefined,
'`sessionId` must be a non-empty string when provided',
);
}
if (typeof sessionId === 'string' && !this.ownsSession(sessionId)) {
throw RequestError.invalidParams(
undefined,
`Session not owned by this channel: ${sessionId}`,
);
}
const response = await send(
payload,
typeof sessionId === 'string' ? { sessionId } : undefined,
);
return { payload: response as Record<string, unknown> };
}

Expand Down
8 changes: 7 additions & 1 deletion packages/acp-bridge/src/bridgeOptions.ts
Original file line number Diff line number Diff line change
Expand Up @@ -477,9 +477,15 @@ export interface BridgeOptions {
* package free of an MCP-SDK dependency; the serve layer passes a
* `JSONRPCMessage`.
*/
export interface ClientMcpMessageContext {
sessionId?: string;
}

export type ClientMcpMessageSender = (
serverName: string,
) => ((payload: unknown) => Promise<unknown>) | undefined;
) =>
| ((payload: unknown, context?: ClientMcpMessageContext) => Promise<unknown>)
| undefined;

/** Ceiling on a sub-session prompt arriving over `extMethod`. The child is a
* separate process, so this is a trust boundary — mirrors the scheduled-task
Expand Down
Loading
Loading