diff --git a/packages/subagent/subagent-acp/src/run.ts b/packages/subagent/subagent-acp/src/run.ts index a0adcb1f4a..0a2984cad6 100644 --- a/packages/subagent/subagent-acp/src/run.ts +++ b/packages/subagent/subagent-acp/src/run.ts @@ -283,12 +283,59 @@ function startupFailure( if (child.pid <= 0) { return new AcpRunFailure({ stage: 'process', category: 'process-start' }, error) } - return new AcpRunFailure( - outcome === undefined - ? { stage, category: 'transport' } - : { stage, category: 'process-exit', outcome }, - error, - ) + return new AcpRunFailure(acpProcessFailureFacts(stage, stage, outcome), error) +} + +/** + * Classify an ACP operation failure from its process outcome. + * @param stage - active protocol stage when no process outcome was observed. + * @param processExitStage - diagnostic stage used when the process exited. + * @param outcome - observed child exit, or undefined while the child remains live. + * @returns fixed failure facts suitable for model-visible diagnostics. + */ +export function acpProcessFailureFacts( + stage: Extract, + processExitStage: Extract, + outcome: SubprocessOutcome | undefined, +): AcpFailureFacts { + return outcome === undefined + ? { stage, category: 'transport' } + : { stage: processExitStage, category: 'process-exit', outcome } +} + +/** + * Observe a child outcome until it settles, the caller aborts, or the grace elapses. + * @param pid - child process id; non-positive ids represent spawn failure. + * @param processDone - child outcome promise. + * @param processOutcome - outcome already observed by the run, if any. + * @param graceMs - maximum observation window. + * @param signal - optional caller cancellation signal. + * @returns the observed outcome, or undefined when observation is interrupted. + */ +export async function observeAcpProcessOutcome( + pid: number, + processDone: Promise, + processOutcome: SubprocessOutcome | undefined, + graceMs: number, + signal?: AbortSignal, +): Promise { + if (processOutcome !== undefined || pid <= 0) return processOutcome + const timeout = AbortSignal.timeout(Math.ceil(graceMs)) + const bound = signal === undefined ? timeout : AbortSignal.any([signal, timeout]) + const aborted = Promise.withResolvers() + const onObservationAbort = (): void => { aborted.resolve(undefined) } + bound.addEventListener('abort', onObservationAbort, { once: true }) + /* v8 ignore next -- closes the event-loop race between listener registration and the preceding derived-signal check. */ + if (bound.aborted) onObservationAbort() + try { + return await Promise.race([processDone, aborted.promise]) + } catch { + // The active protocol failure remains authoritative when exit observation fails. + /* v8 ignore next -- a published child.done cannot reject; spawn rejection is consumed before publication. */ + return processOutcome + } finally { + bound.removeEventListener('abort', onObservationAbort) + } } /** Map one remote terminal reason to the optional safe failure line it needs. */ @@ -373,25 +420,8 @@ 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 <= 0) return processOutcome - const timeout = AbortSignal.timeout(Math.ceil(spec.disposeGraceMs)) - const bound = signal === undefined ? timeout : AbortSignal.any([signal, timeout]) - const aborted = Promise.withResolvers() - const onObservationAbort = (): void => { aborted.resolve(undefined) } - bound.addEventListener('abort', onObservationAbort, { once: true }) - /* v8 ignore next -- closes the event-loop race between listener registration and the preceding derived-signal check. */ - if (bound.aborted) onObservationAbort() - try { - return await Promise.race([processDone, aborted.promise]) - } catch { - // The active protocol failure remains authoritative when exit observation fails. - /* v8 ignore next -- a published child.done cannot reject; spawn rejection is consumed before publication. */ - return processOutcome - } finally { - bound.removeEventListener('abort', onObservationAbort) - } - } + const observeProcessOutcome = (signal?: AbortSignal): Promise => + observeAcpProcessOutcome(child.pid, processDone, processOutcome, spec.disposeGraceMs, signal) // Startup rollback and the published handle share one process teardown. let processDisposal: Promise | undefined @@ -562,9 +592,7 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe } catch (error: unknown) { if (!flags.cancelled) { const outcome = await observeProcessOutcome(request.signal) - const facts = outcome === undefined - ? { stage: 'prompt', category: 'transport' } as const - : { stage: 'process', category: 'process-exit', outcome } as const + const facts = acpProcessFailureFacts('prompt', 'process', outcome) diagnostic = diagnosticText(facts, latestPermission) } throw error diff --git a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts index deb00b2453..41ceabf3ee 100644 --- a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts +++ b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts @@ -10,7 +10,7 @@ import type { Agent } from '@deepseek-ai/dsh-agent' import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout' import type { SubprocessHandle, SubprocessOutcome } from '@deepseek-ai/dsh-subprocess' import * as acp from '../src/index.ts' -import { acpStopReason, acpContentText, DEFAULT_DISPOSE_EOF_GRACE_MS, DEFAULT_DISPOSE_GRACE_MS, disposeAcpChild, startAcpRun, toAcpPrompt, type AcpRunSpec } from '../src/run.ts' +import { acpStopReason, acpContentText, acpProcessFailureFacts, DEFAULT_DISPOSE_EOF_GRACE_MS, DEFAULT_DISPOSE_GRACE_MS, disposeAcpChild, observeAcpProcessOutcome, startAcpRun, toAcpPrompt, type AcpRunSpec } from '../src/run.ts' import LocalSubprocessRuntime from '@deepseek-ai/dsh-subprocess-local' import { spawnSubprocess } from '@deepseek-ai/dsh-subprocess-local/src/spawn.ts' @@ -121,19 +121,6 @@ function tapBoundedExitWait(child: SubprocessHandle, onWait: () => void): Subpro } } -function hideProcessOutcome(child: SubprocessHandle): SubprocessHandle { - return { - pid: child.pid, - stdin: child.stdin, - stdout: child.stdout, - stderr: child.stderr, - collected: child.collected, - done: new Promise(() => {}), - terminate: () => { child.terminate() }, - waitForExit: (signal?: AbortSignal) => child.waitForExit(signal), - } -} - function replaceProcessOutcome(child: SubprocessHandle, outcome: SubprocessOutcome): SubprocessHandle { return { pid: child.pid, @@ -176,6 +163,34 @@ describe('acpContentText / toAcpPrompt', () => { }) }) +describe('ACP process failure observation', () => { + it('classifies transport and process-exit failures without process timing', () => { + expect(acpProcessFailureFacts('initialize', 'initialize', undefined)).toEqual({ + stage: 'initialize', + category: 'transport', + }) + const outcome: SubprocessOutcome = { exitCode: 9, signal: null } + expect(acpProcessFailureFacts('prompt', 'process', outcome)).toEqual({ + stage: 'process', + category: 'process-exit', + outcome, + }) + }) + + it('lets caller cancellation interrupt process observation', async () => { + const controller = new AbortController() + const observed = observeAcpProcessOutcome( + 1, + new Promise(() => {}), + undefined, + 10_000, + controller.signal, + ) + controller.abort() + await expect(observed).resolves.toBeUndefined() + }) +}) + describe('child env layering (through the subprocess seam)', () => { it('drops credential-shaped ambient vars but keeps the explicit extras', async () => { process.env.ACP_TEST_AMBIENT_SECRET_TOKEN = 'leak-me' @@ -624,7 +639,7 @@ describe('dsh-subagent-acp', () => { env: { MOCK_CLOSE_PROTOCOL_ON_INITIALIZE: '1' }, disposeEofGraceMs: 50, disposeGraceMs: 50, - spawn: spec => hideProcessOutcome(spawnSubprocess(spec)), + spawn: spawnSubprocess, }).catch((cause: unknown) => cause) expect(error).toBeInstanceOf(Error) expect((error as Error).message).toBe( @@ -971,7 +986,7 @@ describe('dsh-subagent-acp', () => { env: { MOCK_CLOSE_PROTOCOL_ON_PROMPT: '1' }, disposeEofGraceMs: 100, disposeGraceMs: 100, - spawn: spec => hideProcessOutcome(spawnSubprocess(spec)), + spawn: spawnSubprocess, }) const result = await run.result expect(result).toEqual({