feat: proxy() — typed ACP proxies, the TS counterpart of the Rust Proxy role

README.md

@@ -32,6 +32,8 @@ If you're building an [Agent](https://agentclientprotocol.com/protocol/overview#
 
 If you're building a [Client](https://agentclientprotocol.com/protocol/overview#client), start with `client({ name })`, register client-side handlers such as `requestPermission(...)` and `sessionUpdate(...)`, then run your agent workflow with `connectWith(stream, async (ctx) => ...)`.
 
+If you're building something that sits between the two — a logger, an authorization gate, a message transformer — start with `proxy()`. Register typed handlers with `onRequestFromClient(...)` / `onNotificationFromAgent(...)` (the method names say which direction they intercept) just like the agent/client builders, then call `connect({ client, agent })`. Anything you don't claim is forwarded untouched in both directions; handlers can rewrite, answer, or drop messages before they cross.
+
 ### Study a Production Implementation
 
 For a complete, production-ready implementation, check out the [Gemini CLI Agent](https://github.com/google-gemini/gemini-cli/blob/main/packages/cli/src/zed-integration/zedIntegration.ts).

src/acp.ts

@@ -21,6 +21,19 @@ export {
   PROTOCOL_VERSION,
 } from "./schema/index.js";
 export * from "./stream.js";
+export { proxy } from "./proxy.js";
+// ProxyBuilder and ProxySideBuilder are type-only on purpose: proxy() is the
+// sole factory, so their constructors are not part of the public contract.
+export type {
+  ProxyBuilder,
+  ProxyHandle,
+  ProxyNotificationContext,
+  ProxyNotificationHandler,
+  ProxyRequestContext,
+  ProxyRequestHandler,
+  ProxySideConnection,
+  ProxyStreams,
+} from "./proxy.js";
 export { RequestError } from "./jsonrpc.js";
 export type {
   AnyMessage,
@@ -30,19 +43,27 @@ export type {
   ErrorResponse,
   JsonRpcId,
   MaybePromise,
+  ParamsParser,
   Result,
   SendRequestOptions,
 } from "./jsonrpc.js";
 
 import type { Stream } from "./stream.js";
-import { Connection, Handled, HandlerRegistration } from "./jsonrpc.js";
+import {
+  Connection,
+  Handled,
+  HandlerRegistration,
+  linkClosed,
+  parseParams,
+} from "./jsonrpc.js";
 import type {
   AnyMessage,
   ConnectionBuilder,
   ConnectionContext,
   HandleResult,
   IncomingMessage,
   JsonRpcId,
+  ParamsParser,
   JsonRpcHandler,
   MaybePromise,
   SendRequestOptions,
@@ -61,7 +82,14 @@ function isStream(value: unknown): value is Stream {
   );
 }
 
-function memoryStreamPair(): [Stream, Stream] {
+/**
+ * Creates a pair of in-memory streams wired back to back.
+ *
+ * Messages written to one stream are read from the other. Use this to
+ * connect ACP endpoints in the same process — for example an app to a
+ * `proxy(...)` side, or endpoints under test — without a transport.
+ */
+export function inMemoryStreamPair(): [Stream, Stream] {
   const leftToRight = new TransformStream<AnyMessage>();
   const rightToLeft = new TransformStream<AnyMessage>();
   return [
@@ -911,14 +939,6 @@ export type AppOptions = {
  * A Zod schema can be passed directly because schemas expose a compatible
  * `parse(...)` method.
  */
-export type ParamsParser<Params> =
-  | {
-      /**
-       * Parses raw JSON-RPC params into the handler's typed params.
-       */
-      parse: (params: unknown) => Params;
-    }
-  | ((params: unknown) => Params);
 
 /**
  * Common context passed to agent-side handlers.
@@ -1034,21 +1054,6 @@ export type ClientConnectHandler = (
   connection: ClientConnection,
 ) => MaybePromise<void>;
 
-function parseParams<Params>(
-  parser: ParamsParser<Params> | undefined,
-  params: unknown,
-): Params {
-  if (!parser) {
-    return params as Params;
-  }
-
-  if (typeof parser === "function") {
-    return parser(params);
-  }
-
-  return parser.parse(params);
-}
-
 type AcpRequestSpec<Params, Response, WireResponse = Response> = {
   method: string;
   params?: ParamsParser<Params>;
@@ -2011,12 +2016,11 @@ export class AgentApp {
       return state;
     }
 
-    const [thisStream, peerStream] = memoryStreamPair();
+    const [thisStream, peerStream] = inMemoryStreamPair();
     const peerRawConnection = target[appBuilder]().connect(peerStream);
     const peerConnection = clientConnection(peerRawConnection);
     const state = this.openStreamConnection(thisStream);
-    void state.rawConnection.closed.then(() => peerConnection.close());
-    void peerRawConnection.closed.then(() => state.connection.close());
+    linkClosed(state.rawConnection, peerRawConnection);
     try {
       target[runClientConnectHandlers](peerConnection);
       this[runAgentConnectHandlers](state.connection);
@@ -2255,12 +2259,11 @@ export class ClientApp {
       return state;
     }
 
-    const [thisStream, peerStream] = memoryStreamPair();
+    const [thisStream, peerStream] = inMemoryStreamPair();
     const peerRawConnection = target[appBuilder]().connect(peerStream);
     const peerConnection = agentConnection(peerRawConnection);
     const state = this.openStreamConnection(thisStream);
-    void state.rawConnection.closed.then(() => peerConnection.close());
-    void peerRawConnection.closed.then(() => state.connection.close());
+    linkClosed(state.rawConnection, peerRawConnection);
     try {
       target[runAgentConnectHandlers](peerConnection);
       this[runClientConnectHandlers](state.connection);

src/examples/README.md

@@ -4,6 +4,7 @@ This directory contains examples using the [ACP](https://agentclientprotocol.com
 
 - [`agent.ts`](./agent.ts) - A minimal agent implementation that simulates LLM interaction
 - [`client.ts`](./client.ts) - A minimal client implementation that spawns the [`agent.ts`](./agent.ts) as a subprocess
+- [`proxy.ts`](./proxy.ts) - A pass-through proxy that wraps any agent command and logs the messages crossing it in both directions
 - [`http-server.ts`](./http-server.ts) - A minimal ACP Streamable HTTP server with WebSocket upgrade support
 - [`http-client.ts`](./http-client.ts) - A minimal client using `createHttpStream`
 - [`ws-client.ts`](./ws-client.ts) - A minimal client using `createWebSocketStream`

src/examples/proxy.ts

@@ -0,0 +1,70 @@
+#!/usr/bin/env node
+
+import { spawn } from "node:child_process";
+import { dirname, join } from "node:path";
+import { Readable, Writable } from "node:stream";
+import { fileURLToPath } from "node:url";
+
+import * as acp from "../acp.js";
+
+// A minimal ACP proxy: it presents as an agent on its own stdio, spawns the
+// real agent as a subprocess, and forwards every message between the two
+// while logging traffic to stderr. Point an ACP client (like Zed) at this
+// script exactly as it would run the agent directly — neither side needs to
+// know the proxy is there.
+//
+// Usage: proxy.ts [command args...]
+// Runs the example agent from this directory when no command is given.
+
+function logAndForward(
+  direction: string,
+): acp.ProxyRequestHandler<unknown, unknown> {
+  return ({ method, params, forward }) => {
+    console.error(`[proxy] ${direction} request: ${method}`);
+    return forward(params);
+  };
+}
+
+function logAndForwardNotification(
+  direction: string,
+): acp.ProxyNotificationHandler<unknown> {
+  return async ({ method, params, forward }) => {
+    console.error(`[proxy] ${direction} notification: ${method}`);
+    await forward(params);
+  };
+}
+
+// Spawn the wrapped agent: the command from argv, or the example agent.
+const npxCmd = process.platform === "win32" ? "npx.cmd" : "npx";
+const exampleAgent = join(dirname(fileURLToPath(import.meta.url)), "agent.ts");
+const [command, ...args] =
+  process.argv.length > 2
+    ? process.argv.slice(2)
+    : [npxCmd, "tsx", exampleAgent];
+const agentProcess = spawn(command, args, {
+  stdio: ["pipe", "pipe", "inherit"],
+});
+
+// "*" catches anything without an exact registration; typed interception is
+// also available, e.g.
+// .onRequestFromClient("session/prompt", async ({ params, forward }) => ...).
+const handle = acp
+  .proxy()
+  .onRequestFromClient("*", logAndForward("client → agent"))
+  .onNotificationFromClient("*", logAndForwardNotification("client → agent"))
+  .onRequestFromAgent("*", logAndForward("agent → client"))
+  .onNotificationFromAgent("*", logAndForwardNotification("agent → client"))
+  .connect({
+    client: acp.ndJsonStream(
+      Writable.toWeb(process.stdout),
+      Readable.toWeb(process.stdin) as ReadableStream<Uint8Array>,
+    ),
+    agent: acp.ndJsonStream(
+      Writable.toWeb(agentProcess.stdin!),
+      Readable.toWeb(agentProcess.stdout!) as ReadableStream<Uint8Array>,
+    ),
+  });
+
+agentProcess.once("exit", () => handle.close());
+await handle.closed;
+agentProcess.kill();

src/jsonrpc.test.ts

@@ -1,8 +1,13 @@
 import { describe, expect, it, vi } from "vitest";
 
-import { Connection, RequestError, isJsonRpcMessage } from "./jsonrpc.js";
+import {
+  Connection,
+  RequestError,
+  isJsonRpcMessage,
+  linkClosed,
+} from "./jsonrpc.js";
 import type { AnyMessage, RequestResponder } from "./jsonrpc.js";
-import type { Stream } from "./stream.js";
+import { inMemoryStreamPair } from "./acp.js";
 
 type ConnectionInternals = {
   pendingResponses: Map<string | number | null, unknown>;
@@ -466,17 +471,20 @@ describe("JSON-RPC malformed peer messages", () => {
   });
 });
 
-function memoryStreamPair(): [Stream, Stream] {
-  const leftToRight = new TransformStream<AnyMessage>();
-  const rightToLeft = new TransformStream<AnyMessage>();
-  return [
-    {
-      readable: rightToLeft.readable,
-      writable: leftToRight.writable,
-    },
-    {
-      readable: leftToRight.readable,
-      writable: rightToLeft.writable,
-    },
-  ];
-}
+const memoryStreamPair = inMemoryStreamPair;
+
+describe("linkClosed", () => {
+  it("closes the paired connection with the same reason", async () => {
+    const [aStream] = inMemoryStreamPair();
+    const [bStream] = inMemoryStreamPair();
+    const a = new Connection(aStream, []);
+    const b = new Connection(bStream, []);
+    linkClosed(a, b);
+
+    const pendingOnB = b.sendRequest("anything", {});
+    a.close(new Error("a went away"));
+
+    await expect(pendingOnB).rejects.toThrow("a went away");
+    await b.closed;
+  });
+});

src/jsonrpc.ts

@@ -225,6 +225,40 @@ type ConnectionPendingResponse = {
  */
 export type MaybePromise<T> = T | Promise<T>;
 
+/**
+ * Parser converting raw JSON-RPC params into a handler's typed params.
+ *
+ * Accepts either a plain function or an object with a `parse` method (such
+ * as a Zod schema).
+ */
+export type ParamsParser<Params> =
+  | {
+      /**
+       * Parses raw JSON-RPC params into the handler's typed params.
+       */
+      parse: (params: unknown) => Params;
+    }
+  | ((params: unknown) => Params);
+
+/**
+ * Runs a {@link ParamsParser} over raw params; without a parser the params
+ * pass through untouched.
+ */
+export function parseParams<Params>(
+  parser: ParamsParser<Params> | undefined,
+  params: unknown,
+): Params {
+  if (!parser) {
+    return params as Params;
+  }
+
+  if (typeof parser === "function") {
+    return parser(params);
+  }
+
+  return parser.parse(params);
+}
+
 /**
  * Incoming request passed to JSON-RPC handlers.
  */
@@ -502,6 +536,16 @@ export class RequestResponder<Resp = unknown> {
     return this.respondWithResult({ error: errorResponse });
   }
 
+  /**
+   * Sends the JSON-RPC response for an error thrown while handling the
+   * request: `RequestError` keeps its code/message/data, an abort after
+   * cancellation maps to request-cancelled, anything else becomes a generic
+   * internal error.
+   */
+  respondWithThrown(error: unknown): Promise<void> {
+    return this.respondWithResult(errorToRequestResult(error, this.signal));
+  }
+
   /**
    * Sends a complete JSON-RPC result payload.
    */
@@ -987,9 +1031,7 @@ export class Connection {
       }
 
       if (current.kind === "request" && !current.responder.responded) {
-        await current.responder.respondWithResult(
-          errorToRequestResult(error, current.responder.signal),
-        );
+        await current.responder.respondWithThrown(error);
       } else {
         const response = errorToResult(error);
         if ("error" in response) {
@@ -1114,6 +1156,17 @@ export class Connection {
   }
 }
 
+/**
+ * Closes each connection when the other closes.
+ *
+ * The close reason is propagated so pending requests on the surviving side
+ * reject with the true cause rather than a generic connection-closed error.
+ */
+export function linkClosed(a: Connection, b: Connection): void {
+  void a.closed.then(() => b.close(a.signal.reason));
+  void b.closed.then(() => a.close(b.signal.reason));
+}
+
 /**
  * Builder for a lower-level handler-based JSON-RPC connection.
  */