Skip to content

fix(channels): return only final ACP response text #6615

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wenshao merged 7 commits into from
Jul 10, 2026
Merged

fix(channels): return only final ACP response text #6615

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
732bee8
fix(channels): return only final ACP response text
qqqys Jul 9, 2026
cf18ad4
fix(channels): align daemon response boundaries
qqqys Jul 9, 2026
0cf0112
fix(channels): bound daemon permission responses
qqqys Jul 9, 2026
f20f950
fix(channels): reset adapter buffers on response boundaries
qqqys Jul 9, 2026
a6d31c3
fix(channels): bound auto-approved tool responses
qqqys Jul 10, 2026
aee6a87
Merge remote-tracking branch 'upstream/main' into automation/pr-6615
qqqys Jul 10, 2026
f41e881
fix(channels): clear adapter state at response boundaries
qqqys Jul 10, 2026
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
129 changes: 129 additions & 0 deletions packages/channels/base/src/AcpBridge.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,7 @@ type TestableAcpBridge = AcpBridge & {
connection: {
extMethod: ReturnType<typeof vi.fn>;
newSession?: ReturnType<typeof vi.fn>;
prompt?: ReturnType<typeof vi.fn>;
};
channelLoopMcpServer: unknown;
channelLoopToolHandlers: ChannelLoopToolHandler[];
Expand All @@ -102,6 +103,8 @@ type TestableAcpBridge = AcpBridge & {
method: string,
params: Record<string, unknown>,
): Promise<unknown>;
handleSessionUpdate(params: Record<string, unknown>): void;
requestPermission(params: Record<string, unknown>): Promise<unknown>;
handleClientMcpMessage(params: Record<string, unknown>): Promise<unknown>;
registerChannelLoopMcpServer(): Promise<void>;
resolveChannelLoopToolHandler(sessionId: string): ChannelLoopToolHandler;
Expand Down Expand Up @@ -246,6 +249,132 @@ describe('AcpBridge', () => {
).resolves.toStrictEqual({ messages: [] });
});

it('returns only the final turn text after tool calls', async () => {
const bridge = new AcpBridge({
cliEntryPath: '/tmp/qwen',
cwd: '/tmp',
}) as unknown as TestableAcpBridge;
bridge.child = { killed: false, exitCode: null };
bridge.connection = {
extMethod: vi.fn(),
prompt: vi.fn(async () => {
bridge.emit('textChunk', 's-1', 'Let me search. ');
bridge.handleSessionUpdate({
sessionId: 's-1',
update: {
sessionUpdate: 'tool_call',
toolCallId: 'call-1',
kind: 'search',
title: 'Search',
status: 'pending',
},
});
bridge.emit('textChunk', 's-1', 'Now I will read. ');
bridge.handleSessionUpdate({
sessionId: 's-1',
update: {
sessionUpdate: 'tool_call',
toolCallId: 'call-2',
kind: 'read',
title: 'Read',
status: 'pending',
},
});
bridge.emit('textChunk', 's-1', 'Final answer.');
}),
};

await expect(bridge.prompt('s-1', 'question')).resolves.toBe(
'Final answer.',
);
});

it('preserves text when tool calls are not pending', async () => {
const bridge = new AcpBridge({
cliEntryPath: '/tmp/qwen',
cwd: '/tmp',
}) as unknown as TestableAcpBridge;
bridge.child = { killed: false, exitCode: null };
bridge.connection = {
extMethod: vi.fn(),
prompt: vi.fn(async () => {
bridge.emit('textChunk', 's-1', 'Before. ');
bridge.handleSessionUpdate({
sessionId: 's-1',
update: {
sessionUpdate: 'tool_call',
toolCallId: 'call-1',
kind: 'search',
title: 'Search',
status: 'completed',
},
});
bridge.emit('textChunk', 's-1', 'After.');
}),
};

await expect(bridge.prompt('s-1', 'question')).resolves.toBe(
'Before. After.',
);
});

it('treats plan updates as turn boundaries for TodoWrite-only rounds', async () => {
const bridge = new AcpBridge({
cliEntryPath: '/tmp/qwen',
cwd: '/tmp',
}) as unknown as TestableAcpBridge;
bridge.child = { killed: false, exitCode: null };
bridge.connection = {
extMethod: vi.fn(),
prompt: vi.fn(async () => {
bridge.emit('textChunk', 's-1', 'Updating todos. ');
bridge.handleSessionUpdate({
sessionId: 's-1',
update: {
sessionUpdate: 'plan',
entries: [{ content: 'Task', status: 'pending' }],
},
});
bridge.emit('textChunk', 's-1', 'Done.');
}),
};

await expect(bridge.prompt('s-1', 'question')).resolves.toBe('Done.');
});

it('treats permission requests as turn boundaries', async () => {
const bridge = new AcpBridge({
cliEntryPath: '/tmp/qwen',
cwd: '/tmp',
}) as unknown as TestableAcpBridge;
bridge.child = { killed: false, exitCode: null };
bridge.on('permissionRequest', (event) => {
void bridge.respondToPermission(event.requestId, {
outcome: { outcome: 'selected', optionId: 'proceed_once' },
});
});
bridge.connection = {
extMethod: vi.fn(),
prompt: vi.fn(async () => {
bridge.emit('textChunk', 's-1', 'I need permission. ');
await bridge.requestPermission({
sessionId: 's-1',
toolCall: {
toolCallId: 'tool-1',
kind: 'shell',
title: 'Run command',
},
options: [{ optionId: 'proceed_once', name: 'Allow' }],
});
bridge.emit('textChunk', 's-1', 'Final answer.');
}),
};

await expect(bridge.prompt('s-1', 'question')).resolves.toBe(
'Final answer.',
);
});

it('rejects channel loop tool calls when no handler matches the session', () => {
const bridge = new AcpBridge({
cliEntryPath: '/tmp/qwen',
Expand Down
17 changes: 17 additions & 0 deletions packages/channels/base/src/AcpBridge.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -222,7 +222,11 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge {
const onChunk = (sid: string, chunk: string) => {
if (sid === sessionId) chunks.push(chunk);
};
const clearChunks = (sid: string) => {
if (sid === sessionId) chunks.length = 0;
};
this.on('textChunk', onChunk);
this.on('responseBoundary', clearChunks);

const prompt: Array<Record<string, unknown>> = [];
if (options?.imageBase64 && options.imageMimeType) {
Expand All @@ -241,6 +245,7 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge {
});
} finally {
this.off('textChunk', onChunk);
this.off('responseBoundary', clearChunks);
}

return chunks.join('');
Expand Down Expand Up @@ -316,9 +321,16 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge {
status: (update['status'] as string) || 'pending',
rawInput: update['rawInput'] as Record<string, unknown> | undefined,
};
if (event.status === 'pending') {
this.emitResponseBoundary(sessionId);
}
this.emit('toolCall', event);
break;
}
case 'plan': {
this.emitResponseBoundary(sessionId);
break;
}
case 'available_commands_update': {
if (Array.isArray(update['availableCommands'])) {
this._availableCommands = (
Expand Down Expand Up @@ -375,6 +387,7 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge {
}, ACP_PERMISSION_RESPONSE_TIMEOUT_MS);
timeout.unref?.();
this.pendingPermissions.set(requestId, { sessionId, resolve, timeout });
this.emitResponseBoundary(sessionId);
this.emit('permissionRequest', {
requestId,
sessionId,
Expand All @@ -383,6 +396,10 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge {
});
}

@wenshao wenshao Jul 10, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

[Suggestion] emitResponseBoundary and the clearChunks callback silently zero the chunks array with no log, counter, or diagnostic. When debugging boundary-related issues (e.g., empty responses), there is no way to distinguish "agent generated no text" from "boundaries discarded text" from "bug swallowed chunks" without attaching a debugger.

Consider adding a debug log:

private emitResponseBoundary(sessionId: string): void {
  process.stderr.write(`[${this.constructor.name}] responseBoundary for session ${sessionId}\n`);
  this.emit('responseBoundary', sessionId);
}

— qwen3.7-max via Qwen Code /review

private emitResponseBoundary(sessionId: string): void {
this.emit('responseBoundary', sessionId);
}

private resolvePendingPermissions(sessionId?: string): void {
const response: RequestPermissionResponse = {
outcome: { outcome: 'cancelled' },
Expand Down
1 change: 1 addition & 0 deletions packages/channels/base/src/ChannelAgentBridge.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ export interface PermissionResolvedEvent {
interface ChannelAgentBridgeEventMap {
sessionDied: [SessionDiedEvent];
textChunk: [sessionId: string, chunk: string];
responseBoundary: [sessionId: string];
toolCall: [ToolCallEvent];
permissionRequest: [PermissionRequestEvent];
permissionResolved: [PermissionResolvedEvent];
Expand Down
25 changes: 25 additions & 0 deletions packages/channels/base/src/ChannelBase.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7720,6 +7720,31 @@ describe('ChannelBase', () => {
expect(ch.sent.length).toBeGreaterThanOrEqual(1);
});

it('drops buffered block stream text at response boundaries', async () => {
(bridge.prompt as ReturnType<typeof vi.fn>).mockImplementation(
(sid: string) => {
(bridge as unknown as EventEmitter).emit(
'textChunk',
sid,
'intermediate ',
);
(bridge as unknown as EventEmitter).emit('responseBoundary', sid);
(bridge as unknown as EventEmitter).emit('textChunk', sid, 'final');
return Promise.resolve('final');
},
);

const ch = createChannel({
blockStreaming: 'on',
blockStreamingChunk: { minChars: 100, maxChars: 1000 },
blockStreamingCoalesce: { idleMs: 0 },
});

await ch.handleInbound(envelope());

expect(ch.sent.map((message) => message.text)).toEqual(['final']);
});

it('does not emit buffered stream text after cancellation', async () => {
vi.useFakeTimers();
try {
Expand Down
17 changes: 17 additions & 0 deletions packages/channels/base/src/ChannelBase.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -792,8 +792,15 @@ export abstract class ChannelBase {
releaseHeldChunks();
}
};
const onResponseBoundary = (sid: string) => {
if (sid !== sessionId || promptState.cancelled) {
return;
}
heldChunks.length = 0;
};

@wenshao wenshao Jul 9, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

[Critical] onResponseBoundary unconditionally clears heldChunks without checking promptState.cancelPending. During the cancel-pending window, onChunk correctly holds chunks (guarded by if (!promptState.cancelPending) releaseHeldChunks()), but if a responseBoundary fires while cancel is in flight (e.g., the agent emits a pending tool_call before the cancel RPC settles), heldChunks is wiped. If the cancel then fails, releaseHeldChunks() replays an empty array — text that should have been delivered to the user is silently lost.

Apply the same cancelPending guard at both onResponseBoundary locations (line ~795 and ~3558):

Suggested change
};
const onResponseBoundary = (sid: string) => {
if (sid !== sessionId || promptState.cancelled || promptState.cancelPending) {
return;
}
heldChunks.length = 0;
};

When the cancel succeeds, chunks are discarded via cancelled anyway; when it fails, they're preserved for replay.

— qwen3.7-max via Qwen Code /review

const promptBridge = this.bridge;
promptBridge.on('textChunk', onChunk);
promptBridge.on('responseBoundary', onResponseBoundary);

try {
const response = await this.runLoopBridgePrompt(
Expand Down Expand Up @@ -884,6 +891,7 @@ export abstract class ChannelBase {
throw err;
} finally {
promptBridge.off('textChunk', onChunk);
promptBridge.off('responseBoundary', onResponseBoundary);
const stillCurrent = this.activePrompts.get(sessionId) === promptState;
if (!promptState.clearEvicted) {
try {
Expand Down Expand Up @@ -3547,8 +3555,16 @@ export abstract class ChannelBase {
releaseHeldChunks();
}
};
const onResponseBoundary = (sid: string) => {
if (sid !== sessionId || promptState.cancelled) {
return;
}
heldChunks.length = 0;
streamer?.stop();

@wenshao wenshao Jul 9, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

[Critical] onResponseBoundary clears heldChunks at the ChannelBase level but does not notify adapters to reset their own text accumulators. In non-block-streaming mode, releaseHeldChunks() feeds adapter overrides via this.onResponseChunk():

  • Feishu: cardState.accumulatedText (FeishuAdapter.ts:905)
  • QQBot: streamState.buffer (QQChannel.ts:494)

When the boundary fires, heldChunks is cleared but the adapter's accumulated intermediate text persists. The final response from bridge.prompt() is then delivered via onResponseComplete — the user sees both the intermediate card text and the final answer, which is the exact bug this PR aims to fix leaking through a different path.

In block-streaming mode this is not an issue: Feishu's onResponseChunk returns early (if (this.config.blockStreaming === 'on') return) and QQBot similarly skips.

Suggested fix: Add an adapter-level reset hook (e.g., a protected onResponseBoundary(sessionId) method on ChannelBase that adapters can override) and call it from the boundary handler. Feishu should reset cardState.accumulatedText; QQBot should reset streamState.buffer. This applies to both onResponseBoundary locations.

— qwen3.7-max via Qwen Code /review

};
const promptBridge = this.bridge;
promptBridge.on('textChunk', onChunk);
promptBridge.on('responseBoundary', onResponseBoundary);

try {
const response = await promptBridge.prompt(sessionId, promptToSend, {
Expand Down Expand Up @@ -3615,6 +3631,7 @@ export abstract class ChannelBase {
throw err;
} finally {
promptBridge.off('textChunk', onChunk);
promptBridge.off('responseBoundary', onResponseBoundary);
streamer?.stop();
// Identity guard: a turn that wedged past /clear's bounded wait gets
// EVICTED — /clear gives up on active.done, deletes activePrompts, and a
Expand Down
Loading
Loading