From 542ba4f9a491e26d4546656e4ba81eb46de2e23e Mon Sep 17 00:00:00 2001 From: Brad Groux <3053586+BradGroux@users.noreply.github.com> Date: Sun, 30 Aug 2026 07:19:29 -0500 Subject: [PATCH] fix: contain ACP teardown write failures --- .../src/__tests__/acp-json-rpc-peer.test.ts | 36 +++++++++++++++++++ shared/src/utils/acp-json-rpc-peer.ts | 20 ++++++----- 2 files changed, 48 insertions(+), 8 deletions(-) create mode 100644 server/src/__tests__/acp-json-rpc-peer.test.ts diff --git a/server/src/__tests__/acp-json-rpc-peer.test.ts b/server/src/__tests__/acp-json-rpc-peer.test.ts new file mode 100644 index 00000000..11c4cee7 --- /dev/null +++ b/server/src/__tests__/acp-json-rpc-peer.test.ts @@ -0,0 +1,36 @@ +import { describe, expect, it } from 'vitest'; +import { AcpJsonRpcPeer } from '@veritas-kanban/shared'; + +describe('ACP JSON-RPC peer writes', () => { + it('contains transport teardown failures for detached request replies', async () => { + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => unhandled.push(reason); + process.on('unhandledRejection', onUnhandled); + + try { + const peer = new AcpJsonRpcPeer({ + write: async () => { + throw new Error('stream destroyed'); + }, + onRequest: async () => ({}), + }); + + peer.acceptChunk('{"jsonrpc":"2.0","id":1,"method":"session/request_permission"}\n'); + await new Promise((resolve) => setImmediate(resolve)); + + expect(unhandled).toEqual([]); + } finally { + process.off('unhandledRejection', onUnhandled); + } + }); + + it('keeps caller-awaited write failures observable', async () => { + const peer = new AcpJsonRpcPeer({ + write: async () => { + throw new Error('stream destroyed'); + }, + }); + + await expect(peer.notify('session/cancel', {})).rejects.toThrow('stream destroyed'); + }); +}); diff --git a/shared/src/utils/acp-json-rpc-peer.ts b/shared/src/utils/acp-json-rpc-peer.ts index 509dd55f..30c6270d 100644 --- a/shared/src/utils/acp-json-rpc-peer.ts +++ b/shared/src/utils/acp-json-rpc-peer.ts @@ -76,7 +76,7 @@ export class AcpJsonRpcPeer { try { record = JSON.parse(line); } catch { - void this.send(jsonRpcError(null, -32700, 'Parse error')); + this.sendDetached(jsonRpcError(null, -32700, 'Parse error')); continue; } this.acceptRecord(record); @@ -117,7 +117,7 @@ export class AcpJsonRpcPeer { } if (typeof value.method !== 'string' || !value.method.trim()) { - if (id !== undefined) void this.send(jsonRpcError(id, -32600, 'Invalid Request')); + if (id !== undefined) this.sendDetached(jsonRpcError(id, -32600, 'Invalid Request')); return; } if (id === undefined) { @@ -127,14 +127,18 @@ export class AcpJsonRpcPeer { return; } if (!this.options.onRequest) { - void this.send(jsonRpcError(id, -32601, 'Method not found')); + this.sendDetached(jsonRpcError(id, -32601, 'Method not found')); return; } - void Promise.resolve(this.options.onRequest(value.method, value.params, id)) - .then((result) => this.send({ jsonrpc: '2.0', id, result: result ?? {} })) - .catch((error) => - this.send(jsonRpcError(id, -32603, boundedMessage(error, 'Internal error'))) - ); + void Promise.resolve(this.options.onRequest(value.method, value.params, id)).then( + (result) => this.sendDetached({ jsonrpc: '2.0', id, result: result ?? {} }), + (error) => + this.sendDetached(jsonRpcError(id, -32603, boundedMessage(error, 'Internal error'))) + ); + } + + private sendDetached(record: AcpJsonRpcMessage): void { + void this.send(record).catch(() => undefined); } private send(record: AcpJsonRpcMessage): Promise {