mirror of
https://github.com/Gitlawb/openclaude.git
synced 2026-08-24 02:34:15 -05:00
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.
This commit is contained in:
Bogdan
@@ -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<string, unknown>,
|
||||
): Promise<void> {
|
||||
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 })
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
|
||||
@@ -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',
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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<string[]> {
|
||||
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<void>(resolve => {
|
||||
release = resolve
|
||||
})
|
||||
let firstWrite!: () => void
|
||||
const wroteFirstChunk = new Promise<void>(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)
|
||||
})
|
||||
@@ -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<void>
|
||||
|
||||
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<void> {
|
||||
await faultInjector?.(stage, context)
|
||||
}
|
||||
|
||||
async function resolveWriteTarget(requestedPath: string): Promise<string> {
|
||||
let currentPath = resolve(requestedPath)
|
||||
const visited = new Set<string>()
|
||||
|
||||
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<string | Uint8Array> {
|
||||
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<void> {
|
||||
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<string | Uint8Array>,
|
||||
signal: AbortSignal | undefined,
|
||||
context: AtomicReplaceFaultContext,
|
||||
): Promise<void> {
|
||||
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<void> {
|
||||
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<string | Uint8Array>,
|
||||
options: AtomicReplaceOptions = {},
|
||||
): Promise<void> {
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -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, unknown> = {}): 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<string> {
|
||||
const filePath = join(testRoot, name)
|
||||
await writeFile(filePath, content)
|
||||
switchSession(SESSION_ID as never, testRoot)
|
||||
resetProjectForTesting()
|
||||
setSessionFileForTesting(filePath)
|
||||
return filePath
|
||||
}
|
||||
|
||||
async function prepareHydration(): Promise<string> {
|
||||
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<void> | 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<void>(resolve => {
|
||||
paused = resolve
|
||||
})
|
||||
const releasePromise = new Promise<void>(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<void>(resolve => {
|
||||
paused = resolve
|
||||
})
|
||||
const releasePromise = new Promise<void>(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<void>(resolve => {
|
||||
paused = resolve
|
||||
})
|
||||
const releasePromise = new Promise<void>(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<void>(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<string, unknown> = { 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',
|
||||
)
|
||||
})
|
||||
+735
-160
File diff suppressed because it is too large
Load Diff
@@ -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<string, number>()
|
||||
const syncHeldLockCounts = new Map<string, number>()
|
||||
|
||||
async function resolveTranscriptMutationTarget(
|
||||
requestedPath: string,
|
||||
): Promise<string> {
|
||||
let currentPath = resolve(requestedPath)
|
||||
const visited = new Set<string>()
|
||||
|
||||
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<string>()
|
||||
|
||||
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<string, number>,
|
||||
targetPath: string,
|
||||
): void {
|
||||
counts.set(targetPath, (counts.get(targetPath) ?? 0) + 1)
|
||||
}
|
||||
|
||||
function decrementHeldLock(
|
||||
counts: Map<string, number>,
|
||||
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<T>(
|
||||
requestedPath: string,
|
||||
operation: (signal: AbortSignal) => Promise<T>,
|
||||
): Promise<T> {
|
||||
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<T>(
|
||||
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<boolean> {
|
||||
const targetPath = await resolveTranscriptMutationTarget(requestedPath)
|
||||
return (
|
||||
(asyncHeldLockCounts.get(targetPath) ?? 0) > 0 ||
|
||||
(syncHeldLockCounts.get(targetPath) ?? 0) > 0
|
||||
)
|
||||
}
|
||||
Reference in New Issue
Block a user