diff --git a/server/services/agentCliSpawning.js b/server/services/agentCliSpawning.js index 4f6ae6835a..7696c1f3e3 100644 --- a/server/services/agentCliSpawning.js +++ b/server/services/agentCliSpawning.js @@ -425,15 +425,44 @@ export async function spawnDirectly({ env: childEnv }); - // spawn() can hand back a handle with no pid or stdio when command lookup - // fails. Listen immediately so that failure cannot become an unhandled error - // while the async setup below is still yielding. + // Every child listener is registered HERE, in the same tick as spawn(), into + // forwarding shims that buffer until the real handler bodies exist further + // down (#5791). Those bodies close over state built by the async setup that + // follows — and that setup yields for real state-file I/O (`await + // updateAgent(..., { pid })`), not just a microtask. A CLI that starts and + // exits immediately (bad flag, instant auth refusal) emits its entire life + // cycle inside that window: without the shims its stdout/stderr is lost and + // its `close` lands on nothing, leaving the run record, the execution lane + // and the `activeAgents` entry non-terminal until the orphan reaper notices. + // Buffering rather than hoisting the handler bodies keeps the change local. + // + // spawn() can also hand back a handle with no pid or stdio when command + // lookup fails, hence the optional chaining on the stream registrations. let pendingSpawnError = null; let handleSpawnError = null; + // One ordered queue for both streams: buffering them separately would lose + // the relative order of interleaved stdout/stderr chunks in the transcript. + const pendingChildOutput = []; + let pendingClose = null; + let handleStdout = null; + let handleStderr = null; + let handleClose = null; claudeProcess.on('error', (err) => { if (handleSpawnError) void handleSpawnError(err); else pendingSpawnError = err; }); + claudeProcess.stdout?.on('data', (data) => { + if (handleStdout) handleStdout(data); + else pendingChildOutput.push({ stream: 'stdout', data }); + }); + claudeProcess.stderr?.on('data', (data) => { + if (handleStderr) handleStderr(data); + else pendingChildOutput.push({ stream: 'stderr', data }); + }); + claudeProcess.on('close', (code) => { + if (handleClose) void handleClose(code); + else pendingClose = { code }; + }); // Same reasoning for the stdin pipe: a child that exits before reading it // emits EPIPE, and an unlistened stream 'error' out here would crash the // server. The 'error'/'exit' handlers below settle the run with the real cause. @@ -579,7 +608,7 @@ export async function spawnDirectly({ } }, 3000); - claudeProcess.stdout.on('data', (data) => { + handleStdout = (data) => { try { const text = data.toString(); // Detect fallback signals SYNCHRONOUSLY, before enqueuing any transcript @@ -616,9 +645,9 @@ export async function spawnDirectly({ } catch (err) { console.error(`❌ agentCli stdout handler failed: ${err.message}`); } - }); + }; - claudeProcess.stderr.on('data', (data) => { + handleStderr = (data) => { try { const text = data.toString(); // Synchronous fallback detection before the serialized write (see stdout). @@ -640,7 +669,7 @@ export async function spawnDirectly({ } catch (err) { console.error(`❌ agentCli stderr handler failed: ${err.message}`); } - }); + }; handleSpawnError = async (err) => { // Runs outside the request lifecycle — an uncaught throw from the awaited @@ -682,9 +711,8 @@ export async function spawnDirectly({ activeAgents.delete(agentId); } }; - if (pendingSpawnError) void handleSpawnError(pendingSpawnError); - claudeProcess.on('close', async (code) => { + handleClose = async (code) => { // Runs outside the request lifecycle — a throw from outputBatcher.flush, // analyzeAgentFailure, or finalizeAgent would re-escape this async handler // as an unhandled rejection and crash the process. The inner try/finally @@ -975,7 +1003,22 @@ export async function spawnDirectly({ activeAgents.delete(agentId); } } - }); + }; + + // Replay whatever the child emitted while the async setup above was yielding + // (#5791). Output first, so the transcript is complete before a buffered + // terminal event finalizes the run — and it is enqueued onto + // `transcriptWriteTail` here, before either terminal handler's + // `drainTranscriptWrites()` captures that tail. + for (const { stream, data } of pendingChildOutput) { + if (stream === 'stderr') handleStderr(data); + else handleStdout(data); + } + pendingChildOutput.length = 0; + // A failed spawn emits 'error' AND then 'close'; the error path already + // finalizes the run, so replay exactly one terminal event, never both. + if (pendingSpawnError) void handleSpawnError(pendingSpawnError); + else if (pendingClose) void handleClose(pendingClose.code); return agentId; } diff --git a/server/services/agentCliSpawning.test.js b/server/services/agentCliSpawning.test.js index e39e285c8c..6b5655af3b 100644 --- a/server/services/agentCliSpawning.test.js +++ b/server/services/agentCliSpawning.test.js @@ -680,6 +680,44 @@ describe('stream error containment', () => { }); }); + // ─── Fast-exiting child: events emitted during the post-spawn await (#5791) ─ + + it('does not drop stdout or close emitted while the post-spawn await is still yielding', async () => { + // The child's whole life cycle can land between spawn() and the point where + // the real stdout/stderr/close handler bodies are installed: the + // `updateAgent(agentId, { pid })` in between is real state-file I/O, so it + // yields for a macrotask, not just a microtask. A CLI that dies instantly + // (bad flag, instant auth refusal) exits inside that window — before the fix + // its output was lost and its `close` landed on nothing, leaving the run + // record, the execution lane and the activeAgents entry non-terminal until + // the orphan reaper eventually noticed. + const { activeAgents } = await import('./agentState.js'); + const { finalizeAgent } = await import('./agentFinalization.js'); + activeAgents.clear(); + finalizeAgent.mockClear(); + agentStateMocks.updateAgent.mockImplementationOnce(async () => { + // Emit from inside the same macrotask yield the real updateAgent creates. + await new Promise((r) => setImmediate(r)); + fakeProcess.stdout.emit('data', Buffer.from( + '{"type":"stream_event","event":{"type":"content_block_delta","delta":{"type":"text_delta","text":"unknown flag --nope\\n"}}}\n' + )); + fakeProcess.emit('close', 2); + }); + + await expect(spawnDirectly(minimalArgs)).resolves.toBe('agent-test'); + await new Promise((r) => setTimeout(r, 30)); + + expect(finalizeAgent).toHaveBeenCalledWith(expect.objectContaining({ + agentId: 'agent-test', + exitCode: 2, + success: false, + // The output the child managed to write before exiting is exactly what + // explains the failure, so it has to survive the buffering too. + outputBuffer: expect.stringContaining('unknown flag --nope'), + })); + expect(activeAgents.has('agent-test')).toBe(false); + }); + // ─── Lifecycle ledger — the first-output boundary (#4540) ───────────────── it('records ONE run.output on the first real byte, and never again', async () => { diff --git a/server/services/agentManagement.test.js b/server/services/agentManagement.test.js index d1bf487217..9b5f0aa034 100644 --- a/server/services/agentManagement.test.js +++ b/server/services/agentManagement.test.js @@ -1350,15 +1350,18 @@ describe('close-handler skip-finalization — source contract', () => { } it('CLI close handler guards with pausedAgents.has and returns before finalizeAgent', () => { - // The guard appears in the claudeProcess.on('close', ...) callback. - const closeIdx = AGENT_CLI_SRC.indexOf("claudeProcess.on('close'"); - expect(closeIdx, "claudeProcess 'close' handler must exist").toBeGreaterThan(-1); + // The real body lives in `handleClose`, not in the `claudeProcess.on('close')` + // registration — that registration is a forwarding shim attached in the same + // tick as spawn() so a fast-exiting child's close event isn't dropped while + // the async setup is still yielding (#5791). + const closeIdx = AGENT_CLI_SRC.indexOf('handleClose = async (code)'); + expect(closeIdx, 'CLI handleClose handler must exist').toBeGreaterThan(-1); // Extract the full callback body via brace-balancing rather than a fixed // slice — a try/catch crash-guard wrapper can push finalizeAgent past any // fixed window (see #1825). - const closeBody = extractFunctionBody(AGENT_CLI_SRC, "claudeProcess.on('close'"); - expect(closeBody, "claudeProcess 'close' handler body must be extractable").toBeTruthy(); + const closeBody = extractFunctionBody(AGENT_CLI_SRC, 'handleClose = async (code)'); + expect(closeBody, 'CLI handleClose handler body must be extractable').toBeTruthy(); // Guard present expect(closeBody).toMatch(/pausedAgents\.has\(agentId\)/);