diff --git a/packages/experimental/cua/src/native-client.ts b/packages/experimental/cua/src/native-client.ts index dfede3fc19..726a8c17c4 100644 --- a/packages/experimental/cua/src/native-client.ts +++ b/packages/experimental/cua/src/native-client.ts @@ -1,4 +1,4 @@ -/** Bounded JSON-line transport to one owned native helper process. */ +/** Byte-bounded JSON-line transport with linear frame assembly for one owned helper. */ import { randomUUID } from 'node:crypto' import type { SubprocessHandle, SubprocessRuntime } from '@deepseek-ai/dsh-subprocess' import type { ToolExecution } from '@deepseek-ai/dsh-tools' @@ -38,7 +38,8 @@ export interface NativeClientConfig { export class NativeClient { private readonly handle: SubprocessHandle private readonly config: NativeClientConfig - private tail = Buffer.alloc(0) + private readonly chunks: Buffer[] = [] + private bufferedBytes = 0 private failure: NativeError | undefined private disposal: Promise | undefined private active: { id: string; resolve: (value: JsonValue) => void; reject: (error: NativeError) => void } | undefined @@ -137,7 +138,8 @@ export class NativeClient { private fail(error: NativeError): void { if (this.failure) return this.failure = error - this.tail = Buffer.alloc(0) + this.chunks.length = 0 + this.bufferedBytes = 0 this.active?.reject(error) this.active = undefined this.handle.terminate() @@ -145,20 +147,24 @@ export class NativeClient { private receive(chunk: Buffer): void { if (this.failure) return - if (this.tail.length + chunk.length > this.config.maxResponseBytes) { + if (this.bufferedBytes + chunk.length > this.config.maxResponseBytes) { this.fail(new NativeError('NATIVE_RESPONSE_LIMIT', 'Helper response exceeded the configured byte limit')) return } - this.tail = Buffer.concat([this.tail, chunk]) - const newline = this.tail.indexOf(10) + if (chunk.length === 0) return + this.chunks.push(chunk) + this.bufferedBytes += chunk.length + const newline = chunk.indexOf(10) if (newline < 0) return try { - if (newline !== this.tail.length - 1) throw new Error('One request must produce exactly one response frame') - const line = new TextDecoder('utf-8', { fatal: true }).decode(this.tail.subarray(0, newline)) + if (newline !== chunk.length - 1) throw new Error('One request must produce exactly one response frame') + const frame = Buffer.concat(this.chunks, this.bufferedBytes) + const line = new TextDecoder('utf-8', { fatal: true }).decode(frame.subarray(0, frame.length - 1)) const reply = Reply.parse(JSON.parse(line)) const active = this.active if (!active || reply.id !== active.id) throw new Error('Helper reply has no matching request') - this.tail = Buffer.alloc(0) + this.chunks.length = 0 + this.bufferedBytes = 0 this.active = undefined if ('error' in reply) { const error = new NativeError(reply.error.code, reply.error.message) diff --git a/packages/experimental/cua/tests/native-client.spec.ts b/packages/experimental/cua/tests/native-client.spec.ts index a3d54dcb59..07f4f2b373 100644 --- a/packages/experimental/cua/tests/native-client.spec.ts +++ b/packages/experimental/cua/tests/native-client.spec.ts @@ -127,6 +127,27 @@ describe('native helper transport', () => { await rejected }) + it('assembles fragmented large responses without repeatedly copying their prefixes', async () => { + const f = fixture({ maxResponseBytes: 1024 * 1024 }) + const pending = f.client.request('listApps', {}, new AbortController().signal) + const value = 'x'.repeat(512 * 1024) + const bytes = Buffer.from(JSON.stringify({ id: f.id(), result: value }) + '\n') + const concatenate = Buffer.concat.bind(Buffer) + let copiedBytes = 0 + const copying = vi.spyOn(Buffer, 'concat').mockImplementation((chunks, length) => { + copiedBytes += length ?? chunks.reduce((total, chunk) => total + chunk.byteLength, 0) + return concatenate(chunks, length) + }) + try { + f.stdout.emit('data', Buffer.alloc(0)) + for (let offset = 0; offset < bytes.length; offset += 64 * 1024) f.stdout.write(bytes.subarray(offset, offset + 64 * 1024)) + } finally { copying.mockRestore() } + expect(await pending === value).toBe(true) + // Permit an ownership copy plus one complete-frame assembly, independent of pipe chunk count. + expect(copiedBytes).toBeLessThanOrEqual(bytes.length * 2) + expect(f.client.disposed).toBe(false) + }) + it('bounds request bytes before dispatch and keeps acquired handles after a rejected request', async () => { const f = fixture({ maxRequestBytes: 128 }) const envelope = JSON.stringify({ id: '0'.repeat(36), method: 'typeText', params: { text: '' } }) + '\n'