diff --git a/src/main/native-chat/ssh-transcript-host-routing.test.ts b/src/main/native-chat/ssh-transcript-host-routing.test.ts index f48f2f2255c..30b10a2034d 100644 --- a/src/main/native-chat/ssh-transcript-host-routing.test.ts +++ b/src/main/native-chat/ssh-transcript-host-routing.test.ts @@ -285,7 +285,6 @@ describe('native chat SSH transcript host routing (#13663)', () => { return { size: bytes.length, identity: '1:7:11', - identityReliable: true, mtimeMs: 1, ctimeMs: 1 } @@ -770,7 +769,18 @@ describe('native chat SSH transcript host routing (#13663)', () => { const transcriptPath = '/tmp/large-append-session.jsonl' const initial = Buffer.from(claudeLine('initial', 'first')) const files = new Map([[transcriptPath, initial]]) - const { provider, readFile, readFileRange } = memoryProvider(files) + const { provider, readFile, readFileRange, stat } = memoryProvider(files) + stat.mockImplementation(async (filePath: string): Promise => { + const bytes = files.get(filePath)! + return { + size: bytes.length, + type: 'file', + mtime: bytes.length, + mtimeMs: bytes.length, + dev: 7, + ino: 11 + } + }) mocks.getProvider.mockReturnValue(provider) const appended: string[] = [] const replacements: string[][] = [] @@ -787,13 +797,24 @@ describe('native chat SSH transcript host routing (#13663)', () => { try { await vi.waitFor(() => expect(readFileRange).toHaveBeenCalled()) readFileRange.mockClear() + const partial = Buffer.from('{"type":') + files.set(transcriptPath, Buffer.concat([initial, partial])) + await vi.waitFor(() => + expect(readFileRange).toHaveBeenCalledWith( + transcriptPath, + initial.length, + partial.length, + expect.anything() + ) + ) + readFileRange.mockClear() const append = Buffer.from( Array.from({ length: 3_000 }, (_, index) => claudeLine(`append-${index}`, `bounded remote append ${index}`) ).join('') ) expect(append.length).toBeGreaterThan(MAX_FILE_RANGE_READ_BYTES) - files.set(transcriptPath, Buffer.concat([initial, append])) + files.set(transcriptPath, Buffer.concat([initial, partial, append])) await vi.waitFor(() => expect(replacements.at(-1)).toContain('append-2999')) expect(replacements.at(-1)).toHaveLength(40) @@ -803,6 +824,9 @@ describe('native chat SSH transcript host routing (#13663)', () => { expect(requestedLengths.reduce((sum, length) => sum + length, 0)).toBeLessThanOrEqual( append.length + 256 ) + const afterReplacement = Buffer.from(claudeLine('after-replacement', 'still live')) + files.set(transcriptPath, Buffer.concat([initial, partial, append, afterReplacement])) + await vi.waitFor(() => expect(appended).toContain('after-replacement'), { timeout: 3_000 }) expect(readFile).not.toHaveBeenCalled() } finally { subscription.unsubscribe() diff --git a/src/main/native-chat/ssh-transcript-range-fs.test.ts b/src/main/native-chat/ssh-transcript-range-fs.test.ts index d81cb78458e..d0113912d76 100644 --- a/src/main/native-chat/ssh-transcript-range-fs.test.ts +++ b/src/main/native-chat/ssh-transcript-range-fs.test.ts @@ -39,9 +39,11 @@ function mutableProvider(initial: string) { } as unknown as IFilesystemProvider return { provider, - replace(next: string) { + replace(next: string, touch = true) { bytes = Buffer.from(next) - mtimeMs++ + if (touch) { + mtimeMs++ + } } } } @@ -53,13 +55,12 @@ beforeEach(() => { describe('SSH transcript range stability', () => { it.each([ ['same-length rewrite', 'new generation'], - ['growing rewrite', 'new generation with more bytes'], - ['concurrent append', 'old generation append'] + ['growing rewrite', 'new generation with more bytes'] ])('invalidates a %s', async (_label, replacement) => { const remote = mutableProvider('old generation') mocks.provider = remote.provider const rangeFs = await createSshTranscriptRangeFs('ssh-owner') - const openingStamp = await rangeFs.stat(TRANSCRIPT_PATH) + const openingStamp = await rangeFs.stat(TRANSCRIPT_PATH, undefined, true) await rangeFs.read(TRANSCRIPT_PATH, 0, 3) remote.replace(replacement) @@ -68,4 +69,28 @@ describe('SSH transcript range stability', () => { TranscriptRangeReadInvalidatedError ) }) + + it('accepts a concurrent append that preserves the opening boundary', async () => { + const remote = mutableProvider('old generation') + mocks.provider = remote.provider + const rangeFs = await createSshTranscriptRangeFs('ssh-owner') + const openingStamp = await rangeFs.stat(TRANSCRIPT_PATH, undefined, true) + + remote.replace('old generation append') + + await expect(rangeFs.assertStable(TRANSCRIPT_PATH, openingStamp)).resolves.toBeUndefined() + }) + + it('rejects a growing rewrite even when remote mtime is coarse', async () => { + const remote = mutableProvider('old generation') + mocks.provider = remote.provider + const rangeFs = await createSshTranscriptRangeFs('ssh-owner') + const openingStamp = await rangeFs.stat(TRANSCRIPT_PATH, undefined, true) + + remote.replace('new generation with more bytes', false) + + await expect(rangeFs.assertStable(TRANSCRIPT_PATH, openingStamp)).rejects.toBeInstanceOf( + TranscriptRangeReadInvalidatedError + ) + }) }) diff --git a/src/main/native-chat/ssh-transcript-range-fs.ts b/src/main/native-chat/ssh-transcript-range-fs.ts index 679dcdc438d..92bb5e33cc2 100644 --- a/src/main/native-chat/ssh-transcript-range-fs.ts +++ b/src/main/native-chat/ssh-transcript-range-fs.ts @@ -13,6 +13,8 @@ type ProviderGeneration = { generation: number } +const STABILITY_BOUNDARY_BYTES = 64 + function requireSameProvider(connectionId: string, expected: ProviderGeneration): void { const current = getSshFilesystemProviderSnapshot(connectionId) if ( @@ -41,49 +43,85 @@ export async function createSshTranscriptRangeFs( const providerSnapshot: ProviderGeneration = snapshot requireSameProvider(connectionId, providerSnapshot) const provider = providerSnapshot.provider - async function stat(filePath: string, statSignal?: AbortSignal): Promise { + async function read( + filePath: string, + position: number, + length: number, + readSignal?: AbortSignal + ) { + readSignal?.throwIfAborted() + const parts: Buffer[] = [] + let remaining = length + let cursor = position + while (remaining > 0) { + requireSameProvider(connectionId, providerSnapshot) + const window = Math.min(remaining, MAX_FILE_RANGE_READ_BYTES) + const result = await provider.readFileRange!(filePath, cursor, window, { + signal: readSignal + }) + requireSameProvider(connectionId, providerSnapshot) + const bytes = result.bytes.subarray(0, result.bytesRead) + parts.push(bytes) + cursor += bytes.length + remaining -= bytes.length + if (bytes.length < window) { + break + } + } + return parts.length === 1 ? parts[0]! : Buffer.concat(parts) + } + async function stat( + filePath: string, + statSignal?: AbortSignal, + captureBoundary = false + ): Promise { statSignal?.throwIfAborted() requireSameProvider(connectionId, providerSnapshot) const stamp = await provider.stat(filePath, { signal: statSignal }) requireSameProvider(connectionId, providerSnapshot) const mtimeMs = stamp.mtimeMs ?? stamp.mtime + const boundaryLength = Math.min(stamp.size, STABILITY_BOUNDARY_BYTES) + const boundaryFingerprint = + captureBoundary && boundaryLength > 0 + ? (await read(filePath, stamp.size - boundaryLength, boundaryLength, statSignal)).toString( + 'base64' + ) + : undefined return { size: stamp.size, identity: `${providerSnapshot.generation}:${stamp.dev ?? 0}:${stamp.ino ?? 0}`, - identityReliable: stamp.dev !== undefined && stamp.ino !== undefined, mtimeMs, - ctimeMs: mtimeMs + ctimeMs: mtimeMs, + ...(boundaryFingerprint ? { boundaryFingerprint } : {}) } } const rangeFs: TranscriptRangeFs = { stat, - async read(filePath, position, length, readSignal) { - readSignal?.throwIfAborted() - const parts: Buffer[] = [] - let remaining = length - let cursor = position - while (remaining > 0) { - requireSameProvider(connectionId, providerSnapshot) - const window = Math.min(remaining, MAX_FILE_RANGE_READ_BYTES) - const result = await provider.readFileRange!(filePath, cursor, window, { - signal: readSignal - }) - requireSameProvider(connectionId, providerSnapshot) - const bytes = result.bytes.subarray(0, result.bytesRead) - parts.push(bytes) - cursor += bytes.length - remaining -= bytes.length - if (bytes.length < window) { - break - } - } - return parts.length === 1 ? parts[0]! : Buffer.concat(parts) - }, + read, async assertStable(filePath, openingStamp, stableSignal) { const closingStamp = await stat(filePath, stableSignal) const identityChanged = closingStamp.identity !== openingStamp.identity const versionChanged = closingStamp.mtimeMs !== openingStamp.mtimeMs - if (identityChanged || versionChanged) { + if (identityChanged || closingStamp.size < openingStamp.size) { + throw new TranscriptRangeReadInvalidatedError() + } + if (closingStamp.size === openingStamp.size) { + if (versionChanged) { + throw new TranscriptRangeReadInvalidatedError() + } + return + } + if (!openingStamp.boundaryFingerprint) { + throw new TranscriptRangeReadInvalidatedError() + } + const boundaryLength = Math.min(openingStamp.size, STABILITY_BOUNDARY_BYTES) + const boundary = await read( + filePath, + openingStamp.size - boundaryLength, + boundaryLength, + stableSignal + ) + if (boundary.toString('base64') !== openingStamp.boundaryFingerprint) { throw new TranscriptRangeReadInvalidatedError() } } diff --git a/src/main/native-chat/transcript-incremental-reader.ts b/src/main/native-chat/transcript-incremental-reader.ts index c6b625abf76..81f76a57b89 100644 --- a/src/main/native-chat/transcript-incremental-reader.ts +++ b/src/main/native-chat/transcript-incremental-reader.ts @@ -40,7 +40,7 @@ export async function readIncrementalTranscriptMessages( signal?: AbortSignal, rangeFs?: TranscriptRangeFs ): Promise { - const openingStamp = rangeFs ? await rangeFs.stat(filePath, signal) : null + const openingStamp = rangeFs ? await rangeFs.stat(filePath, signal, true) : null const end = openingStamp?.size ?? (await wslGatedStat(filePath, 'exact', signal)).size if (end <= state.offset) { return [] diff --git a/src/main/native-chat/transcript-range-fs.ts b/src/main/native-chat/transcript-range-fs.ts index 2cdd668874a..156a0b2fba5 100644 --- a/src/main/native-chat/transcript-range-fs.ts +++ b/src/main/native-chat/transcript-range-fs.ts @@ -1,9 +1,9 @@ export type TranscriptFileStamp = { size: number identity: string - identityReliable: boolean mtimeMs: number ctimeMs: number + boundaryFingerprint?: string } export class TranscriptRangeReadInvalidatedError extends Error { @@ -14,7 +14,11 @@ export class TranscriptRangeReadInvalidatedError extends Error { } export type TranscriptRangeFs = { - stat(filePath: string, signal?: AbortSignal): Promise + stat( + filePath: string, + signal?: AbortSignal, + captureBoundary?: boolean + ): Promise read(filePath: string, position: number, length: number, signal?: AbortSignal): Promise assertStable( filePath: string, diff --git a/src/main/native-chat/transcript-tail-reader.ts b/src/main/native-chat/transcript-tail-reader.ts index f46a6e3bf5c..9cb0e3ac8e7 100644 --- a/src/main/native-chat/transcript-tail-reader.ts +++ b/src/main/native-chat/transcript-tail-reader.ts @@ -73,7 +73,7 @@ export async function readNativeChatTranscriptTailFile( oversizedRecordCount?: number }> { signal?.throwIfAborted() - const openingStamp = rangeFs ? await rangeFs.stat(filePath, signal) : null + const openingStamp = rangeFs ? await rangeFs.stat(filePath, signal, true) : null const end = Math.min( openingStamp?.size ?? (await wslGatedStat(filePath, 'exact', signal)).size, endOffset ?? Number.MAX_SAFE_INTEGER diff --git a/src/main/native-chat/transcript-watch-engine.ts b/src/main/native-chat/transcript-watch-engine.ts index 58f4087c706..df0b9af026b 100644 --- a/src/main/native-chat/transcript-watch-engine.ts +++ b/src/main/native-chat/transcript-watch-engine.ts @@ -33,9 +33,7 @@ const ROTATION_RETRY_MS = 25 const MAX_ROTATION_RETRY_MS = 2_000 let activeWatcherCount = 0 -export function getActiveNativeChatWatcherCount(): number { - return activeWatcherCount -} +export const getActiveNativeChatWatcherCount = (): number => activeWatcherCount /** Installs a live tail on an already-resolved path; missing files return null. */ export async function installTranscriptWatcher( @@ -191,6 +189,7 @@ export async function installTranscriptWatcher( return } if (replacementSnapshot && onReplace) { + resetIncrementalTranscriptState(state) state.offset = replacementSnapshot.consumedTo state.pendingStart = state.offset onReplace(