diff --git a/src/entrypoints/sdk/sessions.ts b/src/entrypoints/sdk/sessions.ts index 36d11d64fa..1913fc323a 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 86e208b0a9..e761aadc12 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 0000000000..61455effac --- /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 0000000000..1057a6a4d3 --- /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 0000000000..751a6f4ad8 --- /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 b66b947b23..6975205473 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 { - 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 }) + 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 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 + + 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 + } - for (const { entry, resolve } of batch) { - const line = jsonStringify(entry) + '\n' + 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 + } - 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() + const line = jsonStringify(operation.entry) + '\n' + if (content.length + line.length >= this.MAX_CHUNK_BYTES) { + await this.flushQueuedAppends(filePath, content, queuedAppends) + content = '' + queuedAppends = [] } - resolvers.length = 0 - content = '' + 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 - - 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) - - // 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) + 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 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() + } + + 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 0000000000..77a56b4fa3 --- /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 + ) +}