fix: flush late transcript writes

src/utils/__tests__/sessionStorageFlush.test.ts

@@ -0,0 +1,62 @@
+import { afterEach, beforeEach, describe, expect, it } from 'bun:test'
+import * as fs from 'node:fs/promises'
+import * as os from 'node:os'
+import * as path from 'node:path'
+
+import {
+  enqueueSessionEntryAfterPendingForTesting,
+  flushSessionStorage,
+  resetProjectForTesting,
+} from '../sessionStorage.js'
+import type { CustomTitleMessage } from '../../types/logs.js'
+
+const originalConfigDir = process.env.CLAUDE_CONFIG_DIR
+
+async function createTmpDir(): Promise<string> {
+  const dir = path.join(
+    os.tmpdir(),
+    `session-storage-flush-${Date.now()}-${Math.random().toString(36).slice(2)}`,
+  )
+  await fs.mkdir(dir, { recursive: true })
+  return dir
+}
+
+describe('sessionStorage flush', () => {
+  let tmpDir: string
+
+  beforeEach(async () => {
+    tmpDir = await createTmpDir()
+    process.env.CLAUDE_CONFIG_DIR = tmpDir
+    resetProjectForTesting()
+  })
+
+  afterEach(async () => {
+    resetProjectForTesting()
+    if (originalConfigDir === undefined) {
+      delete process.env.CLAUDE_CONFIG_DIR
+    } else {
+      process.env.CLAUDE_CONFIG_DIR = originalConfigDir
+    }
+    await fs.rm(tmpDir, { recursive: true, force: true }).catch(() => {})
+  })
+
+  it('drains writes that are queued by pending operations during flush', async () => {
+    const transcriptPath = path.join(tmpDir, 'late-enqueue.jsonl')
+    const entry: CustomTitleMessage = {
+      type: 'custom-title',
+      customTitle: 'late enqueue',
+      sessionId: '11111111-1111-4111-8111-111111111111',
+    }
+    const writePromise = enqueueSessionEntryAfterPendingForTesting(
+      transcriptPath,
+      entry,
+      10,
+    )
+
+    await flushSessionStorage()
+    await writePromise
+
+    const content = await fs.readFile(transcriptPath, 'utf-8')
+    expect(content).toContain('"customTitle":"late enqueue"')
+  })
+})

src/utils/sessionStorage.ts

@@ -487,6 +487,24 @@ export function setSessionFileForTesting(path: string): void {
   getProject().sessionFile = path
 }
 
+/** @internal Test hook for flush races where tracked work enqueues late. */
+export async function enqueueSessionEntryAfterPendingForTesting(
+  path: string,
+  entry: Entry,
+  delayMs = 0,
+): Promise<void> {
+  const projectForTesting = getProject() as unknown as {
+    trackWrite<T>(fn: () => Promise<T>): Promise<T>
+    enqueueWrite(filePath: string, entry: Entry): Promise<void>
+  }
+  await projectForTesting.trackWrite(async () => {
+    if (delayMs > 0) {
+      await new Promise(resolve => setTimeout(resolve, delayMs))
+    }
+    void projectForTesting.enqueueWrite(path, entry)
+  })
+}
+
 type InternalEventWriter = (
   eventType: string,
   payload: Record<string, unknown>,
@@ -839,25 +857,36 @@ class Project {
   }
 
   async flush(): Promise<void> {
-    // Cancel pending timer
+    while (true) {
-    if (this.flushTimer) {
+      // Cancel pending timer so process shutdown does not wait for the next
-      clearTimeout(this.flushTimer)
+      // interval tick before draining already queued transcript writes.
-      this.flushTimer = null
+      if (this.flushTimer) {
-    }
+        clearTimeout(this.flushTimer)
-    // Wait for any in-flight drain to finish
+        this.flushTimer = null
-    if (this.activeDrain) {
+      }
-      await this.activeDrain
-    }
-    // Drain anything remaining in the queues
-    await this.drainWriteQueue()
 
-    // Wait for non-queue tracked operations (e.g. removeMessageByUuid)
+      // Wait for any in-flight drain to finish before taking ownership of the
-    if (this.pendingWriteCount === 0) {
+      // remaining queue.
-      return
+      if (this.activeDrain) {
+        await this.activeDrain
+      }
+
+      await this.drainWriteQueue()
+
+      if (this.pendingWriteCount === 0) {
+        // A tracked writer may have enqueued after the drain above and then
+        // completed. Loop once more so flush() only returns after that late
+        // enqueue has also been written.
+        if (!this.flushTimer && !this.activeDrain && this.writeQueues.size === 0) {
+          return
+        }
+        continue
+      }
+
+      await new Promise<void>(resolve => {
+        this.flushResolvers.push(resolve)
+      })
     }
-    return new Promise<void>(resolve => {
-      this.flushResolvers.push(resolve)
-    })
   }
 
   /**