diff --git a/packages/server/src/server/agent/providers/acp-agent.test.ts b/packages/server/src/server/agent/providers/acp-agent.test.ts index 18672c755..6b1f2501a 100644 --- a/packages/server/src/server/agent/providers/acp-agent.test.ts +++ b/packages/server/src/server/agent/providers/acp-agent.test.ts @@ -5,6 +5,7 @@ import { ACPAgentSession, type SpawnedACPProcess, type SessionStateResponse, + createLoggedNdJsonStream, deriveModelDefinitionsFromACP, deriveModesFromACP, mapACPUsage, @@ -35,6 +36,46 @@ function createSession(): ACPAgentSession { ); } +describe("createLoggedNdJsonStream", () => { + test("routes malformed ACP stdout through the provider logger instead of console.error", async () => { + const input = new TransformStream(); + const output = new TransformStream(); + const logger = { + warn: vi.fn(), + }; + const consoleError = vi.spyOn(console, "error").mockImplementation(() => {}); + + const stream = createLoggedNdJsonStream(output.writable, input.readable, { + logger: logger as unknown as ReturnType, + provider: "gemini", + }); + const reader = stream.readable.getReader(); + const writer = input.writable.getWriter(); + + await writer.write( + new TextEncoder().encode( + 'Please visit the following URL to authorize the application:\n{"jsonrpc":"2.0","method":"ok","params":{}}\n', + ), + ); + + const parsed = await reader.read(); + + expect(parsed.value).toEqual({ jsonrpc: "2.0", method: "ok", params: {} }); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ + provider: "gemini", + linePreview: "Please visit the following URL to authorize the application:", + }), + "ACP agent emitted non-JSON stdout; ignoring line", + ); + expect(consoleError).not.toHaveBeenCalled(); + + await writer.close(); + reader.releaseLock(); + consoleError.mockRestore(); + }); +}); + describe("mapACPUsage", () => { test("maps ACP usage fields into Paseo usage", () => { expect( diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index d4a333ace..988defdac 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/packages/server/src/server/agent/providers/acp-agent.ts @@ -6,8 +6,8 @@ import { Readable, Writable } from "node:stream"; import { ClientSideConnection, PROTOCOL_VERSION, - ndJsonStream, type AgentCapabilities as ACPAgentCapabilities, + type AnyMessage, type Client as ACPClient, type ClientCapabilities as ACPClientCapabilities, type ConfigOptionUpdate, @@ -46,6 +46,7 @@ import { type UsageUpdate, type WaitForTerminalExitRequest, type WriteTextFileRequest, + type Stream as ACPStream, } from "@agentclientprotocol/sdk"; import type { Logger } from "pino"; @@ -112,6 +113,75 @@ const COPILOT_AUTOPILOT_MODE = "https://agentclientprotocol.com/protocol/session // NO_BROWSER is honored by Gemini CLI; other ACP agents ignore it. const PROBE_ENV: Record = { NO_BROWSER: "true" }; +export function createLoggedNdJsonStream( + output: WritableStream, + input: ReadableStream, + options: { logger: Logger; provider: string }, +): ACPStream { + const textEncoder = new TextEncoder(); + const textDecoder = new TextDecoder(); + + const readable = new ReadableStream({ + async start(controller) { + let content = ""; + const reader = input.getReader(); + try { + while (true) { + const { value, done } = await reader.read(); + if (done) { + break; + } + if (!value) { + continue; + } + + content += textDecoder.decode(value, { stream: true }); + const lines = content.split("\n"); + content = lines.pop() || ""; + + for (const line of lines) { + const trimmedLine = line.trim(); + if (!trimmedLine) { + continue; + } + + try { + const message = JSON.parse(trimmedLine) as AnyMessage; + controller.enqueue(message); + } catch (error) { + options.logger.warn( + { + err: error, + provider: options.provider, + linePreview: + trimmedLine.length > 500 ? `${trimmedLine.slice(0, 500)}...` : trimmedLine, + }, + "ACP agent emitted non-JSON stdout; ignoring line", + ); + } + } + } + } finally { + reader.releaseLock(); + controller.close(); + } + }, + }); + + const writable = new WritableStream({ + async write(message) { + const writer = output.getWriter(); + try { + await writer.write(textEncoder.encode(`${JSON.stringify(message)}\n`)); + } finally { + writer.releaseLock(); + } + }, + }); + + return { readable, writable }; +} + type ACPAgentClientOptions = { provider: string; logger: Logger; @@ -517,9 +587,10 @@ export class ACPAgentClient implements AgentClient { throw new Error(`${this.provider} ACP process did not expose stdio pipes`); } - const stream = ndJsonStream( + const stream = createLoggedNdJsonStream( Writable.toWeb(child.stdin) as WritableStream, Readable.toWeb(child.stdout) as ReadableStream, + { logger: this.logger, provider: this.provider }, ); const connection = new ClientSideConnection(() => this.buildProbeClient(), stream); const initialize = (await Promise.race([ @@ -1371,9 +1442,10 @@ export class ACPAgentSession implements AgentSession, ACPClient { throw new Error(`${this.provider} ACP process did not expose stdio pipes`); } - const stream = ndJsonStream( + const stream = createLoggedNdJsonStream( Writable.toWeb(child.stdin) as WritableStream, Readable.toWeb(child.stdout) as ReadableStream, + { logger: this.logger, provider: this.provider }, ); const connection = new ClientSideConnection(() => this, stream); const initialize = await connection.initialize({