From d834904e5a3fc295484e78bcf564ca01adf3b1b8 Mon Sep 17 00:00:00 2001 From: Bogdan Date: Fri, 7 Aug 2026 04:57:32 +0300 Subject: [PATCH] fix(session): make transcript replacements crash-safe (#2094) * fix(session): make transcript replacements crash-safe Complete transcript rewrites could truncate live JSONL files before preserved data was durable, risking unrecoverable resume history after an interrupted write. Commit replacements through exclusive sibling temp files and serialize them with all transcript append paths so readers observe either the old file or the complete replacement. * fix(session): preserve concurrent transcript updates Abort tombstone commits when the scanned transcript changes before replacement, and keep existing local history when remote foreground hydration returns no entries. Harden the associated portability, option coverage, queue timing, and diagnostics. * test(session): match hydration reader signature Pass the explicit optional subagent reader in the empty-hydration regression so a fresh TypeScript build sees the complete helper signature. * fix(session): coordinate transcript writers across processes Hold a same-directory cooperative lock across transcript replacement and final-line truncation, and make session plus SDK append paths participate. Exercise the post-validation/pre-rename race deterministically so external appends land after the complete commit. * test(session): provide empty hydration subagent reader * fix(session): scope transcript lock ownership Separate async and synchronous lock ownership so unrelated sync appends cannot bypass an in-flight replacement. Route aliased in-process appends through the queue, propagate lock compromise through AbortSignal, and cover both symlink-alias and rename-boundary races. --- src/entrypoints/sdk/sessions.ts | 10 +- src/services/PromptSuggestion/speculation.ts | 11 +- src/utils/atomicReplace.test.ts | 273 ++++++ src/utils/atomicReplace.ts | 294 ++++++ .../sessionStorage.atomicReplace.test.ts | 807 ++++++++++++++++ src/utils/sessionStorage.ts | 895 ++++++++++++++---- src/utils/transcriptFileLock.ts | 200 ++++ 7 files changed, 2318 insertions(+), 172 deletions(-) create mode 100644 src/utils/atomicReplace.test.ts create mode 100644 src/utils/atomicReplace.ts create mode 100644 src/utils/sessionStorage.atomicReplace.test.ts create mode 100644 src/utils/transcriptFileLock.ts diff --git a/src/entrypoints/sdk/sessions.ts b/src/entrypoints/sdk/sessions.ts index 36d11d64f..1913fc323 100644 --- a/src/entrypoints/sdk/sessions.ts +++ b/src/entrypoints/sdk/sessions.ts @@ -18,6 +18,7 @@ import { resolveSessionFilePath, } from '../../utils/sessionStoragePortable.js' import { readJSONLFile } from '../../utils/json.js' +import { withTranscriptFileLock } from '../../utils/transcriptFileLock.js' import { assertValidSessionId, type JsonlEntry, @@ -222,12 +223,11 @@ async function appendJsonlEntry( entry: Record, ): Promise { const line = JSON.stringify(entry) + '\n' - try { + await mkdir(dirname(filePath), { mode: 0o700, recursive: true }) + await withTranscriptFileLock(filePath, async signal => { + signal.throwIfAborted() await appendFile(filePath, line, { mode: 0o600 }) - } catch { - await mkdir(dirname(filePath), { mode: 0o700, recursive: true }) - await appendFile(filePath, line, { mode: 0o600 }) - } + }) } // ============================================================================ diff --git a/src/services/PromptSuggestion/speculation.ts b/src/services/PromptSuggestion/speculation.ts index 86e208b0a..e761aadc1 100644 --- a/src/services/PromptSuggestion/speculation.ts +++ b/src/services/PromptSuggestion/speculation.ts @@ -1,6 +1,6 @@ import { randomUUID } from 'crypto' import { rm } from 'fs' -import { appendFile, copyFile, mkdir } from 'fs/promises' +import { copyFile, mkdir } from 'fs/promises' import { dirname, isAbsolute, join, relative } from 'path' import { getCwdState } from '../../bootstrap/state.js' import type { CompletionBoundary } from '../../state/AppStateStore.js' @@ -45,8 +45,7 @@ import { } from '../../utils/messages.js' import { getClaudeTempDir } from '../../utils/permissions/filesystem.js' import { extractReadFilesFromMessages } from '../../utils/queryHelpers.js' -import { getTranscriptPath } from '../../utils/sessionStorage.js' -import { jsonStringify } from '../../utils/slowOperations.js' +import { recordSpeculationAccept } from '../../utils/sessionStorage.js' import { type AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, logEvent, @@ -797,11 +796,9 @@ export async function acceptSpeculation( timestamp: new Date().toISOString(), timeSavedMs, } - void appendFile(getTranscriptPath(), jsonStringify(entry) + '\n', { - mode: 0o600, - }).catch(() => { + void recordSpeculationAccept(entry).catch(() => { logForDebugging( - '[Speculation] Failed to write speculation-accept to transcript', + '[Speculation] Failed to queue speculation-accept for transcript', ) }) } diff --git a/src/utils/atomicReplace.test.ts b/src/utils/atomicReplace.test.ts new file mode 100644 index 000000000..61455effa --- /dev/null +++ b/src/utils/atomicReplace.test.ts @@ -0,0 +1,273 @@ +import { afterEach, beforeEach, expect, test } from 'bun:test' +import { + chmod, + lstat, + mkdtemp, + readFile, + readlink, + readdir, + rm, + stat, + symlink, + writeFile, +} from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { basename, join } from 'node:path' + +import { + type AtomicReplaceFaultStage, + replaceFileAtomic, + resetAtomicReplaceFaultInjectorForTesting, + setAtomicReplaceFaultInjectorForTesting, + setAtomicReplaceWriteLimitForTesting, +} from './atomicReplace.js' +import { + acquireSharedMutationLock, + releaseSharedMutationLock, +} from '../test/sharedMutationLock.js' + +const tempDirs: string[] = [] + +async function tempTarget(initial?: string): Promise<{ + dir: string + target: string +}> { + const dir = await mkdtemp(join(tmpdir(), 'openclaude-atomic-replace-')) + tempDirs.push(dir) + const target = join(dir, 'transcript.jsonl') + if (initial !== undefined) await writeFile(target, initial) + return { dir, target } +} + +async function tempFiles(dir: string, target: string): Promise { + const prefix = `.${basename(target)}.tmp-` + return (await readdir(dir)).filter(name => name.startsWith(prefix)) +} + +beforeEach(async () => { + await acquireSharedMutationLock('utils/atomicReplace.test.ts') +}) + +afterEach(async () => { + try { + resetAtomicReplaceFaultInjectorForTesting() + setAtomicReplaceWriteLimitForTesting(undefined) + await Promise.all( + tempDirs.splice(0).map(dir => rm(dir, { recursive: true, force: true })), + ) + } finally { + releaseSharedMutationLock() + } +}) + +test('replaces from strings, bytes, and streamed chunks', async () => { + const { target } = await tempTarget('old') + + await replaceFileAtomic(target, 'string') + expect(await readFile(target, 'utf8')).toBe('string') + + await replaceFileAtomic(target, new TextEncoder().encode('bytes')) + expect(await readFile(target, 'utf8')).toBe('bytes') + + async function* chunks() { + yield 'stream-' + yield new TextEncoder().encode('complete') + } + await replaceFileAtomic(target, chunks()) + expect(await readFile(target, 'utf8')).toBe('stream-complete') +}) + +test('retries deterministic short low-level writes until the chunk is complete', async () => { + const { target } = await tempTarget('old-complete') + setAtomicReplaceWriteLimitForTesting(3) + + await replaceFileAtomic(target, 'new-complete-transcript') + + expect(await readFile(target, 'utf8')).toBe('new-complete-transcript') +}) + +test('a zero-progress low-level write preserves the original and cleans the temp', async () => { + const { dir, target } = await tempTarget('old-complete') + setAtomicReplaceWriteLimitForTesting(0) + + await expect(replaceFileAtomic(target, 'new')).rejects.toThrow( + 'Atomic replacement made no progress while writing', + ) + expect(await readFile(target, 'utf8')).toBe('old-complete') + expect(await tempFiles(dir, target)).toEqual([]) +}) + +test('preserves an existing restrictive mode and creates new files as 0600', async () => { + if (process.platform === 'win32') return + + const existing = await tempTarget('old') + await chmod(existing.target, 0o640) + await replaceFileAtomic(existing.target, 'new') + expect((await stat(existing.target)).mode & 0o777).toBe(0o640) + + const created = await tempTarget() + await replaceFileAtomic(created.target, 'new') + expect((await stat(created.target)).mode & 0o777).toBe(0o600) +}) + +test('an explicit mode overrides preservation for an existing target', async () => { + if (process.platform === 'win32') return + + const existing = await tempTarget('old') + await chmod(existing.target, 0o640) + + await replaceFileAtomic(existing.target, 'new', { mode: 0o600 }) + + expect((await stat(existing.target)).mode & 0o777).toBe(0o600) +}) + +test('preserveMode false applies the private default to an existing target', async () => { + if (process.platform === 'win32') return + + const existing = await tempTarget('old') + await chmod(existing.target, 0o644) + + await replaceFileAtomic(existing.target, 'new', { preserveMode: false }) + + expect((await stat(existing.target)).mode & 0o777).toBe(0o600) +}) + +test('full flush commits complete content', async () => { + const { target } = await tempTarget('old') + + await replaceFileAtomic(target, 'new-complete', { flush: 'full' }) + + expect(await readFile(target, 'utf8')).toBe('new-complete') +}) + +test('writes through live and dangling relative symlinks without replacing them', async () => { + if (process.platform === 'win32') return + + const live = await tempTarget() + const liveTarget = join(live.dir, 'live-target.jsonl') + await writeFile(liveTarget, 'old-live') + await symlink(basename(liveTarget), live.target) + await replaceFileAtomic(live.target, 'new-live') + expect((await lstat(live.target)).isSymbolicLink()).toBe(true) + expect(await readlink(live.target)).toBe(basename(liveTarget)) + expect(await readFile(liveTarget, 'utf8')).toBe('new-live') + + const dangling = await tempTarget() + const danglingTarget = join(dangling.dir, 'created-through-link.jsonl') + await symlink(basename(danglingTarget), dangling.target) + await replaceFileAtomic(dangling.target, 'new-dangling') + expect((await lstat(dangling.target)).isSymbolicLink()).toBe(true) + expect(await readFile(danglingTarget, 'utf8')).toBe('new-dangling') +}) + +test('an already-aborted replacement preserves the original', async () => { + const { target } = await tempTarget('old') + const controller = new AbortController() + controller.abort() + + await expect( + replaceFileAtomic(target, 'new', { signal: controller.signal }), + ).rejects.toBeDefined() + expect(await readFile(target, 'utf8')).toBe('old') +}) + +test('an abort at the rename boundary preserves the original', async () => { + const { dir, target } = await tempTarget('old') + const controller = new AbortController() + setAtomicReplaceFaultInjectorForTesting(stage => { + if (stage === 'rename') controller.abort(new Error('lock compromised')) + }) + + await expect( + replaceFileAtomic(target, 'new', { signal: controller.signal }), + ).rejects.toThrow('lock compromised') + expect(await readFile(target, 'utf8')).toBe('old') + expect(await tempFiles(dir, target)).toEqual([]) +}) + +const preRenameFaults: AtomicReplaceFaultStage[] = [ + 'temp-open', + 'stream-write', + 'data-flush', + 'chmod', + 'close', + 'rename', +] + +for (const faultStage of preRenameFaults) { + test(`${faultStage} failure preserves the original and cleans the temp`, async () => { + const { dir, target } = await tempTarget('old-complete') + setAtomicReplaceFaultInjectorForTesting(stage => { + if (stage === faultStage) throw new Error(`fault:${stage}`) + }) + + async function* replacement() { + yield 'partial-' + yield 'replacement' + } + + await expect(replaceFileAtomic(target, replacement())).rejects.toThrow( + `fault:${faultStage}`, + ) + expect(await readFile(target, 'utf8')).toBe('old-complete') + expect(await tempFiles(dir, target)).toEqual([]) + }) +} + +test('cleanup failure does not mask the primary failure or modify the target', async () => { + const { dir, target } = await tempTarget('old') + setAtomicReplaceFaultInjectorForTesting(stage => { + if (stage === 'rename') throw new Error('primary rename fault') + if (stage === 'cleanup') throw new Error('cleanup fault') + }) + + await expect(replaceFileAtomic(target, 'new')).rejects.toThrow( + 'primary rename fault', + ) + expect(await readFile(target, 'utf8')).toBe('old') + expect((await tempFiles(dir, target)).length).toBe(1) +}) + +test('directory sync failure is post-commit and leaves the complete new file', async () => { + const { target } = await tempTarget('old') + setAtomicReplaceFaultInjectorForTesting(stage => { + if (stage === 'directory-sync') throw new Error('directory sync fault') + }) + + await replaceFileAtomic(target, 'new-complete') + expect(await readFile(target, 'utf8')).toBe('new-complete') +}) + +test('concurrent readers observe only complete old or complete new bytes', async () => { + const oldContent = 'old-complete-transcript' + const newContent = 'new-complete-transcript' + const { target } = await tempTarget(oldContent) + + let release!: () => void + const gate = new Promise(resolve => { + release = resolve + }) + let firstWrite!: () => void + const wroteFirstChunk = new Promise(resolve => { + firstWrite = resolve + }) + let writes = 0 + setAtomicReplaceFaultInjectorForTesting(stage => { + if (stage === 'stream-write' && writes++ === 0) firstWrite() + }) + + async function* slowReplacement() { + yield 'new-complete-' + await gate + yield 'transcript' + } + + const replacing = replaceFileAtomic(target, slowReplacement()) + await wroteFirstChunk + for (let i = 0; i < 25; i++) { + expect(await readFile(target, 'utf8')).toBe(oldContent) + } + release() + await replacing + expect(await readFile(target, 'utf8')).toBe(newContent) +}) diff --git a/src/utils/atomicReplace.ts b/src/utils/atomicReplace.ts new file mode 100644 index 000000000..1057a6a4d --- /dev/null +++ b/src/utils/atomicReplace.ts @@ -0,0 +1,294 @@ +import { randomBytes } from 'node:crypto' +import type { FileHandle } from 'node:fs/promises' +import { + lstat, + open, + readlink, + realpath, + rename, + stat, + unlink, +} from 'node:fs/promises' +import { + basename, + dirname, + isAbsolute, + join, + resolve, +} from 'node:path' +import { getErrnoCode } from './errors.js' + +export type AtomicReplaceOptions = { + mode?: number + preserveMode?: boolean + signal?: AbortSignal + flush?: 'data' | 'full' + /** Refuse to commit if a caller's existing-file snapshot is stale. */ + expectedTargetSize?: number +} + +export type AtomicReplaceFaultStage = + | 'temp-open' + | 'stream-write' + | 'data-flush' + | 'chmod' + | 'close' + | 'rename' + | 'directory-sync' + | 'cleanup' + +export type AtomicReplaceFaultContext = { + requestedPath: string + targetPath: string + tempPath?: string +} + +type AtomicReplaceFaultInjector = ( + stage: AtomicReplaceFaultStage, + context: AtomicReplaceFaultContext, +) => void | Promise + +let faultInjector: AtomicReplaceFaultInjector | undefined +let writeLimitForTesting: number | undefined + +/** @internal Test-only deterministic failure injection. */ +export function setAtomicReplaceFaultInjectorForTesting( + injector: AtomicReplaceFaultInjector, +): void { + faultInjector = injector +} + +/** @internal Reset test-only deterministic failure injection. */ +export function resetAtomicReplaceFaultInjectorForTesting(): void { + faultInjector = undefined +} + +/** @internal Limit each low-level write for deterministic short-write tests. */ +export function setAtomicReplaceWriteLimitForTesting( + limit: number | undefined, +): void { + writeLimitForTesting = limit +} + +async function injectFault( + stage: AtomicReplaceFaultStage, + context: AtomicReplaceFaultContext, +): Promise { + await faultInjector?.(stage, context) +} + +async function resolveWriteTarget(requestedPath: string): Promise { + let currentPath = resolve(requestedPath) + const visited = new Set() + + for (let depth = 0; depth < 40; depth++) { + if (visited.has(currentPath)) { + throw new Error( + `Cannot atomically replace circular symlink: ${requestedPath}`, + ) + } + visited.add(currentPath) + + let fileStat + try { + fileStat = await lstat(currentPath) + } catch (error) { + if (getErrnoCode(error) === 'ENOENT') return currentPath + throw error + } + + if (!fileStat.isSymbolicLink()) return currentPath + + try { + return await realpath(currentPath) + } catch { + const linkTarget = await readlink(currentPath) + currentPath = isAbsolute(linkTarget) + ? linkTarget + : resolve(dirname(currentPath), linkTarget) + } + } + + throw new Error(`Cannot atomically replace symlink chain: ${requestedPath}`) +} + +function isAsyncIterable( + data: unknown, +): data is AsyncIterable { + return ( + typeof data === 'object' && + data !== null && + Symbol.asyncIterator in data && + typeof data[Symbol.asyncIterator] === 'function' + ) +} + +async function writeChunkFully( + handle: FileHandle, + chunk: string | Uint8Array, + signal: AbortSignal | undefined, + context: AtomicReplaceFaultContext, +): Promise { + const buffer = + typeof chunk === 'string' ? Buffer.from(chunk, 'utf8') : Buffer.from(chunk) + let offset = 0 + + while (offset < buffer.length) { + signal?.throwIfAborted() + const remaining = buffer.length - offset + const writeLength = + writeLimitForTesting === undefined + ? remaining + : Math.min(remaining, writeLimitForTesting) + const { bytesWritten } = await handle.write( + buffer, + offset, + writeLength, + null, + ) + if (bytesWritten === 0) { + throw new Error('Atomic replacement made no progress while writing') + } + offset += bytesWritten + await injectFault('stream-write', context) + } +} + +async function writeReplacement( + handle: FileHandle, + data: string | Uint8Array | AsyncIterable, + signal: AbortSignal | undefined, + context: AtomicReplaceFaultContext, +): Promise { + if (isAsyncIterable(data)) { + for await (const chunk of data) { + await writeChunkFully(handle, chunk, signal, context) + } + return + } + + await writeChunkFully(handle, data, signal, context) +} + +async function syncDirectoryBestEffort( + targetPath: string, + context: AtomicReplaceFaultContext, +): Promise { + if (process.platform === 'win32') return + + let directoryHandle: FileHandle | undefined + try { + directoryHandle = await open(dirname(targetPath), 'r') + await injectFault('directory-sync', context) + await directoryHandle.sync() + } catch { + // The rename is already committed. Some filesystems do not support + // syncing directory handles, so directory durability is best-effort. + } finally { + try { + await directoryHandle?.close() + } catch { + // Best-effort directory sync must not turn a committed write into failure. + } + } +} + +/** + * Replace a regular file through an exclusive sibling temp and atomic rename. + * No failure before rename modifies or unlinks the target. + */ +export async function replaceFileAtomic( + requestedPath: string, + data: string | Uint8Array | AsyncIterable, + options: AtomicReplaceOptions = {}, +): Promise { + options.signal?.throwIfAborted() + + const targetPath = await resolveWriteTarget(requestedPath) + let replacementMode = options.mode ?? 0o600 + + try { + const targetStat = await stat(targetPath) + if (!targetStat.isFile()) { + throw new Error( + `Atomic replacement target is not a regular file: ${requestedPath}`, + ) + } + if (options.mode === undefined && options.preserveMode !== false) { + replacementMode = targetStat.mode & 0o7777 + } + } catch (error) { + if (getErrnoCode(error) !== 'ENOENT') throw error + } + + const tempPath = join( + dirname(targetPath), + `.${basename(targetPath)}.tmp-${randomBytes(16).toString('hex')}`, + ) + const context: AtomicReplaceFaultContext = { + requestedPath, + targetPath, + tempPath, + } + + let handle: FileHandle | undefined + let tempCreated = false + let committed = false + + try { + options.signal?.throwIfAborted() + await injectFault('temp-open', context) + handle = await open(tempPath, 'wx', 0o600) + tempCreated = true + + await writeReplacement(handle, data, options.signal, context) + options.signal?.throwIfAborted() + + await injectFault('data-flush', context) + if (options.flush === 'full') { + await handle.sync() + } else { + await handle.datasync() + } + + await injectFault('chmod', context) + await handle.chmod(replacementMode) + + await handle.close() + handle = undefined + await injectFault('close', context) + + options.signal?.throwIfAborted() + if ( + options.expectedTargetSize !== undefined && + (await stat(targetPath)).size !== options.expectedTargetSize + ) { + throw new Error('Atomic replacement target changed before commit') + } + await injectFault('rename', context) + options.signal?.throwIfAborted() + await rename(tempPath, targetPath) + committed = true + + await syncDirectoryBestEffort(targetPath, context) + } catch (error) { + if (handle) { + try { + await handle.close() + } catch { + // Preserve the operation error; cleanup below still gets a chance. + } + } + + if (tempCreated && !committed) { + try { + await injectFault('cleanup', context) + await unlink(tempPath) + } catch { + // Preserve the primary operation error. + } + } + + throw error + } +} diff --git a/src/utils/sessionStorage.atomicReplace.test.ts b/src/utils/sessionStorage.atomicReplace.test.ts new file mode 100644 index 000000000..751a6f4ad --- /dev/null +++ b/src/utils/sessionStorage.atomicReplace.test.ts @@ -0,0 +1,807 @@ +import { afterEach, beforeEach, expect, mock, spyOn, test } from 'bun:test' +import type { UUID } from 'node:crypto' +import { + chmod, + lstat, + mkdir, + mkdtemp, + open, + readFile, + rm, + stat, + symlink, + writeFile, +} from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { dirname, join } from 'node:path' +import { + getOriginalCwd, + getSessionId, + isSessionPersistenceDisabled, + setOriginalCwd, + setSessionPersistenceDisabled, + switchSession, +} from '../bootstrap/state.js' +import { renameSession } from '../entrypoints/sdk/sessions.js' +import * as sessionIngress from '../services/api/sessionIngress.js' +import { + acquireSharedMutationLock, + releaseSharedMutationLock, +} from '../test/sharedMutationLock.js' +import type { Message } from '../types/message.js' +import { + resetAtomicReplaceFaultInjectorForTesting, + setAtomicReplaceFaultInjectorForTesting, +} from './atomicReplace.js' +import { + getClaudeConfigHomeDirOverrideForTesting, + setClaudeConfigHomeDirForTesting, +} from './envUtils.js' +import { isTranscriptFileLockHeldForTesting } from './transcriptFileLock.js' +import { + buildConversationChain, + flushSessionStorage, + getAgentTranscriptPath, + getProjectDir, + getTranscriptPathForSession, + hydrateFromCCRv2InternalEvents, + hydrateRemoteSession, + loadTranscriptFile, + recordGoalState, + recordSpeculationAccept, + recordTranscript, + removeTranscriptMessage, + resetTranscriptRewriteHooksForTesting, + resetProjectForTesting, + saveCustomTitle, + setInternalEventReader, + setSessionFileForTesting, + setTranscriptRewriteHooksForTesting, +} from './sessionStorage.js' + +const SESSION_ID = '10000000-0000-4000-8000-000000000001' +const OTHER_SESSION_ID = '10000000-0000-4000-8000-000000000002' +const TARGET = '20000000-0000-4000-8000-000000000001' as UUID +const KEEP_1 = '20000000-0000-4000-8000-000000000002' as UUID +const KEEP_2 = '20000000-0000-4000-8000-000000000003' as UUID +const TIMESTAMP = '2026-08-05T00:00:00.000Z' + +let testRoot = '' +let originalCwd = '' +let originalSessionId = '' +let originalConfigOverride: string | undefined +let originalPersistenceDisabled = false +let originalNodeEnv: string | undefined +let originalTestPersistence: string | undefined +let originalPersistence: string | undefined +let originalDiagnosticsFile: string | undefined + +function line(uuid: UUID, extra: Record = {}): string { + return JSON.stringify({ type: 'test', uuid, ...extra }) +} + +function message(uuid: UUID, content: string): Message { + return { + type: 'user', + uuid, + timestamp: TIMESTAMP, + message: { role: 'user', content }, + isMeta: false, + } +} + +async function useTranscript( + content: string | Uint8Array, + name = 'session.jsonl', +): Promise { + const filePath = join(testRoot, name) + await writeFile(filePath, content) + switchSession(SESSION_ID as never, testRoot) + resetProjectForTesting() + setSessionFileForTesting(filePath) + return filePath +} + +async function prepareHydration(): Promise { + const configDir = join(testRoot, 'config') + const workspaceDir = join(testRoot, 'workspace') + await mkdir(workspaceDir, { recursive: true }) + setClaudeConfigHomeDirForTesting(configDir) + setOriginalCwd(workspaceDir) + switchSession(SESSION_ID as never) + resetProjectForTesting() + const transcriptPath = getTranscriptPathForSession(SESSION_ID) + await mkdir(dirname(transcriptPath), { recursive: true, mode: 0o700 }) + return transcriptPath +} + +beforeEach(async () => { + await acquireSharedMutationLock('utils/sessionStorage.atomicReplace.test.ts') + testRoot = await mkdtemp(join(tmpdir(), 'openclaude-atomic-session-')) + originalCwd = getOriginalCwd() + originalSessionId = getSessionId() + originalConfigOverride = getClaudeConfigHomeDirOverrideForTesting() + originalPersistenceDisabled = isSessionPersistenceDisabled() + originalNodeEnv = process.env.NODE_ENV + originalTestPersistence = process.env.TEST_ENABLE_SESSION_PERSISTENCE + originalPersistence = process.env.ENABLE_SESSION_PERSISTENCE + originalDiagnosticsFile = process.env.CLAUDE_CODE_DIAGNOSTICS_FILE + process.env.NODE_ENV = 'development' + process.env.TEST_ENABLE_SESSION_PERSISTENCE = 'true' + process.env.ENABLE_SESSION_PERSISTENCE = 'true' + setSessionPersistenceDisabled(false) +}) + +afterEach(async () => { + try { + resetAtomicReplaceFaultInjectorForTesting() + resetTranscriptRewriteHooksForTesting() + mock.restore() + resetProjectForTesting() + switchSession(originalSessionId as never) + setOriginalCwd(originalCwd) + setClaudeConfigHomeDirForTesting(originalConfigOverride) + setSessionPersistenceDisabled(originalPersistenceDisabled) + if (originalNodeEnv === undefined) delete process.env.NODE_ENV + else process.env.NODE_ENV = originalNodeEnv + if (originalTestPersistence === undefined) { + delete process.env.TEST_ENABLE_SESSION_PERSISTENCE + } else { + process.env.TEST_ENABLE_SESSION_PERSISTENCE = originalTestPersistence + } + if (originalPersistence === undefined) { + delete process.env.ENABLE_SESSION_PERSISTENCE + } else { + process.env.ENABLE_SESSION_PERSISTENCE = originalPersistence + } + if (originalDiagnosticsFile === undefined) { + delete process.env.CLAUDE_CODE_DIAGNOSTICS_FILE + } else { + process.env.CLAUDE_CODE_DIAGNOSTICS_FILE = originalDiagnosticsFile + } + await rm(testRoot, { recursive: true, force: true }) + } finally { + releaseSharedMutationLock() + } +}) + +test('final-line tombstones preserve LF, CRLF, and no-final-newline conventions', async () => { + for (const [separator, finalNewline] of [ + ['\n', true], + ['\n', false], + ['\r\n', true], + ['\r\n', false], + ] as const) { + const prefix = line(KEEP_1) + const original = `${prefix}${separator}${line(TARGET)}${finalNewline ? separator : ''}` + const filePath = await useTranscript(original, `final-${separator.length}-${finalNewline}.jsonl`) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + + expect(await readFile(filePath, 'utf8')).toBe( + finalNewline ? `${prefix}${separator}` : prefix, + ) + } +}) + +test('middle-line tombstone preserves surrounding bytes and malformed lines', async () => { + const malformed = '{not valid json but must survive}\r\n' + const prefix = `${line(KEEP_1, { spacing: 'kept' })}\r\n${malformed}` + const suffix = `${line(KEEP_2, { nested: { uuid: TARGET } })}\r\n` + const filePath = await useTranscript(`${prefix}${line(TARGET)}\r\n${suffix}`) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + + expect(await readFile(filePath, 'utf8')).toBe(prefix + suffix) +}) + +test('tail search ignores a nested uuid in a later entry', async () => { + const original = `${line(TARGET)}\n${line(KEEP_1, { nested: { uuid: TARGET } })}\n` + const filePath = await useTranscript(original) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + + expect(await readFile(filePath, 'utf8')).toBe( + `${line(KEEP_1, { nested: { uuid: TARGET } })}\n`, + ) +}) + +test('slow tombstone removes a target outside the tail window', async () => { + const suffix = `${line(KEEP_1, { payload: 'x'.repeat(70 * 1024) })}\n${line(KEEP_2)}\n` + const filePath = await useTranscript(`${line(TARGET)}\n${suffix}`) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + + expect(await readFile(filePath, 'utf8')).toBe(suffix) +}) + +test('slow tombstone handles a target line longer than the tail window', async () => { + const targetLine = line(TARGET, { payload: 'x'.repeat(70 * 1024) }) + const suffix = `${line(KEEP_1)}\n` + const filePath = await useTranscript(`${targetLine}\n${suffix}`) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + + expect(await readFile(filePath, 'utf8')).toBe(suffix) +}) + +test('large slow rewrite streams bounded chunks instead of building a second copy', async () => { + const suffix = `${line(KEEP_1, { payload: 'x'.repeat(8 * 1024 * 1024) })}\n` + const filePath = await useTranscript(`${line(TARGET)}\n${suffix}`) + let writeCount = 0 + setAtomicReplaceFaultInjectorForTesting((stage, context) => { + if (stage === 'stream-write' && context.requestedPath === filePath) { + writeCount++ + } + }) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + + expect((await stat(filePath)).size).toBe(Buffer.byteLength(suffix)) + // Prove bounded streaming without coupling the test to the current chunk size. + expect(writeCount).toBeGreaterThan(10) + expect(await readFile(filePath, 'utf8')).toBe(suffix) +}) + +test('empty file and missing target remain byte-for-byte unchanged', async () => { + const emptyPath = await useTranscript('', 'empty.jsonl') + await removeTranscriptMessage(TARGET) + expect(await readFile(emptyPath)).toEqual(Buffer.alloc(0)) + + const original = `${line(KEEP_1)}\n${line(KEEP_2)}\n` + const missingPath = await useTranscript(original, 'missing.jsonl') + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + expect(await readFile(missingPath, 'utf8')).toBe(original) +}) + +test('maximum rewrite guard leaves an oversized transcript unchanged', async () => { + const filePath = await useTranscript(`${line(TARGET)}\n`, 'guard.jsonl') + const handle = await open(filePath, 'r+') + await handle.truncate(50 * 1024 * 1024 + 1) + await handle.close() + const before = await stat(filePath) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + + expect((await stat(filePath)).size).toBe(before.size) + const head = Buffer.alloc(Buffer.byteLength(line(TARGET))) + const readHandle = await open(filePath, 'r') + await readHandle.read(head, 0, head.length, 0) + await readHandle.close() + expect(head.toString()).toBe(line(TARGET)) +}) + +test.each(['stream-write', 'data-flush', 'rename'] as const)( + '%s failure preserves the old transcript and cleans its temp file', + async stage => { + const original = `${line(KEEP_1)}\n${line(TARGET)}\n${line(KEEP_2)}\n` + const filePath = await useTranscript(original) + setAtomicReplaceFaultInjectorForTesting((actualStage, context) => { + if (actualStage === stage && context.requestedPath === filePath) { + throw new Error(`injected ${stage}`) + } + }) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + + expect(await readFile(filePath, 'utf8')).toBe(original) + expect( + (await Array.fromAsync(new Bun.Glob('.*.tmp-*').scan(testRoot))).length, + ).toBe(0) + }, +) + +test('an external append after validation waits for the tombstone commit', async () => { + const original = `${line(KEEP_1)}\n${line(TARGET)}\n${line(KEEP_2)}\n` + const appended = `${JSON.stringify({ + type: 'custom-title', + customTitle: 'external SDK append', + sessionId: SESSION_ID, + })}\n` + const filePath = await prepareHydration() + await writeFile(filePath, original) + setSessionFileForTesting(filePath) + let injected = false + let appendPromise: Promise | undefined + setAtomicReplaceFaultInjectorForTesting(async (stage, context) => { + if (!injected && stage === 'rename' && context.requestedPath === filePath) { + injected = true + expect(await isTranscriptFileLockHeldForTesting(filePath)).toBe(true) + appendPromise = renameSession(SESSION_ID, 'external SDK append', { + dir: getOriginalCwd(), + }) + } + }) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + expect(appendPromise).toBeDefined() + await appendPromise + + expect(await readFile(filePath, 'utf8')).toBe( + `${line(KEEP_1)}\n${line(KEEP_2)}\n${appended}`, + ) + expect( + (await Array.fromAsync(new Bun.Glob('.*.tmp-*').scan(testRoot))).length, + ).toBe(0) + await expect(lstat(`${filePath}.lock`)).rejects.toMatchObject({ code: 'ENOENT' }) +}) + +test('a synchronous append through a resolved symlink target waits for the rewrite lock', async () => { + if (process.platform === 'win32') return + + const realPath = join(testRoot, 'sync-real.jsonl') + const linkPath = join(testRoot, 'sync-linked.jsonl') + const original = `${line(KEEP_1)}\n${line(TARGET)}\n${line(KEEP_2)}\n` + await writeFile(realPath, original) + await symlink('sync-real.jsonl', linkPath) + switchSession(SESSION_ID as never, testRoot) + resetProjectForTesting() + setSessionFileForTesting(linkPath) + let injected = false + setAtomicReplaceFaultInjectorForTesting(async (stage, context) => { + if (!injected && stage === 'rename' && context.requestedPath === linkPath) { + injected = true + expect(await isTranscriptFileLockHeldForTesting(linkPath)).toBe(true) + await saveCustomTitle( + SESSION_ID as UUID, + 'synchronous symlink append', + realPath, + ) + } + }) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + + expect((await lstat(linkPath)).isSymbolicLink()).toBe(true) + const result = await readFile(realPath, 'utf8') + expect(result).toBe( + `${line(KEEP_1)}\n${line(KEEP_2)}\n${JSON.stringify({ + type: 'custom-title', + customTitle: 'synchronous symlink append', + sessionId: SESSION_ID, + })}\n`, + ) +}) + +test.each(['stream-write', 'data-flush', 'rename'] as const)( + 'slow-path %s failure preserves the old transcript and cleans its temp file', + async stage => { + const suffix = `${line(KEEP_1, { payload: 'x'.repeat(70 * 1024) })}\n` + const original = `${line(TARGET)}\n${suffix}` + const filePath = await useTranscript(original) + setAtomicReplaceFaultInjectorForTesting((actualStage, context) => { + if (actualStage === stage && context.requestedPath === filePath) { + throw new Error(`injected slow ${stage}`) + } + }) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + + expect(await readFile(filePath, 'utf8')).toBe(original) + expect( + (await Array.fromAsync(new Bun.Glob('.*.tmp-*').scan(testRoot))).length, + ).toBe(0) + }, +) + +test('middle tombstone preserves restrictive mode and follows the live symlink target', async () => { + if (process.platform === 'win32') return + + const realPath = join(testRoot, 'real.jsonl') + const linkPath = join(testRoot, 'linked.jsonl') + const original = `${line(KEEP_1)}\n${line(TARGET)}\n${line(KEEP_2)}\n` + await writeFile(realPath, original) + await chmod(realPath, 0o640) + await symlink('real.jsonl', linkPath) + switchSession(SESSION_ID as never, testRoot) + resetProjectForTesting() + setSessionFileForTesting(linkPath) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + + expect((await lstat(linkPath)).isSymbolicLink()).toBe(true) + expect(await readFile(realPath, 'utf8')).toBe(`${line(KEEP_1)}\n${line(KEEP_2)}\n`) + expect((await stat(realPath)).mode & 0o777).toBe(0o640) +}) + +test('serialized appends before and during a paused rewrite survive in order', async () => { + const original = `${line(KEEP_1)}\n${line(TARGET)}\n${line(KEEP_2)}\n` + const filePath = await useTranscript(original) + let release!: () => void + let paused!: () => void + const pausedPromise = new Promise(resolve => { + paused = resolve + }) + const releasePromise = new Promise(resolve => { + release = resolve + }) + let blocked = false + setAtomicReplaceFaultInjectorForTesting(async (stage, context) => { + if (!blocked && stage === 'stream-write' && context.requestedPath === filePath) { + blocked = true + paused() + await releasePromise + } + }) + + await recordTranscript([ + message('20000000-0000-4000-8000-000000000004' as UUID, 'before rewrite'), + ]) + const priorGoalWrite = recordGoalState( + { + id: 'goal-before-rewrite', + condition: 'serialize before replacement', + status: 'active', + createdAt: TIMESTAMP, + updatedAt: TIMESTAMP, + startedAt: TIMESTAMP, + turnCount: 0, + maxTurns: 10, + evaluatorFailures: 0, + }, + SESSION_ID as UUID, + ) + const removal = removeTranscriptMessage(TARGET) + await pausedPromise + await saveCustomTitle(SESSION_ID as UUID, 'during rewrite', filePath) + const goalWrite = recordGoalState( + { + id: 'goal-during-rewrite', + condition: 'preserve queued data', + status: 'active', + createdAt: TIMESTAMP, + updatedAt: TIMESTAMP, + startedAt: TIMESTAMP, + turnCount: 0, + maxTurns: 10, + evaluatorFailures: 0, + }, + SESSION_ID as UUID, + ) + release() + await Promise.all([priorGoalWrite, removal, goalWrite]) + await flushSessionStorage() + + const result = await readFile(filePath, 'utf8') + expect(result).not.toContain(TARGET) + expect(result).toContain('before rewrite') + expect(result).toContain('goal-before-rewrite') + expect(result).toContain('during rewrite') + expect(result).toContain('goal-during-rewrite') + expect(result.indexOf('during rewrite')).toBeLessThan( + result.indexOf('goal-during-rewrite'), + ) + expect(result.indexOf('goal-before-rewrite')).toBeLessThan( + result.indexOf('during rewrite'), + ) +}) + +test('a rewrite barrier is active as soon as hydration is enqueued', async () => { + const transcriptPath = await prepareHydration() + const remote = [message(KEEP_1, 'remote foreground')] + spyOn(sessionIngress, 'getSessionLogs').mockResolvedValue(remote as never) + setTranscriptRewriteHooksForTesting({ + enqueued(filePath) { + if (filePath === transcriptPath) { + void saveCustomTitle( + SESSION_ID as UUID, + 'queued after hydration barrier', + transcriptPath, + ) + } + }, + }) + + expect(await hydrateRemoteSession(SESSION_ID, 'https://ingress.test')).toBe( + true, + ) + await flushSessionStorage() + + const result = await readFile(transcriptPath, 'utf8') + expect(result).toContain('remote foreground') + expect(result).toContain('queued after hydration barrier') + expect(result.indexOf('remote foreground')).toBeLessThan( + result.indexOf('queued after hydration barrier'), + ) +}) + +test('speculation acceptance is queued behind an active transcript rewrite', async () => { + const original = `${line(KEEP_1)}\n${line(TARGET)}\n${line(KEEP_2)}\n` + const filePath = await useTranscript(original) + let release!: () => void + let paused!: () => void + const pausedPromise = new Promise(resolve => { + paused = resolve + }) + const releasePromise = new Promise(resolve => { + release = resolve + }) + let blocked = false + setAtomicReplaceFaultInjectorForTesting(async (stage, context) => { + if (!blocked && stage === 'stream-write' && context.requestedPath === filePath) { + blocked = true + paused() + await releasePromise + } + }) + + const removal = removeTranscriptMessage(TARGET) + await pausedPromise + await recordSpeculationAccept({ + type: 'speculation-accept', + timestamp: TIMESTAMP, + timeSavedMs: 123, + }) + release() + await removal + await flushSessionStorage() + + const result = await readFile(filePath, 'utf8') + expect(result).not.toContain(TARGET) + expect(result).toContain('"type":"speculation-accept"') + expect(result).toContain('"timeSavedMs":123') +}) + +test('a failed queued append settles a following tombstone and releases its barrier', async () => { + const original = `${line(KEEP_1)}\n${line(TARGET)}\n${line(KEEP_2)}\n` + const filePath = await useTranscript(original) + setTranscriptRewriteHooksForTesting({ + beforeFileAppend(appendPath) { + if (appendPath === filePath) throw new Error('injected append failure') + }, + }) + + await recordTranscript([ + message('20000000-0000-4000-8000-000000000006' as UUID, 'queued first'), + ]) + const removal = removeTranscriptMessage(TARGET) + await expect(flushSessionStorage()).rejects.toThrow('injected append failure') + + await removal + + resetTranscriptRewriteHooksForTesting() + await saveCustomTitle(SESSION_ID as UUID, 'barrier released', filePath) + expect(await readFile(filePath, 'utf8')).toContain('barrier released') +}) + +test('a long rewrite keeps drains single-flight across late appends, tombstones, and flush', async () => { + const secondTarget = OTHER_SESSION_ID as UUID + const lateUuid = '20000000-0000-4000-8000-000000000005' as UUID + const original = `${line(KEEP_1)}\n${line(TARGET)}\n${line(secondTarget)}\n${line(KEEP_2)}\n` + const filePath = await useTranscript(original) + let release!: () => void + let paused!: () => void + const pausedPromise = new Promise(resolve => { + paused = resolve + }) + const releasePromise = new Promise(resolve => { + release = resolve + }) + let blocked = false + setAtomicReplaceFaultInjectorForTesting(async (stage, context) => { + if (!blocked && stage === 'stream-write' && context.requestedPath === filePath) { + blocked = true + paused() + await releasePromise + } + }) + + const firstRemoval = removeTranscriptMessage(TARGET) + await pausedPromise + await recordTranscript([message(lateUuid, 'late queued append')]) + const secondRemoval = removeTranscriptMessage(secondTarget) + release() + await Promise.all([firstRemoval, secondRemoval]) + await flushSessionStorage() + + const result = await readFile(filePath, 'utf8') + expect(result).not.toContain(TARGET) + expect(result).not.toContain(secondTarget) + expect(result).toContain(lateUuid) + expect(result).toContain('late queued append') +}) + +test('a microtask append at rewrite completion is not stranded behind the barrier', async () => { + const original = `${line(KEEP_1)}\n${line(TARGET)}\n${line(KEEP_2)}\n` + const filePath = await useTranscript(original) + let queuedLateAppend!: () => void + const lateAppendQueued = new Promise(resolve => { + queuedLateAppend = resolve + }) + setTranscriptRewriteHooksForTesting({ + beforeBarrierRelease(rewritePath) { + if (rewritePath !== filePath) return + queueMicrotask(() => { + void saveCustomTitle( + SESSION_ID as UUID, + 'microtask at barrier release', + filePath, + ) + queuedLateAppend() + }) + }, + }) + + await removeTranscriptMessage(TARGET) + await lateAppendQueued + await flushSessionStorage() + + const result = await readFile(filePath, 'utf8') + expect(result).not.toContain(TARGET) + expect(result).toContain('microtask at barrier release') +}) + +test('two concurrent tombstones serialize without resurrecting either entry', async () => { + const original = `${line(KEEP_1)}\n${line(TARGET)}\n${line(KEEP_2)}\n${line(OTHER_SESSION_ID as UUID)}\n` + const filePath = await useTranscript(original) + + await Promise.all([ + removeTranscriptMessage(TARGET), + removeTranscriptMessage(OTHER_SESSION_ID as UUID), + ]) + await flushSessionStorage() + + expect(await readFile(filePath, 'utf8')).toBe(`${line(KEEP_1)}\n${line(KEEP_2)}\n`) +}) + +test('slow duplicate removal preserves a missing final newline', async () => { + const longFinalTarget = line(TARGET, { payload: 'x'.repeat(70 * 1024) }) + const original = `${line(TARGET)}\n${line(KEEP_1)}\n${longFinalTarget}` + const filePath = await useTranscript(original) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + + expect(await readFile(filePath, 'utf8')).toBe(line(KEEP_1)) +}) + +test('resume loader reads the byte-preserved tombstone result', async () => { + const first = { ...message(KEEP_1, 'keep one'), parentUuid: null } + const removed = { ...message(TARGET, 'remove me'), parentUuid: KEEP_1 } + const last = { ...message(KEEP_2, 'keep two'), parentUuid: KEEP_1 } + const original = `${JSON.stringify(first)}\n{malformed but preserved}\n${JSON.stringify(removed)}\n${JSON.stringify(last)}\n` + const filePath = await useTranscript(original) + + await removeTranscriptMessage(TARGET) + await flushSessionStorage() + const loaded = await loadTranscriptFile(filePath, { keepAllLeaves: true }) + + expect(loaded.messages.has(KEEP_1)).toBe(true) + expect(loaded.messages.has(TARGET)).toBe(false) + expect(loaded.messages.has(KEEP_2)).toBe(true) + expect(loaded.messages.get(KEEP_2)?.parentUuid).toBe(KEEP_1) + expect( + buildConversationChain(loaded.messages, loaded.messages.get(KEEP_2)!).map( + entry => entry.uuid, + ), + ).toEqual([KEEP_1, KEEP_2]) + expect(await readFile(filePath, 'utf8')).toContain('{malformed but preserved}\n') +}) + +test('v1 foreground hydration commits complete content and mode atomically', async () => { + const transcriptPath = await prepareHydration() + const remote = [message(KEEP_1, 'remote one'), message(KEEP_2, 'remote two')] + spyOn(sessionIngress, 'getSessionLogs').mockResolvedValue(remote as never) + + expect(await hydrateRemoteSession(SESSION_ID, 'https://ingress.test')).toBe(true) + + expect(await readFile(transcriptPath, 'utf8')).toBe( + `${remote.map(entry => JSON.stringify(entry)).join('\n')}\n`, + ) + expect((await stat(transcriptPath)).mode & 0o777).toBe(0o600) +}) + +test('v1 null fetch, serialization failure, and rename failure preserve old content', async () => { + const transcriptPath = await prepareHydration() + const original = `${line(KEEP_1)}\n` + await writeFile(transcriptPath, original) + const getLogs = spyOn(sessionIngress, 'getSessionLogs') + + getLogs.mockResolvedValueOnce(null) + expect(await hydrateRemoteSession(SESSION_ID, 'https://ingress.test')).toBe(false) + expect(await readFile(transcriptPath, 'utf8')).toBe(original) + + const circular: Record = { uuid: TARGET } + circular.self = circular + getLogs.mockResolvedValueOnce([circular] as never) + expect(await hydrateRemoteSession(SESSION_ID, 'https://ingress.test')).toBe(false) + expect(await readFile(transcriptPath, 'utf8')).toBe(original) + + getLogs.mockResolvedValueOnce([message(KEEP_2, 'replacement')] as never) + setAtomicReplaceFaultInjectorForTesting((stage, context) => { + if (stage === 'rename' && context.requestedPath === transcriptPath) { + throw new Error('injected rename') + } + }) + expect(await hydrateRemoteSession(SESSION_ID, 'https://ingress.test')).toBe(false) + expect(await readFile(transcriptPath, 'utf8')).toBe(original) +}) + +test('empty foreground hydration preserves an existing transcript', async () => { + const transcriptPath = await prepareHydration() + const original = `${line(KEEP_1)}\n` + await writeFile(transcriptPath, original) + spyOn(sessionIngress, 'getSessionLogs').mockResolvedValue([] as never) + + expect(await hydrateRemoteSession(SESSION_ID, 'https://ingress.test')).toBe(false) + expect(await readFile(transcriptPath, 'utf8')).toBe(original) + + resetProjectForTesting() + setInternalEventReader(async () => [], async () => []) + + expect(await hydrateFromCCRv2InternalEvents(SESSION_ID)).toBe(false) + expect(await readFile(transcriptPath, 'utf8')).toBe(original) +}) + +test('CCR subagent transcripts commit independently and suppress full-success diagnostics', async () => { + const transcriptPath = await prepareHydration() + const diagnosticsPath = join(testRoot, 'diagnostics.jsonl') + process.env.CLAUDE_CODE_DIAGNOSTICS_FILE = diagnosticsPath + const agentA = 'agent-a' + const agentB = 'agent-b' + setInternalEventReader( + async () => [{ payload: message(KEEP_1, 'foreground') as never }], + async () => [ + { agent_id: agentA, payload: { type: 'test', uuid: KEEP_1 } }, + { agent_id: agentB, payload: { type: 'test', uuid: KEEP_2 } }, + ], + ) + const agentAPath = getAgentTranscriptPath(agentA as never) + const agentBPath = getAgentTranscriptPath(agentB as never) + await mkdir(dirname(agentAPath), { recursive: true, mode: 0o700 }) + await writeFile(agentAPath, 'old-a\n') + await writeFile(agentBPath, 'old-b\n') + setAtomicReplaceFaultInjectorForTesting((stage, context) => { + if (stage === 'rename' && context.requestedPath === agentBPath) { + throw new Error('agent-b rename failed') + } + }) + + expect(await hydrateFromCCRv2InternalEvents(SESSION_ID)).toBe(false) + + expect(await readFile(transcriptPath, 'utf8')).toContain('foreground') + expect(await readFile(agentAPath, 'utf8')).toContain(KEEP_1) + expect(await readFile(agentBPath, 'utf8')).toBe('old-b\n') + const diagnostics = await readFile(diagnosticsPath, 'utf8') + expect(diagnostics).not.toContain('hydrate_ccr_v2_completed') + expect(diagnostics).toContain('hydrate_ccr_v2_subagent_write_fail') +}) + +test('CCR distinguishes failed subagent fetch from a successful empty fetch', async () => { + const transcriptPath = await prepareHydration() + const diagnosticsPath = join(testRoot, 'subagent-read-diagnostics.jsonl') + process.env.CLAUDE_CODE_DIAGNOSTICS_FILE = diagnosticsPath + setInternalEventReader( + async () => [{ payload: message(KEEP_1, 'foreground') as never }], + async () => null, + ) + expect(await hydrateFromCCRv2InternalEvents(SESSION_ID)).toBe(false) + expect(await readFile(transcriptPath, 'utf8')).toContain('foreground') + const failedDiagnostics = await readFile(diagnosticsPath, 'utf8') + expect(failedDiagnostics).toContain('hydrate_ccr_v2_subagent_read_fail') + expect(failedDiagnostics).not.toContain('hydrate_ccr_v2_completed') + + resetProjectForTesting() + await writeFile(diagnosticsPath, '') + setInternalEventReader( + async () => [{ payload: message(KEEP_2, 'foreground') as never }], + async () => [], + ) + expect(await hydrateFromCCRv2InternalEvents(SESSION_ID)).toBe(true) + expect(await readFile(diagnosticsPath, 'utf8')).toContain( + 'hydrate_ccr_v2_completed', + ) +}) diff --git a/src/utils/sessionStorage.ts b/src/utils/sessionStorage.ts index b66b947b2..697520547 100644 --- a/src/utils/sessionStorage.ts +++ b/src/utils/sessionStorage.ts @@ -53,6 +53,7 @@ import { type SerializedMessage, type SessionBranchEntry, sortLogs, + type SpeculationAcceptMessage, type TranscriptMessage, } from '../types/logs.js' import type { @@ -65,6 +66,7 @@ import type { } from '../types/message.js' import type { QueueOperationMessage } from '../types/messageQueueTypes.js' import { uniq } from './array.js' +import { replaceFileAtomic } from './atomicReplace.js' import { registerCleanup } from './cleanupRegistry.js' import { updateSessionName } from './concurrentSessions.js' import { getCwd } from './cwd.js' @@ -92,6 +94,11 @@ import { } from './sessionStoragePortable.js' import { shouldSkipSessionPersistence } from './sessionPersistencePolicy.js' import { jsonParse, jsonStringify } from './slowOperations.js' +import { + isTranscriptFileLockHeldByAsyncOperation, + withTranscriptFileLock, + withTranscriptFileLockSync, +} from './transcriptFileLock.js' import type { ContentReplacementRecord } from './toolResultStorage.js' import { validateUuid } from './uuid.js' @@ -130,9 +137,318 @@ type Transcript = ( * marker. Kept in sync with sessionStoragePortable.ts — generic pattern avoids * an ever-growing allowlist that falls behind as new notification types ship. */ -// 50MB — prevents OOM in the tombstone slow path which reads + rewrites the -// entire session file. Session files can grow to multiple GB (inc-3930). +// 50 MB — bounds the tombstone slow scan. Session files can grow to multiple +// GB (inc-3930), while a target outside the tail window is exceptionally rare. const MAX_TOMBSTONE_REWRITE_BYTES = 50 * 1024 * 1024 +const TRANSCRIPT_COPY_CHUNK_BYTES = 64 * 1024 + +type TranscriptByteRange = { start: number; end: number } + +type TranscriptLineSpan = TranscriptByteRange & { + hasTerminator: boolean +} + +type QueuedAppend = { + kind: 'append' + entry: Entry + resolve: () => void + reject: (error: unknown) => void +} + +type QueuedRewrite = { + kind: 'rewrite' + rewrite: (signal: AbortSignal) => Promise + resolve: () => void + reject: (error: unknown) => void +} + +type QueuedDirectAppend = { + kind: 'direct-append' + data: string + resolve?: () => void + reject?: (error: unknown) => void +} + +type TranscriptWriteOperation = + | QueuedAppend + | QueuedDirectAppend + | QueuedRewrite + +type TranscriptRewriteHooksForTesting = { + enqueued?: (filePath: string) => void + beforeBarrierRelease?: (filePath: string) => void + beforeFileAppend?: (filePath: string) => void +} + +let transcriptRewriteHooksForTesting: TranscriptRewriteHooksForTesting = {} + +/** @internal Deterministic rewrite scheduler hooks for concurrency tests. */ +export function setTranscriptRewriteHooksForTesting( + hooks: TranscriptRewriteHooksForTesting, +): void { + transcriptRewriteHooksForTesting = hooks +} + +/** @internal Reset deterministic rewrite scheduler hooks. */ +export function resetTranscriptRewriteHooksForTesting(): void { + transcriptRewriteHooksForTesting = {} +} + +type SessionFileHandle = Awaited> + +function transcriptLineHasUuid(lineBytes: Uint8Array, targetUuid: UUID): boolean { + try { + const entry = jsonParse(Buffer.from(lineBytes).toString('utf8').trim()) + return ( + typeof entry === 'object' && + entry !== null && + 'uuid' in entry && + entry.uuid === targetUuid + ) + } catch { + return false + } +} + +async function readTranscriptRange( + handle: SessionFileHandle, + start: number, + end: number, +): Promise { + const result = Buffer.allocUnsafe(end - start) + let offset = 0 + while (offset < result.length) { + const { bytesRead } = await handle.read( + result, + offset, + result.length - offset, + start + offset, + ) + if (bytesRead === 0) { + throw new Error('Transcript changed while scanning tombstone target') + } + offset += bytesRead + } + return result +} + +function findTailTombstoneSpan( + tail: Buffer, + tailStart: number, + targetUuid: UUID, +): TranscriptLineSpan | undefined { + const needle = Buffer.from(`"uuid":"${targetUuid}"`) + let searchFrom = tail.length - needle.length + + while (searchFrom >= 0) { + const matchIndex = tail.lastIndexOf(needle, searchFrom) + if (matchIndex < 0) return undefined + + const previousNewline = tail.lastIndexOf(0x0a, matchIndex) + if (previousNewline < 0 && tailStart !== 0) { + // The candidate line began before the tail window. A bounded slow scan + // is required to validate its top-level UUID. + return undefined + } + + const lineStart = previousNewline + 1 + const nextNewline = tail.indexOf(0x0a, matchIndex + needle.length) + const lineContentEnd = nextNewline >= 0 ? nextNewline : tail.length + if ( + transcriptLineHasUuid( + tail.subarray(lineStart, lineContentEnd), + targetUuid, + ) + ) { + return { + start: tailStart + lineStart, + end: tailStart + (nextNewline >= 0 ? nextNewline + 1 : tail.length), + hasTerminator: nextNewline >= 0, + } + } + + // The needle belonged to a nested value on an unrelated line. Skip the + // entire line so another occurrence on it cannot be mistaken for a key. + searchFrom = lineStart - 1 + } + + return undefined +} + +async function scanTranscriptTombstoneSpans( + filePath: string, + fileSize: number, + targetUuid: UUID, +): Promise { + const needle = `"uuid":"${targetUuid}"` + const handle = await fsOpen(filePath, 'r') + const spans: TranscriptLineSpan[] = [] + const buffer = Buffer.allocUnsafe(TRANSCRIPT_COPY_CHUNK_BYTES) + let offset = 0 + let lineStart = 0 + let lineHasNeedle = false + let needleCarry = '' + + const finishLine = async (contentEnd: number, hasTerminator: boolean) => { + if (lineHasNeedle) { + const lineBytes = await readTranscriptRange(handle, lineStart, contentEnd) + if (transcriptLineHasUuid(lineBytes, targetUuid)) { + spans.push({ + start: lineStart, + end: hasTerminator ? contentEnd + 1 : contentEnd, + hasTerminator, + }) + } + } + lineStart = hasTerminator ? contentEnd + 1 : contentEnd + lineHasNeedle = false + needleCarry = '' + } + + try { + while (offset < fileSize) { + const length = Math.min(buffer.length, fileSize - offset) + const { bytesRead } = await handle.read(buffer, 0, length, offset) + if (bytesRead === 0) { + throw new Error('Transcript changed while scanning tombstone target') + } + + let segmentStart = 0 + while (segmentStart < bytesRead) { + const newline = buffer.indexOf(0x0a, segmentStart) + const segmentEnd = newline >= 0 && newline < bytesRead ? newline : bytesRead + // UUID needles are ASCII. Latin-1 maps each byte one-to-one so chunk + // boundaries cannot corrupt the substring scan or alter copied bytes. + const segmentText = buffer.toString('latin1', segmentStart, segmentEnd) + const searchText = needleCarry + segmentText + if (searchText.includes(needle)) lineHasNeedle = true + needleCarry = searchText.slice(-(needle.length - 1)) + + if (newline < 0 || newline >= bytesRead) break + await finishLine(offset + newline, true) + segmentStart = newline + 1 + } + + offset += bytesRead + } + + if (lineStart < fileSize) await finishLine(fileSize, false) + return spans + } finally { + await handle.close() + } +} + +function rangesExcludingSpans( + fileSize: number, + spans: TranscriptLineSpan[], +): TranscriptByteRange[] { + const ranges: TranscriptByteRange[] = [] + let cursor = 0 + for (const span of spans) { + if (cursor < span.start) ranges.push({ start: cursor, end: span.start }) + cursor = span.end + } + if (cursor < fileSize) ranges.push({ start: cursor, end: fileSize }) + return ranges +} + +async function* streamTranscriptRanges( + filePath: string, + ranges: TranscriptByteRange[], +): AsyncGenerator { + const handle = await fsOpen(filePath, 'r') + const buffer = Buffer.allocUnsafe(TRANSCRIPT_COPY_CHUNK_BYTES) + try { + for (const range of ranges) { + let position = range.start + while (position < range.end) { + const length = Math.min(buffer.length, range.end - position) + const { bytesRead } = await handle.read(buffer, 0, length, position) + if (bytesRead === 0) { + throw new Error('Transcript changed while building replacement') + } + position += bytesRead + yield buffer.subarray(0, bytesRead) + } + } + } finally { + await handle.close() + } +} + +async function finalTombstoneBoundary( + filePath: string, + span: TranscriptLineSpan, +): Promise { + if (span.hasTerminator || span.start === 0) return span.start + + const handle = await fsOpen(filePath, 'r') + try { + const separator = Buffer.allocUnsafe(2) + const readStart = Math.max(0, span.start - separator.length) + const { bytesRead } = await handle.read( + separator, + 0, + span.start - readStart, + readStart, + ) + const bytes = separator.subarray(0, bytesRead) + if (bytes.at(-1) !== 0x0a) return span.start + return bytes.at(-2) === 0x0d ? span.start - 2 : span.start - 1 + } finally { + await handle.close() + } +} + +async function adjustFinalUnterminatedRemovalSpan( + filePath: string, + fileSize: number, + spans: TranscriptLineSpan[], +): Promise { + const finalSpan = spans.at(-1) + if ( + !finalSpan || + finalSpan.end !== fileSize || + finalSpan.hasTerminator + ) { + return spans + } + + const boundary = await finalTombstoneBoundary(filePath, finalSpan) + if (boundary === finalSpan.start) return spans + return [ + ...spans.slice(0, -1), + { ...finalSpan, start: boundary }, + ] +} + +async function truncateFinalTranscriptLine( + filePath: string, + span: TranscriptLineSpan, + expectedSize: number, + signal?: AbortSignal, +): Promise { + signal?.throwIfAborted() + const boundary = await finalTombstoneBoundary(filePath, span) + const handle = await fsOpen(filePath, 'r+') + try { + if ((await handle.stat()).size !== expectedSize) { + throw new Error('Transcript changed before final tombstone truncate') + } + signal?.throwIfAborted() + await handle.truncate(boundary) + await handle.datasync() + } finally { + await handle.close() + } +} + +async function* serializeTranscriptEntries( + entries: Iterable, +): AsyncGenerator { + for (const entry of entries) yield jsonStringify(entry) + '\n' +} const SKIP_FIRST_PROMPT_PATTERN = /^(?:\s*<[a-z][\w-]*[\s>]|\[Request interrupted by user[^\]]*\])/ @@ -588,12 +904,13 @@ class Project { private internalSubagentEventReader: InternalEventReader | null = null private pendingWriteCount: number = 0 private flushResolvers: Array<() => void> = [] - // Per-file write queues. Each entry carries a resolve callback so - // callers of enqueueWrite can optionally await their specific write. - private writeQueues = new Map< - string, - Array<{ entry: Entry; resolve: () => void }> - >() + // Appends and complete-file rewrites share one per-file operation queue. + // This makes a rewrite an ordering barrier: earlier appends reach the old + // inode before it is copied, and later appends reach the replacement. + private writeQueues = new Map() + private rewriteBarrierFiles = new Set() + private pendingRewriteCounts = new Map() + private pendingDirectAppends = new Map>>() private flushTimer: ReturnType | null = null private activeDrain: Promise | null = null private FLUSH_INTERVAL_MS = 100 @@ -609,6 +926,9 @@ class Project { this.flushTimer = null this.activeDrain = null this.writeQueues = new Map() + this.rewriteBarrierFiles = new Set() + this.pendingRewriteCounts = new Map() + this.pendingDirectAppends = new Map() } private incrementPendingWrites(): void { @@ -636,85 +956,272 @@ class Project { } private enqueueWrite(filePath: string, entry: Entry): Promise { - return new Promise(resolve => { + const append = new Promise((resolve, reject) => { let queue = this.writeQueues.get(filePath) if (!queue) { queue = [] this.writeQueues.set(filePath, queue) } - queue.push({ entry, resolve }) + queue.push({ kind: 'append', entry, resolve, reject }) + this.scheduleDrain() + }) + // Most append call sites are intentionally fire-and-forget. Mark failures + // handled here while preserving rejection for the callers that do await. + void append.catch(() => {}) + return append + } + + private enqueueRewrite( + filePath: string, + rewrite: (signal: AbortSignal) => Promise, + ): Promise { + return new Promise((resolve, reject) => { + let queue = this.writeQueues.get(filePath) + if (!queue) { + queue = [] + this.writeQueues.set(filePath, queue) + } + this.rewriteBarrierFiles.add(filePath) + this.pendingRewriteCounts.set( + filePath, + (this.pendingRewriteCounts.get(filePath) ?? 0) + 1, + ) + queue.push({ kind: 'rewrite', rewrite, resolve, reject }) + transcriptRewriteHooksForTesting.enqueued?.(filePath) this.scheduleDrain() }) } private scheduleDrain(): void { - if (this.flushTimer) { - return - } - this.flushTimer = setTimeout(async () => { + if (this.flushTimer || this.activeDrain) return + this.flushTimer = setTimeout(() => { this.flushTimer = null - this.activeDrain = this.drainWriteQueue() - await this.activeDrain - this.activeDrain = null - // If more items arrived during drain, schedule again - if (this.writeQueues.size > 0) { - this.scheduleDrain() - } + void this.startDrain() }, this.FLUSH_INTERVAL_MS) } - private async appendToFile(filePath: string, data: string): Promise { + private startDrain(): Promise { + if (this.activeDrain) return this.activeDrain + + const drain = this.drainWriteQueue() + this.activeDrain = drain + void drain.then( + () => this.finishDrain(drain), + error => { + this.finishDrain(drain) + logError(error) + }, + ) + return drain + } + + private finishDrain(drain: Promise): void { + if (this.activeDrain !== drain) return + this.activeDrain = null + if (this.writeQueues.size > 0) this.scheduleDrain() + } + + private async appendDirectlyToFile( + filePath: string, + data: string, + ): Promise { + transcriptRewriteHooksForTesting.beforeFileAppend?.(filePath) + await mkdir(dirname(filePath), { recursive: true, mode: 0o700 }) + await withTranscriptFileLock(filePath, async signal => { + signal.throwIfAborted() + await fsAppendFile(filePath, data, { mode: 0o600 }) + }) + } + + private appendToFile(filePath: string, data: string): Promise { + if (this.rewriteBarrierFiles.has(filePath)) { + return this.enqueueDirectAppend(filePath, data) + } + + const append = this.appendDirectlyToFile(filePath, data) + let pending = this.pendingDirectAppends.get(filePath) + if (!pending) { + pending = new Set() + this.pendingDirectAppends.set(filePath, pending) + } + pending.add(append) + const removePending = () => { + pending?.delete(append) + if (pending?.size === 0) this.pendingDirectAppends.delete(filePath) + } + void append.then(removePending, removePending) + return append + } + + private enqueueDirectAppend(filePath: string, data: string): Promise { + return new Promise((resolve, reject) => { + this.queueDirectAppend(filePath, { data, resolve, reject }) + }) + } + + private queueDirectAppend( + filePath: string, + append: Omit, + ): void { + let queue = this.writeQueues.get(filePath) + if (!queue) { + queue = [] + this.writeQueues.set(filePath, queue) + } + queue.push({ kind: 'direct-append', ...append }) + this.scheduleDrain() + } + + /** @internal Route synchronous metadata appends around a queued rewrite. */ + _deferSynchronousAppend(filePath: string, data: string): boolean { + if ( + !this.rewriteBarrierFiles.has(filePath) && + !isTranscriptFileLockHeldByAsyncOperation(filePath) + ) { + return false + } + this.queueDirectAppend(filePath, { data }) + return true + } + + private async runRewriteOperation( + filePath: string, + rewrite: (signal: AbortSignal) => Promise, + ): Promise { + const earlierDirectAppends = this.pendingDirectAppends.get(filePath) + if (earlierDirectAppends) await Promise.all(earlierDirectAppends) + await withTranscriptFileLock(filePath, signal => rewrite(signal)) + transcriptRewriteHooksForTesting.beforeBarrierRelease?.(filePath) + } + + private finishRewrite(filePath: string): void { + const remaining = (this.pendingRewriteCounts.get(filePath) ?? 1) - 1 + if (remaining === 0) this.pendingRewriteCounts.delete(filePath) + else this.pendingRewriteCounts.set(filePath, remaining) + } + + private async flushQueuedAppends( + filePath: string, + content: string, + operations: QueuedAppend[], + ): Promise { + // Every operation corresponds to serialized content, so an empty batch has + // no promises to settle. + if (content.length === 0) return try { - await fsAppendFile(filePath, data, { mode: 0o600 }) - } catch { - // Directory may not exist — some NFS-like filesystems return - // unexpected error codes, so don't discriminate on code. - await mkdir(dirname(filePath), { recursive: true, mode: 0o700 }) - await fsAppendFile(filePath, data, { mode: 0o600 }) + await this.appendDirectlyToFile(filePath, content) + } catch (error) { + for (const operation of operations) operation.reject(error) + throw error + } + for (const operation of operations) operation.resolve() + } + + private rejectQueuedOperations( + filePath: string, + operations: TranscriptWriteOperation[], + error: unknown, + ): void { + for (const operation of operations) { + if (operation.kind === 'append') { + operation.reject(error) + } else if (operation.kind === 'rewrite') { + operation.reject(error) + this.finishRewrite(filePath) + } else if (operation.reject) { + operation.reject(error) + } else { + logError(error) + } } } private async drainWriteQueue(): Promise { for (const [filePath, queue] of this.writeQueues) { if (queue.length === 0) { + if (!this.pendingRewriteCounts.has(filePath)) { + this.rewriteBarrierFiles.delete(filePath) + } + this.writeQueues.delete(filePath) continue } const batch = queue.splice(0) let content = '' - const resolvers: Array<() => void> = [] + let queuedAppends: QueuedAppend[] = [] + let operationIndex = 0 - for (const { entry, resolve } of batch) { - const line = jsonStringify(entry) + '\n' - - if (content.length + line.length >= this.MAX_CHUNK_BYTES) { - // Flush chunk and resolve its entries before starting a new one - await this.appendToFile(filePath, content) - for (const r of resolvers) { - r() + try { + for (; operationIndex < batch.length; operationIndex++) { + const operation = batch[operationIndex]! + if (operation.kind === 'rewrite') { + await this.flushQueuedAppends(filePath, content, queuedAppends) + content = '' + queuedAppends = [] + try { + await this.runRewriteOperation(filePath, operation.rewrite) + operation.resolve() + } catch (error) { + operation.reject(error) + } finally { + this.finishRewrite(filePath) + } + continue } - resolvers.length = 0 - content = '' + + if (operation.kind === 'direct-append') { + await this.flushQueuedAppends(filePath, content, queuedAppends) + content = '' + queuedAppends = [] + try { + await this.appendDirectlyToFile(filePath, operation.data) + operation.resolve?.() + } catch (error) { + if (operation.reject) operation.reject(error) + else logError(error) + } + continue + } + + const line = jsonStringify(operation.entry) + '\n' + if (content.length + line.length >= this.MAX_CHUNK_BYTES) { + await this.flushQueuedAppends(filePath, content, queuedAppends) + content = '' + queuedAppends = [] + } + content += line + queuedAppends.push(operation) } - content += line - resolvers.push(resolve) - } - - if (content.length > 0) { - await this.appendToFile(filePath, content) - for (const r of resolvers) { - r() + await this.flushQueuedAppends(filePath, content, queuedAppends) + content = '' + queuedAppends = [] + } catch (error) { + for (const operation of queuedAppends) operation.reject(error) + this.rejectQueuedOperations( + filePath, + batch.slice(operationIndex), + error, + ) + throw error + } finally { + if (!this.pendingRewriteCounts.has(filePath) && queue.length === 0) { + this.rewriteBarrierFiles.delete(filePath) } + if (queue.length === 0) this.writeQueues.delete(filePath) } } + } - // Clean up empty queues - for (const [filePath, queue] of this.writeQueues) { - if (queue.length === 0) { - this.writeQueues.delete(filePath) - } - } + async replaceTranscriptFile( + filePath: string, + data: string | Uint8Array | AsyncIterable, + ): Promise { + return this.trackWrite(() => + this.enqueueRewrite(filePath, signal => + replaceFileAtomic(filePath, data, { signal }), + ), + ) } resetSessionFile(): void { @@ -888,17 +1395,22 @@ class Project { } async flush(): Promise { - // Cancel pending timer if (this.flushTimer) { clearTimeout(this.flushTimer) this.flushTimer = null } - // Wait for any in-flight drain to finish - if (this.activeDrain) { - await this.activeDrain + + while (this.activeDrain || this.writeQueues.size > 0) { + if (this.activeDrain) { + await this.activeDrain + continue + } + if (this.flushTimer) { + clearTimeout(this.flushTimer) + this.flushTimer = null + } + await this.startDrain() } - // Drain anything remaining in the queues - await this.drainWriteQueue() // Wait for non-queue tracked operations (e.g. removeMessageByUuid) if (this.pendingWriteCount === 0) { @@ -913,88 +1425,115 @@ class Project { * Remove a message from the transcript by UUID. * Used for tombstoning orphaned messages from failed streaming attempts. * - * The target is almost always the most recently appended entry, so we - * read only the tail, locate the line, and splice it out with a - * positional write + truncate instead of rewriting the whole file. + * A final line can be committed with one durable truncate. Any removal + * that preserves later bytes is built in a sibling file and renamed over + * the transcript so interruption never exposes a truncated live file. */ async removeMessageByUuid(targetUuid: UUID): Promise { + if (this.sessionFile === null) return + const sessionFile = this.sessionFile + return this.trackWrite(async () => { - if (this.sessionFile === null) return try { - let fileSize = 0 - const fh = await fsOpen(this.sessionFile, 'r+') - try { - const { size } = await fh.stat() - fileSize = size - if (size === 0) return + await this.enqueueRewrite(sessionFile, async signal => { + try { + const handle = await fsOpen(sessionFile, 'r') + let fileSize = 0 + let tailSpan: TranscriptLineSpan | undefined + try { + fileSize = (await handle.stat()).size + if (fileSize === 0) return - const chunkLen = Math.min(size, LITE_READ_BUF_SIZE) - const tailStart = size - chunkLen - const buf = Buffer.allocUnsafe(chunkLen) - const { bytesRead } = await fh.read(buf, 0, chunkLen, tailStart) - const tail = buf.subarray(0, bytesRead) + const chunkLength = Math.min(fileSize, LITE_READ_BUF_SIZE) + const tailStart = fileSize - chunkLength + const buffer = Buffer.allocUnsafe(chunkLength) + const { bytesRead } = await handle.read( + buffer, + 0, + chunkLength, + tailStart, + ) + tailSpan = findTailTombstoneSpan( + buffer.subarray(0, bytesRead), + tailStart, + targetUuid, + ) + } finally { + await handle.close() + } - // Entries are serialized via JSON.stringify (no key-value - // whitespace). Search for the full `"uuid":"..."` pattern, not - // just the bare UUID, so we do not match the same value sitting - // in `parentUuid` of a child entry. UUIDs are pure ASCII so a - // byte-level search is correct. - const needle = `"uuid":"${targetUuid}"` - const matchIdx = tail.lastIndexOf(needle) - - if (matchIdx >= 0) { - // 0x0a never appears inside a UTF-8 multi-byte sequence, so - // byte-scanning for line boundaries is safe even if the chunk - // starts mid-character. - const prevNl = tail.lastIndexOf(0x0a, matchIdx) - // If the preceding newline is outside our chunk and we did not - // read from the start of the file, the line is longer than the - // window - fall through to the slow path. - if (prevNl >= 0 || tailStart === 0) { - const lineStart = prevNl + 1 // 0 when prevNl === -1 - const nextNl = tail.indexOf(0x0a, matchIdx + needle.length) - const lineEnd = nextNl >= 0 ? nextNl + 1 : bytesRead - - const absLineStart = tailStart + lineStart - const afterLen = bytesRead - lineEnd - // Truncate first, then re-append the trailing lines. In the - // common case (target is the last entry) afterLen is 0 and - // this is a single ftruncate. - await fh.truncate(absLineStart) - if (afterLen > 0) { - await fh.write(tail, lineEnd, afterLen, absLineStart) + if (tailSpan) { + if (tailSpan.end === fileSize) { + await truncateFinalTranscriptLine( + sessionFile, + tailSpan, + fileSize, + signal, + ) + return } + await replaceFileAtomic( + sessionFile, + streamTranscriptRanges( + sessionFile, + rangesExcludingSpans(fileSize, [tailSpan]), + ), + { expectedTargetSize: fileSize, signal }, + ) return } - } - } finally { - await fh.close() - } - // Slow path: target was not in the last 64KB. Rare - requires many - // large entries to have landed between the write and the tombstone. - if (fileSize > MAX_TOMBSTONE_REWRITE_BYTES) { - logForDebugging( - `Skipping tombstone removal: session file too large (${formatFileSize(fileSize)})`, - { level: 'warn' }, - ) - return - } - const content = await readFile(this.sessionFile, { encoding: 'utf-8' }) - const lines = content.split('\n').filter((line: string) => { - if (!line.trim()) return true - try { - const entry = jsonParse(line) - return entry.uuid !== targetUuid - } catch { - return true // Keep malformed lines + // Slow path: the target is outside the tail window, its line is + // longer than that window, or tail candidates only contained a + // nested UUID. The guard retains the existing 50 MB policy. + if (fileSize > MAX_TOMBSTONE_REWRITE_BYTES) { + logForDebugging( + 'Skipping tombstone removal: session file too large ' + + `(${formatFileSize(fileSize)})`, + { level: 'warn' }, + ) + return + } + + const spans = await scanTranscriptTombstoneSpans( + sessionFile, + fileSize, + targetUuid, + ) + if (spans.length === 0) return + if (spans.length === 1 && spans[0]!.end === fileSize) { + await truncateFinalTranscriptLine( + sessionFile, + spans[0]!, + fileSize, + signal, + ) + return + } + + const removalSpans = await adjustFinalUnterminatedRemovalSpan( + sessionFile, + fileSize, + spans, + ) + + await replaceFileAtomic( + sessionFile, + streamTranscriptRanges( + sessionFile, + rangesExcludingSpans(fileSize, removalSpans), + ), + { expectedTargetSize: fileSize, signal }, + ) + } catch (error) { + // Tombstones are best-effort: a missing file or failed atomic + // replacement leaves the previous transcript intact. + logForDebugging(`Tombstone removal failed: ${error}`) } }) - await writeFile(this.sessionFile, lines.join('\n'), { - encoding: 'utf8', - }) } catch { - // Silently ignore errors - the file might not exist yet + // An earlier direct append may also fail while the rewrite barrier is + // waiting for it. Preserve the historical best-effort contract. } }) } @@ -1605,6 +2144,12 @@ export async function recordContentReplacement( await getProject().insertContentReplacement(replacements, agentId) } +export async function recordSpeculationAccept( + entry: SpeculationAcceptMessage, +): Promise { + await getProject().appendEntry(entry) +} + export async function recordGoalState( goal: GoalStateEntry['goal'], sessionId: UUID = getSessionId() as UUID, @@ -1708,8 +2253,15 @@ export async function hydrateRemoteSession( const project = getProject() try { - const remoteLogs = - (await sessionIngress.getSessionLogs(sessionId, ingressUrl)) || [] + const remoteLogs = await sessionIngress.getSessionLogs( + sessionId, + ingressUrl, + ) + if (remoteLogs === null || remoteLogs.length === 0) { + logForDebugging('Remote session hydration returned no transcript') + logForDiagnosticsNoPII('error', 'hydrate_remote_session_read_fail') + return false + } // Ensure the project directory and session file exist const projectDir = getProjectDir(getOriginalCwd()) @@ -1717,10 +2269,10 @@ export async function hydrateRemoteSession( const sessionFile = getTranscriptPathForSession(sessionId) - // Replace local logs with remote logs. writeFile truncates, so no - // unlink is needed; an empty remoteLogs array produces an empty file. - const content = remoteLogs.map(e => jsonStringify(e) + '\n').join('') - await writeFile(sessionFile, content, { encoding: 'utf8', mode: 0o600 }) + await project.replaceTranscriptFile( + sessionFile, + serializeTranscriptEntries(remoteLogs), + ) logForDebugging(`Hydrated ${remoteLogs.length} entries from remote`) return remoteLogs.length > 0 @@ -1760,8 +2312,8 @@ export async function hydrateFromCCRv2InternalEvents( try { // Fetch foreground events const events = await reader() - if (!events) { - logForDebugging('Failed to read internal events for resume') + if (!events || events.length === 0) { + logForDebugging('CCR v2 hydration returned no foreground transcript') logForDiagnosticsNoPII('error', 'hydrate_ccr_v2_read_fail') return false } @@ -1769,10 +2321,12 @@ export async function hydrateFromCCRv2InternalEvents( const projectDir = getProjectDir(getOriginalCwd()) await mkdir(projectDir, { recursive: true, mode: 0o700 }) - // Write foreground transcript + // Commit the foreground transcript before reporting hydration success. const sessionFile = getTranscriptPathForSession(sessionId) - const fgContent = events.map(e => jsonStringify(e.payload) + '\n').join('') - await writeFile(sessionFile, fgContent, { encoding: 'utf8', mode: 0o600 }) + await project.replaceTranscriptFile( + sessionFile, + serializeTranscriptEntries(events.map(event => event.payload)), + ) logForDebugging( `Hydrated ${events.length} foreground entries from CCR v2 internal events`, @@ -1780,10 +2334,16 @@ export async function hydrateFromCCRv2InternalEvents( // Fetch and write subagent events let subagentEventCount = 0 + let subagentWriteFailures = 0 const subagentReader = project.getInternalSubagentEventReader() if (subagentReader) { const subagentEvents = await subagentReader() - if (subagentEvents && subagentEvents.length > 0) { + if (subagentEvents === null) { + logForDebugging('Failed to read CCR v2 subagent events for resume') + logForDiagnosticsNoPII('error', 'hydrate_ccr_v2_subagent_read_fail') + return false + } + if (subagentEvents.length > 0) { subagentEventCount = subagentEvents.length // Group by agent_id const byAgent = new Map[]>() @@ -1798,25 +2358,41 @@ export async function hydrateFromCCRv2InternalEvents( list.push(e.payload) } - // Write each agent's transcript to its own file - for (const [agentId, entries] of byAgent) { - const agentFile = getAgentTranscriptPath(asAgentId(agentId)) - await mkdir(dirname(agentFile), { recursive: true, mode: 0o700 }) - const agentContent = entries - .map(p => jsonStringify(p) + '\n') - .join('') - await writeFile(agentFile, agentContent, { - encoding: 'utf8', - mode: 0o600, - }) - } - - logForDebugging( - `Hydrated ${subagentEvents.length} subagent entries across ${byAgent.size} agents`, + // Agent files are independent commits. Queue them together so a + // failed transcript does not prevent or corrupt its siblings. + await Promise.all( + Array.from(byAgent, async ([agentId, entries]) => { + const agentFile = getAgentTranscriptPath(asAgentId(agentId)) + try { + await mkdir(dirname(agentFile), { recursive: true, mode: 0o700 }) + await project.replaceTranscriptFile( + agentFile, + serializeTranscriptEntries(entries), + ) + } catch { + subagentWriteFailures++ + logForDebugging( + 'Failed to hydrate one CCR v2 subagent transcript', + ) + logForDiagnosticsNoPII( + 'error', + 'hydrate_ccr_v2_subagent_write_fail', + ) + } + }), ) + + if (subagentWriteFailures === 0) { + logForDebugging( + `Hydrated ${subagentEvents.length} subagent entries ` + + `across ${byAgent.size} agents`, + ) + } } } + if (subagentWriteFailures > 0) return false + logForDiagnosticsNoPII('info', 'hydrate_ccr_v2_completed', { duration_ms: Date.now() - startMs, event_count: events.length, @@ -2949,12 +3525,11 @@ function appendEntryToFile( ): void { const fs = getFsImplementation() const line = jsonStringify(entry) + '\n' - try { + if (project?._deferSynchronousAppend(fullPath, line)) return + fs.mkdirSync(dirname(fullPath), { mode: 0o700 }) + withTranscriptFileLockSync(fullPath, () => { fs.appendFileSync(fullPath, line, { mode: 0o600 }) - } catch { - fs.mkdirSync(dirname(fullPath), { mode: 0o700 }) - fs.appendFileSync(fullPath, line, { mode: 0o600 }) - } + }) } /** diff --git a/src/utils/transcriptFileLock.ts b/src/utils/transcriptFileLock.ts new file mode 100644 index 000000000..77a56b4fa --- /dev/null +++ b/src/utils/transcriptFileLock.ts @@ -0,0 +1,200 @@ +import { lstatSync, readlinkSync } from 'node:fs' +import { lstat, readlink } from 'node:fs/promises' +import { dirname, isAbsolute, resolve } from 'node:path' +import { getErrnoCode } from './errors.js' +import * as lockfile from './lockfile.js' + +const TRANSCRIPT_LOCK_STALE_MS = 30_000 +const TRANSCRIPT_LOCK_WAIT_MS = 30_000 +const syncWaitBuffer = new Int32Array(new SharedArrayBuffer(4)) +const asyncHeldLockCounts = new Map() +const syncHeldLockCounts = new Map() + +async function resolveTranscriptMutationTarget( + requestedPath: string, +): Promise { + let currentPath = resolve(requestedPath) + const visited = new Set() + + for (let depth = 0; depth < 40; depth++) { + if (visited.has(currentPath)) { + throw new Error(`Cannot lock circular transcript symlink: ${requestedPath}`) + } + visited.add(currentPath) + + let fileStat + try { + fileStat = await lstat(currentPath) + } catch (error) { + if (getErrnoCode(error) === 'ENOENT') return currentPath + throw error + } + if (!fileStat.isSymbolicLink()) return currentPath + + const linkTarget = await readlink(currentPath) + currentPath = isAbsolute(linkTarget) + ? linkTarget + : resolve(dirname(currentPath), linkTarget) + } + + throw new Error(`Cannot lock transcript symlink chain: ${requestedPath}`) +} + +function resolveTranscriptMutationTargetSync(requestedPath: string): string { + let currentPath = resolve(requestedPath) + const visited = new Set() + + for (let depth = 0; depth < 40; depth++) { + if (visited.has(currentPath)) { + throw new Error(`Cannot lock circular transcript symlink: ${requestedPath}`) + } + visited.add(currentPath) + + let fileStat + try { + fileStat = lstatSync(currentPath) + } catch (error) { + if (getErrnoCode(error) === 'ENOENT') return currentPath + throw error + } + if (!fileStat.isSymbolicLink()) return currentPath + + const linkTarget = readlinkSync(currentPath) + currentPath = isAbsolute(linkTarget) + ? linkTarget + : resolve(dirname(currentPath), linkTarget) + } + + throw new Error(`Cannot lock transcript symlink chain: ${requestedPath}`) +} + +function incrementHeldLock( + counts: Map, + targetPath: string, +): void { + counts.set(targetPath, (counts.get(targetPath) ?? 0) + 1) +} + +function decrementHeldLock( + counts: Map, + targetPath: string, +): void { + const remaining = (counts.get(targetPath) ?? 1) - 1 + if (remaining === 0) counts.delete(targetPath) + else counts.set(targetPath, remaining) +} + +function asyncLockOptions( + targetPath: string, + onCompromised: (error: Error) => void, +) { + return { + lockfilePath: `${targetPath}.lock`, + realpath: false, + stale: TRANSCRIPT_LOCK_STALE_MS, + update: 5_000, + retries: { + retries: 240, + factor: 1.1, + minTimeout: 5, + maxTimeout: 250, + randomize: true, + }, + onCompromised, + } +} + +function acquireTranscriptLockSync(targetPath: string): () => void { + const deadline = Date.now() + TRANSCRIPT_LOCK_WAIT_MS + let retryDelay = 5 + + while (true) { + try { + return lockfile.lockSync(targetPath, { + lockfilePath: `${targetPath}.lock`, + realpath: false, + stale: TRANSCRIPT_LOCK_STALE_MS, + update: 5_000, + }) + } catch (error) { + if (getErrnoCode(error) !== 'ELOCKED' || Date.now() >= deadline) { + throw error + } + Atomics.wait(syncWaitBuffer, 0, 0, retryDelay) + retryDelay = Math.min(retryDelay * 2, 100) + } + } +} + +/** Serialize a complete transcript mutation with writers in other processes. */ +export async function withTranscriptFileLock( + requestedPath: string, + operation: (signal: AbortSignal) => Promise, +): Promise { + const targetPath = await resolveTranscriptMutationTarget(requestedPath) + const controller = new AbortController() + const release = await lockfile.lock( + targetPath, + asyncLockOptions(targetPath, error => controller.abort(error)), + ) + incrementHeldLock(asyncHeldLockCounts, targetPath) + try { + controller.signal.throwIfAborted() + return await operation(controller.signal) + } finally { + try { + try { + await release() + } catch (error) { + if ( + !controller.signal.aborted || + getErrnoCode(error) !== 'ERELEASED' + ) { + throw error + } + } + } finally { + decrementHeldLock(asyncHeldLockCounts, targetPath) + } + } +} + +/** Synchronous counterpart for shutdown and metadata append call sites. */ +export function withTranscriptFileLockSync( + requestedPath: string, + operation: () => T, +): T { + const targetPath = resolveTranscriptMutationTargetSync(requestedPath) + if ((syncHeldLockCounts.get(targetPath) ?? 0) > 0) return operation() + + const release = acquireTranscriptLockSync(targetPath) + incrementHeldLock(syncHeldLockCounts, targetPath) + try { + return operation() + } finally { + try { + release() + } finally { + decrementHeldLock(syncHeldLockCounts, targetPath) + } + } +} + +/** Return whether this process currently holds an async mutation lock. */ +export function isTranscriptFileLockHeldByAsyncOperation( + requestedPath: string, +): boolean { + const targetPath = resolveTranscriptMutationTargetSync(requestedPath) + return (asyncHeldLockCounts.get(targetPath) ?? 0) > 0 +} + +/** @internal Verify exact lock coverage in deterministic concurrency tests. */ +export async function isTranscriptFileLockHeldForTesting( + requestedPath: string, +): Promise { + const targetPath = await resolveTranscriptMutationTarget(requestedPath) + return ( + (asyncHeldLockCounts.get(targetPath) ?? 0) > 0 || + (syncHeldLockCounts.get(targetPath) ?? 0) > 0 + ) +}