fix(acp): bound unread raw notification retention

packages/effect-acp/src/agent.ts

@@ -36,7 +36,9 @@ export class AcpAgent extends Context.Service<
   {
     readonly raw: {
       /**
-       * Stream of inbound ACP notifications observed on the connection.
+       * Stream of inbound ACP notifications observed on the connection. Unread
+       * ones are kept best-effort: the newest 32 whose JSON lines fit in 16Mi
+       * characters, always including the latest. Ends when the connection does.
        */
       readonly notifications: Stream.Stream<AcpProtocol.AcpIncomingNotification>;
       /**

packages/effect-acp/src/client.ts

@@ -51,6 +51,11 @@ export interface AcpClientOptions {
 }
 
 type AcpClientRaw = {
+  /**
+   * Inbound notifications. Unread ones are kept best-effort: the newest 32 whose
+   * JSON lines fit in 16Mi characters, always including the latest. Ends with the
+   * connection.
+   */
   readonly notifications: Stream.Stream<AcpProtocol.AcpIncomingNotification>;
   readonly request: (method: string, payload: unknown) => Effect.Effect<unknown, AcpError.AcpError>;
   readonly notify: (method: string, payload: unknown) => Effect.Effect<void, AcpError.AcpError>;

packages/effect-acp/src/protocol.test.ts

@@ -64,6 +64,16 @@ const JsonRpcErrorResponse = Schema.Struct({
 });
 const encodeUnknownJsonString = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
 const encoder = new TextEncoder();
+const MIB = 1024 * 1024;
+const decodeIndex = Schema.decodeUnknownSync(Schema.Struct({ index: Schema.Number }));
+const largeNotification = (index: number, textLength: number) =>
+  encoder.encode(
+    `${encodeUnknownJsonString({
+      jsonrpc: "2.0",
+      method: "x/large",
+      params: { index, text: "x".repeat(textLength) },
+    })}\n`,
+  );
 const mockPeerPath = Effect.map(Effect.service(Path.Path), (path) =>
   path.join(import.meta.dirname, "../test/fixtures/acp-mock-peer.ts"),
 );
@@ -183,6 +193,138 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
     }),
   );
 
+  it.effect("keeps the newest unread raw notifications whose JSON fits in 16Mi characters", () =>
+    Effect.gen(function* () {
+      const { stdio, input } = yield* makeInMemoryStdio();
+      const handled = yield* Queue.unbounded<number>();
+      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
+        stdio,
+        serverRequestMethods: new Set(),
+        onNotification: (notification) =>
+          Queue.offer(handled, decodeIndex(notification.params).index).pipe(Effect.asVoid),
+      });
+
+      // Ten 3 MiB notifications: only the newest five fit in the budget.
+      for (let index = 0; index < 10; index++) {
+        yield* Queue.offer(input, largeNotification(index, 3 * MIB));
+        assert.equal(yield* Queue.take(handled), index);
+      }
+
+      const retained = yield* transport.incoming.pipe(Stream.take(5), Stream.runCollect);
+      assert.deepEqual(
+        retained.map((notification) => decodeIndex(notification.params).index),
+        [5, 6, 7, 8, 9],
+      );
+    }),
+  );
+
+  it.effect("keeps a single unread raw notification that alone exceeds the budget", () =>
+    Effect.gen(function* () {
+      const { stdio, input } = yield* makeInMemoryStdio();
+      const handled = yield* Queue.unbounded<number>();
+      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
+        stdio,
+        serverRequestMethods: new Set(),
+        onNotification: (notification) =>
+          Queue.offer(handled, decodeIndex(notification.params).index).pipe(Effect.asVoid),
+      });
+
+      yield* Queue.offer(input, largeNotification(0, 1));
+      yield* Queue.take(handled);
+      yield* Queue.offer(input, largeNotification(1, 17 * MIB));
+      yield* Queue.take(handled);
+
+      const [retained] = yield* transport.incoming.pipe(Stream.take(1), Stream.runCollect);
+      assert.equal(decodeIndex(retained?.params).index, 1);
+    }),
+  );
+
+  it.effect("delivers every raw notification to a reader that keeps up, then ends", () =>
+    Effect.gen(function* () {
+      const { stdio, input } = yield* makeInMemoryStdio();
+      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
+        stdio,
+        serverRequestMethods: new Set(),
+      });
+      const received = yield* Queue.unbounded<number>();
+      const reader = yield* transport.incoming.pipe(
+        Stream.map((notification) => decodeIndex(notification.params).index),
+        Stream.tap((index) => Queue.offer(received, index)),
+        Stream.runCollect,
+        Effect.forkScoped,
+      );
+
+      // 30 MiB in total, well past the unread budget.
+      for (let index = 0; index < 10; index++) {
+        yield* Queue.offer(input, largeNotification(index, 3 * MIB));
+        assert.equal(yield* Queue.take(received), index);
+      }
+      yield* Queue.end(input);
+
+      assert.deepEqual(yield* Fiber.join(reader), [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
+    }),
+  );
+
+  it.effect("ends the raw notification stream when the connection scope closes", () =>
+    Effect.gen(function* () {
+      const { stdio } = yield* makeInMemoryStdio();
+      const connection = yield* Scope.make();
+      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
+        stdio,
+        serverRequestMethods: new Set(),
+      }).pipe(Scope.provide(connection));
+      const reader = yield* transport.incoming.pipe(Stream.runCollect, Effect.forkScoped);
+
+      yield* Scope.close(connection, Exit.void);
+
+      assert.deepEqual(yield* Fiber.join(reader), []);
+    }),
+  );
+
+  it.effect("keeps the ended raw backlog intact when notifications arrive after termination", () =>
+    Effect.gen(function* () {
+      const { stdio, input } = yield* makeInMemoryStdio();
+      const handled = yield* Queue.unbounded<number>();
+      const terminated = yield* Deferred.make<void>();
+      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
+        stdio: Stdio.make({
+          args: stdio.args,
+          stdin: stdio.stdin,
+          stdout: () => Sink.forEach(() => Effect.die("stdout failed")),
+          stderr: () => Sink.drain,
+        }),
+        serverRequestMethods: new Set(),
+        onNotification: (notification) =>
+          Queue.offer(handled, decodeIndex(notification.params).index).pipe(Effect.asVoid),
+        onTermination: () => Deferred.succeed(terminated, undefined).pipe(Effect.asVoid),
+      });
+      const deliver = (index: number) =>
+        Queue.offer(
+          input,
+          encoder.encode(
+            `${encodeUnknownJsonString({ jsonrpc: "2.0", method: "x/late", params: { index } })}\n`,
+          ),
+        ).pipe(Effect.andThen(Queue.take(handled)));
+
+      for (let index = 0; index < 32; index++) {
+        assert.equal(yield* deliver(index), index);
+      }
+      // A failed write terminates the connection while the reader keeps running.
+      yield* transport.notify("x/fail", {}).pipe(Effect.exit);
+      yield* Deferred.await(terminated);
+      assert.equal(yield* deliver(32), 32);
+
+      const retained = yield* transport.incoming.pipe(
+        Stream.map((notification) => decodeIndex(notification.params).index),
+        Stream.runCollect,
+      );
+      assert.deepEqual(
+        retained,
+        Array.from({ length: 32 }, (_, index) => index),
+      );
+    }),
+  );
+
   it.effect("decodes standard JSON-RPC errors into typed ACP request failures", () =>
     Effect.gen(function* () {
       const { stdio, input, output } = yield* makeInMemoryStdio();

packages/effect-acp/src/protocol.ts

@@ -128,6 +128,17 @@ interface AcpPendingRequest {
   readonly method: string;
 }
 
+/** A decoded wire message and the length of the JSON line it arrived on. */
+interface AcpDecodedWireMessage {
+  readonly message: RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded;
+  readonly lineLength: number;
+}
+
+interface AcpBufferedRawNotification {
+  readonly notification: AcpIncomingNotification;
+  readonly lineLength: number;
+}
+
 interface AcpOutgoingWrite {
   readonly acknowledgement?: Deferred.Deferred<void, AcpError.AcpError>;
   readonly payload: string | Uint8Array;
@@ -153,19 +164,31 @@ const makeStrictNdJsonRpcParser = () => {
   const decoder = new TextDecoder();
   let buffer = "";
   return {
-    decode: (bytes: Uint8Array | string): ReadonlyArray<unknown> => {
+    // Every message in a batch line is charged the whole line.
+    decode: (bytes: Uint8Array | string): ReadonlyArray<AcpDecodedWireMessage> => {
       buffer += typeof bytes === "string" ? bytes : decoder.decode(bytes, { stream: true });
       const lines = buffer.split("\n");
       buffer = lines.pop() ?? "";
-      return lines.flatMap((line) => codec.decode(line));
+      return lines.flatMap((line) =>
+        codec.decode(line).map((message) => ({
+          message: normalizeAcpJsonRpcError(
+            message as RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded,
+          ),
+          lineLength: line.length,
+        })),
+      );
     },
     encode: (response: Parameters<typeof codec.encode>[0]) => {
       const encoded = codec.encode(response);
       return encoded === undefined ? undefined : `${encoded}\n`;
     },
   };
 };
+// Unread raw notifications are best-effort. The newest is always kept; older
+// ones are dropped once the backlog passes either limit. The length limit counts
+// JSON line characters (UTF-16 code units), not bytes or heap size.
 const MAX_BUFFERED_RAW_NOTIFICATIONS = 32;
+const MAX_BUFFERED_RAW_NOTIFICATION_JSON_LENGTH = 16 * 1024 * 1024;
 // Outbound JSON-RPC notification: no `id`, so peers never treat it as a request.
 const encodeJsonRpcNotification = Schema.encodeUnknownExit(
   Schema.fromJsonString(
@@ -213,9 +236,14 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
   const parser = makeStrictNdJsonRpcParser();
   const serverQueue = yield* Queue.unbounded<RpcMessage.FromClientEncoded>();
   const clientQueue = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
-  const notificationQueue = yield* Queue.sliding<AcpIncomingNotification>(
-    MAX_BUFFERED_RAW_NOTIFICATIONS,
-  );
+  const rawNotifications = yield* Queue.unbounded<AcpBufferedRawNotification, Cause.Done<void>>();
+  // JSON length of the notifications offered since the queue was last empty.
+  // `incoming` takes the whole backlog on every pull, so an empty queue holds
+  // nothing, and a read can only make this an overestimate.
+  let rawNotificationsJsonLength = 0;
+  // Raw readers finish with the connection: termination ends the queue, and so
+  // does closing the scope, which interrupts the reader before it terminates.
+  yield* Effect.addFinalizer(() => Queue.end(rawNotifications));
   const disconnects = yield* Queue.unbounded<number>();
   const outgoing = yield* Queue.unbounded<AcpOutgoingWrite, AcpError.AcpError | Cause.Done<void>>();
   const outgoingWriterState = yield* Ref.make<AcpOutgoingWriterState>({
@@ -384,8 +412,27 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
       ),
     );
 
-  const dispatchNotification = (notification: AcpIncomingNotification) =>
-    Queue.offer(notificationQueue, notification).pipe(
+  // Offer first so an ended queue rejects the notification without evicting the
+  // backlog readers are still draining. Offer and eviction share one sync block,
+  // so readers never see more than the limits allow.
+  const bufferRawNotification = (notification: AcpIncomingNotification, lineLength: number) =>
+    Effect.sync(() => {
+      if (Queue.sizeUnsafe(rawNotifications) === 0) rawNotificationsJsonLength = 0;
+      if (!Queue.offerUnsafe(rawNotifications, { notification, lineLength })) return;
+      rawNotificationsJsonLength += lineLength;
+      while (
+        Queue.sizeUnsafe(rawNotifications) > MAX_BUFFERED_RAW_NOTIFICATIONS ||
+        (Queue.sizeUnsafe(rawNotifications) > 1 &&
+          rawNotificationsJsonLength > MAX_BUFFERED_RAW_NOTIFICATION_JSON_LENGTH)
+      ) {
+        const dropped = Queue.takeUnsafe(rawNotifications);
+        if (dropped === undefined || Exit.isFailure(dropped)) break;
+        rawNotificationsJsonLength -= dropped.value.lineLength;
+      }
+    });
+
+  const dispatchNotification = (notification: AcpIncomingNotification, lineLength: number) =>
+    bufferRawNotification(notification, lineLength).pipe(
       Effect.andThen(
         // A failing or dying handler must not stop the reader, or every later
         // message on the connection goes unanswered.
@@ -415,6 +462,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
       return [
         Effect.gen(function* () {
           yield* Queue.offer(disconnects, 0);
+          yield* Queue.end(rawNotifications);
           const error = yield* classify();
           yield* Deferred.fail(terminationFailure, error);
           yield* failAllExtPending(error);
@@ -499,7 +547,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
       );
   };
 
-  const handleRequestEncoded = (message: RpcMessage.RequestEncoded) => {
+  const handleRequestEncoded = (message: RpcMessage.RequestEncoded, lineLength: number) => {
     if (message.id === "") {
       if (message.tag === CLIENT_METHODS.session_update) {
         return decodeSessionUpdate(message.payload).pipe(
@@ -518,7 +566,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
               cause,
             ),
           ),
-          Effect.flatMap(dispatchNotification),
+          Effect.flatMap((notification) => dispatchNotification(notification, lineLength)),
         );
       }
       if (
@@ -542,14 +590,13 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
               cause,
             ),
           ),
-          Effect.flatMap(dispatchNotification),
+          Effect.flatMap((notification) => dispatchNotification(notification, lineLength)),
         );
       }
-      return dispatchNotification({
-        _tag: "ExtNotification",
-        method: message.tag,
-        params: message.payload,
-      });
+      return dispatchNotification(
+        { _tag: "ExtNotification", method: message.tag, params: message.payload },
+        lineLength,
+      );
     }
 
     const observeIncoming =
@@ -625,12 +672,13 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
       }),
     );
 
-  const routeDecodedMessage = (
-    message: RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded,
-  ): Effect.Effect<void, AcpError.AcpError> => {
+  const routeDecodedMessage = ({
+    message,
+    lineLength,
+  }: AcpDecodedWireMessage): Effect.Effect<void, AcpError.AcpError> => {
     switch (message._tag) {
       case "Request":
-        return handleRequestEncoded(message);
+        return handleRequestEncoded(message, lineLength);
       case "Exit":
         return handleExitEncoded(message);
       case "Chunk":
@@ -672,25 +720,22 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
       ).pipe(
         Effect.flatMap(() =>
           Effect.try({
-            try: () =>
-              (
-                parser.decode(data) as ReadonlyArray<
-                  RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded
-                >
-              ).map(normalizeAcpJsonRpcError),
+            try: () => parser.decode(data),
             catch: (cause) =>
               new AcpError.AcpProtocolParseError({
                 operation: "decode-wire-message",
                 cause,
               }),
           }),
         ),
-        Effect.tap((messages) =>
-          logProtocol({
-            direction: "incoming",
-            stage: "decoded",
-            payload: messages,
-          }),
+        Effect.tap((decoded) =>
+          options.logIncoming
+            ? logProtocol({
+                direction: "incoming",
+                stage: "decoded",
+                payload: decoded.map(({ message }) => message),
+              })
+            : Effect.void,
         ),
         Effect.tapErrorTag("AcpProtocolParseError", (error) =>
           logProtocol({
@@ -708,8 +753,8 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
             },
           }),
         ),
-        Effect.flatMap((messages) =>
-          Effect.forEach(messages, routeDecodedMessage, {
+        Effect.flatMap((decoded) =>
+          Effect.forEach(decoded, routeDecodedMessage, {
             discard: true,
           }),
         ),
@@ -970,7 +1015,9 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
     clientProtocol,
     serverProtocol,
     get incoming() {
-      return Stream.fromQueue(notificationQueue);
+      return Stream.fromQueue(rawNotifications).pipe(
+        Stream.map(({ notification }) => notification),
+      );
     },
     request: sendRequest,
     notify: sendNotification,