diff --git a/.changeset/fix-headless-force-exit.md b/.changeset/fix-headless-force-exit.md new file mode 100644 index 0000000000..2a00ca4cbf --- /dev/null +++ b/.changeset/fix-headless-force-exit.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Force-exit headless runs (`kimi -p`) so a stray ref'd handle left over from the run can't keep a completed run alive until an external timeout, and bound prompt cleanup so a wedged shutdown step can't hang shutdown. diff --git a/apps/kimi-code/src/cli/headless-exit.ts b/apps/kimi-code/src/cli/headless-exit.ts new file mode 100644 index 0000000000..1805728443 --- /dev/null +++ b/apps/kimi-code/src/cli/headless-exit.ts @@ -0,0 +1,96 @@ +import type { Writable } from 'node:stream'; + +import { HEADLESS_FORCE_EXIT_GRACE_MS, HEADLESS_STDIO_DRAIN_TIMEOUT_MS } from '#/constant/app'; + +/** Minimal process surface needed to force a headless run to terminate. */ +export interface ExitableProcess { + exit(code?: number): void; +} + +/** + * Schedule a best-effort force-exit for a completed headless (`kimi -p`) run. + * + * Print mode does not call `process.exit()`; it relies on the Node event loop + * draining once the run is done. If a stray ref'd handle survives shutdown — a + * lingering socket (e.g. a connection blackholed by a restrictive firewall, or + * an HTTP/2 session kept alive by PING), an un-cleared timer, or a child whose + * pipes stay open — the loop never empties and the process hangs until an + * external timeout kills it. + * + * This arms an **unref'd** fallback timer: a healthy run drains and exits + * naturally before it fires (so behaviour is unchanged), and the timer itself + * never keeps the loop alive. It only force-exits a run whose loop is already + * wedged. The exit code is read lazily at fire time so callers may set + * `process.exitCode` after scheduling (e.g. a goal turn mapping its terminal + * status to a non-zero code). + * + * Returns the timer handle so callers/tests can `clearTimeout` it. + */ +export function scheduleHeadlessForceExit( + proc: ExitableProcess, + getExitCode: () => number, + graceMs: number = HEADLESS_FORCE_EXIT_GRACE_MS, +): NodeJS.Timeout { + const timer = setTimeout(() => { + proc.exit(getExitCode()); + }, graceMs); + timer.unref?.(); + return timer; +} + +/** Resolve once a stream's currently-buffered writes have flushed to its sink. */ +function flushStream(stream: Writable): Promise { + return new Promise((resolve) => { + try { + // An empty write's callback fires after all previously-queued writes have + // been flushed (writes are ordered), which is the documented way to know a + // stream's buffer has drained. + stream.write('', () => resolve()); + } catch { + resolve(); + } + }); +} + +/** + * Wait for buffered output on the given streams to flush, bounded by `timeoutMs`. + * + * A slow or piped consumer that hasn't read all of stdout/stderr yet leaves the + * pipe as a legitimate ref'd handle keeping the loop alive. Flushing before any + * force-exit prevents truncating output from an otherwise-successful run. The + * wait is bounded so a permanently-stuck consumer can't re-introduce the hang. + */ +export async function drainStdio( + streams: readonly Writable[], + timeoutMs: number = HEADLESS_STDIO_DRAIN_TIMEOUT_MS, +): Promise { + let timer: NodeJS.Timeout | undefined; + const timeout = new Promise((resolve) => { + timer = setTimeout(resolve, timeoutMs); + timer.unref?.(); + }); + try { + await Promise.race([Promise.all(streams.map(flushStream)).then(() => undefined), timeout]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} + +/** + * Finalize a completed headless run: flush stdio, then arm the force-exit + * backstop. + * + * Draining first means in-flight legitimate output is fully written before the + * backstop can fire, and — since drained stdio no longer holds the loop — only a + * genuinely leaked handle can keep it alive afterwards, which is exactly what + * the backstop is for. + */ +export async function finalizeHeadlessRun( + proc: ExitableProcess, + streams: readonly Writable[], + getExitCode: () => number, + options: { drainTimeoutMs?: number; graceMs?: number } = {}, +): Promise { + await drainStdio(streams, options.drainTimeoutMs ?? HEADLESS_STDIO_DRAIN_TIMEOUT_MS); + scheduleHeadlessForceExit(proc, getExitCode, options.graceMs); +} diff --git a/apps/kimi-code/src/cli/run-prompt.ts b/apps/kimi-code/src/cli/run-prompt.ts index 79a301ba15..6f9379259c 100644 --- a/apps/kimi-code/src/cli/run-prompt.ts +++ b/apps/kimi-code/src/cli/run-prompt.ts @@ -19,7 +19,7 @@ import { } from '@moonshot-ai/kimi-code-sdk'; import { resolve } from 'pathe'; -import { CLI_SHUTDOWN_TIMEOUT_MS } from '#/constant/app'; +import { CLI_SHUTDOWN_TIMEOUT_MS, PROMPT_CLEANUP_TIMEOUT_MS } from '#/constant/app'; import type { CLIOptions, PromptOutputFormat } from './options'; import { @@ -32,6 +32,44 @@ import { import { createCliTelemetryBootstrap, initializeCliTelemetry } from './telemetry'; import { createKimiCodeHostIdentity } from './version'; +/** + * Await `promise`, but stop waiting after `timeoutMs`. + * + * The timeout only bounds how long we WAIT — it does not change the outcome: + * - if `promise` settles first, its result is propagated (a rejection throws), + * so a cleanup step that actually fails in time still surfaces; + * - if the timeout wins, we resolve (give up waiting) and swallow the abandoned + * promise's eventual late rejection so it can't surface as an unhandled + * rejection. + * + * Used to bound shutdown so a wedged cleanup step can't keep a completed + * headless run alive, without silently swallowing a cleanup that fails fast. The + * timer is unref'd so it never keeps the loop alive on its own. + */ +async function raceWithTimeout(promise: Promise, timeoutMs: number): Promise { + let timedOut = false; + let timer: ReturnType | undefined; + // Attach the catch eagerly (synchronously) so `promise` is always consumed and + // a late rejection can never become an unhandled rejection. Before the timeout + // wins, the handler rethrows so a real cleanup failure still propagates. + const guarded = promise.catch((error: unknown) => { + if (timedOut) return; + throw error; + }); + const timedOutSignal = new Promise((resolve) => { + timer = setTimeout(() => { + timedOut = true; + resolve(); + }, timeoutMs); + timer.unref?.(); + }); + try { + await Promise.race([guarded, timedOutSignal]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} + interface PromptOutput { readonly columns?: number | undefined; write(chunk: string): boolean; @@ -96,7 +134,7 @@ export async function runPrompt( let removeTerminationCleanup: (() => void) | undefined; let cleanupPromise: Promise | undefined; const cleanupPromptRun = async (): Promise => { - cleanupPromise ??= (async () => { + const pending = (cleanupPromise ??= (async () => { removeTerminationCleanup?.(); setCrashPhase('shutdown'); try { @@ -105,8 +143,13 @@ export async function runPrompt( await shutdownTelemetry({ timeoutMs: CLI_SHUTDOWN_TIMEOUT_MS }); await harness.close(); } - })(); - await cleanupPromise; + })()); + // Bound cleanup so a wedged shutdown step (e.g. a SessionEnd hook, MCP + // shutdown, or a connection blackholed by a restrictive firewall) cannot + // keep a completed headless run alive forever. The cleanup keeps running in + // the background if it overruns; the caller (`kimi -p`) force-exits shortly + // after, so any straggling work is torn down with the process. + await raceWithTimeout(pending, PROMPT_CLEANUP_TIMEOUT_MS); }; removeTerminationCleanup = installPromptTerminationCleanup(promptProcess, cleanupPromptRun); diff --git a/apps/kimi-code/src/constant/app.ts b/apps/kimi-code/src/constant/app.ts index d6cbede4ba..e38ea6a686 100644 --- a/apps/kimi-code/src/constant/app.ts +++ b/apps/kimi-code/src/constant/app.ts @@ -14,6 +14,26 @@ export const WEB_UI_MODE = 'web'; // Give telemetry a short flush window without making CLI exit feel stuck. export const CLI_SHUTDOWN_TIMEOUT_MS = 3000; +// Upper bound on headless (`kimi -p`) shutdown. A wedged cleanup step (e.g. a +// SessionEnd hook, an MCP shutdown, or a connection blackholed by a restrictive +// firewall) must not keep a completed run alive indefinitely — once this elapses +// we stop waiting on cleanup and let the run return. +export const PROMPT_CLEANUP_TIMEOUT_MS = 8000; + +// Grace after a headless run has fully completed (turn done, cleanup attempted) +// before force-exiting. `kimi -p` otherwise relies on the event loop draining to +// exit; a stray ref'd handle (socket/timer/child) left over from the run would +// wedge it. The guard timer is unref'd, so a healthy run still exits naturally +// well before this fires. +export const HEADLESS_FORCE_EXIT_GRACE_MS = 2000; + +// Max time to wait for buffered stdout/stderr to flush before arming the +// force-exit fallback. A slow/piped consumer's still-draining stdio is a +// legitimate ref'd handle — flushing first prevents the fallback from +// truncating completed output. Bounded so a permanently-stuck consumer can't +// re-introduce the hang. +export const HEADLESS_STDIO_DRAIN_TIMEOUT_MS = 10000; + // Published npm package name; this can differ from the executable command. export const NPM_PACKAGE_NAME = '@moonshot-ai/kimi-code'; diff --git a/apps/kimi-code/src/main.ts b/apps/kimi-code/src/main.ts index e944725900..a54bfa2bdc 100644 --- a/apps/kimi-code/src/main.ts +++ b/apps/kimi-code/src/main.ts @@ -23,6 +23,7 @@ import { } from '@moonshot-ai/kimi-telemetry'; import { createProgram } from './cli/commands'; +import { finalizeHeadlessRun } from './cli/headless-exit'; import type { CLIOptions } from './cli/options'; import { OptionConflictError, validateOptions } from './cli/options'; import { runPrompt } from './cli/run-prompt'; @@ -38,7 +39,22 @@ import { cleanupStaleNativeCacheForCurrent } from './native/native-assets'; import { installNativeModuleHook } from './native/module-hook'; import { runNativeAssetSmokeIfRequested } from './native/smoke'; -export async function handleMainCommand(opts: CLIOptions, version: string): Promise { +/** + * Outcome of a CLI command run, reported back to the process entrypoint. + * + * `handleMainCommand` is a reusable, unit-tested handler — it must not terminate + * the process itself. It reports here whether a headless (`kimi -p`) run + * completed so the entrypoint (the only place that owns the process) can arm the + * force-exit fallback. + */ +export interface MainCommandOutcome { + readonly headlessCompleted: boolean; +} + +export async function handleMainCommand( + opts: CLIOptions, + version: string, +): Promise { let validated: ReturnType; try { validated = validateOptions(opts); @@ -60,10 +76,11 @@ export async function handleMainCommand(opts: CLIOptions, version: string): Prom if (validated.uiMode === 'print') { await runPrompt(validated.options, version); - return; + return { headlessCompleted: true }; } await runShell(validated.options, version); + return { headlessCompleted: false }; } /** `kimi migrate`: launch the migration screen only, then exit. */ @@ -139,17 +156,32 @@ export function main(): void { const program = createProgram( version, (opts) => { - void handleMainCommand(opts, version).catch(async (error: unknown) => { - const operation = opts.prompt !== undefined ? 'run prompt' : 'start shell'; - await logStartupFailure(operation, error); - process.stderr.write( - formatStartupError(error, { - operation, - }), - ); - process.stderr.write(`See log: ${resolveGlobalLogPath(resolveKimiHome())}\n`); - process.exit(1); - }); + void handleMainCommand(opts, version) + .then(async (outcome) => { + // Only the process entrypoint disposes of the process. Print mode + // relies on the event loop draining to exit; flush any buffered output + // and then arm an unref'd fallback so a stray ref'd handle left over + // from the run can't wedge a completed `kimi -p` until an external + // timeout. A healthy run drains and exits before the fallback fires. + if (outcome.headlessCompleted) { + await finalizeHeadlessRun( + process, + [process.stdout, process.stderr], + () => Number(process.exitCode) || 0, + ); + } + }) + .catch(async (error: unknown) => { + const operation = opts.prompt !== undefined ? 'run prompt' : 'start shell'; + await logStartupFailure(operation, error); + process.stderr.write( + formatStartupError(error, { + operation, + }), + ); + process.stderr.write(`See log: ${resolveGlobalLogPath(resolveKimiHome())}\n`); + process.exit(1); + }); }, () => { void handleMigrateCommand(version).catch(async (error: unknown) => { diff --git a/apps/kimi-code/test/cli/headless-exit.test.ts b/apps/kimi-code/test/cli/headless-exit.test.ts new file mode 100644 index 0000000000..a977a70d2d --- /dev/null +++ b/apps/kimi-code/test/cli/headless-exit.test.ts @@ -0,0 +1,125 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { Writable } from 'node:stream'; + +import { drainStdio, finalizeHeadlessRun, scheduleHeadlessForceExit } from '#/cli/headless-exit'; + +describe('scheduleHeadlessForceExit', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it('force-exits with the lazily-resolved exit code after the grace period', () => { + vi.useFakeTimers(); + const exit = vi.fn(); + let code = 0; + const handle = scheduleHeadlessForceExit({ exit }, () => code, 2000); + // The exit code can be set after scheduling (e.g. a goal turn maps its + // terminal status to process.exitCode); it must be read at fire time. + code = 7; + + expect(exit).not.toHaveBeenCalled(); + vi.advanceTimersByTime(1999); + expect(exit).not.toHaveBeenCalled(); + vi.advanceTimersByTime(1); + expect(exit).toHaveBeenCalledWith(7); + + clearTimeout(handle); + }); + + it('schedules an unref\'d timer so a healthy run still exits naturally', () => { + // Real timers: an un-unref'd guard would itself keep the event loop alive, + // turning the fix into a regression (every healthy run would wait the full + // grace before exiting). hasRef() must be false. + const exit = vi.fn(); + const handle = scheduleHeadlessForceExit({ exit }, () => 0, 60_000); + expect((handle as { hasRef?: () => boolean }).hasRef?.()).toBe(false); + clearTimeout(handle); + }); + + it('does not fire once cancelled via clearTimeout', () => { + vi.useFakeTimers(); + const exit = vi.fn(); + const handle = scheduleHeadlessForceExit({ exit }, () => 0, 2000); + clearTimeout(handle); + vi.advanceTimersByTime(5000); + expect(exit).not.toHaveBeenCalled(); + }); +}); + +describe('drainStdio', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it('resolves once buffered output has flushed', async () => { + let flush: (() => void) | undefined; + const stream = { + write: vi.fn((_chunk: string, cb: () => void) => { + flush = cb; + return false; + }), + } as unknown as Writable; + + let resolved = false; + const done = drainStdio([stream], 5000).then(() => { + resolved = true; + }); + await Promise.resolve(); + expect(resolved).toBe(false); // still draining + + flush?.(); // consumer caught up + await done; + expect(resolved).toBe(true); + }); + + it('gives up after the timeout when the consumer never drains', async () => { + vi.useFakeTimers(); + // write() never invokes its flush callback — a permanently-stuck consumer. + const stream = { write: vi.fn(() => false) } as unknown as Writable; + + let resolved = false; + const done = drainStdio([stream], 3000).then(() => { + resolved = true; + }); + await vi.advanceTimersByTimeAsync(2999); + expect(resolved).toBe(false); + await vi.advanceTimersByTimeAsync(1); + await done; + expect(resolved).toBe(true); + }); +}); + +describe('finalizeHeadlessRun', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it('flushes stdio before arming the force-exit so buffered output is not truncated', async () => { + vi.useFakeTimers(); + let flush: (() => void) | undefined; + const stream = { + write: vi.fn((_chunk: string, cb: () => void) => { + flush = cb; + return false; + }), + } as unknown as Writable; + const exit = vi.fn(); + + const done = finalizeHeadlessRun({ exit }, [stream], () => 0, { + drainTimeoutMs: 5000, + graceMs: 2000, + }); + + // Output is still draining: even well past the force-exit grace, we must NOT + // have armed/fired the exit — doing so would truncate the buffered output. + await vi.advanceTimersByTimeAsync(4000); + expect(exit).not.toHaveBeenCalled(); + + // Consumer catches up → drain completes → only now is the backstop armed. + flush?.(); + await done; + expect(exit).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(2000); + expect(exit).toHaveBeenCalledWith(0); + }); +}); diff --git a/apps/kimi-code/test/cli/main.test.ts b/apps/kimi-code/test/cli/main.test.ts index 52aba94b16..04d6556cf1 100644 --- a/apps/kimi-code/test/cli/main.test.ts +++ b/apps/kimi-code/test/cli/main.test.ts @@ -32,6 +32,7 @@ const mocks = vi.hoisted(() => { })), initializeCliTelemetry: vi.fn(), handleUpgrade: vi.fn(), + finalizeHeadlessRun: vi.fn(), log: { info: vi.fn(), warn: vi.fn(), @@ -127,6 +128,10 @@ vi.mock('../../src/cli/run-prompt', () => ({ runPrompt: mocks.runPrompt, })); +vi.mock('../../src/cli/headless-exit', () => ({ + finalizeHeadlessRun: mocks.finalizeHeadlessRun, +})); + class ExitCalled extends Error { constructor(readonly code: number) { super(`exit(${code})`); @@ -147,6 +152,20 @@ function defaultOpts(): CLIOptions { }; } +async function waitForAssertion(assertion: () => void): Promise { + let lastError: unknown; + for (let attempt = 0; attempt < 20; attempt += 1) { + try { + assertion(); + return; + } catch (error) { + lastError = error; + await new Promise((resolve) => setTimeout(resolve, 0)); + } + } + throw lastError; +} + async function runHandleMainCommand(opts: CLIOptions): Promise { const exitSpy = vi.spyOn(process, 'exit').mockImplementation((code?: string | number | null) => { throw new ExitCalled(Number(code ?? 0)); @@ -235,6 +254,53 @@ describe('main entry command handling', () => { expect(runShell).not.toHaveBeenCalled(); }); + it('does not force-exit from the reusable handler in print mode', async () => { + const opts: CLIOptions = { ...defaultOpts(), prompt: 'explain the repo' }; + mocks.validateOptions.mockReturnValue({ options: opts, uiMode: 'print' }); + mocks.runUpdatePreflight.mockResolvedValue('continue'); + mocks.runPrompt.mockResolvedValue(void 0); + + const outcome = await handleMainCommand(opts, '0.0.1-alpha.2'); + + // Process disposition belongs to the entrypoint, never to this reusable, + // unit-tested handler: arming a process.exit here would kill the test runner + // or any embedding host. The handler only reports what ran. + expect(mocks.finalizeHeadlessRun).not.toHaveBeenCalled(); + expect(outcome).toEqual({ headlessCompleted: true }); + }); + + it('reports no headless completion for interactive (shell) mode', async () => { + const opts = defaultOpts(); + mocks.validateOptions.mockReturnValue({ options: opts, uiMode: 'shell' }); + mocks.runUpdatePreflight.mockResolvedValue('continue'); + mocks.runShell.mockResolvedValue(void 0); + + const outcome = await handleMainCommand(opts, '0.0.1-alpha.2'); + + expect(outcome).toEqual({ headlessCompleted: false }); + expect(mocks.finalizeHeadlessRun).not.toHaveBeenCalled(); + }); + + it('arms the force-exit fallback at the entrypoint after a completed headless run', async () => { + const opts: CLIOptions = { ...defaultOpts(), prompt: 'explain the repo' }; + mocks.validateOptions.mockReturnValue({ options: opts, uiMode: 'print' }); + mocks.runUpdatePreflight.mockResolvedValue('continue'); + mocks.runPrompt.mockResolvedValue(void 0); + mocks.finalizeHeadlessRun.mockResolvedValue(void 0); + + main(); + const programArgs = mocks.createProgram.mock.calls[0] as unknown as unknown[]; + const mainAction = programArgs[1] as (opts: CLIOptions) => void; + mainAction(opts); + + await waitForAssertion(() => { + expect(mocks.finalizeHeadlessRun).toHaveBeenCalledTimes(1); + }); + // The exit code is resolved lazily so a goal turn that sets process.exitCode wins. + const forceExitArgs = mocks.finalizeHeadlessRun.mock.calls[0] as unknown as unknown[]; + expect(typeof forceExitArgs[2]).toBe('function'); + }); + it('keeps shell mode update preflight interactive by default', async () => { const opts = defaultOpts(); mocks.validateOptions.mockReturnValue({ options: opts, uiMode: 'shell' }); diff --git a/apps/kimi-code/test/cli/run-prompt.test.ts b/apps/kimi-code/test/cli/run-prompt.test.ts index 0d1fc3ea16..41673cc51d 100644 --- a/apps/kimi-code/test/cli/run-prompt.test.ts +++ b/apps/kimi-code/test/cli/run-prompt.test.ts @@ -2,6 +2,7 @@ import type { createKimiDeviceId as createKimiDeviceIdFn } from '@moonshot-ai/ki import { afterEach, describe, expect, it, vi } from 'vitest'; import { runPrompt } from '#/cli/run-prompt'; +import { PROMPT_CLEANUP_TIMEOUT_MS } from '#/constant/app'; type CreateKimiDeviceId = typeof createKimiDeviceIdFn; @@ -218,6 +219,93 @@ describe('runPrompt', () => { expect(mocks.harnessClose).toHaveBeenCalled(); }); + it('completes even if harness.close() never resolves (cleanup is time-bounded)', async () => { + vi.useFakeTimers(); + try { + const stdout = writer(); + const stderr = writer(); + // Simulate a shutdown step that hangs (e.g. a wedged SessionEnd hook or a + // blackholed connection in a firewalled sandbox). A completed headless run + // must not stay alive forever waiting on cleanup. + mocks.harnessClose.mockReturnValueOnce(new Promise(() => {})); + + let settled = false; + const done = runPrompt(opts(), '1.2.3-test', { + stdout, + stderr, + process: fakeProcess(), + }).then(() => { + settled = true; + }); + + await vi.advanceTimersByTimeAsync(PROMPT_CLEANUP_TIMEOUT_MS + 100); + await done; + + expect(settled).toBe(true); + expect(mocks.harnessClose).toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it('propagates a cleanup failure that settles before the timeout', async () => { + const stdout = writer(); + const stderr = writer(); + // A cleanup step that fails fast (e.g. a permission restore or harness close + // hitting a persistence error) must surface — not be silently swallowed by + // the timeout guard — otherwise the run reports success while shutdown + // actually failed (e.g. a resumed session left in `auto`). + mocks.harnessClose.mockRejectedValueOnce(new Error('close failed')); + + await expect( + runPrompt(opts(), '1.2.3-test', { stdout, stderr, process: fakeProcess() }), + ).rejects.toThrow('close failed'); + }); + + it('ignores a cleanup rejection that lands after the timeout', async () => { + vi.useFakeTimers(); + try { + const stdout = writer(); + const stderr = writer(); + // Cleanup overruns the bound and only rejects later. The run already gave + // up waiting and resolved; that late rejection must not flip it to a + // failure (nor surface as an unhandled rejection). + mocks.harnessClose.mockReturnValueOnce( + new Promise((_, reject) => { + const timer = setTimeout( + () => reject(new Error('late close')), + PROMPT_CLEANUP_TIMEOUT_MS + 5000, + ); + timer.unref?.(); + }), + ); + + let settled: 'resolved' | 'rejected' | undefined; + const done = runPrompt(opts(), '1.2.3-test', { + stdout, + stderr, + process: fakeProcess(), + }).then( + () => { + settled = 'resolved'; + }, + () => { + settled = 'rejected'; + }, + ); + + await vi.advanceTimersByTimeAsync(PROMPT_CLEANUP_TIMEOUT_MS + 100); + await done; + expect(settled).toBe('resolved'); + + await vi.advanceTimersByTimeAsync(5000); + await Promise.resolve(); + expect(settled).toBe('resolved'); + } finally { + vi.useRealTimers(); + } + }); + it('stops prompt startup when session creation fails', async () => { const stdout = writer(); const stderr = writer();