From 215c8c813283b9bcdc75d984761c8e5847f014b2 Mon Sep 17 00:00:00 2001 From: pku-xht Date: Wed, 26 Aug 2026 18:51:14 +0800 Subject: [PATCH] fix(subagent): wait for real startup evidence --- packages/subagent/subagent-acp/src/run.ts | 41 +++++--- .../subagent-acp/tests/subagent-acp.spec.ts | 99 ++++++++++++++++++- .../subagent/subagent-claude-code/src/run.ts | 59 +++++++++-- .../tests/subagent-claude-code.spec.ts | 69 ++++++++++++- 4 files changed, 234 insertions(+), 34 deletions(-) diff --git a/packages/subagent/subagent-acp/src/run.ts b/packages/subagent/subagent-acp/src/run.ts index 0951df6745..47fbb436e5 100644 --- a/packages/subagent/subagent-acp/src/run.ts +++ b/packages/subagent/subagent-acp/src/run.ts @@ -285,11 +285,11 @@ function reportFailure(spec: AcpRunSpec, error: unknown): void { function startupFailure( error: unknown, stage: Extract, - 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 => { - 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() @@ -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, diff --git a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts index 690c4f6d2a..a362932b49 100644 --- a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts +++ b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts @@ -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() + 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, diff --git a/packages/subagent/subagent-claude-code/src/run.ts b/packages/subagent/subagent-claude-code/src/run.ts index 5ddf5ea703..6dba308789 100644 --- a/packages/subagent/subagent-claude-code/src/run.ts +++ b/packages/subagent/subagent-claude-code/src/run.ts @@ -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, +): AsyncGenerator { + 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 | undefined let query: Query | undefined + let queryMessages: AsyncIterable | 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(() => {}), + (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((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', diff --git a/packages/subagent/subagent-claude-code/tests/subagent-claude-code.spec.ts b/packages/subagent/subagent-claude-code/tests/subagent-claude-code.spec.ts index ae048ed6b4..4253308fc0 100644 --- a/packages/subagent/subagent-claude-code/tests/subagent-claude-code.spec.ts +++ b/packages/subagent/subagent-claude-code/tests/subagent-claude-code.spec.ts @@ -262,6 +262,7 @@ function queryFrom( function waitingQuery(signal: AbortSignal, close = vi.fn()): Query { async function* stream(): AsyncGenerator { + yield { type: 'system', subtype: 'init' } as SDKMessage await new Promise((_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 { + 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 { + await new Promise(() => {}) + } + 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 { + 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()