Contain malformed ACP stdout logging

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<Uint8Array, Uint8Array>();
+    const output = new TransformStream<Uint8Array, Uint8Array>();
+    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<typeof createTestLogger>,
+      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(

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<string, string> = { NO_BROWSER: "true" };
 
+export function createLoggedNdJsonStream(
+  output: WritableStream<Uint8Array>,
+  input: ReadableStream<Uint8Array>,
+  options: { logger: Logger; provider: string },
+): ACPStream {
+  const textEncoder = new TextEncoder();
+  const textDecoder = new TextDecoder();
+
+  const readable = new ReadableStream<AnyMessage>({
+    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<AnyMessage>({
+    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<Uint8Array>,
       Readable.toWeb(child.stdout) as ReadableStream<Uint8Array>,
+      { 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<Uint8Array>,
       Readable.toWeb(child.stdout) as ReadableStream<Uint8Array>,
+      { logger: this.logger, provider: this.provider },
     );
     const connection = new ClientSideConnection(() => this, stream);
     const initialize = await connection.initialize({