Skip to content
246 changes: 246 additions & 0 deletions packages/acp-bridge/src/bridge.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ import { TurnBoundaryCompactionEngine } from './compactionEngine.js';
import {
CHANNEL_STARTUP_PROFILE_META_KEY,
CHANNEL_STARTUP_PROFILE_VERSION,
WORKTREE_MCP_DEFER_META_KEY,
LOAD_REPLAY_HIDE_INHERITED_META_KEY,
} from './bridgeTypes.js';
import {
Expand Down Expand Up @@ -1047,6 +1048,24 @@ describe('createAcpSessionBridge', () => {
});
});

it('marks worktree session creation to defer MCP discovery', async () => {
const handle = makeChannel();
const bridge = makeBridge({
sessionScope: 'thread',
channelFactory: async () => handle.channel,
});

await bridge.spawnOrAttach({
workspaceCwd: WS_A,
worktree: { slug: 'task-a', path: WS_B, branch: 'worktree-task-a' },
});

expect(handle.agent.newSessionCalls[0]?._meta).toMatchObject({
[WORKTREE_MCP_DEFER_META_KEY]: true,
});
await bridge.shutdown();
});

it('does not fail initialization when span enrichment throws', async () => {
const handle = makeChannel({
initializeImpl: async () => ({
Expand Down Expand Up @@ -11626,6 +11645,15 @@ describe('createAcpSessionBridge', () => {
aborted: false,
});
expect(shellSpy).toHaveBeenCalledTimes(1);
expect(shellSpy).toHaveBeenCalledWith(
'echo hello',
WS_A,
expect.any(Function),
expect.any(AbortSignal),
false,
{ terminalWidth: 120, terminalHeight: 40 },
{ streamStdout: true },
);
const it = events[Symbol.asyncIterator]();
const first = await it.next();
expect(first.value?.type).toBe('user_shell_command');
Expand All @@ -11635,6 +11663,224 @@ describe('createAcpSessionBridge', () => {
await bridge.shutdown();
shellSpy.mockRestore();
});

it('executes direct shell in each session effective cwd', async () => {
const shellSpy = mockShellExecute();
const handle = makeChannel({
extMethodImpl: async (method, params) => {
if (method === SERVE_CONTROL_EXT_METHODS.sessionCd) {
return {
previousCwd: WS_A,
newCwd: (params as { path: string }).path,
warnings: [],
};
}
return {};
},
});
const bridge = makeBridge({
sessionShellCommandEnabled: true,
channelFactory: async () => handle.channel,
});
const firstSession = await bridge.spawnOrAttach({
workspaceCwd: WS_A,
sessionScope: 'thread',
});
const secondSession = await bridge.spawnOrAttach({
workspaceCwd: WS_A,
sessionScope: 'thread',
});

await Promise.all([
bridge.changeSessionCwd(firstSession.sessionId, { path: WS_A }),
bridge.changeSessionCwd(secondSession.sessionId, { path: WS_B }),
]);
await Promise.all([
bridge.executeShellCommand(
firstSession.sessionId,
'echo first',
undefined,
{ clientId: firstSession.clientId },
),
bridge.executeShellCommand(
secondSession.sessionId,
'echo second',
undefined,
{ clientId: secondSession.clientId },
),
]);

expect(shellSpy).toHaveBeenCalledTimes(2);
expect(
shellSpy.mock.calls.map(([command, cwd]) => [command, cwd]),
).toEqual(
expect.arrayContaining([
['echo first', WS_A],
['echo second', WS_B],
]),
);

await bridge.shutdown();
shellSpy.mockRestore();
});

it('waits for a pending cwd change before executing direct shell', async () => {
const shellSpy = mockShellExecute();
const cdResult = deferred<{
previousCwd: string;
newCwd: string;
warnings: string[];
}>();
const handle = makeChannel({
extMethodImpl: async (method) => {
if (method === SERVE_CONTROL_EXT_METHODS.sessionCd) {
return cdResult.promise;
}
return {};
},
});
const bridge = makeBridge({
sessionShellCommandEnabled: true,
channelFactory: async () => handle.channel,
});
const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });

const cd = bridge.changeSessionCwd(session.sessionId, { path: WS_B });
await vi.waitFor(() =>
expect(handle.agent.extMethodCalls).toContainEqual({
method: SERVE_CONTROL_EXT_METHODS.sessionCd,
params: {
sessionId: session.sessionId,
path: WS_B,
},
}),
);
const shell = bridge.executeShellCommand(
session.sessionId,
'echo after-cd',
undefined,
{ clientId: session.clientId },
);

await Promise.resolve();
expect(shellSpy).not.toHaveBeenCalled();
cdResult.resolve({ previousCwd: WS_A, newCwd: WS_B, warnings: [] });
await Promise.all([cd, shell]);

expect(shellSpy.mock.calls[0]?.[1]).toBe(WS_B);

await bridge.shutdown();
shellSpy.mockRestore();
});

it('executes direct shell in previous cwd when a pending cd fails', async () => {
const shellSpy = mockShellExecute();
const cdResult = deferred<{
previousCwd: string;
newCwd: string;
warnings: string[];
}>();
const handle = makeChannel({
extMethodImpl: async (method) => {
if (method === SERVE_CONTROL_EXT_METHODS.sessionCd) {
return cdResult.promise;
}
return {};
},
});
const bridge = makeBridge({
sessionShellCommandEnabled: true,
channelFactory: async () => handle.channel,
});
const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });

const cd = bridge.changeSessionCwd(session.sessionId, { path: WS_B });
await vi.waitFor(() =>
expect(handle.agent.extMethodCalls).toContainEqual({
method: SERVE_CONTROL_EXT_METHODS.sessionCd,
params: {
sessionId: session.sessionId,
path: WS_B,
},
}),
);
const shell = bridge.executeShellCommand(
session.sessionId,
'echo after-failed-cd',
undefined,
{ clientId: session.clientId },
);

await Promise.resolve();
expect(shellSpy).not.toHaveBeenCalled();
cdResult.reject(new Error('cd failed'));
await expect(cd).rejects.toThrow();
await shell;

expect(shellSpy.mock.calls[0]?.[1]).toBe(WS_A);

await bridge.shutdown();
shellSpy.mockRestore();
});

it('returns an aborted direct shell without waiting for a hung cwd change', async () => {
const shellSpy = mockShellExecute();
const cdResult = deferred<{
previousCwd: string;
newCwd: string;
warnings: string[];
}>();
const handle = makeChannel({
extMethodImpl: async (method) => {
if (method === SERVE_CONTROL_EXT_METHODS.sessionCd) {
return cdResult.promise;
}
return {};
},
});
const bridge = makeBridge({
sessionShellCommandEnabled: true,
channelFactory: async () => handle.channel,
});
const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });

const cd = bridge.changeSessionCwd(session.sessionId, { path: WS_B });
await vi.waitFor(() =>
expect(handle.agent.extMethodCalls).toContainEqual({
method: SERVE_CONTROL_EXT_METHODS.sessionCd,
params: {
sessionId: session.sessionId,
path: WS_B,
},
}),
);

const abort = new AbortController();
const shell = bridge.executeShellCommand(
session.sessionId,
'echo aborted-cd',
abort.signal,
{ clientId: session.clientId },
);

await Promise.resolve();
expect(shellSpy).not.toHaveBeenCalled();
abort.abort();

// The cd extMethod never settles, yet the aborted command must return
// promptly instead of parking on the cwd queue forever.
await expect(shell).resolves.toEqual({
exitCode: null,
output: '',
aborted: true,
});
expect(shellSpy).not.toHaveBeenCalled();

cdResult.resolve({ previousCwd: WS_A, newCwd: WS_B, warnings: [] });
await cd;
await bridge.shutdown();
shellSpy.mockRestore();
});
});

describe('setSessionApprovalMode (#4175 Wave 4 PR 17)', () => {
Expand Down
58 changes: 48 additions & 10 deletions packages/acp-bridge/src/bridge.ts
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,7 @@ import {
LOAD_REPLAY_VERSION,
PROMPT_CANCEL_METHOD,
TODO_STOP_GUARD_QUEUE_RELEASE_METHOD,
WORKTREE_MCP_DEFER_META_KEY,
} from './bridgeTypes.js';
import { getChannelStartupProfileAttributes } from './channel-startup-profile.js';
import type {
Expand Down Expand Up @@ -471,6 +472,7 @@ interface ChannelInfo {
interface SessionEntry {
sessionId: string;
workspaceCwd: string;
effectiveCwd: string;
createdAt: string;
displayName?: string;
/** Id of the session that spawned this one (via `create_sub_session`).
Expand All @@ -494,6 +496,8 @@ interface SessionEntry {
recordingDegraded: boolean;
/** Set synchronously while agent-owned state and its writer lease close. */
closing: boolean;
/** Tail of cwd changes that direct shell commands must not overtake. */
cwdChangeQueue: Promise<void>;
/**
* Tail of the per-session prompt queue. Each new prompt chains off the
* resolved (or rejected) state of this promise so prompts run one at a
Expand Down Expand Up @@ -2651,17 +2655,24 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
async () => {
// This legacy-named helper sanitizes and injects trace metadata
// for any ACP request, not only prompts.
const request = telemetry.injectPromptContext({
cwd: boundWorkspace,
mcpServers: [],
...(sourceType
? { _meta: sessionSourceRequestMeta(sourceType, sourceId) }
: {}),
});
const response = await withTimeout(
ci.connection.newSession(
telemetry.injectPromptContext({
cwd: boundWorkspace,
mcpServers: [],
...(sourceType
? {
_meta: sessionSourceRequestMeta(sourceType, sourceId),
}
: {}),
}),
worktree
? {
...request,
_meta: {
...(isRecord(request._meta) ? request._meta : {}),
[WORKTREE_MCP_DEFER_META_KEY]: true,
},
}
: request,
),
initTimeoutMs,
'newSession',
Expand Down Expand Up @@ -3885,6 +3896,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
const entry: SessionEntry = {
sessionId,
workspaceCwd,
effectiveCwd: workspaceCwd,
createdAt: new Date().toISOString(),
...(options.parentSessionId
? { parentSessionId: options.parentSessionId }
Expand All @@ -3903,6 +3915,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
}),
recordingDegraded: false,
closing: false,
cwdChangeQueue: Promise.resolve(),
promptQueue: Promise.resolve(),
pendingPromptCount: 0,
pendingPromptList: [],
Expand Down Expand Up @@ -6483,6 +6496,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {

// State update inside the queue lambda — always executes when
// the extMethod settles, regardless of caller timeout.
entry.effectiveCwd = extResult.newCwd;
if (extResult.previousCwd !== extResult.newCwd) {
entry.events.publish({
type: 'session_cwd_changed',
Expand All @@ -6504,6 +6518,10 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
() => undefined,
() => undefined,
);
entry.cwdChangeQueue = cdPromise.then(
() => undefined,
() => undefined,
);
Comment thread
qwen-code-dev-bot marked this conversation as resolved.
Resolved

// Timeout is caller-facing only: surfaces a deadline exceeded error
// to the HTTP client without advancing the queue prematurely.
Expand Down Expand Up @@ -7910,7 +7928,27 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
return { exitCode: null, output: '', aborted: true };
}

const cwd = entry.workspaceCwd;
// Race the cwd queue against the caller's abort signal so a shell
// command cannot park forever on a changeSessionCwd extMethod that
// never settles (agent crash / deadlock / partitioned ACP channel).
let abortResolve: (() => void) | undefined;
const onAbort = () => abortResolve?.();
try {
await Promise.race([
entry.cwdChangeQueue,
new Promise<void>((resolve) => {
abortResolve = resolve;
if (signal?.aborted) return resolve();
signal?.addEventListener('abort', onAbort, { once: true });
}),
]);
} finally {
signal?.removeEventListener('abort', onAbort);
}
if (signal?.aborted) {
return { exitCode: null, output: '', aborted: true };
}
const cwd = entry.effectiveCwd;

entry.events.publish({
type: 'user_shell_command',
Expand Down
Loading
Loading