perf: decode fragmented CLI replies without repeated scans (#18909)

* perf: decode fragmented CLI replies without rescanning accumulated text

* bench: require an explicit CLI framing baseline

config/scripts/benchmark-cli-response-framing.mjs

@@ -0,0 +1,128 @@
+import assert from 'node:assert/strict'
+import { execFileSync } from 'node:child_process'
+import { EventEmitter } from 'node:events'
+import { readFileSync } from 'node:fs'
+import Module from 'node:module'
+import { dirname, resolve } from 'node:path'
+import { performance } from 'node:perf_hooks'
+import { build } from 'esbuild'
+
+// Run from the worktree root: node config/scripts/benchmark-cli-response-framing.mjs <base-ref>
+const sourcePath = 'src/cli/runtime/transport.ts'
+const baselineRef = process.argv[2]
+assert.ok(baselineRef, 'Pass the pre-change transport revision as base-ref.')
+const beforeSource = execFileSync('git', ['show', `${baselineRef}:${sourcePath}`], {
+  encoding: 'utf8'
+})
+let chunks = []
+
+async function loadTransport(source) {
+  const built = await build({
+    stdin: { contents: source, loader: 'ts', resolveDir: dirname(resolve(sourcePath)) },
+    bundle: true,
+    platform: 'node',
+    format: 'cjs',
+    write: false,
+    logLevel: 'silent'
+  })
+  const module = new Module(resolve(sourcePath))
+  const originalRequire = module.require.bind(module)
+  module.require = (name) => {
+    if (name === 'node:crypto') {
+      return { randomUUID: () => 'benchmark-request' }
+    }
+    if (name !== 'node:net') {
+      return originalRequire(name)
+    }
+    return {
+      createConnection() {
+        const socket = new EventEmitter()
+        socket.setEncoding = () => {}
+        socket.end = () => {}
+        socket.destroy = () => {}
+        socket.write = () => {
+          for (const chunk of chunks) {
+            socket.emit('data', chunk)
+          }
+        }
+        queueMicrotask(() => socket.emit('connect'))
+        return socket
+      }
+    }
+  }
+  module._compile(built.outputFiles[0].text, resolve(sourcePath))
+  return module.exports.sendRequest
+}
+
+const before = await loadTransport(beforeSource)
+const after = await loadTransport(readFileSync(sourcePath, 'utf8'))
+const metadata = {
+  runtimeId: 'benchmark-runtime',
+  authToken: 'benchmark-token',
+  transports: [{ kind: 'unix', endpoint: 'injected-socket' }]
+}
+const run = (sendRequest) => sendRequest(metadata, 'terminal.read', {}, 30000)
+
+async function measure(sendRequest, payloadBytes, repetitions) {
+  const warmup = await run(sendRequest)
+  assert.equal(warmup.result.data.length, payloadBytes)
+  const samples = []
+  for (let sample = 0; sample < 5; sample++) {
+    const start = performance.now()
+    for (let iteration = 0; iteration < repetitions; iteration++) {
+      await run(sendRequest)
+    }
+    samples.push((performance.now() - start) / repetitions)
+  }
+  return samples.sort((a, b) => a - b)[2]
+}
+
+async function searchedCharacters(sendRequest) {
+  const original = String.prototype.indexOf
+  let searched = 0
+  String.prototype.indexOf = function (needle, position) {
+    if (needle === '\n') {
+      searched += this.length - (position ?? 0)
+    }
+    return original.call(this, needle, position)
+  }
+  try {
+    await run(sendRequest)
+  } finally {
+    String.prototype.indexOf = original
+  }
+  return searched
+}
+
+const rows = []
+for (const [payloadBytes, chunkChars, repetitions] of [
+  [32, 65536, 1000],
+  [1024 * 1024, 2 * 1024 * 1024, 20],
+  [1024 * 1024, 65536, 10],
+  [1024 * 1024, 4096, 5],
+  [4 * 1024 * 1024, 4096, 2],
+  [4 * 1024 * 1024, 256, 1]
+]) {
+  const line = `${JSON.stringify({
+    id: 'benchmark-request',
+    ok: true,
+    result: { data: 'x'.repeat(payloadBytes) },
+    _meta: { runtimeId: 'benchmark-runtime' }
+  })}\n`
+  chunks = []
+  for (let offset = 0; offset < line.length; offset += chunkChars) {
+    chunks.push(line.slice(offset, offset + chunkChars))
+  }
+  const beforeMs = await measure(before, payloadBytes, repetitions)
+  const afterMs = await measure(after, payloadBytes, repetitions)
+  rows.push({
+    payloadBytes,
+    chunkChars,
+    beforeMs: +beforeMs.toFixed(6),
+    afterMs: +afterMs.toFixed(6),
+    speedup: +(beforeMs / afterMs).toFixed(2),
+    beforeSearchedCharacters: await searchedCharacters(before),
+    afterSearchedCharacters: await searchedCharacters(after)
+  })
+}
+console.log(JSON.stringify({ node: process.version, baselineRef, rows }, null, 2))

src/cli/runtime/transport-framing.test.ts

@@ -0,0 +1,150 @@
+import { EventEmitter } from 'node:events'
+import { StringDecoder } from 'node:string_decoder'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import type { RuntimeMetadata } from '../../shared/runtime-bootstrap'
+import { sendRequest } from './transport'
+
+const { createConnection } = vi.hoisted(() => ({ createConnection: vi.fn() }))
+vi.mock('node:net', () => ({ createConnection }))
+vi.mock('node:crypto', () => ({ randomUUID: () => 'request-1' }))
+
+const metadata: RuntimeMetadata = {
+  runtimeId: 'runtime-1',
+  pid: 123,
+  transports: [{ kind: 'unix', endpoint: 'test-only' }],
+  authToken: 'token',
+  startedAt: 1
+}
+const reply = (result: unknown) =>
+  `${JSON.stringify({ id: 'request-1', ok: true, result, _meta: { runtimeId: 'runtime-1' } })}\n`
+
+class TestSocket extends EventEmitter {
+  setEncoding = vi.fn()
+  write = vi.fn()
+  end = vi.fn()
+  destroy = vi.fn()
+}
+let socket: TestSocket
+
+beforeEach(() => {
+  socket = new TestSocket()
+  createConnection.mockReturnValue(socket)
+})
+afterEach(() => {
+  vi.restoreAllMocks()
+  vi.useRealTimers()
+})
+
+describe('CLI runtime response framing', () => {
+  it.each([1, 7, 256, 4096])(
+    'reads a fragmented response with %i-character chunks',
+    async (size) => {
+      const result = { data: '界😀'.repeat(10000) }
+      const encoded = reply(result)
+      const pending = sendRequest(metadata, 'terminal.read', {}, 30000)
+      for (let offset = 0; offset < encoded.length; offset += size) {
+        socket.emit('data', encoded.slice(offset, offset + size))
+      }
+      await expect(pending).resolves.toMatchObject({ result })
+      expect(socket.setEncoding).toHaveBeenCalledExactlyOnceWith('utf8')
+      expect(socket.end).toHaveBeenCalledOnce()
+    }
+  )
+
+  it('accepts Unicode split across socket bytes using the existing UTF-8 decoder', async () => {
+    const pending = sendRequest(metadata, 'terminal.read', {}, 30000)
+    const decoder = new StringDecoder('utf8')
+    for (const byte of Buffer.from(reply({ data: '界😀é' }))) {
+      socket.emit('data', decoder.write(Buffer.from([byte])))
+    }
+    socket.emit('data', decoder.end())
+    await expect(pending).resolves.toMatchObject({ result: { data: '界😀é' } })
+  })
+
+  it('searches each fragment once without rescanning the accumulated reply', async () => {
+    const encoded = reply({ data: 'x'.repeat(1024 * 1024) })
+    const pending = sendRequest(metadata, 'terminal.read', {}, 30000)
+    const originalIndexOf = String.prototype.indexOf
+    let searchedCharacters = 0
+    const search = vi
+      .spyOn(String.prototype, 'indexOf')
+      .mockImplementation(function (this: string, value, position) {
+        if (value === '\n') {
+          searchedCharacters += this.length - (position ?? 0)
+        }
+        return originalIndexOf.call(this, value, position)
+      })
+    try {
+      for (let offset = 0; offset < encoded.length; offset += 256) {
+        socket.emit('data', encoded.slice(offset, offset + 256))
+      }
+    } finally {
+      search.mockRestore()
+    }
+    await expect(pending).resolves.toMatchObject({ ok: true })
+    expect(searchedCharacters).toBe(encoded.length)
+  })
+
+  it('refreshes keepalives across chunks and ignores blanks and data after the final frame', async () => {
+    vi.useFakeTimers()
+    const pending = sendRequest(metadata, 'terminal.read', {}, 100)
+    await vi.advanceTimersByTimeAsync(90)
+    socket.emit('data', ' \r\n{"_keep')
+    socket.emit('data', 'alive":true}\n\t\n')
+    await vi.advanceTimersByTimeAsync(90)
+    socket.emit('data', `${reply({ data: 'done' })}invalid JSON\n`)
+    socket.emit('data', 'more ignored data')
+    await expect(pending).resolves.toMatchObject({ result: { data: 'done' } })
+    expect(socket.end).toHaveBeenCalledOnce()
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it.each([
+    ['broken JSON\n', 'invalid_runtime_response'],
+    ['{}\n', 'invalid_runtime_response'],
+    ['{"id":"other","ok":true,"result":{}}\n', 'invalid_runtime_response'],
+    [
+      '{"id":"request-1","ok":true,"result":{},"_meta":{"runtimeId":"other"}}\n',
+      'runtime_unavailable'
+    ]
+  ])(
+    'rejects a fragmented invalid first frame before subsequent valid frames',
+    async (line, code) => {
+      const pending = sendRequest(metadata, 'terminal.read', {}, 30000)
+      socket.emit('data', line.slice(0, 2))
+      socket.emit('data', line.slice(2) + reply({ data: 'ignored' }))
+      await expect(pending).rejects.toMatchObject({ code })
+      expect(socket.end).toHaveBeenCalledOnce()
+    }
+  )
+
+  it('preserves terminal failure envelopes', async () => {
+    const pending = sendRequest(metadata, 'terminal.read', {}, 30000)
+    socket.emit('data', '{"id":"request-1","ok":false,"error":{"code":"bad","message":"no"}}\n')
+    await expect(pending).resolves.toMatchObject({
+      ok: false,
+      error: { code: 'bad', message: 'no' }
+    })
+  })
+
+  it('rejects close with an incomplete frame and does not parse later data', async () => {
+    const pending = sendRequest(metadata, 'terminal.read', {}, 30000)
+    socket.emit('data', '{"id":')
+    socket.emit('close')
+    socket.emit('data', reply({ data: 'ignored' }))
+    await expect(pending).rejects.toMatchObject({ code: 'runtime_unavailable' })
+    expect(socket.end).toHaveBeenCalledOnce()
+  })
+
+  it('destroys a timed out socket holding an incomplete frame', async () => {
+    vi.useFakeTimers()
+    const pending = sendRequest(metadata, 'terminal.read', {}, 100)
+    const rejected = expect(pending).rejects.toMatchObject({ code: 'runtime_timeout' })
+    socket.emit('data', '{"id":')
+    await vi.advanceTimersByTimeAsync(100)
+    socket.emit('data', reply({ data: 'ignored' }))
+    await rejected
+    expect(socket.destroy).toHaveBeenCalledOnce()
+    expect(vi.getTimerCount()).toBe(0)
+  })
+})

src/cli/runtime/transport.ts

@@ -31,7 +31,7 @@ export async function sendRequest<TResult>(
       return
     }
     const socket = createConnection(transport.endpoint)
-    let buffer = ''
+    let lineSegments: string[] = []
     let settled = false
     const requestId = randomUUID()
 
@@ -40,6 +40,7 @@ export async function sendRequest<TResult>(
         return
       }
       settled = true
+      lineSegments = []
       socket.destroy()
       reject(
         new RuntimeClientError(
@@ -56,6 +57,7 @@ export async function sendRequest<TResult>(
         return
       }
       settled = true
+      lineSegments = []
       clearTimeout(timeout)
       socket.end()
       if (result.ok === false) {
@@ -89,18 +91,27 @@ export async function sendRequest<TResult>(
       })
     })
     socket.on('data', (chunk: string) => {
-      buffer += chunk
       // Why: the server may interleave `{"_keepalive":true}\n` frames with the
       // final success/failure frame to keep both idle timers alive during a
       // long-poll (see design doc §3.1). Read frames in a loop until we see a
       // terminal frame. Each keepalive refreshes the client-side timer so a
       // 10 min wait doesn't trip the 60 s default ceiling.
-      let newlineIndex = buffer.indexOf('\n')
-      while (newlineIndex !== -1 && !settled) {
-        const line = buffer.slice(0, newlineIndex)
-        buffer = buffer.slice(newlineIndex + 1)
+      let cursor = 0
+      while (cursor < chunk.length && !settled) {
+        const newlineIndex = chunk.indexOf('\n', cursor)
+        if (newlineIndex === -1) {
+          lineSegments.push(chunk.slice(cursor))
+          return
+        }
+        const segment = chunk.slice(cursor, newlineIndex)
+        let line = segment
+        if (lineSegments.length > 0) {
+          lineSegments.push(segment)
+          line = lineSegments.join('')
+          lineSegments = []
+        }
+        cursor = newlineIndex + 1
         if (line.trim().length === 0) {
-          newlineIndex = buffer.indexOf('\n')
           continue
         }
 
@@ -124,7 +135,6 @@ export async function sendRequest<TResult>(
         // major). See §7 risk #9.
         if (isKeepaliveFrame(raw)) {
           timeout.refresh()
-          newlineIndex = buffer.indexOf('\n')
           continue
         }
 
@@ -150,7 +160,6 @@ export async function sendRequest<TResult>(
         const frame = parsed.data
         if ('_keepalive' in frame) {
           timeout.refresh()
-          newlineIndex = buffer.indexOf('\n')
           continue
         }