perf(core): cut runtime sync I/O on tool hot path by 91%

packages/cli/src/nonInteractiveCli.test.ts

@@ -273,6 +273,38 @@ describe('runNonInteractive', () => {
     expect(mockShutdownTelemetry).toHaveBeenCalled();
   });
 
+  it('on EPIPE, destroys stdout and returns normally instead of process.exit', async () => {
+    // Regression: process.exit(0) on EPIPE bypassed runExitCleanup → flush()
+    // and dropped queued JSONL writes for `qwen -p ... | head -1` patterns.
+    // process.exit is mocked to throw in beforeEach, so reaching the
+    // assertion also proves the bypass route is gone.
+    setupMetricsMock();
+    const stdoutDestroySpy = vi
+      .spyOn(process.stdout, 'destroy')
+      .mockReturnValue(process.stdout);
+
+    mockGeminiClient.sendMessageStream.mockImplementation(
+      async function* mockStream(): AsyncGenerator<ServerGeminiStreamEvent> {
+        process.stdout.emit(
+          'error',
+          Object.assign(new Error('EPIPE'), { code: 'EPIPE' }),
+        );
+        yield { type: GeminiEventType.Content, value: 'Hello' };
+        yield {
+          type: GeminiEventType.Finished,
+          value: {
+            reason: undefined,
+            usageMetadata: { totalTokenCount: 0 },
+          },
+        };
+      },
+    );
+
+    await runNonInteractive(mockConfig, mockSettings, 'test', 'p1');
+
+    expect(stdoutDestroySpy).toHaveBeenCalled();
+  });
+
   it('should handle a single tool call and respond', async () => {
     setupMetricsMock();
     const toolCallEvent: ServerGeminiStreamEvent = {

packages/cli/src/nonInteractiveCli.ts

@@ -175,16 +175,22 @@ export async function runNonInteractive(
     let totalApiDurationMs = 0;
     const startTime = Date.now();
 
+    const geminiClient = config.getGeminiClient();
+    const abortController = options.abortController ?? new AbortController();
+
+    // EPIPE: don't process.exit here — that bypasses the caller's
+    // runExitCleanup → flush() and drops queued JSONL writes. Destroy
+    // stdout instead and let the natural return drive cleanup. (Aborting
+    // is also wrong: the abort path runs handleCancellationError → exit
+    // 130 and re-introduces the same bypass.)
+    let pipeBroken = false;
     const stdoutErrorHandler = (err: NodeJS.ErrnoException) => {
-      if (err.code === 'EPIPE') {
-        process.stdout.removeListener('error', stdoutErrorHandler);
-        process.exit(0);
+      if (err.code === 'EPIPE' && !pipeBroken) {
+        pipeBroken = true;
+        process.stdout.destroy();
       }
     };
 
-    const geminiClient = config.getGeminiClient();
-    const abortController = options.abortController ?? new AbortController();
-
     // Setup signal handlers for graceful shutdown
     const shutdownHandler = () => {
       debugLogger.debug('[runNonInteractive] Shutdown signal received');

packages/cli/src/utils/cleanup.test.ts

@@ -5,19 +5,19 @@
  */
 
 import { vi } from 'vitest';
-import { registerCleanup, runExitCleanup } from './cleanup';
+import {
+  _resetCleanupFunctionsForTest,
+  registerCleanup,
+  runExitCleanup,
+} from './cleanup';
 
 describe('cleanup', () => {
-  const originalCleanupFunctions = global['cleanupFunctions'];
-
   beforeEach(() => {
-    // Isolate cleanup functions for each test
-    global['cleanupFunctions'] = [];
-  });
-
-  afterAll(() => {
-    // Restore original cleanup functions
-    global['cleanupFunctions'] = originalCleanupFunctions;
+    // The previous `global['cleanupFunctions'] = []` setup was dead code —
+    // the array is module-private, not on `global`. Tests passed by accident
+    // because `runExitCleanup` itself clears at the end. A test that throws
+    // before reaching `runExitCleanup` would leak state into the next case.
+    _resetCleanupFunctionsForTest();
   });
 
   it('should run a registered synchronous function', async () => {
@@ -65,4 +65,79 @@ describe('cleanup', () => {
     expect(errorFn).toHaveBeenCalledTimes(1);
     expect(successFn).toHaveBeenCalledTimes(1);
   });
+
+  describe('timeout failsafes', () => {
+    // Without these the async-jsonl flush() could hang exit forever on slow
+    // disks / dead sockets — sync writes were inherently bounded, async aren't.
+
+    it('caps a hung cleanup at the per-fn timeout and proceeds to the next one', async () => {
+      const hangFn = vi.fn(() => new Promise<void>(() => {}));
+      const nextFn = vi.fn();
+
+      registerCleanup(hangFn);
+      registerCleanup(nextFn);
+
+      const start = Date.now();
+      await runExitCleanup({
+        _testPerFnTimeoutMs: 50,
+        _testOverallTimeoutMs: 5_000,
+      });
+      const elapsed = Date.now() - start;
+
+      expect(hangFn).toHaveBeenCalledTimes(1);
+      expect(nextFn).toHaveBeenCalledTimes(1);
+      expect(elapsed).toBeLessThan(500);
+    });
+
+    it('caps overall wall-clock time when many cleanups all hang', async () => {
+      // 100 × 50ms perFn ≈ 5000ms drain — structurally impossible for "drain
+      // finished naturally" to satisfy < 800ms, so the upper bound proves
+      // wallClock actually fired. Lower bound proves we waited for it and
+      // didn't short-circuit. 800ms slack absorbs CI scheduler jitter.
+      for (let i = 0; i < 100; i++) {
+        registerCleanup(() => new Promise<void>(() => {}));
+      }
+
+      const start = Date.now();
+      await runExitCleanup({
+        _testPerFnTimeoutMs: 50,
+        _testOverallTimeoutMs: 100,
+      });
+      const elapsed = Date.now() - start;
+
+      expect(elapsed).toBeLessThan(800);
+      expect(elapsed).toBeGreaterThanOrEqual(80);
+    });
+
+    it('still calls fast cleanups normally when timeouts are configured', async () => {
+      const fastFn = vi.fn().mockResolvedValue(undefined);
+      registerCleanup(fastFn);
+
+      await runExitCleanup({
+        _testPerFnTimeoutMs: 1_000,
+        _testOverallTimeoutMs: 2_000,
+      });
+
+      expect(fastFn).toHaveBeenCalledTimes(1);
+    });
+
+    it('does not let a rejected cleanup poison the chain', async () => {
+      // The original `for…await` already swallowed sync throws; this guards
+      // the new withTimeout wrapper against rejected-async-cleanup leaks.
+      const rejectFn = vi.fn().mockRejectedValue(new Error('boom'));
+      const nextFn = vi.fn();
+
+      registerCleanup(rejectFn);
+      registerCleanup(nextFn);
+
+      await expect(
+        runExitCleanup({
+          _testPerFnTimeoutMs: 50,
+          _testOverallTimeoutMs: 1_000,
+        }),
+      ).resolves.toBeUndefined();
+      expect(rejectFn).toHaveBeenCalledTimes(1);
+      expect(nextFn).toHaveBeenCalledTimes(1);
+    });
+  });
 });

packages/cli/src/utils/cleanup.ts

@@ -14,15 +14,93 @@ export function registerCleanup(fn: (() => void) | (() => Promise<void>)) {
   cleanupFunctions.push(fn);
 }
 
-export async function runExitCleanup() {
-  for (const fn of cleanupFunctions) {
-    try {
-      await fn();
-    } catch (_) {
-      // Ignore errors during cleanup.
+/**
+ * Per-cleanup ceiling. Caps any single hung cleanup (slow disk on
+ * `chatRecording.flush`, MCP disconnect on a dead socket, telemetry HTTP
+ * stall) so it can't starve the rest of the cleanup chain.
+ */
+const PER_CLEANUP_TIMEOUT_MS = 2_000;
+
+/**
+ * Wall-clock ceiling for the whole cleanup pass. Pre-async-jsonl, sync
+ * fs writes were inherently bounded by their syscall return; with the
+ * write queue moved off-thread, an unbounded `await flush()` could now
+ * hang exit indefinitely. This ceiling guarantees the process always
+ * exits within a bounded time, even if a cleanup never resolves.
+ */
+const OVERALL_CLEANUP_TIMEOUT_MS = 5_000;
+
+/**
+ * Awaits `promise`, but resolves to `undefined` if `ms` elapses first.
+ * Rejection collapses to the same undefined resolution — caller treats
+ * cleanup errors as best-effort. Timer is unrefed so it can't keep the
+ * event loop alive on its own.
+ */
+function withTimeout<T>(promise: Promise<T>, ms: number): Promise<T | void> {
+  return new Promise((resolve) => {
+    const timer = setTimeout(() => resolve(undefined), ms);
+    timer.unref?.();
+    promise.then(
+      (value) => {
+        clearTimeout(timer);
+        resolve(value);
+      },
+      () => {
+        clearTimeout(timer);
+        resolve(undefined);
+      },
+    );
+  });
+}
+
+export interface RunExitCleanupOptions {
+  /** TEST ONLY — override per-cleanup-function timeout (default 2s). */
+  _testPerFnTimeoutMs?: number;
+  /** TEST ONLY — override overall wall-clock timeout (default 5s). */
+  _testOverallTimeoutMs?: number;
+}
+
+export async function runExitCleanup(
+  options: RunExitCleanupOptions = {},
+): Promise<void> {
+  const perFn = options._testPerFnTimeoutMs ?? PER_CLEANUP_TIMEOUT_MS;
+  const overall = options._testOverallTimeoutMs ?? OVERALL_CLEANUP_TIMEOUT_MS;
+
+  const drain = (async () => {
+    for (const fn of cleanupFunctions) {
+      try {
+        await withTimeout(Promise.resolve().then(fn), perFn);
+      } catch (_) {
+        // Ignore errors during cleanup.
+      }
     }
+  })();
+
+  // clearTimeout when drain wins; unref keeps the handle from blocking exit.
+  let wallClockTimer: NodeJS.Timeout | undefined;
+  const wallClock = new Promise<void>((resolve) => {
+    wallClockTimer = setTimeout(() => resolve(), overall);
+    wallClockTimer.unref?.();
+  });
+
+  try {
+    await Promise.race([drain, wallClock]);
+  } finally {
+    if (wallClockTimer) clearTimeout(wallClockTimer);
+    cleanupFunctions.length = 0; // Clear the array
   }
-  cleanupFunctions.length = 0; // Clear the array
+}
+
+/**
+ * Test-only: clear the registered cleanup functions array. Module-private
+ * state otherwise leaks across vitest cases — the previous test isolation
+ * via `global['cleanupFunctions']` was a no-op (the array isn't on global)
+ * and only happened to work because `runExitCleanup` itself clears at the
+ * end. Naming follows the `_reset*ForTest` convention from
+ * d6485964c (paths, jsonl-utils, ripGrep).
+ */
+export function _resetCleanupFunctionsForTest(): void {
+  cleanupFunctions.length = 0;
 }
 
 export async function cleanupCheckpoints() {

packages/core/src/config/config.ts

@@ -1610,9 +1610,11 @@ export class Config {
       return;
     }
     try {
-      // Finalize the current session's metadata before cleanup.
+      // Finalize the current session's metadata before cleanup, then drain
+      // the async write queue so no records are lost on exit.
       try {
         this.chatRecordingService?.finalize();
+        await this.chatRecordingService?.flush();
       } catch {
         // Best-effort — don't block shutdown
       }

packages/core/src/services/chatRecordingService.autoTitle.test.ts

@@ -68,7 +68,7 @@ async function flushMicrotasks(): Promise<void> {
 
 function findCustomTitleRecord(): ChatRecord | undefined {
   return vi
-    .mocked(jsonl.writeLineSync)
+    .mocked(jsonl.writeLine)
     .mock.calls.map((c) => c[1] as ChatRecord)
     .find((r) => r.type === 'system' && r.subtype === 'custom_title');
 }
@@ -133,6 +133,10 @@ describe('ChatRecordingService - auto-title trigger', () => {
     vi.spyOn(fs, 'existsSync').mockReturnValue(false);
 
     chatRecordingService = new ChatRecordingService(mockConfig);
+
+    // writeLine is async; mockResolvedValue lets the writeChain settle when
+    // tests await flushMicrotasks() / chatRecordingService.flush().
+    vi.mocked(jsonl.writeLine).mockResolvedValue(undefined);
   });
 
   afterEach(() => {
@@ -174,7 +178,8 @@ describe('ChatRecordingService - auto-title trigger', () => {
 
   it('does not overwrite a manual title', async () => {
     chatRecordingService.recordCustomTitle('chose-this-myself', 'manual');
-    vi.mocked(jsonl.writeLineSync).mockClear();
+    await chatRecordingService.flush();
+    vi.mocked(jsonl.writeLine).mockClear();
 
     chatRecordingService.recordAssistantTurn({
       model: 'qwen-plus',
@@ -326,10 +331,13 @@ describe('ChatRecordingService - auto-title trigger', () => {
     expect(svc.getCurrentCustomTitle()).toBe('Auto-generated title');
     expect(svc.getCurrentTitleSource()).toBe('auto');
 
-    // finalize() was called by the constructor — the re-appended record
-    // must carry titleSource: 'auto', not 'manual'.
+    // finalize() was called by the constructor — drain the queued async
+    // write before inspecting the mock.
+    await svc.flush();
+
+    // The re-appended record must carry titleSource: 'auto', not 'manual'.
     const finalizeRecord = vi
-      .mocked(jsonl.writeLineSync)
+      .mocked(jsonl.writeLine)
       .mock.calls.map((c) => c[1] as ChatRecord)
       .find((r) => r.type === 'system' && r.subtype === 'custom_title');
     expect(finalizeRecord?.systemPayload).toEqual({
@@ -362,9 +370,10 @@ describe('ChatRecordingService - auto-title trigger', () => {
 
     expect(svc.getCurrentCustomTitle()).toBe('User chose this');
     expect(svc.getCurrentTitleSource()).toBe('manual');
+    await svc.flush();
 
     const finalizeRecord = vi
-      .mocked(jsonl.writeLineSync)
+      .mocked(jsonl.writeLine)
       .mock.calls.map((c) => c[1] as ChatRecord)
       .find((r) => r.type === 'system' && r.subtype === 'custom_title');
     expect(finalizeRecord?.systemPayload).toEqual({
@@ -395,9 +404,10 @@ describe('ChatRecordingService - auto-title trigger', () => {
     // Must stay undefined so the JSONL isn't upgraded to a misleading
     // `titleSource: 'manual'` we can't actually verify.
     expect(svc.getCurrentTitleSource()).toBeUndefined();
+    await svc.flush();
 
     const finalizeRecord = vi
-      .mocked(jsonl.writeLineSync)
+      .mocked(jsonl.writeLine)
       .mock.calls.map((c) => c[1] as ChatRecord)
       .find((r) => r.type === 'system' && r.subtype === 'custom_title');
     // Payload must NOT contain a titleSource field when source is unknown.
@@ -494,7 +504,8 @@ describe('ChatRecordingService - auto-title trigger', () => {
 
     // User renames while the title LLM call is still pending.
     chatRecordingService.recordCustomTitle('user-chosen', 'manual');
-    vi.mocked(jsonl.writeLineSync).mockClear();
+    await chatRecordingService.flush();
+    vi.mocked(jsonl.writeLine).mockClear();
 
     // Now the LLM call returns a title.
     resolveLlm({ ok: true, title: 'Auto Title', modelUsed: 'qwen-turbo' });