perf(protocol): make raw event retention opt-in

docs/internals/providers.md

@@ -60,6 +60,29 @@ the active runtime and starts the selected provider without reusing an incompati
 bridge is not the Codex turn path, and AgentController never falls back to the legacy Codex loop when
 a Mastra session is absent.
 
+## Raw protocol observation
+
+The [ACP protocol](../../packages/effect-acp/src/protocol.ts) and
+[Codex app-server protocol](../../packages/effect-codex-app-server/src/protocol.ts) retain raw
+observations only when configured before connection. `AcpClientOptions` and `AcpAgentOptions` expose
+`rawNotificationBufferSize` for `raw.notifications`. `CodexAppServerClientOptions` exposes that option
+and `rawRequestBufferSize` for `raw.requests`. Pass them to the package's `make` or layer constructor.
+
+- `0`, the default, retains nothing and completes the raw stream immediately.
+- A positive safe integer `N` keeps the newest `N` unread events in a sliding queue. Overflow drops
+  the oldest observation, including when a reader starts late or falls behind.
+- `"unbounded"` explicitly restores the legacy lossless FIFO within the connection scope. An absent
+  or slow reader can then retain the whole session.
+
+These are work-sharing streams, not broadcasts. Enabled streams drain when input ends; closing the
+connection scope discards the buffer and interrupts readers. Observation never blocks callback
+dispatch. Notification callbacks, request handlers, and replies keep their existing behavior even
+when raw observation is disabled or drops events. Reading `raw.requests` does not send a reply.
+Use handlers for required protocol work, not a lossy observation stream.
+
+This bounds optional transport observations, not provider transcripts or the Mastra Codex/Kimi turn
+path described above.
+
 ## Composio runtime
 
 Composio is an integration provider. Its toolkits appear as named plugins such as Gmail, but Akeru

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

@@ -7,6 +7,7 @@ import * as Queue from "effect/Queue";
 import * as Ref from "effect/Ref";
 import * as Schema from "effect/Schema";
 import * as Scope from "effect/Scope";
+import * as Stream from "effect/Stream";
 
 import { assert, it } from "@effect/vitest";
 
@@ -179,6 +180,42 @@ it.effect("effect-acp agent handles core agent requests and outbound client requ
   }),
 );
 
+it.effect(
+  "effect-acp agent opts into bounded late raw observation without changing cancel callbacks",
+  () =>
+    Effect.gen(function* () {
+      const { stdio, input } = yield* makeInMemoryStdio();
+      const agent = yield* AcpAgent.make(stdio, { rawNotificationBufferSize: 2 });
+      const received = yield* Deferred.make<void>();
+      let handled = 0;
+      yield* agent.handleCancel(() =>
+        Effect.suspend(() => {
+          handled++;
+          return handled === 3
+            ? Deferred.succeed(received, undefined).pipe(Effect.asVoid)
+            : Effect.void;
+        }),
+      );
+      for (const index of [0, 1, 2]) {
+        yield* Queue.offer(
+          input,
+          yield* encodeJsonl(SessionCancelNotification, {
+            jsonrpc: "2.0",
+            method: "session/cancel",
+            params: { sessionId: `session-${index}` },
+          }),
+        );
+      }
+      yield* Deferred.await(received);
+      const replay = yield* agent.raw.notifications.pipe(Stream.take(2), Stream.runCollect);
+      assert.equal(handled, 3);
+      assert.deepEqual(
+        replay.map((notification) => notification.params),
+        [{ sessionId: "session-1" }, { sessionId: "session-2" }],
+      );
+    }),
+);
+
 it.effect("effect-acp agent uses distinct ids for RPC calls and extension requests", () =>
   Effect.gen(function* () {
     const { stdio, input, output } = yield* makeInMemoryStdio();

packages/effect-acp/src/agent.ts

@@ -22,7 +22,10 @@ import {
 } from "./_internal/shared.ts";
 import * as AcpTerminal from "./terminal.ts";
 
-export interface AcpAgentOptions {
+export interface AcpAgentOptions extends Pick<
+  AcpProtocol.AcpPatchedProtocolOptions,
+  "rawNotificationBufferSize"
+> {
   readonly logIncoming?: boolean;
   readonly logOutgoing?: boolean;
   readonly logger?: (event: AcpProtocol.AcpProtocolLogEvent) => Effect.Effect<void, never>;
@@ -33,7 +36,9 @@ export class AcpAgent extends Context.Service<
   {
     readonly raw: {
       /**
-       * Stream of inbound ACP notifications observed on the connection.
+       * Opt-in, work-sharing stream of inbound ACP notifications, not a broadcast.
+       * Configure rawNotificationBufferSize before connecting; disabled streams end immediately.
+       * Enabled streams drain on input termination and are interrupted when the connection scope closes.
        */
       readonly notifications: Stream.Stream<AcpProtocol.AcpIncomingNotification>;
       /**
@@ -280,6 +285,7 @@ export const make = Effect.fn("effect-acp/AcpAgent.make")(function* (
   const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
     stdio,
     serverRequestMethods: new Set(AcpRpcs.AgentRpcs.requests.keys()),
+    rawNotificationBufferSize: options.rawNotificationBufferSize ?? 0,
     ...(options.logIncoming !== undefined ? { logIncoming: options.logIncoming } : {}),
     ...(options.logOutgoing !== undefined ? { logOutgoing: options.logOutgoing } : {}),
     ...(options.logger ? { logger: options.logger } : {}),

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

@@ -1,6 +1,7 @@
 import * as Path from "effect/Path";
 import * as Cause from "effect/Cause";
 import * as Effect from "effect/Effect";
+import * as Deferred from "effect/Deferred";
 import * as Exit from "effect/Exit";
 import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
@@ -31,6 +32,10 @@ const ExtRequest = jsonRpcRequest("x/test", Schema.Struct({ hello: Schema.String
 const ExtResponse = jsonRpcResponse(Schema.Struct({ ok: Schema.Boolean }));
 const PromptRequest = jsonRpcRequest("session/prompt", AcpSchema.PromptRequest);
 const PromptResponse = jsonRpcResponse(AcpSchema.PromptResponse);
+const SessionUpdateNotification = jsonRpcNotification(
+  "session/update",
+  AcpSchema.SessionNotification,
+);
 const decodePromptRequestLine = Schema.decodeEffect(Schema.fromJsonString(PromptRequest));
 const XAiPromptCompleteNotification = jsonRpcNotification(
   "_x.ai/session/prompt_complete",
@@ -71,6 +76,43 @@ function concatBytes(chunks: ReadonlyArray<Uint8Array>): Uint8Array {
 }
 
 it.layer(NodeServices.layer)("effect-acp client", (it) => {
+  it.effect("handles callback-only session traffic without retaining raw notifications", () =>
+    Effect.gen(function* () {
+      const { stdio, input } = yield* makeInMemoryStdio();
+      const acp = yield* AcpClient.make(stdio);
+      const handled = yield* Deferred.make<void>();
+      const count = 10_000;
+      let updates = 0;
+      yield* acp.handleSessionUpdate(() =>
+        Effect.suspend(() => {
+          updates++;
+          return updates === count
+            ? Deferred.succeed(handled, undefined).pipe(Effect.asVoid)
+            : Effect.void;
+        }),
+      );
+      for (let index = 0; index < count; index++) {
+        yield* Queue.offer(
+          input,
+          yield* encodeJsonl(SessionUpdateNotification, {
+            jsonrpc: "2.0",
+            method: "session/update",
+            params: {
+              sessionId: "stress-session",
+              update: {
+                sessionUpdate: "agent_message_chunk",
+                content: { type: "text", text: `${index}:${"x".repeat(1024)}` },
+              },
+            },
+          }),
+        );
+      }
+      yield* Deferred.await(handled);
+      assert.equal(updates, count);
+      assert.deepEqual(yield* Stream.runCollect(acp.raw.notifications), []);
+    }),
+  );
+
   const makeHandle = (env?: Record<string, string>) =>
     Effect.gen(function* () {
       const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
@@ -90,7 +132,9 @@ it.layer(NodeServices.layer)("effect-acp client", (it) => {
       const typedNotifications = yield* Ref.make<Array<unknown>>([]);
       const handle = yield* makeHandle();
       const scope = yield* Scope.make();
-      const acpLayer = AcpClient.layerChildProcess(handle);
+      const acpLayer = AcpClient.layerChildProcess(handle, {
+        rawNotificationBufferSize: "unbounded",
+      });
       const context = yield* Layer.buildWithScope(acpLayer, scope);
 
       const ext = yield* Effect.gen(function* () {

packages/effect-acp/src/client.ts

@@ -23,13 +23,20 @@ import {
 } from "./_internal/shared.ts";
 import { makeChildStdio, makeTerminationError } from "./_internal/stdio.ts";
 
-export interface AcpClientOptions {
+export interface AcpClientOptions extends Pick<
+  AcpProtocol.AcpPatchedProtocolOptions,
+  "rawNotificationBufferSize"
+> {
   readonly logIncoming?: boolean;
   readonly logOutgoing?: boolean;
   readonly logger?: (event: AcpProtocol.AcpProtocolLogEvent) => Effect.Effect<void, never>;
 }
 
 type AcpClientRaw = {
+  /**
+   * Opt-in, work-sharing stream, not a broadcast. Configure rawNotificationBufferSize before connecting.
+   * Disabled streams end immediately; enabled streams drain on input termination and are interrupted on scope close.
+   */
   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>;
@@ -401,6 +408,7 @@ export const make = Effect.fn("effect-acp/AcpClient.make")(function* (
     stdio: stdio,
     ...(terminationError ? { terminationError } : {}),
     serverRequestMethods: new Set(AcpRpcs.ClientRpcs.requests.keys()),
+    rawNotificationBufferSize: options.rawNotificationBufferSize ?? 0,
     ...(options.logIncoming !== undefined ? { logIncoming: options.logIncoming } : {}),
     ...(options.logOutgoing !== undefined ? { logOutgoing: options.logOutgoing } : {}),
     ...(options.logger ? { logger: options.logger } : {}),

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

@@ -3,6 +3,8 @@ import * as AcpError from "./errors.ts";
 import * as Effect from "effect/Effect";
 import * as Deferred from "effect/Deferred";
 import * as Fiber from "effect/Fiber";
+import * as Exit from "effect/Exit";
+import * as Scope from "effect/Scope";
 import * as Queue from "effect/Queue";
 import * as Schema from "effect/Schema";
 import * as Stream from "effect/Stream";
@@ -68,6 +70,149 @@ const makeHandle = (env?: Record<string, string>) =>
   });
 
 it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
+  for (const bufferSize of [undefined, 0, 8] as const) {
+    it.effect(
+      `bounds callback-only raw retention with buffer size ${bufferSize ?? "default"}`,
+      () =>
+        Effect.gen(function* () {
+          const { stdio, input, output } = yield* makeInMemoryStdio();
+          const terminated = yield* Deferred.make<void>();
+          const count = 10_000;
+          const text = "x".repeat(1024);
+          let handled = 0;
+          const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
+            stdio,
+            serverRequestMethods: new Set(),
+            ...(bufferSize === undefined ? {} : { rawNotificationBufferSize: bufferSize }),
+            onNotification: () =>
+              Effect.suspend(() => {
+                handled++;
+                return handled === 1
+                  ? Effect.fail(AcpError.AcpRequestError.internalError("handler failed"))
+                  : Effect.void;
+              }),
+            onExtRequest: () => Effect.succeed({ ok: true }),
+            onTermination: () => Deferred.succeed(terminated, undefined).pipe(Effect.asVoid),
+          });
+
+          for (let index = 0; index < count; index++) {
+            yield* Queue.offer(
+              input,
+              encoder.encode(
+                `${encodeUnknownJsonString({
+                  jsonrpc: "2.0",
+                  method: "x/stress",
+                  params: { index, text },
+                })}\n`,
+              ),
+            );
+          }
+          yield* Queue.offer(
+            input,
+            yield* encodeJsonl(ExtRequest, {
+              jsonrpc: "2.0",
+              id: 7,
+              method: "x/test",
+              params: { hello: "world" },
+              headers: [],
+            }),
+          );
+          assert.deepEqual(yield* decodeExtResponse(yield* Queue.take(output)), {
+            jsonrpc: "2.0",
+            id: 7,
+            result: { ok: true },
+          });
+          yield* Queue.end(input);
+          yield* Deferred.await(terminated);
+
+          assert.equal(handled, count);
+          const retained = yield* Stream.runCollect(transport.incoming);
+          assert.deepEqual(
+            retained.map((notification) => notification.params),
+            Array.from({ length: bufferSize ?? 0 }, (_, offset) => ({
+              index: count - (bufferSize ?? 0) + offset,
+              text,
+            })),
+          );
+          assert.deepEqual(yield* Stream.runCollect(transport.incoming), []);
+        }),
+    );
+  }
+
+  it.effect("preserves opt-in late replay after a reader is interrupted", () =>
+    Effect.gen(function* () {
+      const { stdio, input } = yield* makeInMemoryStdio();
+      const observed = yield* Deferred.make<void>();
+      const terminated = yield* Deferred.make<void>();
+      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
+        stdio,
+        serverRequestMethods: new Set(),
+        rawNotificationBufferSize: "unbounded",
+        onTermination: () => Deferred.succeed(terminated, undefined).pipe(Effect.asVoid),
+      });
+      const reader = yield* transport.incoming.pipe(
+        Stream.runForEach(() =>
+          Deferred.succeed(observed, undefined).pipe(Effect.andThen(Effect.never)),
+        ),
+        Effect.forkScoped,
+      );
+      yield* Queue.offer(input, encoder.encode('{"jsonrpc":"2.0","method":"x/first"}\n'));
+      yield* Deferred.await(observed);
+      yield* Fiber.interrupt(reader);
+
+      yield* Queue.offer(
+        input,
+        encoder.encode(
+          '{"jsonrpc":"2.0","method":"x/second"}\n{"jsonrpc":"2.0","method":"x/third"}\n',
+        ),
+      );
+      yield* Queue.end(input);
+      yield* Deferred.await(terminated);
+      const replay = yield* Stream.runCollect(transport.incoming);
+      assert.deepEqual(
+        replay.map((notification) => notification.method),
+        ["x/second", "x/third"],
+      );
+      assert.deepEqual(yield* Stream.runCollect(transport.incoming), []);
+    }),
+  );
+
+  it.effect("drains raw observations and completes waiting readers on decode failure", () =>
+    Effect.gen(function* () {
+      const { stdio, input } = yield* makeInMemoryStdio();
+      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
+        stdio,
+        serverRequestMethods: new Set(),
+        rawNotificationBufferSize: 8,
+      });
+      const reader = yield* transport.incoming.pipe(Stream.runCollect, Effect.forkScoped);
+      yield* Queue.offer(input, encoder.encode('{"jsonrpc":"2.0","method":"x/first"}\n'));
+      yield* Queue.offer(input, encoder.encode("{malformed}\n"));
+      assert.deepEqual(
+        (yield* Fiber.join(reader)).map((notification) => notification.method),
+        ["x/first"],
+      );
+    }),
+  );
+
+  it.effect("interrupts raw readers outside the connection scope when it closes", () =>
+    Effect.gen(function* () {
+      const { stdio } = yield* makeInMemoryStdio();
+      const scope = yield* Scope.make();
+      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
+        stdio,
+        serverRequestMethods: new Set(),
+        rawNotificationBufferSize: 8,
+      }).pipe(Effect.provideService(Scope.Scope, scope));
+      const reader = yield* transport.incoming.pipe(
+        Stream.runCollect,
+        Effect.forkScoped({ startImmediately: true }),
+      );
+      yield* Scope.close(scope, Exit.void);
+      assert.isTrue(Exit.hasInterrupts(yield* Fiber.await(reader)));
+    }),
+  );
+
   it.effect(
     "emits exact JSON-RPC notifications and decodes inbound session/update and elicitation completion",
     () =>
@@ -76,6 +221,7 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
         const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
           stdio,
           serverRequestMethods: new Set(),
+          rawNotificationBufferSize: "unbounded",
         });
 
         const notifications =