cli: skip checkpoint writes when the state is unchanged, and one fsync group per drain

cli/src/hooks/__tests__/use-send-message.test.tsx

@@ -12,6 +12,7 @@ import { useAgentValidation } from '../use-agent-validation'
 import { setProjectRoot, tryGetProjectRoot } from '../../project-files'
 import { useChatStore } from '../../state/chat-store'
 import { stopActiveRun } from '../../utils/active-run'
+import { readChatMeta } from '../../utils/chat-meta'
 import { setChatDirOverrideForTesting } from '../../utils/run-state-storage'
 
 import type { RunState } from '@codebuff/sdk'
@@ -271,6 +272,59 @@ describe('useSendMessage continuation state', () => {
     }
   })
 
+  test('refreshes the chat sidecar at turn end, not on checkpoint drains', async () => {
+    const { setup, root } = await mountHost()
+    const run = sendMessageFromHost!({
+      content: 'summarize this',
+      agentMode: 'DEFAULT',
+    })
+
+    try {
+      await waitFor('SDK run', () => runCalls.length === 1)
+      // The turn-start save writes the sidecar once, so /history lists the
+      // in-flight chat: a valid (non-stale) sidecar describes the transcript.
+      await waitFor('turn-start save', () => readChatMeta(testRoot) !== null)
+
+      // Mid-turn: a checkpoint drain persists new messages but no longer
+      // refreshes the sidecar — it now reads stale against the rewritten
+      // messages file, and readChatMeta legitimately falls back to the full
+      // parse. That is the designed staleness window.
+      useChatStore.getState().setMessages([
+        ...useChatStore.getState().messages,
+        {
+          id: 'msg-steer',
+          variant: 'user',
+          content: 'one more thing',
+          timestamp: new Date().toISOString(),
+        } as any,
+      ])
+      // A new snapshot object, like a later step boundary's: the checkpoint
+      // skips only a byte-identical re-drain of the turn-start save, which
+      // this is not.
+      runCalls[0].runConfig.onStateSnapshot(makeRunState('checkpoint'))
+      await waitFor('checkpoint drain', () =>
+        fs
+          .readFileSync(path.join(testRoot, 'chat-messages.json'), 'utf8')
+          .includes('one more thing'),
+      )
+      expect(readChatMeta(testRoot)).toBeNull()
+
+      runCalls[0].resolve(makeRunState('final'))
+      await run
+
+      // Turn end must leave a fresh sidecar behind.
+      const meta = readChatMeta(testRoot)
+      expect(meta).not.toBeNull()
+      expect(meta!.firstPrompt).toBe('summarize this')
+      expect(meta!.messageCount).toBeGreaterThan(0)
+    } finally {
+      settlePendingRuns()
+      await run
+      flushSync(() => root.unmount())
+      setup.renderer.destroy()
+    }
+  })
+
   test('follow-ups inherit rejected and aborted run snapshots without stale replacement', async () => {
     const { setup, root } = await mountHost()
     const runs: Promise<void>[] = []

cli/src/hooks/use-send-message.ts

@@ -40,6 +40,7 @@ import {
   checkpointIntervalMsForChat,
   clearLiveChatStateProvider,
   loadMostRecentChatState,
+  refreshChatMetaSync,
   resolveCurrentChatDir,
   saveChatState,
   scheduleCheckpointSave,
@@ -1065,6 +1066,16 @@ export const useSendMessage = ({
             useChatStore.getState().messages,
             runChatDir,
           )
+          // The periodic checkpoint no longer rewrites the sidecar, so
+          // refresh it here where a turn ends: /history then reads the
+          // settled chat's summary from chat-meta.json without the full-parse
+          // fallback. (The save above covers every case that wrote messages;
+          // a sponsored turn whose checkpoint already persisted this exact
+          // transcript writes nothing and needs the refresh.)
+          refreshChatMetaSync(
+            runChatDir,
+            useChatStore.getState().messages,
+          )
         }
       } catch (error) {
         runFunnel.finish(false)

cli/src/utils/__tests__/run-state-storage.test.ts

@@ -13,6 +13,8 @@ import {
   scheduleCheckpointSave,
   settleCheckpointSave,
 } from '../run-state-storage'
+import { setAtomicWriteFsForTesting } from '../write-file-atomic'
+import type { AtomicWriteFs } from '../write-file-atomic'
 import type { ChatMessage } from '../../types/chat'
 import type { RunState } from '@codebuff/sdk'
 
@@ -385,6 +387,257 @@ describe('scheduleCheckpointSave (async, coalescing)', () => {
   })
 })
 
+describe('checkpoint drains skip what the disk already holds', () => {
+  const chatDir = path.join(TEST_ROOT, 'codebuff-test-skip-chatdir')
+
+  // A recording delegate (same seam as write-file-atomic's own durability
+  // tests) so "the drain wrote nothing" is asserted at the syscall layer,
+  // not inferred from file contents — an unchanged checkpoint leaves the old
+  // bytes readable either way.
+  const recordingFs = () => {
+    const calls: string[] = []
+    const recording: AtomicWriteFs = {
+      openSync: ((...args: any[]) => {
+        calls.push('openSync')
+        return (fs.openSync as any)(...args)
+      }) as typeof fs.openSync,
+      writeFileSync: ((...args: any[]) => {
+        calls.push('writeFileSync')
+        return (fs.writeFileSync as any)(...args)
+      }) as typeof fs.writeFileSync,
+      fsyncSync: (fd: number) => {
+        calls.push('fsyncSync')
+        return fs.fsyncSync(fd)
+      },
+      closeSync: (fd: number) => {
+        calls.push('closeSync')
+        return fs.closeSync(fd)
+      },
+      renameSync: ((...args: any[]) => {
+        calls.push('renameSync')
+        return (fs.renameSync as any)(...args)
+      }) as typeof fs.renameSync,
+      unlinkSync: ((...args: any[]) => {
+        calls.push('unlinkSync')
+        return (fs.unlinkSync as any)(...args)
+      }) as typeof fs.unlinkSync,
+      promises: {
+        open: async (...args: Parameters<typeof fs.promises.open>) => {
+          calls.push('open')
+          const handle = await fs.promises.open(...args)
+          return {
+            writeFile: (data: any) => {
+              calls.push('writeFile')
+              return handle.writeFile(data)
+            },
+            sync: () => {
+              calls.push('sync')
+              return handle.sync()
+            },
+            close: () => {
+              calls.push('close')
+              return handle.close()
+            },
+          } as any
+        },
+        rename: (from: fs.PathLike, to: fs.PathLike) => {
+          calls.push('rename')
+          return fs.promises.rename(from, to)
+        },
+        unlink: (target: fs.PathLike) => {
+          calls.push('unlink')
+          return fs.promises.unlink(target)
+        },
+      },
+    }
+    return { fs: recording, calls }
+  }
+
+  const runState = () =>
+    ({
+      traceSessionId: 'trace-skip',
+      sessionState: {
+        mainAgentState: {
+          messageHistory: [],
+          contextTokenCount: 0,
+          creditsUsed: 0,
+          stepsRemaining: 300,
+          agentType: 'base',
+        },
+      },
+      output: { type: 'error', message: 'x' },
+    }) as unknown as RunState
+
+  const messages = (content: string): ChatMessage[] => [
+    {
+      id: 'msg-1',
+      variant: 'user',
+      content,
+      timestamp: new Date().toISOString(),
+    },
+  ]
+
+  let recorder: ReturnType<typeof recordingFs>
+
+  beforeEach(() => {
+    fs.rmSync(chatDir, { recursive: true, force: true })
+    recorder = recordingFs()
+    setAtomicWriteFsForTesting(recorder.fs)
+    setChatDirOverrideForTesting(chatDir)
+  })
+
+  afterEach(async () => {
+    await settleCheckpointSave()
+    setAtomicWriteFsForTesting(undefined)
+    setChatDirOverrideForTesting(undefined)
+    fs.rmSync(chatDir, { recursive: true, force: true })
+  })
+
+  const flushCalls = () => recorder.calls.splice(0, recorder.calls.length)
+
+  test('a second drain with the same state performs no writes at all', async () => {
+    const state = runState()
+    const transcript = messages('unchanged')
+
+    scheduleCheckpointSave(state, transcript)
+    await settleCheckpointSave()
+    flushCalls()
+
+    // The SDK's snapshot timer reschedules at the interval even when the run
+    // made no progress; persisting the settled state again would rewrite
+    // byte-identical files.
+    scheduleCheckpointSave(state, transcript)
+    await settleCheckpointSave()
+
+    expect(recorder.calls).toEqual([])
+  })
+
+  test('a drain after an authoritative save with unchanged state is free', async () => {
+    const state = runState()
+    const transcript = messages('finished')
+
+    // Use snapshots the periodic path actually sees: the run-side containers
+    // are shared between the checkpoint state and the completion state.
+    saveChatState(state, transcript)
+    flushCalls()
+
+    scheduleCheckpointSave(state, transcript)
+    await settleCheckpointSave()
+
+    expect(recorder.calls).toEqual([])
+    expect(
+      fs.existsSync(path.join(chatDir, 'chat-messages.json')),
+    ).toBe(true)
+  })
+
+  test('a change anywhere triggers exactly one write pass', async () => {
+    const state = runState()
+
+    scheduleCheckpointSave(state, messages('one'))
+    await settleCheckpointSave()
+    flushCalls()
+
+    // New array + new tail message: what a finished turn's adoption looks
+    // like. One pass = tmp+fsync for both files, both renames, one dir fsync.
+    scheduleCheckpointSave(state, messages('two'))
+    await settleCheckpointSave()
+
+    expect(recorder.calls).toEqual([
+      // run-state.json: prepare, then chat-messages.json: prepare
+      'open', 'writeFile', 'sync', 'close',
+      'open', 'writeFile', 'sync', 'close',
+      // both renames, then the single directory flush
+      'rename', 'rename',
+      'open', 'sync', 'close',
+    ])
+
+    flushCalls()
+    // Streaming grows the tail in place but always through a new array in the
+    // store: same tail object identity inside a new container still writes.
+    const transcript = messages('three')
+    const next = [...transcript]
+    scheduleCheckpointSave(state, next)
+    await settleCheckpointSave()
+    expect(recorder.calls).not.toEqual([])
+  })
+
+  test('appending one character to the tail message writes again', async () => {
+    const state = runState()
+    const transcript = messages('the transcript so far')
+
+    scheduleCheckpointSave(state, transcript)
+    await settleCheckpointSave()
+    flushCalls()
+
+    // The store rebuilds the streaming tail on each batch: new tail object,
+    // same everything else. Even a single appended character must not hash
+    // equal to the persisted state.
+    const grown = [{ ...transcript[0], content: 'the transcript so far.' }]
+    scheduleCheckpointSave(state, grown)
+    await settleCheckpointSave()
+
+    expect(recorder.calls).toContain('rename')
+    const saved = JSON.parse(
+      fs.readFileSync(path.join(chatDir, 'chat-messages.json'), 'utf8'),
+    ) as ChatMessage[]
+    expect(saved[0].content).toBe('the transcript so far.')
+  })
+
+  test('a run-state scalar change (credits used) writes again', async () => {
+    const transcript = messages('same transcript')
+    scheduleCheckpointSave(runState(), transcript)
+    await settleCheckpointSave()
+    flushCalls()
+
+    const moreSpent = runState()
+    moreSpent.sessionState!.mainAgentState.creditsUsed = 42
+    scheduleCheckpointSave(moreSpent, transcript)
+    await settleCheckpointSave()
+
+    expect(recorder.calls).toContain('rename')
+  })
+
+  test('the meta sidecar is not rewritten by checkpoints', async () => {
+    const state = runState()
+    const transcript = messages('checkpointed chat')
+
+    scheduleCheckpointSave(state, transcript)
+    await settleCheckpointSave()
+
+    // The sidecar belongs to turn end now: mid-checkpoint it simply doesn't
+    // exist yet, and /history falls back to parsing the transcript.
+    expect(
+      fs.existsSync(path.join(chatDir, 'chat-meta.json')),
+    ).toBe(false)
+
+    // An authoritative save writes it (turn completion goes through
+    // saveChatState), and a later unchanged checkpoint leaves it alone.
+    saveChatState(state, transcript)
+    expect(fs.existsSync(path.join(chatDir, 'chat-meta.json'))).toBe(true)
+    flushCalls()
+
+    scheduleCheckpointSave(state, transcript)
+    await settleCheckpointSave()
+    expect(recorder.calls).toEqual([])
+  })
+
+  test('checkpoint drains share one directory fsync across both files', async () => {
+    const transcript = messages('probe')
+    scheduleCheckpointSave(runState(), transcript)
+    await settleCheckpointSave()
+
+    // Two files published (two renames), but the directory they live in is
+    // flushed once: 2 file opens/fsyncs + 1 dir open/fsync (opened with 'r',
+    // no writeFile in between).
+    expect(recorder.calls.filter((c) => c === 'rename')).toHaveLength(2)
+    expect(recorder.calls.filter((c) => c === 'writeFile')).toHaveLength(2)
+    // Every fsync pairs with an open: 2 file fsyncs + 1 directory fsync.
+    expect(recorder.calls.filter((c) => c === 'sync')).toHaveLength(3)
+    // Last group is the single directory flush: open, sync, close.
+    expect(recorder.calls.slice(-3)).toEqual(['open', 'sync', 'close'])
+  })
+})
+
 describe('chat switches while saves are pending', () => {
   // Regression tests for /history threads being clobbered: save paths used to
   // be resolved at write time from the mutable current chat id, so a pending