fix(channels): return only final ACP response text

packages/channels/base/src/AcpBridge.test.ts

@@ -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[];
@@ -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;
@@ -246,6 +249,161 @@ 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('returns only the final turn text after auto-approved 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 inspect. ');
+        bridge.handleSessionUpdate({
+          sessionId: 's-1',
+          update: {
+            sessionUpdate: 'tool_call',
+            toolCallId: 'call-1',
+            kind: 'read',
+            title: 'Read',
+            status: 'in_progress',
+          },
+        });
+        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',

packages/channels/base/src/AcpBridge.ts

@@ -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) {
@@ -241,6 +245,7 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge {
       });
     } finally {
       this.off('textChunk', onChunk);
+      this.off('responseBoundary', clearChunks);
     }
 
     return chunks.join('');
@@ -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' || event.status === 'in_progress') {
+          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 = (
@@ -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,
@@ -383,6 +396,10 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge {
     });
   }
 
+  private emitResponseBoundary(sessionId: string): void {
+    this.emit('responseBoundary', sessionId);
+  }
+
   private resolvePendingPermissions(sessionId?: string): void {
     const response: RequestPermissionResponse = {
       outcome: { outcome: 'cancelled' },

packages/channels/base/src/ChannelAgentBridge.ts

@@ -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];

packages/channels/base/src/ChannelBase.test.ts

@@ -48,6 +48,7 @@ class TestChannel extends ChannelBase {
   }> = [];
   responseChunks: Array<{ chatId: string; chunk: string; sessionId: string }> =
     [];
+  responseBoundaries: Array<{ chatId: string; sessionId: string }> = [];
   /** When set, onPromptEnd throws AFTER recording — to exercise the finally guard. */
   throwOnPromptEnd = false;
   responseCompleteGate?: Promise<void>;
@@ -156,6 +157,13 @@ class TestChannel extends ChannelBase {
     this.responseChunks.push({ chatId, chunk, sessionId });
   }
 
+  protected override onResponseBoundary(
+    chatId: string,
+    sessionId: string,
+  ): void {
+    this.responseBoundaries.push({ chatId, sessionId });
+  }
+
   protected override async onResponseComplete(
     chatId: string,
     fullText: string,
@@ -8064,6 +8072,80 @@ 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('preserves held chunks when response boundary fires during cancel', async () => {
+      let resolvePrompt!: (v: string) => void;
+      let rejectCancel!: (e: Error) => void;
+      const pendingPrompt = new Promise<string>((resolve) => {
+        resolvePrompt = resolve;
+      });
+      const pendingCancel = new Promise<void>((_resolve, reject) => {
+        rejectCancel = reject;
+      });
+
+      (bridge.prompt as ReturnType<typeof vi.fn>).mockReturnValue(
+        pendingPrompt,
+      );
+      (bridge.cancelSession as ReturnType<typeof vi.fn>).mockReturnValue(
+        pendingCancel,
+      );
+
+      const ch = createChannel();
+      ch.enableCancelCommand();
+      const prompt = ch.handleInbound(envelope({ text: 'long task' }));
+      for (let i = 0; i < 10 && ch.promptStarts.length === 0; i++) {
+        await Promise.resolve();
+      }
+      expect(ch.promptStarts).toHaveLength(1);
+
+      const cancel = ch.handleInbound(envelope({ text: '/cancel' }));
+      await Promise.resolve();
+
+      (bridge as unknown as EventEmitter).emit(
+        'textChunk',
+        's-1',
+        'held while cancel pending',
+      );
+      (bridge as unknown as EventEmitter).emit('responseBoundary', 's-1');
+
+      rejectCancel(new Error('cancel failed'));
+      await cancel;
+
+      resolvePrompt('final response');
+      await prompt;
+
+      expect(ch.responseBoundaries).toEqual([]);
+      expect(ch.responseChunks).toContainEqual({
+        chatId: 'chat1',
+        chunk: 'held while cancel pending',
+        sessionId: 's-1',
+      });
+    });
+
     it('does not emit buffered stream text after cancellation', async () => {
       vi.useFakeTimers();
       try {