diff --git a/cli/src/hooks/__tests__/use-send-message.test.tsx b/cli/src/hooks/__tests__/use-send-message.test.tsx index b37f99c614..447801c4f9 100644 --- a/cli/src/hooks/__tests__/use-send-message.test.tsx +++ b/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[] = [] diff --git a/cli/src/hooks/use-send-message.ts b/cli/src/hooks/use-send-message.ts index 589f658bc5..4f17ac7ca7 100644 --- a/cli/src/hooks/use-send-message.ts +++ b/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) diff --git a/cli/src/utils/__tests__/run-state-storage.test.ts b/cli/src/utils/__tests__/run-state-storage.test.ts index 5cc92546b9..a94f2ca964 100644 --- a/cli/src/utils/__tests__/run-state-storage.test.ts +++ b/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) => { + 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 + + 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 diff --git a/cli/src/utils/run-state-storage.ts b/cli/src/utils/run-state-storage.ts index 940b44dcc0..fb64167bba 100644 --- a/cli/src/utils/run-state-storage.ts +++ b/cli/src/utils/run-state-storage.ts @@ -13,7 +13,11 @@ import { } from './chat-meta' import { logger } from './logger' import { classifyStringifyError, serializeForPersistence } from './safe-json' -import { writeFileAtomic, writeFileAtomicAsync } from './write-file-atomic' +import { + commitAtomicWrites, + prepareAtomicWrite, + writeFileAtomic, +} from './write-file-atomic' import type { ChatMessage } from '../types/chat' import type { RunState } from '@codebuff/sdk' @@ -399,14 +403,135 @@ export function saveChatState( ) // Sidecar summary so /history can list this chat without parsing the // (unbounded) chat-messages.json. Must be written after the messages - // file: it records the file's size/mtime to detect staleness. + // file: it records the file's size/mtime to detect staleness. Written + // only by the authoritative saves (turn completion / interrupt / exit + // flush); the periodic checkpoint below deliberately leaves the sidecar + // stale, which readChatMeta detects and resolves with a full parse. writeChatMeta(chatDir, messages) } + noteChatStatePersisted(chatDir, runState, messages) } catch (error) { logSaveWriteFailure(error, chatDir, 'Failed to save chat state') } } +// --- Skip checkpoints that would rewrite bytes already on disk -------------- + +/** What makeCheckpointKey captures about one state. */ +type CheckpointKey = { + messages: unknown + messagesLength: number + lastMessage: unknown + runState: unknown + /** Identity marker for scalars that went into runStateScalarDigest. */ + runStateScalarDigest: string +} + +/** + * The state the SDK hands a checkpoint is a shallow snapshot whose message + * objects are shared with the live run (snapshotSessionState): anything that + * mutates them — streaming text appended to the tail message, the CLI's + * setMessages replacing the array — always writes through a new container + * (`blocks: [...]`, tail copy, or a fresh array). So container identity is a + * sound "has anything changed" signal, and it is O(1) where a defensive + * re-serialize of a multi-MB transcript on every drain is not. + * + * The run state gets no such discipline from its producers, so its + * mutation-relevant scalars are digested cheaply instead — values the next + * turn's continuation depends on, compared by content. The rest of the run + * state changes only with the history and its containers carry identity. + */ +function makeCheckpointKey( + runState: RunState, + messages: ChatMessage[], +): CheckpointKey { + const mainAgentState = (runState as { sessionState?: any } | null) + ?.sessionState?.mainAgentState + let runStateScalarDigest = '' + if (mainAgentState && typeof mainAgentState === 'object') { + try { + runStateScalarDigest = JSON.stringify([ + mainAgentState.contextTokenCount ?? null, + mainAgentState.creditsUsed ?? null, + mainAgentState.stepsRemaining ?? null, + mainAgentState.agentType ?? null, + mainAgentState.compactionSummary?.text ?? null, + mainAgentState.pendingCompactionTelemetry ?? null, + ]) + } catch { + // Never let a fingerprint break a save: unstringifyable scalars simply + // stop comparing equal, which errs on the side of writing. + runStateScalarDigest = randomUUID() + } + } + return { + messages, + messagesLength: Array.isArray(messages) ? messages.length : -1, + lastMessage: Array.isArray(messages) ? messages[messages.length - 1] : null, + runState, + runStateScalarDigest, + } +} + +function checkpointKeysEqual(a: CheckpointKey, b: CheckpointKey): boolean { + return ( + a.messages === b.messages && + a.messagesLength === b.messagesLength && + a.lastMessage === b.lastMessage && + a.runState === b.runState && + a.runStateScalarDigest === b.runStateScalarDigest + ) +} + +// What was last durably written, per chat dir. Checkpoint drains compare +// against this before serializing: an unchanged checkpoint — the SDK's +// snapshot timer fired while the run was waiting on a tool or a slow LLM +// call, or the CLI rescheduled the same settled state — would regenerate two +// multi-MB JSON documents and pay their fsyncs without changing a single byte +// on disk. +const persistedCheckpointKeys = new Map() + +function noteChatStatePersisted( + chatDir: string, + runState: RunState, + messages: ChatMessage[], +): void { + persistedCheckpointKeys.set(chatDir, makeCheckpointKey(runState, messages)) +} + +/** + * True when this checkpoint would persist exactly what the last successful + * save already wrote. Only drains consult it — the synchronous, authoritative + * saves (turn completion, interrupt, exit flush) always write: they are the + * final state the user lost if a crash lands before the next checkpoint. + */ +function isCheckpointUnchanged( + chatDir: string, + runState: RunState, + messages: ChatMessage[], +): boolean { + const key = persistedCheckpointKeys.get(chatDir) + return !!key && checkpointKeysEqual(key, makeCheckpointKey(runState, messages)) +} + +/** + * Refresh only the /history sidecar for a chat directory, after the messages + * file on disk already holds exactly `messages` (e.g. that state landed via + * a skipped or regular checkpoint). Authoritative completion paths use this + * when the components of the state were last persisted separately, so chat + * completion always leaves a fresh sidecar behind. + */ +export function refreshChatMetaSync( + chatDir: string, + messages: ChatMessage[], +): void { + try { + writeChatMeta(chatDir, messages) + } catch (error) { + logSaveWriteFailure(error, chatDir, 'Failed to refresh chat meta') + } +} + /** * Async counterpart to saveChatState. Serializes and writes off the caller's * tick so a multi-MB transcript doesn't block the CLI's render/input thread. @@ -416,29 +541,49 @@ async function saveChatStateAsync( messages: ChatMessage[], chatDir: string, ): Promise { + if (isCheckpointUnchanged(chatDir, runState, messages)) { + // Nothing to do at all: no serialize, no writes. An unchanged checkpoint + // tick — the SDK fires at the interval regardless of progress — used to + // cost two full stringifies of the transcript plus three atomic writes + // (run-state, messages, meta) and their fsyncs, all producing the exact + // bytes already on disk. + return + } const serialized = serializeChatState(runState, messages, chatDir) if (!serialized.runStateJson && !serialized.messagesJson) { return } try { await fs.promises.mkdir(chatDir, { recursive: true }) + // Both files went through tmp-file + fsync + rename + dir fsync each, and + // the two dir fsyncs flush the same directory. Prepare both (contents + // durable under their temp names), publish both renames, then pay that + // directory fsync once for the pair. + const pending = [] if (serialized.runStateJson) { - await writeFileAtomicAsync( - path.join(chatDir, RUN_STATE_FILENAME), - serialized.runStateJson, + pending.push( + await prepareAtomicWrite( + path.join(chatDir, RUN_STATE_FILENAME), + serialized.runStateJson, + ), ) } if (serialized.messagesJson) { - await writeFileAtomicAsync( - path.join(chatDir, CHAT_MESSAGES_FILENAME), - serialized.messagesJson, + pending.push( + await prepareAtomicWrite( + path.join(chatDir, CHAT_MESSAGES_FILENAME), + serialized.messagesJson, + ), ) - // Sidecar summary so /history can list this chat without parsing the - // (unbounded) chat-messages.json. Written after the messages file: it - // records that file's size/mtime to detect staleness. The meta write is - // tiny, so keeping it synchronous here is fine. - writeChatMeta(chatDir, messages) } + await commitAtomicWrites(pending) + // The chat-meta sidecar moved out of this path on purpose: it exists to + // summarize a settled chat for /history, is validated against the messages + // file's size/mtime with a full-parse fallback (readChatMeta), and cost a + // third atomic write plus a statSync on every drain. It is refreshed by + // the authoritative save at turn end instead; a mid-turn sidecar that + // reads stale is exactly the case readChatMeta was written to detect. + noteChatStatePersisted(chatDir, runState, messages) } catch (error) { logSaveWriteFailure(error, chatDir, 'Failed to save chat state (async)') } diff --git a/cli/src/utils/write-file-atomic.ts b/cli/src/utils/write-file-atomic.ts index 913f4ccd7f..6d493c8b4a 100644 --- a/cli/src/utils/write-file-atomic.ts +++ b/cli/src/utils/write-file-atomic.ts @@ -169,3 +169,78 @@ export async function writeFileAtomicAsync( throw error } } + +/** + * The async pair of saves in a checkpoint drain both target files and both + * flush the same directory after the rename (one fsync per write). The sibling + * helpers below let a caller rename both files first and pay that directory + * fsync once: flush each tmp file, publish both names, then flush the + * directory that now holds both. + * + * Splitting the write at the rename changes nothing about ordering — the tmp + * file is still flushed to the platter before its name exists — it only stops + * flushing a directory whose entry is about to change again a microtask later. + */ +export type AtomicRename = { + rename: () => Promise + tmpPath: string + targetPath: string +} + +export async function prepareAtomicWrite( + filePath: string, + data: string, +): Promise { + const tmpPath = tempPathFor(filePath) + try { + const handle = await activeFs.promises.open( + tmpPath, + 'wx', + PRIVATE_FILE_MODE, + ) + try { + await handle.writeFile(data) + await handle.sync() + } finally { + await handle.close() + } + return { + tmpPath, + targetPath: filePath, + rename: () => activeFs.promises.rename(tmpPath, filePath), + } + } catch (error) { + try { + await activeFs.promises.unlink(tmpPath) + } catch { + // Ignore cleanup errors; the original error is what matters + } + throw error + } +} + +/** Commit prepared writes: publish every name, then flush the shared + * directory once. `prepareAtomicWrite` already dropped prepared tmp files it + * could not flush, so a failure here (a rename error) only needs the same + * unlink sweep for whatever has not been renamed yet. */ +export async function commitAtomicWrites( + writes: AtomicRename[], +): Promise { + try { + for (const write of writes) { + await write.rename() + } + } catch (error) { + for (const write of writes) { + try { + await activeFs.promises.unlink(write.tmpPath) + } catch { + // Either renamed already or already gone. + } + } + throw error + } + if (writes.length > 0) { + await syncDirectoryAsync(path.dirname(writes[0].targetPath)) + } +}