feat(sdk): TypeScript SDK client + shared wire protocol + SDK subagent backend

- @deepseek-ai/dsh-sdk-protocol: extract the line transport from dsh-jsonrpc and name the request/result/notification wire types both ends share; error responses preserve wire code/data via JsonRpcResponseError. - @deepseek-ai/dsh-sdk-client: TypeScript twin of the Python SDK — spawns the dsh-jsonrpc-agent runtime as a subprocess, drives stdio JSON-RPC turns (DeepSeekHarness high-level API + HarnessClient protocol client), scopes notifications to session trees client-side, and reaps the child through the shared subprocess dispose ladder. - @deepseek-ai/dsh-subagent-sdk: out-of-process subagent backend driving a child harness runtime through the TS SDK; shares cwd resolution with subagent-acp via new dsh-subagent-subprocess cwd helpers. - Keyless unit suites drive real subprocesses (scripted fake runtime peer); 100% per-file coverage on all touched packages.
2026-07-27 03:21:06 +08:00
parent 46e70be34a
commit 4ad37344a8
33 changed files with 2738 additions and 122 deletions
@@ -0,0 +1,179 @@
+#!/usr/bin/env node
+/**
+ * Scripted stand-in for the DeepSeek Harness SDK runtime, driven entirely by
+ * env vars — no model, no network, no harness imports. Speaks the runtime's
+ * newline-delimited JSON-RPC protocol on stdio: answers `initialize`,
+ * `session/prompt` (streaming scripted `session.event` notifications, then
+ * `session.finished`, then the response), and `shutdown`.
+ *
+ * Script vocabulary (all optional):
+ * - `FAKE_TEXT`: assistant text for each turn (default `hello from fake runtime`).
+ * - `FAKE_STATUS`: the `session.finished` status (default `ok`).
+ * - `FAKE_REASON_KIND`: the `session.finished` reason kind (default `completed`; `none` omits the reason).
+ * - `FAKE_SUBAGENT`: also emit a child session (subagent.started + child event + subagent.finished).
+ * - `FAKE_ECHO_CWD`: prefix the assistant text with the process cwd.
+ * - `FAKE_ECHO_ENV`: comma-separated env names to echo as `name=value` lines in the assistant text.
+ * - `FAKE_MALFORMED`: `initialize` returns `{}` (no serverInfo); `prompt` returns `{}` (no accepted).
+ * - `FAKE_MALFORMED_PROMPT`: `initialize` is normal; only `prompt` returns `{}` (no accepted).
+ * - `FAKE_INIT_ERROR`: `initialize` answers a JSON-RPC error response with code 7.
+ * - `FAKE_HANG_INIT`: never answer `initialize` (mid-handshake cancel probe).
+ * - `FAKE_INIT_READY` + `FAKE_INIT_GO`: touch the READY file when `initialize`
+ *   arrives, then poll for the GO file before answering (deterministic
+ *   cancel-during-handshake window).
+ * - `FAKE_HANG_PROMPT`: never answer `session/prompt` (for timeout/dispose tests).
+ * - `FAKE_STREAM_THEN_HANG`: stream a text chunk for the prompt, then never
+ *   finish the turn or answer (partial-output cancel probe). Touches
+ *   `FAKE_STREAM_READY` after the chunk when set.
+ * - `FAKE_IGNORE_EOF` + `FAKE_SIGTERM_FILE`: keep running after stdin EOF; touch the file on SIGTERM (ladder probe).
+ * - `FAKE_TRAP_SIGTERM`: with `FAKE_IGNORE_EOF`, survive SIGTERM too (SIGKILL-rung probe).
+ * - `FAKE_EXIT_BEFORE_INIT`: exit 3 immediately (spawn-then-die probe).
+ * - `FAKE_STDERR`: write this line to stderr at boot (diagnostics-tail probe).
+ * - `FAKE_STDERR_NO_NEWLINE`: write this to stderr WITHOUT a newline (buffer-flush probe).
+ * - `FAKE_RECORD_INIT`: append each `initialize` params JSON to this file (handshake probe).
+ */
+
+import { appendFileSync, existsSync, writeFileSync } from 'node:fs'
+import process from 'node:process'
+import { createInterface } from 'node:readline'
+
+const env = process.env
+
+if (env.FAKE_STDERR !== undefined) process.stderr.write(`${env.FAKE_STDERR}\n`)
+if (env.FAKE_STDERR_NO_NEWLINE !== undefined) process.stderr.write(env.FAKE_STDERR_NO_NEWLINE)
+if (env.FAKE_EXIT_BEFORE_INIT !== undefined) process.exit(3)
+
+if (env.FAKE_IGNORE_EOF !== undefined) {
+  // Simulate a runtime that never quiesces from EOF so the dispose ladder
+  // must escalate; record which rung fired.
+  process.stdin.resume()
+  process.stdin.on('end', () => { setInterval(() => {}, 1_000) })
+  process.on('SIGTERM', () => {
+    if (env.FAKE_SIGTERM_FILE !== undefined) writeFileSync(env.FAKE_SIGTERM_FILE, 'sigterm\n')
+    if (env.FAKE_TRAP_SIGTERM === undefined) process.exit(0)
+  })
+}
+
+function write(message: object): void {
+  process.stdout.write(`${JSON.stringify(message)}\n`)
+}
+
+function notify(method: string, params: object): void {
+  write({ jsonrpc: '2.0', method, params })
+}
+
+let seq = 0
+function event(sessionId: string, type: string, data: object): void {
+  notify('session.event', { sessionId, event: { type, seq: seq++, time: 0, data } })
+}
+
+function assistantText(): string {
+  const parts: string[] = []
+  if (env.FAKE_ECHO_CWD !== undefined) parts.push(`cwd=${process.cwd()}`)
+  for (const name of (env.FAKE_ECHO_ENV ?? '').split(',').filter(entry => entry.length > 0)) {
+    parts.push(`${name}=${env[name] ?? ''}`)
+  }
+  parts.push(env.FAKE_TEXT ?? 'hello from fake runtime')
+  return parts.join('\n')
+}
+
+function runTurn(sessionId: string): void {
+  const text = assistantText()
+  event(sessionId, 'turn/start', { turn: 0 })
+  event(sessionId, 'assistant/chunk', { turn: 0, step: 0, chunk: { type: 'text-delta', index: 0, text } })
+  event(sessionId, 'assistant/message', {
+    turn: 0,
+    step: 0,
+    content: [{ type: 'text', text }],
+    provenance: { provider: 'fake', model: 'fake' },
+  })
+  const reasonKind = env.FAKE_REASON_KIND ?? 'completed'
+  event(sessionId, 'turn/end', { turn: 0, reason: { kind: reasonKind } })
+  if (env.FAKE_SUBAGENT !== undefined) {
+    const childId = `${sessionId}-child`
+    notify('subagent.started', { parentSessionId: sessionId, childSessionId: childId })
+    event(childId, 'assistant/message', {
+      turn: 0,
+      step: 0,
+      content: [{ type: 'text', text: 'child says hi' }],
+      provenance: { provider: 'fake', model: 'fake' },
+    })
+    notify('subagent.finished', {
+      provider: 'spawn',
+      agentId: childId,
+      parentSessionId: sessionId,
+      childSessionId: childId,
+      status: 'ok',
+      stopReason: 'completed',
+      lastAssistantMessage: [{ type: 'text', text: 'child says hi' }],
+    })
+  }
+  notify('session.finished', {
+    sessionId,
+    status: env.FAKE_STATUS ?? 'ok',
+    ...(reasonKind === 'none' ? {} : { reason: { kind: reasonKind } }),
+  })
+}
+
+function sessionIdOf(params: Record<string, unknown> | undefined): string {
+  const value = params?.sessionId
+  return typeof value === 'string' ? value : ''
+}
+
+const reader = createInterface({ input: process.stdin })
+reader.on('line', (line) => {
+  if (line.trim().length === 0) return
+  const frame = JSON.parse(line) as { id?: string | number; method?: string; params?: Record<string, unknown> }
+  if (frame.method === undefined || frame.id === undefined) return
+  const respond = (result: object): void => { write({ jsonrpc: '2.0', id: frame.id, result }) }
+  switch (frame.method) {
+    case 'initialize':
+      if (env.FAKE_RECORD_INIT !== undefined) appendFileSync(env.FAKE_RECORD_INIT, `${JSON.stringify(frame.params)}\n`)
+      if (env.FAKE_HANG_INIT !== undefined) return
+      if (env.FAKE_INIT_READY !== undefined && env.FAKE_INIT_GO !== undefined) {
+        writeFileSync(env.FAKE_INIT_READY, 'ready\n')
+        const go = env.FAKE_INIT_GO
+        const id = frame.id
+        const poll = setInterval(() => {
+          if (!existsSync(go)) return
+          clearInterval(poll)
+          write({ jsonrpc: '2.0', id, result: { serverInfo: { name: 'deepseek-harness-sdk-runtime', version: '0.0.1' } } })
+        }, 5)
+        return
+      }
+      if (env.FAKE_INIT_ERROR !== undefined) {
+        write({ jsonrpc: '2.0', id: frame.id, error: { code: 7, message: 'scripted init failure', data: { hint: 'fake' } } })
+        return
+      }
+      if (env.FAKE_MALFORMED !== undefined) {
+        respond({})
+        return
+      }
+      respond({ serverInfo: { name: 'deepseek-harness-sdk-runtime', version: '0.0.1' } })
+      return
+    case 'session/prompt': {
+      if (env.FAKE_STREAM_THEN_HANG !== undefined) {
+        const sessionId = sessionIdOf(frame.params)
+        event(sessionId, 'assistant/chunk', { turn: 0, step: 0, chunk: { type: 'text-delta', index: 0, text: 'streamed then hung' } })
+        if (env.FAKE_STREAM_READY !== undefined) writeFileSync(env.FAKE_STREAM_READY, 'streamed\n')
+        return
+      }
+      if (env.FAKE_HANG_PROMPT !== undefined) return
+      if (env.FAKE_MALFORMED !== undefined || env.FAKE_MALFORMED_PROMPT !== undefined) {
+        respond({})
+        return
+      }
+      const sessionId = sessionIdOf(frame.params)
+      runTurn(sessionId)
+      respond({ accepted: true })
+      return
+    }
+    case 'shutdown':
+      respond({})
+      // An EOF-ignoring fake also refuses the protocol exit, so the client's
+      // dispose ladder (not this cooperative path) must reap it.
+      if (env.FAKE_IGNORE_EOF === undefined) setImmediate(() => process.exit(0))
+      return
+    default:
+      write({ jsonrpc: '2.0', id: frame.id, error: { code: -32603, message: `unknown method: ${frame.method}` } })
+  }
+})
@@ -0,0 +1,344 @@
+/**
+ * SDK client against a real scripted runtime subprocess
+ * (`tests/fake-runtime.ts`, protocol-only — the only faked boundary is the
+ * model-owning runtime itself). Covers the turn loop, notification routing
+ * and session-tree scoping, error surfaces, timeouts, and the dispose ladder.
+ */
+
+import { mkdtemp, readFile, rm, stat } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import { afterEach, describe, expect, it } from 'vitest'
+import {
+  DeepSeekHarness,
+  finalResponse,
+  HarnessClient,
+  normalizeInput,
+  RequestTimeoutError,
+  SdkProtocolError,
+  TransportClosedError,
+  type HarnessNotification,
+} from '../src/index.ts'
+import { JsonRpcResponseError } from '@deepseek-ai/dsh-sdk-protocol'
+
+const fakeRuntime = fileURLToPath(new URL('./fake-runtime.ts', import.meta.url))
+
+const cleanups: (() => Promise<void>)[] = []
+afterEach(async () => {
+  for (const cleanup of cleanups.splice(0)) await cleanup()
+})
+
+type LaunchOverrides = Partial<ConstructorParameters<typeof HarnessClient>[0]>
+
+/** Launch options running the fake runtime on the current node (type stripping). */
+function fakeLaunch(env: Record<string, string> = {}, extra: LaunchOverrides = {}) {
+  return {
+    command: process.execPath,
+    args: [fakeRuntime],
+    env: { ...process.env as Record<string, string>, ...env },
+    ...extra,
+  }
+}
+
+function harnessWith(env: Record<string, string> = {}, extra: LaunchOverrides = {}): DeepSeekHarness {
+  const harness = new DeepSeekHarness({ launch: fakeLaunch(env, extra) })
+  cleanups.push(() => harness.close())
+  return harness
+}
+
+async function tempDir(prefix: string): Promise<string> {
+  const dir = await mkdtemp(join(tmpdir(), prefix))
+  cleanups.push(() => rm(dir, { recursive: true, force: true }))
+  return dir
+}
+
+describe('DeepSeekHarness', () => {
+  it('runs a turn end to end and reuses the runtime across sessions', async () => {
+    const harness = harnessWith({ FAKE_TEXT: 'turn answer' })
+    const first = await harness.run('say hi')
+    expect(first.status).toBe('ok')
+    expect(first.reason).toEqual({ kind: 'completed' })
+    expect(first.finalResponse).toBe('turn answer')
+    expect(first.events.map(event => event.type)).toEqual(['turn/start', 'assistant/chunk', 'assistant/message', 'turn/end'])
+
+    // Same subprocess, second session: ids differ, protocol state is reusable.
+    const second = await harness.run([{ type: 'text', text: 'again' }])
+    expect(second.status).toBe('ok')
+    expect(second.sessionId).not.toBe(first.sessionId)
+    await harness.close()
+  })
+
+  it('streams notifications to the observer and scopes them to the session tree', async () => {
+    const harness = harnessWith({ FAKE_SUBAGENT: '1' })
+    const seen: HarnessNotification[] = []
+    const result = await harness.run('delegate', {
+      sessionId: 'parent-1',
+      onNotification: (n) => { seen.push(n) },
+    })
+
+    expect(result.status).toBe('ok')
+    // The child session's events arrive through subagent.started lineage.
+    expect(seen.map(n => n.method)).toContain('subagent.started')
+    expect(seen.map(n => n.method)).toContain('subagent.finished')
+    const childEvents = seen.filter(n => n.method === 'session.event' && n.params.sessionId === 'parent-1-child')
+    expect(childEvents.length).toBeGreaterThan(0)
+    // Child events do not count as the parent's own turn events.
+    expect(result.events.every(event => event.type !== 'assistant/message'
+      || (event.data as { content: { type: string; text?: string }[] }).content[0]?.text !== 'child says hi')).toBe(true)
+    await harness.close()
+  })
+
+  it('reports an error status with the turn-end reason', async () => {
+    const harness = harnessWith({ FAKE_STATUS: 'error', FAKE_REASON_KIND: 'max-tokens' })
+    const result = await harness.run('overflow')
+    expect(result.status).toBe('error')
+    expect(result.reason).toEqual({ kind: 'max-tokens' })
+    await harness.close()
+  })
+
+  it('omits the reason when the runtime settled without one', async () => {
+    const harness = harnessWith({ FAKE_STATUS: 'error', FAKE_REASON_KIND: 'none' })
+    const result = await harness.run('no turn')
+    expect(result.status).toBe('error')
+    expect(result.reason).toBeUndefined()
+    await harness.close()
+  })
+
+  it('sends the configured cwd/provider/model in the handshake exactly once', async () => {
+    const dir = await tempDir('sdk-client-init-')
+    const recordFile = join(dir, 'init.jsonl')
+    const harness = new DeepSeekHarness({
+      launch: fakeLaunch({ FAKE_RECORD_INIT: recordFile }),
+      cwd: dir,
+      provider: 'custom-provider',
+      model: 'custom-model',
+    })
+    cleanups.push(() => harness.close())
+    await harness.run('one')
+    await harness.run('two')
+    await harness.close()
+    const records = (await readFile(recordFile, 'utf8')).trim().split('\n').map(line => JSON.parse(line) as object)
+    expect(records).toEqual([{ cwd: dir, provider: 'custom-provider', model: 'custom-model' }])
+  })
+
+  it('propagates a JSON-RPC error response from initialize and closes the runtime', async () => {
+    const harness = harnessWith({ FAKE_INIT_ERROR: '1' })
+    const failure = await harness.run('boom').then(
+      () => { throw new Error('run unexpectedly succeeded') },
+      (error: unknown) => error,
+    )
+    expect(failure).toBeInstanceOf(JsonRpcResponseError)
+    expect(failure).toMatchObject({ code: 7, message: 'scripted init failure', data: { hint: 'fake' } })
+    // The failed handshake reset lets a later start retry instead of wedging.
+    await expect(harness.run('later')).rejects.toThrow()
+  })
+
+  it('rejects a malformed initialize result as a protocol error', async () => {
+    const harness = harnessWith({ FAKE_MALFORMED: '1' })
+    await expect(harness.run('bad')).rejects.toThrow(SdkProtocolError)
+  })
+
+  it('supports await using disposal', async () => {
+    let captured: DeepSeekHarness
+    {
+      await using harness = new DeepSeekHarness({ launch: fakeLaunch() })
+      captured = harness
+      const result = await harness.run('scoped')
+      expect(result.status).toBe('ok')
+    }
+    // After scope exit the runtime is closed: reuse fails loudly.
+    await expect(captured.run('after')).rejects.toThrow(TransportClosedError)
+  })
+})
+
+describe('HarnessClient', () => {
+  it('times out a hung request at the per-call bound', async () => {
+    const client = new HarnessClient(fakeLaunch({ FAKE_HANG_PROMPT: '1' }))
+    cleanups.push(() => client.close())
+    await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' })
+    await expect(client.request('session/prompt', { sessionId: 's', contentBlocks: normalizeInput('hi') }, 200))
+      .rejects.toThrow(RequestTimeoutError)
+    await client.close()
+  })
+
+  it('applies the client-wide request timeout when no per-call bound is given', async () => {
+    const client = new HarnessClient(fakeLaunch({ FAKE_HANG_PROMPT: '1' }, { requestTimeoutMs: 400 }))
+    cleanups.push(() => client.close())
+    // The bound applies from send, so it holds regardless of runtime boot time.
+    await expect(client.prompt('s', normalizeInput('hi'))).rejects.toThrow(RequestTimeoutError)
+    await client.close()
+  })
+
+  it('rejects a malformed prompt acceptance as a protocol error', async () => {
+    const client = new HarnessClient(fakeLaunch({ FAKE_MALFORMED: '1' }))
+    cleanups.push(() => client.close())
+    await expect(client.prompt('s', normalizeInput('hi'))).rejects.toThrow(SdkProtocolError)
+    await client.close()
+  })
+
+  it('fails pending requests with exit code and stderr tail when the runtime dies', async () => {
+    const client = new HarnessClient(fakeLaunch({ FAKE_EXIT_BEFORE_INIT: '1', FAKE_STDERR: 'fatal: scripted death' }))
+    cleanups.push(() => client.close())
+    const failure = await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' }).then(
+      () => { throw new Error('initialize unexpectedly succeeded') },
+      (error: unknown) => error,
+    )
+    expect(failure).toBeInstanceOf(TransportClosedError)
+    expect(String(failure)).toContain('exit code: 3')
+    expect(String(failure)).toContain('fatal: scripted death')
+    // Requests after death fail immediately with the same context.
+    await expect(client.request('initialize', {})).rejects.toThrow('exit code: 3')
+  })
+
+  it('flushes an unterminated stderr line into the tail at close', async () => {
+    const client = new HarnessClient(fakeLaunch({ FAKE_STDERR_NO_NEWLINE: 'no trailing newline', FAKE_EXIT_BEFORE_INIT: '1' }))
+    cleanups.push(() => client.close())
+    const failure = await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' }).then(
+      () => { throw new Error('initialize unexpectedly succeeded') },
+      (error: unknown) => error,
+    )
+    expect(String(failure)).toContain('no trailing newline')
+  })
+
+  it('fails fast when the command does not exist', async () => {
+    const client = new HarnessClient({ command: join(tmpdir(), 'dsh-no-such-runtime-bin') })
+    cleanups.push(() => client.close())
+    await expect(client.request('initialize', {}, 1_000)).rejects.toThrow(TransportClosedError)
+  })
+
+  it('close() is idempotent, reaps the child, and fails later use', async () => {
+    const client = new HarnessClient(fakeLaunch())
+    await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' })
+    await Promise.all([client.close(), client.close()])
+    expect(() => { client.start() }).toThrow(TransportClosedError)
+    await expect(client.request('anything')).rejects.toThrow(TransportClosedError)
+    // Close with no child ever spawned is a no-op.
+    const untouched = new HarnessClient(fakeLaunch())
+    await untouched.close()
+  })
+
+  it('escalates through SIGTERM when the runtime ignores EOF', async () => {
+    const dir = await tempDir('sdk-client-ladder-')
+    const sigtermFile = join(dir, 'sigterm.txt')
+    const client = new HarnessClient(fakeLaunch(
+      { FAKE_IGNORE_EOF: '1', FAKE_SIGTERM_FILE: sigtermFile },
+      { shutdownTimeoutMs: 100, disposeEofGraceMs: 100, disposeGraceMs: 1_000 },
+    ))
+    await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' })
+    await client.close()
+    expect((await stat(sigtermFile)).isFile()).toBe(true)
+  })
+
+  it('escalates to SIGKILL when the runtime traps SIGTERM too', async () => {
+    const client = new HarnessClient(fakeLaunch(
+      { FAKE_IGNORE_EOF: '1', FAKE_TRAP_SIGTERM: '1' },
+      { shutdownTimeoutMs: 100, disposeEofGraceMs: 100, disposeGraceMs: 300 },
+    ))
+    await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' })
+    // Resolves (does not hang or reject): the SIGKILL rung reaped the child.
+    await client.close()
+  })
+
+  it('delivers notifications to unfiltered and filtered subscriptions in wire order', async () => {
+    const client = new HarnessClient(fakeLaunch())
+    cleanups.push(() => client.close())
+    await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' })
+
+    const all = client.subscribe()
+    const finishedOnly = client.subscribe(n => n.method === 'session.finished')
+    await client.prompt('sub-test', normalizeInput('go'))
+
+    const first = await all.next()
+    expect(first.method).toBe('session.event')
+    const finished = await finishedOnly.next()
+    expect(finished.method).toBe('session.finished')
+    expect(finishedOnly.tryNext()).toBeUndefined()
+
+    // Async iteration consumes queued items and then parks.
+    const collected: string[] = []
+    for await (const notification of all) {
+      collected.push(notification.method)
+      if (notification.method === 'session.finished') break
+    }
+    expect(collected.at(-1)).toBe('session.finished')
+
+    all.close()
+    finishedOnly.close()
+    await expect(all.next()).rejects.toThrow('notification subscription closed')
+    await client.close()
+  })
+
+  it('closes subscriptions with the runtime and rejects parked waiters', async () => {
+    const client = new HarnessClient(fakeLaunch())
+    await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' })
+    const subscription = client.subscribe()
+    const parked = subscription.next()
+    await client.close()
+    await expect(parked).rejects.toThrow(TransportClosedError)
+  })
+
+  it('scopes the session tree across multi-hop lineage and ignores foreign sessions', async () => {
+    const client = new HarnessClient(fakeLaunch())
+    cleanups.push(() => client.close())
+    await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' })
+
+    const tree = client.subscribeSessionTree('root')
+    // Lineage edges arrive as subagent.started notifications.
+    const inject = (method: string, params: Record<string, unknown>): void => {
+      (client as unknown as { dispatchNotification(n: HarnessNotification): void }).dispatchNotification({ method, params })
+    }
+    inject('subagent.started', { parentSessionId: 'root', childSessionId: 'child' })
+    inject('subagent.started', { parentSessionId: 'child', childSessionId: 'grandchild' })
+    inject('session.event', { sessionId: 'grandchild', event: { type: 'noop' } })
+    inject('session.event', { sessionId: 'stranger', event: { type: 'noop' } })
+    inject('subagent.started', { parentSessionId: 'other-root', childSessionId: 'other-child' })
+    inject('subagent.finished', { parentSessionId: 'child', childSessionId: 'grandchild' })
+    // Self-loop and empty edges must not corrupt the lineage map.
+    inject('subagent.started', { parentSessionId: 'loop', childSessionId: 'loop' })
+    inject('subagent.started', { parentSessionId: '', childSessionId: 'x' })
+    inject('subagent.finished', { childSessionId: 'root' })
+
+    expect((await tree.next()).method).toBe('subagent.started')
+    expect((await tree.next()).method).toBe('subagent.started')
+    expect((await tree.next()).params.sessionId).toBe('grandchild')
+    expect((await tree.next()).method).toBe('subagent.finished')
+    // The foreign-root edge and stranger event were filtered; next is the root-child edge.
+    expect((await tree.next()).params.childSessionId).toBe('root')
+    tree.close()
+    await client.close()
+  })
+})
+
+describe('stderr tail bound', () => {
+  it('keeps only the newest lines up to the limit', async () => {
+    const manyLines = Array.from({ length: 450 }, (_, i) => `line-${i}`).join('\n')
+    const client = new HarnessClient(fakeLaunch({ FAKE_STDERR: manyLines, FAKE_EXIT_BEFORE_INIT: '1' }))
+    cleanups.push(() => client.close())
+    const failure = await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' }).then(
+      () => { throw new Error('initialize unexpectedly succeeded') },
+      (error: unknown) => error,
+    )
+    const text = String(failure)
+    // The tail is bounded to the newest 400 lines: the oldest are dropped.
+    expect(text).toContain('line-449')
+    expect(text).not.toContain('line-0\n')
+  })
+})
+
+describe('pure helpers', () => {
+  it('normalizeInput wraps strings and passes blocks through', () => {
+    expect(normalizeInput('x')).toEqual([{ type: 'text', text: 'x' }])
+    const blocks = [{ type: 'text' as const, text: 'y' }]
+    expect(normalizeInput(blocks)).toBe(blocks)
+  })
+
+  it('finalResponse reads the last assistant message and tolerates absence', () => {
+    expect(finalResponse([])).toBe('')
+    expect(finalResponse([{ type: 'turn/start', seq: 0, time: 0, data: { turn: 0 } } as never])).toBe('')
+    expect(finalResponse([
+      { type: 'assistant/message', seq: 0, time: 0, data: { content: [{ type: 'text', text: 'first' }] } } as never,
+      { type: 'assistant/message', seq: 1, time: 0, data: { content: [{ type: 'text', text: 'a' }, { type: 'tool-call' }, { type: 'text', text: 'b' }] } } as never,
+    ])).toBe('ab')
+  })
+})