From b3a4e7329f6c31a21b90c589ad84e305aaefc800 Mon Sep 17 00:00:00 2001 From: Wout Stiens <71498452+StiensWout@users.noreply.github.com> Date: Wed, 14 Oct 2026 22:23:45 +0200 Subject: [PATCH 1/3] fix(effect-acp): unread raw notifications no longer pin up to 32 large payloads The raw notification stream kept the last 32 incoming notifications in a sliding queue whether or not anything read them, so a connection receiving 8 MiB notifications held about 256 MiB. Bound the unread backlog by JSON line length as well as count: keep the newest notification and as many older ones as fit in 16 MiB, still capped at 32. Line length comes from the parser, so nothing is re-encoded. Readers that keep up still receive every notification, typed handlers are unchanged, and the stream now ends when the connection terminates or its scope closes. Fixes #18799 Co-Authored-By: Claude Opus 5.5 --- packages/effect-acp/src/agent.ts | 4 +- packages/effect-acp/src/client.ts | 4 + packages/effect-acp/src/protocol.test.ts | 98 ++++++++++++++++++++ packages/effect-acp/src/protocol.ts | 112 ++++++++++++++++------- 4 files changed, 183 insertions(+), 35 deletions(-) diff --git a/packages/effect-acp/src/agent.ts b/packages/effect-acp/src/agent.ts index 2ddf5cc8cc0dc..c581871e2f071 100644 --- a/packages/effect-acp/src/agent.ts +++ b/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 that fit in 16 MiB of JSON, + * always including the latest. Ends when the connection does. */ readonly notifications: Stream.Stream; /** diff --git a/packages/effect-acp/src/client.ts b/packages/effect-acp/src/client.ts index 732e2d26650f6..e109dd2257fcd 100644 --- a/packages/effect-acp/src/client.ts +++ b/packages/effect-acp/src/client.ts @@ -51,6 +51,10 @@ export interface AcpClientOptions { } type AcpClientRaw = { + /** + * Inbound notifications. Unread ones are kept best-effort: the newest 32 that + * fit in 16 MiB of JSON, always including the latest. Ends with the connection. + */ readonly notifications: Stream.Stream; readonly request: (method: string, payload: unknown) => Effect.Effect; readonly notify: (method: string, payload: unknown) => Effect.Effect; diff --git a/packages/effect-acp/src/protocol.test.ts b/packages/effect-acp/src/protocol.test.ts index 9c3f41e5f4c95..b251f02174dde 100644 --- a/packages/effect-acp/src/protocol.test.ts +++ b/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,94 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => { }), ); + it.effect("keeps the newest unread raw notifications that fit in 16 MiB of JSON", () => + Effect.gen(function* () { + const { stdio, input } = yield* makeInMemoryStdio(); + const handled = yield* Queue.unbounded(); + 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(); + 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(); + 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("decodes standard JSON-RPC errors into typed ACP request failures", () => Effect.gen(function* () { const { stdio, input, output } = yield* makeInMemoryStdio(); diff --git a/packages/effect-acp/src/protocol.ts b/packages/effect-acp/src/protocol.ts index 12e85a9e75426..e6d15ac4633fc 100644 --- a/packages/effect-acp/src/protocol.ts +++ b/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; readonly payload: string | Uint8Array; @@ -153,11 +164,19 @@ const makeStrictNdJsonRpcParser = () => { const decoder = new TextDecoder(); let buffer = ""; return { - decode: (bytes: Uint8Array | string): ReadonlyArray => { + // Every message in a batch line is charged the whole line. + decode: (bytes: Uint8Array | string): ReadonlyArray => { 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[0]) => { const encoded = codec.encode(response); @@ -165,7 +184,10 @@ const makeStrictNdJsonRpcParser = () => { }, }; }; +// Unread raw notifications are best-effort. The newest is always kept; older +// ones are dropped once the backlog passes either limit. 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 +235,14 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi const parser = makeStrictNdJsonRpcParser(); const serverQueue = yield* Queue.unbounded(); const clientQueue = yield* Queue.unbounded(); - const notificationQueue = yield* Queue.sliding( - MAX_BUFFERED_RAW_NOTIFICATIONS, - ); + const rawNotifications = yield* Queue.unbounded>(); + // 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(); const outgoing = yield* Queue.unbounded>(); const outgoingWriterState = yield* Ref.make({ @@ -384,8 +411,25 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi ), ); - const dispatchNotification = (notification: AcpIncomingNotification) => - Queue.offer(notificationQueue, notification).pipe( + const bufferRawNotification = (notification: AcpIncomingNotification, lineLength: number) => + Effect.sync(() => { + if (Queue.sizeUnsafe(rawNotifications) === 0) rawNotificationsJsonLength = 0; + while ( + Queue.sizeUnsafe(rawNotifications) >= MAX_BUFFERED_RAW_NOTIFICATIONS || + (Queue.sizeUnsafe(rawNotifications) > 0 && + rawNotificationsJsonLength + lineLength > MAX_BUFFERED_RAW_NOTIFICATION_JSON_LENGTH) + ) { + const dropped = Queue.takeUnsafe(rawNotifications); + if (dropped === undefined || Exit.isFailure(dropped)) break; + rawNotificationsJsonLength -= dropped.value.lineLength; + } + if (Queue.offerUnsafe(rawNotifications, { notification, lineLength })) { + rawNotificationsJsonLength += 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 +459,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 +544,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 +563,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi cause, ), ), - Effect.flatMap(dispatchNotification), + Effect.flatMap((notification) => dispatchNotification(notification, lineLength)), ); } if ( @@ -542,14 +587,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 +669,13 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi }), ); - const routeDecodedMessage = ( - message: RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded, - ): Effect.Effect => { + const routeDecodedMessage = ({ + message, + lineLength, + }: AcpDecodedWireMessage): Effect.Effect => { switch (message._tag) { case "Request": - return handleRequestEncoded(message); + return handleRequestEncoded(message, lineLength); case "Exit": return handleExitEncoded(message); case "Chunk": @@ -672,12 +717,7 @@ 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", @@ -685,12 +725,14 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi }), }), ), - 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 +750,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 +1012,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, From e4d691e98078e3959a8f90d15b47ab23d9a30d9c Mon Sep 17 00:00:00 2001 From: Wout Stiens <71498452+StiensWout@users.noreply.github.com> Date: Wed, 14 Oct 2026 22:59:02 +0200 Subject: [PATCH 2/3] fix(effect-acp): late notifications no longer evict an ended raw backlog When a writer failure ended the raw notification queue while the input reader kept running, each late notification evicted the oldest unread entry before the ended queue rejected the new one. Readers draining the ended stream lost entries. Offer first, and enforce the count and length limits only when the offer is admitted, in the same sync block. Also state the length limit in JSON characters (UTF-16 code units), not bytes or heap size. Co-Authored-By: Claude Opus 5.5 --- packages/effect-acp/src/agent.ts | 4 +-- packages/effect-acp/src/client.ts | 5 +-- packages/effect-acp/src/protocol.test.ts | 46 +++++++++++++++++++++++- packages/effect-acp/src/protocol.ts | 17 +++++---- 4 files changed, 60 insertions(+), 12 deletions(-) diff --git a/packages/effect-acp/src/agent.ts b/packages/effect-acp/src/agent.ts index c581871e2f071..e1088d10fe74e 100644 --- a/packages/effect-acp/src/agent.ts +++ b/packages/effect-acp/src/agent.ts @@ -37,8 +37,8 @@ export class AcpAgent extends Context.Service< readonly raw: { /** * Stream of inbound ACP notifications observed on the connection. Unread - * ones are kept best-effort: the newest 32 that fit in 16 MiB of JSON, - * always including the latest. Ends when the connection does. + * 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; /** diff --git a/packages/effect-acp/src/client.ts b/packages/effect-acp/src/client.ts index e109dd2257fcd..825e129be848c 100644 --- a/packages/effect-acp/src/client.ts +++ b/packages/effect-acp/src/client.ts @@ -52,8 +52,9 @@ export interface AcpClientOptions { type AcpClientRaw = { /** - * Inbound notifications. Unread ones are kept best-effort: the newest 32 that - * fit in 16 MiB of JSON, always including the latest. Ends with the connection. + * 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; readonly request: (method: string, payload: unknown) => Effect.Effect; diff --git a/packages/effect-acp/src/protocol.test.ts b/packages/effect-acp/src/protocol.test.ts index b251f02174dde..edf2cb9d1fdd2 100644 --- a/packages/effect-acp/src/protocol.test.ts +++ b/packages/effect-acp/src/protocol.test.ts @@ -193,7 +193,7 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => { }), ); - it.effect("keeps the newest unread raw notifications that fit in 16 MiB of JSON", () => + 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(); @@ -281,6 +281,50 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => { }), ); + 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(); + const terminated = yield* Deferred.make(); + 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(); diff --git a/packages/effect-acp/src/protocol.ts b/packages/effect-acp/src/protocol.ts index e6d15ac4633fc..c428d56429288 100644 --- a/packages/effect-acp/src/protocol.ts +++ b/packages/effect-acp/src/protocol.ts @@ -185,7 +185,8 @@ const makeStrictNdJsonRpcParser = () => { }; }; // Unread raw notifications are best-effort. The newest is always kept; older -// ones are dropped once the backlog passes either limit. +// 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. @@ -411,21 +412,23 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi ), ); + // 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) > 0 && - rawNotificationsJsonLength + lineLength > MAX_BUFFERED_RAW_NOTIFICATION_JSON_LENGTH) + 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; } - if (Queue.offerUnsafe(rawNotifications, { notification, lineLength })) { - rawNotificationsJsonLength += lineLength; - } }); const dispatchNotification = (notification: AcpIncomingNotification, lineLength: number) => From 32b32d4bab222150026deade08599649dc1adddb Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sat, 31 Oct 2026 22:08:34 +0000 Subject: [PATCH 3/3] fix(acp): preserve main test pruning after merge --- packages/effect-acp/src/protocol.test.ts | 26 ------------------------ 1 file changed, 26 deletions(-) diff --git a/packages/effect-acp/src/protocol.test.ts b/packages/effect-acp/src/protocol.test.ts index 64b0094a79197..fdf33bb8a9c0c 100644 --- a/packages/effect-acp/src/protocol.test.ts +++ b/packages/effect-acp/src/protocol.test.ts @@ -323,32 +323,6 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => { }), ); - it.effect("decodes standard JSON-RPC errors into typed ACP request failures", () => - Effect.gen(function* () { - const { stdio, input, output } = yield* makeInMemoryStdio(); - const transport = yield* AcpProtocol.makeAcpPatchedProtocol({ - stdio, - serverRequestMethods: new Set(), - }); - - const request = yield* transport.request("x/auth", {}).pipe(Effect.forkScoped); - const outbound = yield* decodeJsonRpcRequestId(yield* Queue.take(output)); - yield* Queue.offer( - input, - yield* encodeJsonl(JsonRpcErrorResponse, { - jsonrpc: "2.0", - id: outbound.id, - error: { code: -32000, message: "Authentication required" }, - }), - ); - - const error = yield* Fiber.join(request).pipe(Effect.flip, Effect.orDie); - assert.instanceOf(error, AcpError.AcpRequestError); - assert.equal(error.code, -32000); - assert.equal(error.message, "Authentication required"); - }), - ); - // clanka once sent a usage_update without `used` and `size`. One bad update // must not end a session whose later updates are fine. it.effect("drops a session/update that fails to decode and keeps reading", () =>