diff --git a/src/agent/claude/adapter.ts b/src/agent/claude/adapter.ts index 3d9c0fd..7cbe0c9 100644 --- a/src/agent/claude/adapter.ts +++ b/src/agent/claude/adapter.ts @@ -13,6 +13,7 @@ import { } from './owner-only-settings'; import { cleanupSandboxPlaceholders, snapshotPlaceholders } from './sandbox-placeholders'; import { planRuntimeDirMask, wrapWithRuntimeDirMask } from './runtime-dir-mask'; +import { killProcessTree } from '../../platform/kill-tree'; import { spawnProcess, type SpawnedProcessByStdio } from '../../platform/spawn'; import { buildBridgeSystemPrompt } from '../bridge-system-prompt'; import { buildLarkChannelEnv, type LarkChannelEnvContext } from '../lark-channel-env'; @@ -250,6 +251,7 @@ export class ClaudeAdapter implements AgentAdapter { return { runId: opts.runId, + pid: child.pid, events: holdEmptyDone(createEventStream(child, stderrChunks, () => runtimeError), { holdMs: this.emptyResultHoldMs, }), @@ -265,7 +267,10 @@ export class ClaudeAdapter implements AgentAdapter { graceMs: stopGraceMs, reason: 'grace-period-expired', }); - child.kill('SIGKILL'); + // issue #9 (P2): child.kill() cannot reach past a .cmd shim to + // the real claude.exe or its grandchilds — the orphan that held + // a session file open for two hours was exactly that survivor. + void killProcessTree(child.pid, 'stop-grace-expired'); } resolve(); }, stopGraceMs); @@ -275,6 +280,11 @@ export class ClaudeAdapter implements AgentAdapter { }); }); }, + async forceKillTree() { + // Works even when the child already exited: the event stream can + // still be wedged by a grandchild holding the stdout write handle. + await killProcessTree(child.pid, 'force-kill'); + }, waitForExit(timeoutMs: number): Promise { if (child.exitCode !== null || child.signalCode !== null) { return Promise.resolve(true); @@ -314,13 +324,30 @@ async function* createEventStream( const rl = createInterface({ input: child.stdout, crlfDelay: Infinity }); let sawStdout = false; - let silentExitTimer: ReturnType | undefined; - const closeSilentStdout = (): void => { - silentExitTimer = setTimeout(() => { - if (!sawStdout && !child.stdout.readableEnded) child.stdout.destroy(); - }, 50); + // issue #9: did the CLI emit its terminal `result` event? An exit without + // one means the run never finished — the caller must get an error, not + // silence. + let sawResultEvent = false; + // issue #9 (P0): stdout EOF is not a reliable end-of-stream. claude.exe can + // exit before emitting any terminal event while a grandchild that inherited + // the stdout write handle keeps it open — the for-await below then hangs + // forever (the 2026-11-08 two-hour freeze). When the child exits, give any + // final lines 2s to flush, then destroy stdout so readline ends regardless + // of who still holds the write handle. + const EXIT_FLUSH_GRACE_MS = 2_000; + const closeStdoutAfterExit = (sawStdout: boolean): void => { + setTimeout(() => { + if (!child.stdout.readableEnded) { + log.info('agent', 'stdout-force-close-after-exit', { + pid: child.pid ?? null, + sawStdout, + graceMs: EXIT_FLUSH_GRACE_MS, + }); + child.stdout.destroy(); + } + }, EXIT_FLUSH_GRACE_MS).unref(); }; - child.once('exit', closeSilentStdout); + child.once('exit', () => closeStdoutAfterExit(sawStdout)); try { for await (const line of rl) { sawStdout = true; @@ -333,10 +360,9 @@ async function* createEventStream( continue; } yield* translateEvent(parsed); + if (parsedType(parsed) === 'result') sawResultEvent = true; } } finally { - if (silentExitTimer) clearTimeout(silentExitTimer); - child.removeListener('exit', closeSilentStdout); rl.close(); } @@ -376,6 +402,15 @@ async function* createEventStream( message: `claude runtime error: ${runtimeError.message}`, terminationReason: 'failed', }; + } else if (!sawResultEvent) { + // issue #9: a claude.exe that exits 0 without ever emitting its `result` + // event used to end the stream silently — no terminal for the caller, the + // run state stayed "running" and the slot froze. Synthesize the error. + yield { + type: 'error', + message: 'claude exited without emitting a result event', + terminationReason: 'failed', + }; } } @@ -409,6 +444,10 @@ function writeSystemPromptFile( }; } +function parsedType(parsed: unknown): string | undefined { + return typeof parsed === 'object' && parsed !== null ? (parsed as { type?: string }).type : undefined; +} + function isWindowsCommandNotFoundLine(line: string): boolean { return ( process.platform === 'win32' && diff --git a/src/agent/codex/adapter.ts b/src/agent/codex/adapter.ts index f12ea9c..ec2b2d6 100644 --- a/src/agent/codex/adapter.ts +++ b/src/agent/codex/adapter.ts @@ -4,6 +4,7 @@ import { join } from 'node:path'; import type { SandboxMode } from '../../config/profile-schema'; import { log } from '../../core/logger'; import { buildAgentEnv } from '../../platform/agent-env'; +import { killProcessTree } from '../../platform/kill-tree'; import { spawnProcess, type SpawnedProcessByStdio } from '../../platform/spawn'; import { SpawnFailed } from '../../runtime/errors'; import { prefixBridgeSystemPrompt } from '../bridge-system-prompt'; @@ -183,6 +184,7 @@ export class CodexAdapter implements AgentAdapter { return { runId: opts.runId, + pid: child.pid, events: createEventStream(child, stderrChunks, () => runtimeError, () => stopReason, codexHomeDir), async stop() { if (child.exitCode !== null || child.signalCode !== null) return; @@ -197,7 +199,10 @@ export class CodexAdapter implements AgentAdapter { graceMs: stopGraceMs, reason: 'grace-period-expired', }); - child.kill('SIGKILL'); + // issue #9 (P2): child.kill() cannot reach past a .cmd shim to + // the real codex.exe or its grandchilds — same orphan failure + // mode as the claude adapter. + void killProcessTree(child.pid, 'stop-grace-expired'); } resolve(); }, stopGraceMs); @@ -207,6 +212,11 @@ export class CodexAdapter implements AgentAdapter { }); }); }, + async forceKillTree() { + // Works even when the child already exited: the event stream can + // still be wedged by a grandchild holding the stdout write handle. + await killProcessTree(child.pid, 'force-kill'); + }, waitForExit(timeoutMs: number): Promise { if (child.exitCode !== null || child.signalCode !== null) { return Promise.resolve(true); @@ -250,13 +260,24 @@ async function* createEventStream( let lastUsageFetchAt = 0; let threadId: string | undefined; let lastKnownUsage: { inputTokens?: number; outputTokens?: number; cachedInputTokens?: number } | undefined; - let silentExitTimer: ReturnType | undefined; - const closeSilentStdout = (): void => { - silentExitTimer = setTimeout(() => { - if (!sawStdout && !child.stdout.readableEnded) child.stdout.destroy(); - }, 50); + // issue #9 (P0): stdout EOF is not a reliable end-of-stream — same failure + // mode as the claude adapter. When the child exits, give any final lines 2s + // to flush, then destroy stdout so readline ends regardless of who still + // holds the write handle (a grandchild can keep it open indefinitely). + const EXIT_FLUSH_GRACE_MS = 2_000; + const closeStdoutAfterExit = (sawStdout: boolean): void => { + setTimeout(() => { + if (!child.stdout.readableEnded) { + log.info('agent', 'stdout-force-close-after-exit', { + pid: child.pid ?? null, + sawStdout, + graceMs: EXIT_FLUSH_GRACE_MS, + }); + child.stdout.destroy(); + } + }, EXIT_FLUSH_GRACE_MS).unref(); }; - child.once('exit', closeSilentStdout); + child.once('exit', () => closeStdoutAfterExit(sawStdout)); try { for await (const line of rl) { sawStdout = true; @@ -306,8 +327,6 @@ async function* createEventStream( yield* translator.translate(parsed); } } finally { - if (silentExitTimer) clearTimeout(silentExitTimer); - child.removeListener('exit', closeSilentStdout); rl.close(); // aiduCAT: final guaranteed refresh — run-state keeps whichever usage // snapshot arrives last before/carrying the terminal event. diff --git a/src/agent/hermes/adapter.ts b/src/agent/hermes/adapter.ts index fcd9a47..41ef167 100644 --- a/src/agent/hermes/adapter.ts +++ b/src/agent/hermes/adapter.ts @@ -2,6 +2,7 @@ import { createInterface } from 'node:readline'; import type { Readable, Writable } from 'node:stream'; import { log } from '../../core/logger'; import { buildAgentEnv } from '../../platform/agent-env'; +import { killProcessTree } from '../../platform/kill-tree'; import { spawnProcess, type SpawnedProcessByStdio } from '../../platform/spawn'; import { SpawnFailed } from '../../runtime/errors'; import { prefixBridgeSystemPrompt } from '../bridge-system-prompt'; @@ -161,6 +162,9 @@ export class HermesAdapter implements AgentAdapter { return { runId: opts.runId, + get pid() { + return current?.pid; + }, events, async stop() { stopped = true; @@ -198,6 +202,9 @@ export class HermesAdapter implements AgentAdapter { child.once('exit', onExit); }); }, + async forceKillTree() { + await killProcessTree(current?.pid, 'force-kill'); + }, }; } } diff --git a/src/agent/types.ts b/src/agent/types.ts index b0ba87a..2f5fa53 100644 --- a/src/agent/types.ts +++ b/src/agent/types.ts @@ -78,6 +78,8 @@ export interface AgentRunOptions { export interface AgentRun { readonly runId: string; + /** OS pid of the agent process, once spawned; undefined until/unless it is. */ + readonly pid: number | undefined; readonly events: AsyncIterable; stop(): Promise; /** @@ -92,6 +94,14 @@ export interface AgentRun { * 143 instead of 0; waiting it out lets it exit cleanly. */ waitForExit(timeoutMs: number): Promise; + /** + * issue #9 (P1): the watchdog's last-resort exit. Kills the agent process + * tree (Windows: `taskkill /T /F`, so grandchilds holding the stdout write + * handle die too) and must not throw even if the process is already gone. + * Unlike stop() this works after the child exited — the event stream can + * still be wedged on a grandchild-inherited stdout handle. + */ + forceKillTree(): Promise; } /** diff --git a/src/bot/channel.ts b/src/bot/channel.ts index 672a7e2..ad49c40 100644 --- a/src/bot/channel.ts +++ b/src/bot/channel.ts @@ -1611,7 +1611,7 @@ async function runAgentBatch(deps: RunBatchDeps): Promise { recordSession, async () => {}, sessions, - capability.agentId); + capability.agentId, execution.forceClose); await cotDone; if (cotPublisher.degradedReason) { await sendCotDegradedNotice({ @@ -1724,7 +1724,7 @@ async function runAgentBatch(deps: RunBatchDeps): Promise { }, sessions, // (diagnostics) - capability.agentId); + capability.agentId, execution.forceClose); // f0.3.4: claude runs open the card right away — the initial render is // the "☤Thinking…" placeholder with the model logo in the panel header — // instead of waiting for the first tool call / text (10–200s). Later @@ -1851,7 +1851,7 @@ async function runAgentBatch(deps: RunBatchDeps): Promise { } }, sessions, - capability.agentId); + capability.agentId, execution.forceClose); let mdDelivered = false; try { await awaitRenderAwareStream({ @@ -1907,7 +1907,7 @@ async function runAgentBatch(deps: RunBatchDeps): Promise { recordSession, async () => {}, sessions, - capability.agentId); + capability.agentId, execution.forceClose); await sendFinalReply({ channel, chatId, @@ -2201,12 +2201,19 @@ async function processAgentStream( flush: (state: RunState) => Promise, stats: Pick, agentId: string, + // issue #9 (P1): the executor's hard exit, used when stop() cannot end the + // run (child already dead, event stream wedged on a grandchild stdout + // handle). Undefined in callers without an execution (tests). + forceClose: (reason: string) => Promise | undefined, ): Promise { const runStart = Date.now(); let state: RunState = initialState; // f0.3.5: this turn's TPS, timed here (model generation only, tool windows // removed) — never from the CLI's session-cumulative duration fields. const tpsMeter = new TurnTpsMeter(); + // issue #9 (P1): true once this function reached a terminal state — the + // watchdog's delayed force-close must not fire after a normal end. + let settled = false; // Idle watchdog: claude going silent for `idleTimeoutMs` is treated as // "presumed hung", we stop() and surface a timeout marker on the card. @@ -2235,6 +2242,26 @@ async function processAgentStream( void handle.run.stop().catch(() => { /* stop errors are non-fatal */ }); + // issue #9 (P1): stop() is a no-op once the child already exited, and + // then nothing ever ends the event stream — the slot froze for two + // hours on 2026-11-08. Give the stream a short window to end on its + // own; if this function is still running (no terminal state reached), + // force-close: synthetic terminal + process-tree kill. + const FORCE_CLOSE_AFTER_MS = 10_000; + setTimeout(() => { + if (settled) return; + log.fail( + 'agent', + new Error(`watchdog ${fire.kind} fire did not end the run`), + { scope, action: 'force-close', afterMs: FORCE_CLOSE_AFTER_MS }, + ); + const close = forceClose(`watchdog-${fire.kind}-timeout`); + if (close) { + close.catch((err) => { + log.fail('agent', err, { scope, event: 'watchdog-force-close-failed' }); + }); + } + }, FORCE_CLOSE_AFTER_MS).unref(); }, }); watchdog.start(); @@ -2395,6 +2422,7 @@ async function processAgentStream( log.info('card', 'final', { scope, terminal: state.terminal, interrupted: handle.interrupted }); reportMetric('run_e2e_ms', Date.now() - runStart, { terminal: state.terminal }); await flush(state); + settled = true; if (handle.interrupted) { await handle.run.stop(); } diff --git a/src/platform/kill-tree.ts b/src/platform/kill-tree.ts new file mode 100644 index 0000000..00bd8ce --- /dev/null +++ b/src/platform/kill-tree.ts @@ -0,0 +1,49 @@ +import { log } from '../core/logger'; +import { spawnProcess } from './spawn'; + +/** + * issue #9 (P2): kill an agent process AND everything it spawned. + * + * On Windows, `child.kill()` only terminates the direct child — claude/codex + * resolve to a `.cmd` shim routed through `cmd.exe`, and the real CLI process + * (plus any grandchild that inherited the stdout write handle) survives. That + * is exactly how the 2026-11-08 orphan claude.exe hung around for two hours + * holding a session file. `taskkill /T /F` walks the whole tree instead. + * + * POSIX keeps the plain SIGKILL: killing a process group would hit our own + * group (the child is not spawned detached), and the bug this fixes is + * Windows-specific. + * + * Best-effort: a pid that already exited makes taskkill fail harmlessly, so + * this never throws. + */ +export async function killProcessTree(pid: number | undefined, why: string): Promise { + if (pid === undefined) return; + if (process.platform !== 'win32') { + try { + process.kill(pid, 'SIGKILL'); + } catch { + // already dead / already reaped — nothing to do + } + return; + } + await new Promise((resolve) => { + // windowsHide: the daemon runs headless; a taskkill console flash on the + // desktop would be alarming noise. + const killer = spawnProcess('taskkill', ['/PID', String(pid), '/T', '/F'], { + stdio: 'ignore', + windowsHide: true, + }); + const done = (): void => resolve(); + // taskkill exits non-zero when the pid is already gone — that is success + // for us. Only a spawn failure (taskkill missing) is worth a log line. + killer.on('exit', done); + killer.on('error', () => { + log.warn('run', 'kill-tree-spawn-failed', { pid, why }); + done(); + }); + // Belt and braces: taskkill finishes in milliseconds; never hang a caller + // on it no matter what. + setTimeout(done, 5_000).unref(); + }); +} diff --git a/src/runtime/run-executor.ts b/src/runtime/run-executor.ts index 327d92a..05a1d9f 100644 --- a/src/runtime/run-executor.ts +++ b/src/runtime/run-executor.ts @@ -2,8 +2,9 @@ import { randomUUID } from 'node:crypto'; import type { AgentAdapter, AgentEvent, AgentRun } from '../agent/types'; import { ActiveRuns, type RunHandle } from '../bot/active-runs'; import { ProcessPool } from '../bot/process-pool'; -import type { RunPolicyAllow } from '../policy/run-policy'; import { log } from '../core/logger'; +import { killProcessTree } from '../platform/kill-tree'; +import type { RunPolicyAllow } from '../policy/run-policy'; import { RunRejected, SpawnFailed } from './errors'; export interface RunExecutorDeps { @@ -13,6 +14,8 @@ export interface RunExecutorDeps { createRunId?: () => string; now?: () => number; postDoneExitGraceMs?: number; + /** issue #9 (P2): extra ms on top of postDoneExitGraceMs before a stuck previous run is tree-killed. */ + drainKillBudgetMs?: number; } export interface SubmitRunInput { @@ -45,6 +48,13 @@ export interface RunExecution { handle: RunHandle; subscribe(): AsyncIterable; stop(): Promise; + /** + * issue #9 (P1): the watchdog's hard exit — force-close the event stream + * with a synthetic terminal error and kill the process tree, regardless of + * whether the stream was ever going to end on its own. Guarantees cleanup + * (unregister / pool release / scope release) actually runs. + */ + forceClose(reason: string): Promise; } // fix(2026-10-30): 2s was too tight - claude often needs >2s to flush the @@ -61,6 +71,8 @@ interface ScopeProcess { finished: () => boolean; /** Resolves once the process exited (or was killed). */ exited: Promise; + /** issue #9 (P2): agent pid, for the tree kill on the timeout path. */ + pid: number | undefined; } export class RunExecutor { @@ -70,6 +82,7 @@ export class RunExecutor { private readonly createRunId: () => string; private readonly now: () => number; private readonly postDoneExitGraceMs: number; + private readonly drainKillBudgetMs: number; /** * f031-stab-#3: the agent process currently owning each scope. It stays here * after its turn ends (card final, scope idle) until the process exits, and @@ -85,6 +98,7 @@ export class RunExecutor { this.createRunId = deps.createRunId ?? randomUUID; this.now = deps.now ?? Date.now; this.postDoneExitGraceMs = deps.postDoneExitGraceMs ?? DEFAULT_POST_DONE_EXIT_GRACE_MS; + this.drainKillBudgetMs = deps.drainKillBudgetMs ?? DRAIN_KILL_BUDGET_MS; } async submit(input: SubmitRunInput): Promise { @@ -201,6 +215,7 @@ export class RunExecutor { exited: new Promise((resolve) => { drained = resolve; }), + pid: run.pid, }; this.scopeProcs.set(input.scopeId, proc); let cleaning: Promise | undefined; @@ -256,6 +271,19 @@ export class RunExecutor { await run.waitForExit(this.postDoneExitGraceMs); await cleanup(true); }, + forceClose: async (reason: string) => { + handle.interrupted = true; + // issue #9 (P1): hard exit — synthetic terminal first (so consumers + // wind down and cleanup runs via onTerminal), then kill the tree so a + // grandchild-inherited stdout handle can't wedge the next run either. + fanout.abort(reason, { + type: 'error', + message: `run force-closed: ${reason}`, + terminationReason: 'timeout', + }); + await cleanup(false).catch(() => {}); + await run.forceKillTree(); + }, }; } @@ -264,7 +292,7 @@ export class RunExecutor { const startedAt = this.now(); log.info('run', 'wait-previous-exit', { scope: scopeId }); let timer: NodeJS.Timeout | undefined; - const capMs = this.postDoneExitGraceMs + DRAIN_KILL_BUDGET_MS; + const capMs = this.postDoneExitGraceMs + this.drainKillBudgetMs; const outcome = await Promise.race([ prev.exited.then(() => 'exited' as const), new Promise<'timeout'>((resolve) => { @@ -274,6 +302,11 @@ export class RunExecutor { if (timer) clearTimeout(timer); if (outcome === 'timeout') { log.warn('run', 'previous-exit-timeout', { scope: scopeId, capMs }); + // issue #9 (P2): the wedged previous run must not linger as an orphan + // holding its session file open (the 2026-11-08 two-hour orphan). Kill + // the whole tree, fire-and-forget — the next submit must not wait on + // the killer, only on `exited`/cap, both already accounted for above. + void killProcessTree(prev.pid, 'previous-exit-timeout'); if (this.scopeProcs.get(scopeId) === prev) this.scopeProcs.delete(scopeId); } else { log.info('run', 'previous-exited', { scope: scopeId, waitedMs: this.now() - startedAt }); @@ -326,6 +359,9 @@ class EventFanout { private started = false; private done = false; private error: unknown; + private abortRequested = false; + private abortWaker: (() => void) | undefined; + private syntheticTerminal: AgentEvent | undefined; constructor(source: AsyncIterable, onDone: () => Promise, onTerminal: () => void = () => {}) { this.source = source; @@ -363,6 +399,23 @@ class EventFanout { }; } + /** + * issue #9 (P1): the pump can be parked forever inside the source's + * for-await (child exited, grandchild holds the stdout write handle). + * End the stream anyway: a real next() that later resolves is raced against + * an abort promise, the winner defines the outcome, and a synthetic + * terminal event is appended so every consumer reaches a terminal state. + * A run must always have an exit that does not depend on the stream + * cooperating. + */ + abort(reason: string, terminal: AgentEvent): void { + if (this.abortRequested || this.done) return; + this.abortRequested = true; + this.syntheticTerminal = terminal; + log.warn('run', 'stream-force-closed', { reason }); + this.abortWaker?.(); + } + private start(): void { if (this.started) return; this.started = true; @@ -370,25 +423,64 @@ class EventFanout { } private async pump(): Promise { + const it = this.source[Symbol.asyncIterator](); try { - for await (const event of this.source) { + while (true) { + const abortPromise = this.abortRequested + ? undefined + : new Promise<{ kind: 'aborted' }>((resolve) => { + this.abortWaker = () => resolve({ kind: 'aborted' }); + }); + const nextPromise = it.next(); + const outcome = abortPromise + ? await Promise.race([nextPromise.then((r) => ({ kind: 'next' as const, r })), abortPromise]) + : ({ kind: 'next' as const, r: await nextPromise }); + if (outcome.kind !== 'next') { + // Park the abandoned next() behind .catch so a late rejection + // (stream error after abort) can't become an unhandledRejection. + void nextPromise.catch(() => {}); + break; + } + if (outcome.r.done === true) break; + const event = outcome.r.value; this.buffer.push(event); if (isTerminalEvent(event)) { this.onTerminal(); this.wakeAll(); break; } + // Non-terminal events must reach consumers as they arrive — streaming + // cards, TPS timing and the watchdog all depend on real-time delivery. this.wakeAll(); } } catch (err) { - this.error = err; + // An abort raced a failing source: the synthetic terminal wins — the + // forced close is the point of this path. + if (this.abortRequested) { + this.finishWithSynthetic(); + } else { + this.error = err; + } } finally { + this.abortWaker = undefined; + this.finishWithSynthetic(); await this.onDone(); this.done = true; this.wakeAll(); } } + private finishWithSynthetic(): void { + if (!this.abortRequested) return; + const terminal = this.syntheticTerminal; + if (!terminal) return; + const last = this.buffer[this.buffer.length - 1]; + if (last && isTerminalEvent(last)) return; + this.buffer.push(terminal); + this.onTerminal(); + this.wakeAll(); + } + private wakeAll(): void { for (const wake of [...this.waiters]) wake(); } diff --git a/tests/helpers/fake-agent.ts b/tests/helpers/fake-agent.ts index 28311cd..9a480a6 100644 --- a/tests/helpers/fake-agent.ts +++ b/tests/helpers/fake-agent.ts @@ -17,6 +17,7 @@ class FakeRun implements FakeAgentRun { readonly opts: AgentRunOptions; readonly events: AsyncIterable; readonly waitForExitResult: boolean; + readonly pid = undefined; #stopped = false; #waitForExitCalls = 0; @@ -43,6 +44,10 @@ class FakeRun implements FakeAgentRun { this.#stopped = true; } + async forceKillTree(): Promise { + this.#stopped = true; + } + async waitForExit(): Promise { this.#waitForExitCalls++; return this.waitForExitResult; diff --git a/tests/integration/bot/markdown-stream-startup-failure.test.ts b/tests/integration/bot/markdown-stream-startup-failure.test.ts index 8bb0439..b875f7c 100644 --- a/tests/integration/bot/markdown-stream-startup-failure.test.ts +++ b/tests/integration/bot/markdown-stream-startup-failure.test.ts @@ -771,12 +771,14 @@ describe('f04-bwrap: agent start failure never shows the raw error to the group' const n = ++call; const run = { runId: opts.runId, + pid: undefined, events: (async function* () { if (n === 1) await gate.promise; yield { type: 'final_text', content: `ANSWER_${n}` } as AgentEvent; yield { type: 'done', terminationReason: 'normal' } as AgentEvent; })(), async stop() {}, + async forceKillTree() {}, async waitForExit() { return true; }, diff --git a/tests/integration/commands/reconnect-profile.test.ts b/tests/integration/commands/reconnect-profile.test.ts index 295f0c7..a4b9a89 100644 --- a/tests/integration/commands/reconnect-profile.test.ts +++ b/tests/integration/commands/reconnect-profile.test.ts @@ -124,6 +124,7 @@ class ManualRun implements AgentRun { readonly events: AsyncIterable = { async *[Symbol.asyncIterator]() {}, }; + readonly pid = undefined; stopCalls = 0; waitForExitCalls = 0; @@ -137,6 +138,8 @@ class ManualRun implements AgentRun { this.waitForExitCalls++; return true; } + + async forceKillTree(): Promise {} } function message(content: string): NormalizedMessage { diff --git a/tests/integration/comments/comment-lifecycle.test.ts b/tests/integration/comments/comment-lifecycle.test.ts index 9d15f1a..d1d21e3 100644 --- a/tests/integration/comments/comment-lifecycle.test.ts +++ b/tests/integration/comments/comment-lifecycle.test.ts @@ -289,12 +289,17 @@ class BlockingAgentAdapter implements AgentAdapter { this.runOptions.push(opts); return { runId: opts.runId, + pid: undefined, events: this.events(runIndex), stop: async () => { this.wasStopped = true; this.release(runIndex); }, waitForExit: async () => true, + forceKillTree: async () => { + this.wasStopped = true; + this.release(runIndex); + }, }; } diff --git a/tests/integration/comments/comment-run-flow.test.ts b/tests/integration/comments/comment-run-flow.test.ts index 6c74948..ddf5193 100644 --- a/tests/integration/comments/comment-run-flow.test.ts +++ b/tests/integration/comments/comment-run-flow.test.ts @@ -517,6 +517,7 @@ class BlockingAgentAdapter implements AgentAdapter { this.finishers[index] = finish; return { runId: opts.runId, + pid: undefined, events: this.events(index, done, () => stopped), async stop() { stopped = true; @@ -525,6 +526,10 @@ class BlockingAgentAdapter implements AgentAdapter { async waitForExit() { return true; }, + async forceKillTree() { + stopped = true; + finish(); + }, }; } diff --git a/tests/process/claude-adapter.test.ts b/tests/process/claude-adapter.test.ts index 36d740a..71174de 100644 --- a/tests/process/claude-adapter.test.ts +++ b/tests/process/claude-adapter.test.ts @@ -278,6 +278,50 @@ describe('ClaudeAdapter process contract', () => { new ClaudeAdapter({ binary: 'unused' }).run({ runId: 'run-no-cwd', prompt: 'hi' }), ).toThrow(/cwd is required/); }); + + // issue #9 (P0): claude.exe exits quickly without a terminal event while a + // grandchild keeps the stdout write handle open — the event stream must + // still produce a terminal error within the exit-flush grace, instead of + // hanging until the grandchild dies (which can be hours). + it('ends the stream with an error when claude exits early and a grandchild holds stdout', async () => { + const fake = await createQuickExitWithGrandchildClaude({ holdStdoutMs: 8_000 }); + cleanup.push(fake.dir); + + const run = new ClaudeAdapter({ binary: fake.path }).run({ + runId: 'run-hang', + prompt: 'hang', + cwd: fake.dir, + }); + + const startedAt = Date.now(); + const events = await collect(run.events); + const elapsed = Date.now() - startedAt; + + expect(elapsed).toBeLessThan(7_000); + expect(events.some((e) => e.type === 'error')).toBe(true); + const error = events.find((e) => e.type === 'error') as { message?: string }; + expect(error.message).toMatch(/exited with code|without emitting a result/); + // A grandchild that outlived the run must not keep the test alive. + await run.forceKillTree(); + }, 20_000); + + it('surfaces an error when claude exits 0 without ever emitting a result event', async () => { + const fake = await createFakeClaude({ + lines: [{ type: 'system', subtype: 'init', session_id: 'sess-silent' }], + }); + cleanup.push(fake.dir); + + const run = new ClaudeAdapter({ binary: fake.path }).run({ + runId: 'run-silent-exit', + prompt: 'silent', + cwd: fake.dir, + }); + + const events = await collect(run.events); + const last = events[events.length - 1]; + expect(last?.type).toBe('error'); + expect((last as { message?: string }).message).toMatch(/without emitting a result/); + }); }); async function collect(events: AsyncIterable): Promise { @@ -325,6 +369,53 @@ async function createFakeClaude(options: { return { path, dir, recordPath }; } +// issue #9 (P0): a fake claude that prints one line, exits immediately, and +// leaves a grandchild process holding the stdout write handle for a while — +// reproducing the 2026-11-08 hang without waiting that long in the test. +async function createQuickExitWithGrandchildClaude(options: { + holdStdoutMs: number; +}): Promise { + const dir = await mkdtemp(join(tmpdir(), 'claude-adapter-test-')); + const path = join(dir, 'fake-claude.mjs'); + const holderPath = join(dir, 'stdout-holder.mjs'); + const recordPath = join(dir, 'argv.json'); + await writeFile( + holderPath, + [ + '#!/usr/bin/env node', + `const holdMs = ${options.holdStdoutMs};`, + // Keep the inherited stdout open without writing; exit later. On exit + // the handle closes on its own, which is what the old code waited for. + 'const start = Date.now();', + 'process.stdout.write("");', + 'setTimeout(() => process.exit(0), holdMs).unref();', + 'setInterval(() => {}, 10_000).unref();', + 'process.on("disconnect", () => {});', + ].join('\n'), + 'utf8', + ); + await writeFile( + path, + [ + '#!/usr/bin/env node', + 'import { spawn } from "node:child_process";', + 'import { writeFileSync } from "node:fs";', + `writeFileSync(${JSON.stringify(recordPath)}, JSON.stringify({ argv: process.argv.slice(2) }));`, + // One stdout line (so sawStdout=true, like the real init event) — the + // bug shipped exactly here: the old guard only closed stdout when + // nothing had been printed. + 'console.log(JSON.stringify({ type: "system", subtype: "init", session_id: "sess-hang" }));', + `const holder = spawn(process.execPath, [${JSON.stringify(holderPath)}], { stdio: ["ignore", "inherit", "ignore"], detached: false });`, + 'holder.unref();', + 'setTimeout(() => process.exit(0), 30);', + ].join('\n'), + 'utf8', + ); + await chmod(path, 0o755); + await chmod(holderPath, 0o755); + return { path, dir, recordPath }; +} + async function readRecord(path: string): Promise<{ argv: string[]; stdin: string; diff --git a/tests/process/codex-adapter.test.ts b/tests/process/codex-adapter.test.ts index a3f3b78..8916b3f 100644 --- a/tests/process/codex-adapter.test.ts +++ b/tests/process/codex-adapter.test.ts @@ -426,6 +426,28 @@ describe('CodexAdapter process contract', () => { }), ).toThrow(/cwd is required/); }); + + // issue #9 (P0): codex.exe exits quickly without a terminal event while a + // grandchild keeps the stdout write handle open — the event stream must + // still produce a terminal error within the exit-flush grace. + it('ends the stream with an error when codex exits early and a grandchild holds stdout', async () => { + const fake = await createQuickExitWithGrandchildCodex({ holdStdoutMs: 8_000 }); + cleanup.push(fake.dir); + + const run = new CodexAdapter({ binary: fake.path, profileStateDir: fake.dir }).run({ + runId: 'run-hang', + prompt: 'hang', + cwd: await realpath(fake.dir), + }); + + const startedAt = Date.now(); + const events = await collect(run.events); + const elapsed = Date.now() - startedAt; + + expect(elapsed).toBeLessThan(7_000); + expect(events.some((e) => e.type === 'error')).toBe(true); + await run.forceKillTree(); + }, 20_000); }); async function collect(events: AsyncIterable): Promise { @@ -479,6 +501,43 @@ async function createFakeCodex(options: { return { path, dir, recordPath }; } +async function createQuickExitWithGrandchildCodex(options: { + holdStdoutMs: number; +}): Promise { + const dir = await mkdtemp(join(tmpdir(), 'codex-adapter-test-')); + const path = join(dir, 'fake-codex.mjs'); + const holderPath = join(dir, 'stdout-holder.mjs'); + const recordPath = join(dir, 'argv.json'); + await writeFile( + holderPath, + [ + '#!/usr/bin/env node', + `const holdMs = ${options.holdStdoutMs};`, + 'process.stdout.write("");', + 'setTimeout(() => process.exit(0), holdMs).unref();', + 'setInterval(() => {}, 10_000).unref();', + ].join('\n'), + 'utf8', + ); + await writeFile( + path, + [ + '#!/usr/bin/env node', + 'import { spawn } from "node:child_process";', + 'import { writeFileSync } from "node:fs";', + `writeFileSync(${JSON.stringify(recordPath)}, JSON.stringify({ argv: process.argv.slice(2) }));`, + 'console.log(JSON.stringify({ type: "thread.started", thread_id: "thread-hang" }));', + `const holder = spawn(process.execPath, [${JSON.stringify(holderPath)}], { stdio: ["ignore", "inherit", "ignore"], detached: false });`, + 'holder.unref();', + 'setTimeout(() => process.exit(0), 30);', + ].join('\n'), + 'utf8', + ); + await chmod(path, 0o755); + await chmod(holderPath, 0o755); + return { path, dir, recordPath }; +} + async function readRecord(path: string): Promise<{ argv: string[]; cwd: string; diff --git a/tests/process/kill-tree.test.ts b/tests/process/kill-tree.test.ts new file mode 100644 index 0000000..949530b --- /dev/null +++ b/tests/process/kill-tree.test.ts @@ -0,0 +1,107 @@ +import { mkdtemp, rm, writeFile, chmod } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { spawn } from 'node:child_process'; +import { afterEach, describe, expect, it } from 'vitest'; +import { killProcessTree } from '../../src/platform/kill-tree'; +import { spawnProcess } from '../../src/platform/spawn'; + +// issue #9 (P2): child.kill() cannot reach past a .cmd shim; taskkill /T /F +// must reap the whole tree. On this machine the real claude/codex shims are +// .cmd, so a tree kill that misses grandchildren would leave this test's +// child alive too. + +const cleanupDirs: string[] = []; + +afterEach(async () => { + await Promise.all( + cleanupDirs.splice(0).map((dir) => + rm(dir, { recursive: true, force: true, maxRetries: 5, retryDelay: 25 }), + ), + ); +}); + +function tasklistHasPid(pid: number): Promise { + return new Promise((resolve) => { + const probe = spawnProcess('tasklist', ['/FI', `PID eq ${pid}`, '/FO', 'CSV', '/NH'], { + windowsHide: true, + }); + let out = ''; + probe.stdout?.on('data', (c: Buffer) => (out += c.toString('utf8'))); + probe.on('exit', () => resolve(out.includes(String(pid)))); + probe.on('error', () => resolve(false)); + }); +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +describe('killProcessTree (issue #9 P2)', () => { + it('kills the whole tree: the child spawned by the target dies with it', async () => { + const dir = await mkdtemp(join(tmpdir(), 'kill-tree-test-')); + cleanupDirs.push(dir); + const parentPath = join(dir, 'parent.mjs'); + const childPath = join(dir, 'sleeper.mjs'); + const markerPath = join(dir, 'pid.json'); + await writeFile( + childPath, + [ + 'import { writeFileSync } from "node:fs";', + 'writeFileSync(' + JSON.stringify(markerPath) + ', String(process.pid));', + 'setInterval(() => {}, 60_000);', + ].join('\n'), + 'utf8', + ); + await chmod(childPath, 0o755); + await writeFile( + parentPath, + [ + 'import { spawn } from "node:child_process";', + 'const child = spawn(process.execPath, [' + JSON.stringify(childPath) + '], { stdio: "ignore" });', + 'child.unref();', + 'setInterval(() => {}, 60_000);', + ].join('\n'), + 'utf8', + ); + await chmod(parentPath, 0o755); + + const parent = spawn(process.execPath, [parentPath], { + stdio: 'ignore', + windowsHide: true, + }); + + let childPid: number | undefined; + for (let i = 0; i < 100 && !childPid; i++) { + await sleep(30); + try { + const { readFile } = await import('node:fs/promises'); + const raw = await readFile(markerPath, 'utf8'); + childPid = Number(raw.trim()); + } catch { + // marker not written yet + } + } + expect(childPid).toBeGreaterThan(0); + + await sleep(100); + expect(await tasklistHasPid(childPid!)).toBe(true); + + await killProcessTree(parent.pid, 'kill-tree-test'); + + // Both parent and its child must be gone. Poll briefly: taskkill is fast + // but the test should not flake on scheduling. + let parentGone = false; + let childGone = false; + for (let i = 0; i < 50; i++) { + await sleep(50); + parentGone = !(await tasklistHasPid(parent.pid!)); + childGone = !(await tasklistHasPid(childPid!)); + if (parentGone && childGone) break; + } + expect(parentGone).toBe(true); + expect(childGone).toBe(true); + }, 15_000); + + it('does not throw for an already-dead pid', async () => { + await expect(killProcessTree(-1, 'already-dead')).resolves.toBeUndefined(); + }); +}); diff --git a/tests/unit/runtime/f031-stab-executor.test.ts b/tests/unit/runtime/f031-stab-executor.test.ts index ae806b6..a47248b 100644 --- a/tests/unit/runtime/f031-stab-executor.test.ts +++ b/tests/unit/runtime/f031-stab-executor.test.ts @@ -1,9 +1,19 @@ -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi, beforeEach } from 'vitest'; import type { AgentAdapter, AgentEvent, AgentRun, AgentRunOptions } from '../../../src/agent/types'; import { ActiveRuns } from '../../../src/bot/active-runs'; import { ProcessPool } from '../../../src/bot/process-pool'; import type { RunPolicyAllow } from '../../../src/policy/run-policy'; import { RunExecutor } from '../../../src/runtime/run-executor'; +import { killProcessTree } from '../../../src/platform/kill-tree'; + +// issue #9 (P2): the drain-timeout hard kill must go through killProcessTree. +vi.mock('../../../src/platform/kill-tree', () => ({ + killProcessTree: vi.fn(async () => {}), +})); +const killTreeMock = vi.mocked(killProcessTree); +beforeEach(() => { + killTreeMock.mockClear(); +}); // f031-stab-#3: the old claude must really be gone before the next round on the // same scope starts. Before the fix, cleanup released the scope first and then @@ -13,6 +23,7 @@ import { RunExecutor } from '../../../src/runtime/run-executor'; class ControlledRun implements AgentRun { readonly runId: string; + readonly pid = undefined; readonly events: AsyncIterable; exited = false; stopCalls = 0; @@ -31,6 +42,9 @@ class ControlledRun implements AgentRun { this.stopCalls++; if (this.exitOnStop) this.exit(); } + async forceKillTree(): Promise { + this.exit(); + } waitForExit(timeoutMs: number): Promise { if (this.exited) return Promise.resolve(true); return new Promise((resolve) => { @@ -60,6 +74,45 @@ class ControlledAgent implements AgentAdapter { } } +// issue #9 (P2): a run whose stop() never resolves and whose pid is set — +// the drain-timeout path must hard-kill it by pid. +class HangingStopRun implements AgentRun { + readonly pid = 1234; + readonly events: AsyncIterable; + exited = false; + constructor(runId: string) { + this.runId = runId; + this.events = (async function* () { + yield { type: 'text', delta: 'hi' } as AgentEvent; + yield { type: 'done', terminationReason: 'normal' } as AgentEvent; + })(); + } + readonly runId: string; + async stop(): Promise { + await new Promise(() => {}); + } + async waitForExit(): Promise { + return this.exited ? Promise.resolve(true) : new Promise(() => {}); + } + async forceKillTree(): Promise { + this.exited = true; + } +} + +class HangingStopAgent implements AgentAdapter { + readonly id = 'claude'; + readonly displayName = 'Claude'; + readonly runs: HangingStopRun[] = []; + async isAvailable() { + return true; + } + run(opts: AgentRunOptions): AgentRun { + const r = new HangingStopRun(opts.runId); + this.runs.push(r); + return r; + } +} + function policy(): RunPolicyAllow { return { ok: true, @@ -83,11 +136,18 @@ async function consumeToTerminal(it: AsyncIterable): Promise { const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); -function make(agent: ControlledAgent, graceMs: number) { +function make(agent: ControlledAgent, graceMs: number, drainKillBudgetMs?: number) { const activeRuns = new ActiveRuns(); const pool = new ProcessPool(() => 4); let n = 0; - const executor = new RunExecutor({ agent, pool, activeRuns, postDoneExitGraceMs: graceMs, createRunId: () => `run-${++n}` }); + const executor = new RunExecutor({ + agent, + pool, + activeRuns, + postDoneExitGraceMs: graceMs, + ...(drainKillBudgetMs !== undefined ? { drainKillBudgetMs } : {}), + createRunId: () => `run-${++n}`, + }); return { executor, activeRuns, pool }; } @@ -163,3 +223,92 @@ describe('f031-stab-#3 next round waits for the previous claude to exit', () => await consumeToTerminal(other.subscribe()); }); }); + +// issue #9 (P1): a wedged event stream (child dead, grandchild holds stdout) +// used to freeze the slot forever — the watchdog's stop() was a no-op and +// nothing else could end the run. forceClose is the guaranteed hard exit. +describe('issue #9 forceClose hard exit', () => { + class WedgedRun implements AgentRun { + readonly runId: string; + readonly pid = 4242; + // A stream that never yields and never ends, no matter what. + readonly events: AsyncIterable = new Promise(() => {}) as never; + forceKillCalls = 0; + constructor(runId: string) { + this.runId = runId; + } + async stop(): Promise {} + async waitForExit(): Promise { + return false; + } + async forceKillTree(): Promise { + this.forceKillCalls++; + } + } + + class WedgedAgent implements AgentAdapter { + readonly id = 'claude'; + readonly displayName = 'Claude'; + readonly runs: WedgedRun[] = []; + async isAvailable() { + return true; + } + run(opts: AgentRunOptions): AgentRun { + const r = new WedgedRun(opts.runId); + this.runs.push(r); + return r; + } + } + + it('forceClose injects a terminal error, releases the scope, and kills the tree', async () => { + const agent = new WedgedAgent(); + const { executor, activeRuns } = make(agent as unknown as ControlledAgent, 5_000); + const execution = await executor.submit({ scopeId: 's', policy: policy() }); + + const consumer = (async () => { + const out: AgentEvent[] = []; + for await (const e of execution.subscribe()) { + out.push(e); + if (e.type === 'error' || e.type === 'done') break; + } + return out; + })(); + + await execution.forceClose('watchdog-idle-timeout'); + const events = await consumer; + + const last = events[events.length - 1]; + expect(last?.type).toBe('error'); + expect((last as { message?: string }).message).toContain('force-closed'); + expect((last as { terminationReason?: string }).terminationReason).toBe('timeout'); + // Slot released: the scope can take a new run right away. + expect(activeRuns.get('s')).toBeUndefined(); + const next = await executor.submit({ scopeId: 's', policy: policy() }); + expect(agent.runs).toHaveLength(2); + await next.forceClose('cleanup'); + }); + + it('forceClose is idempotent and safe even if the run already terminated', async () => { + const agent = new WedgedAgent(); + const { executor } = make(agent as unknown as ControlledAgent, 5_000); + const execution = await executor.submit({ scopeId: 's', policy: policy() }); + await execution.forceClose('first'); + await execution.forceClose('second'); + expect(agent.runs[0]!.forceKillCalls).toBeGreaterThanOrEqual(1); + }); + + it('the previous-exit-timeout path calls forceKillTree instead of leaking the process', async () => { + // stop() hangs forever (a wedged run); the scope's next submit hits the + // drain cap and must hard-kill the tree instead of leaking the process. + const agent = new HangingStopAgent(); + const { executor } = make(agent as unknown as ControlledAgent, 20, 100); + const first = await executor.submit({ scopeId: 's', policy: policy() }); + await consumeToTerminal(first.subscribe()); + const t0 = Date.now(); + const second = await executor.submit({ scopeId: 's', policy: policy() }); + expect(Date.now() - t0).toBeLessThan(3_000); + expect(killTreeMock).toHaveBeenCalledWith(agent.runs[0]!.pid, expect.any(String)); + expect(agent.runs).toHaveLength(2); + await consumeToTerminal(second.subscribe()); + }, 10_000); +});