fix(subagent): wait for real startup evidence

2026-08-26 18:51:14 +08:00
1 parent 20f04373c8
commit 215c8c8132
@@ -285,11 +285,11 @@ function reportFailure(spec: AcpRunSpec, error: unknown): void {
 function startupFailure(
  error: unknown,
  stage: Extract<AcpFailureStage, 'initialize' | 'new-session'>,
-  child: SubprocessHandle,
+  processFailure: Error | undefined,
  outcome: SubprocessOutcome | undefined,
 ): AcpRunFailure {
-  if (child.pid === undefined) {
-    return new AcpRunFailure({ stage: 'process', category: 'process-start' }, error)
+  if (processFailure !== undefined) {
+    return new AcpRunFailure({ stage: 'process', category: 'process-start' }, processFailure)
  }
  return new AcpRunFailure(
    /* v8 ignore next -- Windows anonymous pipes cannot expose a live-child protocol close during startup. */
@@ -366,10 +366,17 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe
  }
  /* v8 ignore stop */
  let processOutcome: SubprocessOutcome | undefined
-  const processDone = child.done.then((outcome) => {
-    processOutcome = outcome
-    return outcome
-  })
+  let processFailure: Error | undefined
+  const processDone = child.done.then(
+    (outcome) => {
+      processOutcome = outcome
+      return outcome
+    },
+    (error: unknown) => {
+      processFailure = toError(error)
+      throw processFailure
+    },
+  )

  // Spawn-level failure surfaces as `done` rejecting into the startup race; a
  // clean exit must never win it, so the success arm parks forever. (The ACP
@@ -383,7 +390,7 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe
  spawnFailed.catch(() => { /* observed by the startup race; never unhandled */ })

  const observeProcessOutcome = async (signal?: AbortSignal): Promise<SubprocessOutcome | undefined> => {
-    if (processOutcome !== undefined || child.pid === undefined) return processOutcome
+    if (processOutcome !== undefined) return processOutcome
    const timeout = AbortSignal.timeout(Math.ceil(spec.disposeGraceMs))
    const bound = signal === undefined ? timeout : AbortSignal.any([signal, timeout])
    const aborted = Promise.withResolvers<undefined>()
@@ -507,13 +514,16 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe
    // A child closing its protocol stream can precede whole-tree exit
    // observation. Local cancellation does not need the discarded startup
    // classification; other failures use the configured process grace.
+    const observedOutcome = !cancelledBeforeCleanup && !(error instanceof AcpRunFailure)
+      ? await observeProcessOutcome()
+      : undefined
    const startup = cancelledBeforeCleanup
      ? { kind: 'cancelled' } as const
      : {
        kind: 'failed',
        failure: error instanceof AcpRunFailure
          ? error
-          : startupFailure(error, startupStage, child, await observeProcessOutcome()),
+          : startupFailure(error, startupStage, processFailure, observedOutcome),
      } as const
    if (startup.kind === 'cancelled') {
      // Local cancellation owns the startup outcome; only cleanup failure is
@@ -521,7 +531,7 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe
    } else {
      reportFailure(spec, error instanceof AcpRunFailure
        ? error.cause
-        : error)
+        : processFailure ?? error)
    }
    try {
      await disposeProcess()
@@ -572,13 +582,14 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe
      } catch (error: unknown) {
        if (!flags.cancelled) {
          const outcome = await observeProcessOutcome(request.signal)
-          /* v8 ignore next -- Windows anonymous pipes cannot expose a live-child prompt transport failure. */
-          const facts = outcome === undefined
-            ? { stage: 'prompt', category: 'transport' } as const
-            : { stage: 'process', category: 'process-exit', outcome } as const
+          const facts = processFailure !== undefined
+            ? { stage: 'process', category: 'process-start' } as const
+            : outcome === undefined
+              ? { stage: 'prompt', category: 'transport' } as const
+              : { stage: 'process', category: 'process-exit', outcome } as const
          diagnostic = diagnosticText(facts, latestPermission)
        }
-        throw error
+        throw processFailure ?? error
      }
    },
    collectOutput,
@@ -179,6 +179,19 @@ function replaceProcessOutcome(child: SubprocessHandle, outcome: SubprocessOutco
  }
 }

+function hideProcessPid(child: SubprocessHandle): SubprocessHandle {
+  return {
+    pid: undefined,
+    stdin: child.stdin,
+    stdout: child.stdout,
+    stderr: child.stderr,
+    collected: child.collected,
+    done: child.done,
+    terminate: () => { child.terminate() },
+    waitForExit: (signal?: AbortSignal) => child.waitForExit(signal),
+  }
+}
+
 describe('acpStopReason', () => {
  it('maps each ACP stop reason to the harness vocabulary', () => {
    expect(acpStopReason('end_turn')).toBe('completed')
@@ -734,7 +747,7 @@ describe('dsh-subagent-acp', () => {
      env: { MOCK_CRASH_ON_INITIALIZE: '1' },
      disposeEofGraceMs: DEFAULT_DISPOSE_EOF_GRACE_MS,
      disposeGraceMs: DEFAULT_DISPOSE_GRACE_MS,
-      spawn: spawnSubprocess,
+      spawn: spec => hideProcessPid(spawnSubprocess(spec)),
    }).catch((cause: unknown) => cause)
    expect(error).toBeInstanceOf(Error)
    expect((error as Error).message).toBe(
@@ -751,7 +764,7 @@ describe('dsh-subagent-acp', () => {
      env: {},
      disposeEofGraceMs: 50,
      disposeGraceMs: 50,
-      spawn: spec => closeProtocolImmediately(spawnSubprocess(spec)),
+      spawn: spec => closeProtocolImmediately(hideProcessPid(spawnSubprocess(spec))),
    }).catch((cause: unknown) => cause)
    expect(error).toBeInstanceOf(Error)
    expect((error as Error).message).toBe(
@@ -759,6 +772,37 @@ describe('dsh-subagent-acp', () => {
    )
  })

+  it('reuses a direct outcome already observed before the startup transport closes', async () => {
+    const outcome = { exitCode: 19, signal: null } as const
+    const stdin = new PassThrough()
+    const stdout = new PassThrough()
+    const starting = startAcpRun(request(), {
+      command: 'fake-acp',
+      args: [],
+      cwd: process.cwd(),
+      permission: 'reject',
+      env: {},
+      disposeEofGraceMs: 50,
+      disposeGraceMs: 50,
+      spawn: () => ({
+        pid: undefined,
+        stdin,
+        stdout,
+        stderr: undefined,
+        collected: {},
+        done: Promise.resolve(outcome),
+        terminate: vi.fn(),
+        waitForExit: vi.fn().mockResolvedValue(true),
+      }),
+    })
+    await Promise.resolve()
+    stdout.end()
+
+    await expect(starting).rejects.toThrow(
+      `subagent-acp: ${expectedFailure('stage: initialize; category: process-exit; exit code: 19')}`,
+    )
+  })
+
  it('reaps a child whose session/new response omits the session id', async () => {
    const tmp = mkdtempSync(join(tmpdir(), 'acp-malformed-session-'))
    const flushed = join(tmp, 'flushed')
@@ -1144,8 +1188,16 @@ describe('dsh-subagent-acp', () => {
  })

  it('preserves partial output and structured process facts when the child exits', async () => {
-    const ctx = await setup({ MOCK_TEXT: 'partial answer', MOCK_CRASH_AFTER_CHUNK: '1' })
-    const run = await ctx.subagents.start('acp', request())
+    const run = await startAcpRun(request(), {
+      command: process.execPath,
+      args: [mockServer],
+      cwd: process.cwd(),
+      permission: 'reject',
+      env: { MOCK_TEXT: 'partial answer', MOCK_CRASH_AFTER_CHUNK: '1' },
+      disposeEofGraceMs: DEFAULT_DISPOSE_EOF_GRACE_MS,
+      disposeGraceMs: DEFAULT_DISPOSE_GRACE_MS,
+      spawn: spec => hideProcessPid(spawnSubprocess(spec)),
+    })
    const result = await run.result
    expect(result).toEqual({
      output: [{ type: 'text', text: 'partial answer' }],
@@ -1155,6 +1207,45 @@ describe('dsh-subagent-acp', () => {
    await run.dispose()
  })

+  it('classifies a rejected direct result independently of PID publication', async () => {
+    const processFailure = new Error('remote provider failed before publishing a PID')
+    const direct = Promise.withResolvers<SubprocessOutcome>()
+    let realChild: SubprocessHandle | undefined
+    const errors: Error[] = []
+    const run = await startAcpRun(request(), {
+      command: process.execPath,
+      args: [mockServer],
+      cwd: process.cwd(),
+      permission: 'reject',
+      env: { MOCK_HANG: '1' },
+      disposeEofGraceMs: 100,
+      disposeGraceMs: 100,
+      spawn: (spec) => {
+        const child = spawnSubprocess(spec)
+        realChild = child
+        return closeProtocolOnPrompt({
+          pid: undefined,
+          stdin: child.stdin,
+          stdout: child.stdout,
+          stderr: child.stderr,
+          collected: child.collected,
+          done: direct.promise,
+          terminate: () => { child.terminate() },
+          waitForExit: (signal?: AbortSignal) => child.waitForExit(signal),
+        }, () => { direct.reject(processFailure) })
+      },
+      onError: (error) => { errors.push(error) },
+    })
+    await expect(run.result).resolves.toEqual({
+      output: [],
+      diagnostic: expectedFailure('stage: process; category: process-start'),
+      stopReason: 'error',
+    })
+    expect(errors).toContain(processFailure)
+    await run.dispose()
+    await realChild?.done
+  })
+
  it('reports a signal-only process outcome', async () => {
    const run = await startAcpRun(request(), {
      command: process.execPath,
@@ -259,6 +259,28 @@ export async function consumeClaudeQuery(
  }
 }

+/** Continue one SDK iterator after startup consumed its first message. */
+async function* prefetchedClaudeQuery(
+  first: SDKMessage,
+  iterator: AsyncIterator<SDKMessage>,
+): AsyncGenerator<SDKMessage, void> {
+  let completed = false
+  try {
+    yield first
+    while (true) {
+      const next = await iterator.next()
+      if (next.done) {
+        completed = true
+        return
+      }
+      yield next.value
+    }
+  } finally {
+    /* v8 ignore next -- the official Query iterator always owns return(). */
+    if (!completed) await iterator.return?.()
+  }
+}
+
 /**
 * Close the official query, terminate the managed process tree, and wait for
 * the subprocess owner to prove it is gone.
@@ -379,7 +401,7 @@ export function claudeQueryOptions(
 * Start one official Claude Agent SDK query and publish its one-shot run.
 * @param request - resolved shared subagent request.
 * @param spec - Workspace, environment, process service, and diagnostic policy.
- * @returns the published run after both Query and real CLI handle exist.
+ * @returns the published run after Query, the real CLI handle, and the first SDK message exist.
 */
 export async function startClaudeCodeRun(
  request: SubagentStartRequest,
@@ -408,7 +430,9 @@ export async function startClaudeCodeRun(

  let child: SubprocessHandle | undefined
  let childFailure: Error | undefined
+  let childStartupFailure: Promise<never> | undefined
  let query: Query | undefined
+  let queryMessages: AsyncIterable<SDKMessage> | undefined
  let managedProcess: ManagedClaudeCodeProcess | undefined
  let diagnostic: string | undefined
  const capturePermissionDiagnostic = (value: string): void => {
@@ -426,7 +450,14 @@ export async function startClaudeCodeRun(
  ): void => {
    child = captured
    managedProcess = process
-    void captured.done.catch((error: unknown) => { childFailure = thrown(error) })
+    childStartupFailure = captured.done.then(
+      () => new Promise<never>(() => {}),
+      (error: unknown) => {
+        childFailure = thrown(error)
+        throw childFailure
+      },
+    )
+    void childStartupFailure.catch(() => {})
  }
  try {
    query = officialQuery({
@@ -438,19 +469,26 @@ export async function startClaudeCodeRun(
        capturePermissionDiagnostic,
      ),
    })
-    if (child === undefined) {
+    if (child === undefined || childStartupFailure === undefined) {
      throw new Error(
        'subagent-claude-code: official SDK did not publish a controllable Claude Code process',
      )
    }
-    // A provider may publish no PID and reject `done` through several already-
-    // queued promise reactions. PID absence is not failure; give that complete
-    // synchronous rejection chain one event-loop turn before publication.
-    await new Promise<void>((resolve) => { setImmediate(resolve) })
-    if (childFailure !== undefined) throw childFailure
-    if (controller.signal.aborted) {
+    if (isAborted(controller.signal)) {
      throw new Error('subagent-claude-code: request was aborted before SDK startup')
    }
+    const iterator = query[Symbol.asyncIterator]()
+    const first = await Promise.race([
+      childStartupFailure,
+      iterator.next(),
+    ])
+    if (isAborted(controller.signal)) {
+      throw new Error('subagent-claude-code: request was aborted before SDK startup')
+    }
+    if (first.done) {
+      throw new Error('subagent-claude-code: official SDK query ended before its first message')
+    }
+    queryMessages = prefetchedClaudeQuery(first.value, iterator)
  } catch (error: unknown) {
    request.signal.removeEventListener('abort', onAbort)
    const cancelledBeforeCleanup = controller.signal.aborted
@@ -513,11 +551,12 @@ export async function startClaudeCodeRun(

  const publishedQuery = query
  const publishedChild = child
+  const publishedMessages = queryMessages
  let receivedResult = false
  const result = settleRunResult({
    attempt: async () => {
      try {
-        return await consumeClaudeQuery(publishedQuery, () => {
+        return await consumeClaudeQuery(publishedMessages, () => {
          capturePermissionDiagnostic(unattendedDiagnostic(
            spec.permissionMode,
            'tool permission',
@@ -262,6 +262,7 @@ function queryFrom(

 function waitingQuery(signal: AbortSignal, close = vi.fn()): Query {
  async function* stream(): AsyncGenerator<SDKMessage, void> {
+    yield { type: 'system', subtype: 'init' } as SDKMessage
    await new Promise<never>((_resolve, reject) => {
      const fail = (): void => {
        reject(signal.reason instanceof Error
@@ -330,7 +331,7 @@ beforeEach(() => {
      env: options.env!,
      signal: options.abortController!.signal,
    }))
-    return queryFrom([])
+    return queryFrom([{ type: 'system', subtype: 'init' } as SDKMessage])
  })
 })

@@ -686,7 +687,10 @@ describe('task admission and package contracts', () => {
    child.stdout.end()
    await expect(run.result).resolves.toEqual({
      output: [],
-      diagnostic: expectedFailureDiagnostic('query-run', 'invalid-result'),
+      diagnostic: expectedFailureDiagnostic('query-run', 'invalid-result', {
+        exitCode: 9,
+        signal: null,
+      }),
      stopReason: 'error',
    })
    expect(warn).toHaveBeenCalledWith(
@@ -1213,6 +1217,7 @@ describe('run publication, cancellation, and settlement', () => {
    for (const outcome of outcomes) {
      const child = fakeChild()
      async function* stream(): AsyncGenerator<SDKMessage, void> {
+        yield { type: 'system', subtype: 'init' } as SDKMessage
        child.settle(outcome)
        await Promise.resolve()
        throw new Error('SECRET_TOKEN from process transport')
@@ -1592,7 +1597,7 @@ describe('run publication, cancellation, and settlement', () => {
      .rejects.not.toThrow('live child cleanup failed')
  })

-  it('waits one event-loop turn for a queued provider startup rejection', async () => {
+  it('waits for the first SDK message or a delayed provider startup rejection', async () => {
    const spawnError = Object.assign(
      new Error('spawn /sdk/claude ENOENT'),
      { code: 'ENOENT', path: '/sdk/claude' },
@@ -1601,7 +1606,59 @@ describe('run publication, cancellation, and settlement', () => {
    const close = vi.fn()
    queryMock.mockImplementationOnce(({ options }) => {
      options.spawnClaudeCodeProcess!(sdkSpawnOptions())
-      queueMicrotask(() => { child.fail(spawnError) })
+      async function* stream(): AsyncGenerator<SDKMessage, void> {
+        await new Promise<never>(() => {})
+      }
+      return Object.assign(stream(), { close }) as unknown as Query
+    })
+
+    const startup = startClaudeCodeRun(request(), {
+      cwd: '/workspace',
+      permissionMode: DEFAULT_CLAUDE_CODE_PERMISSION_MODE,
+      env: {},
+      disposeGraceMs: 5,
+      spawn: () => child.handle,
+    })
+    await nextTask()
+    child.fail(spawnError)
+    await expect(startup).rejects.toMatchObject({ cause: spawnError })
+    expect(close).toHaveBeenCalledOnce()
+    expect(child.terminate).toHaveBeenCalledOnce()
+    expect(child.waitForExit).toHaveBeenCalledOnce()
+  })
+
+  it('keeps local cancellation authoritative when it arrives with the first SDK message', async () => {
+    const controller = new AbortController()
+    const child = fakeChild()
+    const close = vi.fn()
+    queryMock.mockImplementationOnce(({ options }) => {
+      options.spawnClaudeCodeProcess!(sdkSpawnOptions())
+      async function* stream(): AsyncGenerator<SDKMessage, void> {
+        controller.abort(new Error('cancelled while the first message arrived'))
+        yield { type: 'system', subtype: 'init' } as SDKMessage
+      }
+      return Object.assign(stream(), { close }) as unknown as Query
+    })
+
+    await expect(startClaudeCodeRun(
+      request(undefined, controller.signal),
+      {
+        cwd: '/workspace',
+        permissionMode: DEFAULT_CLAUDE_CODE_PERMISSION_MODE,
+        env: {},
+        disposeGraceMs: 5,
+        spawn: () => child.handle,
+      },
+    )).rejects.toThrow('aborted before SDK startup')
+    expect(close).toHaveBeenCalledOnce()
+    expect(child.terminate).toHaveBeenCalledOnce()
+  })
+
+  it('rejects an SDK stream that ends before its first message', async () => {
+    const child = fakeChild()
+    const close = vi.fn()
+    queryMock.mockImplementationOnce(({ options }) => {
+      options.spawnClaudeCodeProcess!(sdkSpawnOptions())
      return queryFrom([], undefined, close)
    })

@@ -1612,7 +1669,9 @@ describe('run publication, cancellation, and settlement', () => {
      disposeGraceMs: 5,
      spawn: () => child.handle,
    })
-    await expect(startup).rejects.toMatchObject({ cause: spawnError })
+    await expect(startup).rejects.toThrow(
+      expectedFailureDiagnostic('query-start', 'unknown'),
+    )
    expect(close).toHaveBeenCalledOnce()
    expect(child.terminate).toHaveBeenCalledOnce()
    expect(child.waitForExit).toHaveBeenCalledOnce()