perf(cua): assemble native response frames once

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<void> | 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)

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'