From 433a8a43500c821a9557bd766863b13d840680cb Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 5 Oct 2026 23:15:29 +0200 Subject: [PATCH] fix(server): replay Claude history without temporary transcript copies Keep the original descriptors open until synchronous replay finishes. Indexed passes survive unlink and rename without requiring a second copy of the session on temporary storage. Close every descriptor on success and failure. --- .../claude/agent.history-load.test.ts | 12 ++ .../providers/claude/transcript-history.ts | 181 ++++++++---------- 2 files changed, 92 insertions(+), 101 deletions(-) diff --git a/packages/server/src/server/agent/providers/claude/agent.history-load.test.ts b/packages/server/src/server/agent/providers/claude/agent.history-load.test.ts index 01f743cb71..b961f8cc17 100644 --- a/packages/server/src/server/agent/providers/claude/agent.history-load.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.history-load.test.ts @@ -92,6 +92,18 @@ describe("ClaudeAgentSession persisted history load", () => { ); }); + test("loads readable history when the temporary directory is unavailable", () => { + const file = transcriptPath(); + const entry = { type: "assistant", message: { content: [{ type: "text", text: "saved" }] } }; + mkdirSync(path.dirname(file), { recursive: true }); + writeFileSync(file, JSON.stringify(entry)); + const unavailable = path.join(tempRoot, "unavailable-tmp"); + for (const variable of ["TMPDIR", "TEMP", "TMP"]) vi.stubEnv(variable, unavailable); + withClaudeReplayHistory(file, (history) => { + expect([...history.parentEntries]).toEqual([entry]); + }); + }); + test.each(["parent", "child", "workflow"])( "preserves indexed %s history when its transcript disappears between passes", (target) => { diff --git a/packages/server/src/server/agent/providers/claude/transcript-history.ts b/packages/server/src/server/agent/providers/claude/transcript-history.ts index 9652ab6d32..cba606b0e4 100644 --- a/packages/server/src/server/agent/providers/claude/transcript-history.ts +++ b/packages/server/src/server/agent/providers/claude/transcript-history.ts @@ -1,6 +1,5 @@ import fs from "node:fs"; import path from "node:path"; -import os from "node:os"; import { normalizeProviderReplayTimestamp } from "../../provider-history-timestamps.js"; import { parseClaudeSubagentMeta, @@ -26,99 +25,81 @@ interface RecordLocation { /** Chunk in bytes so UTF-8 and multi-chunk records are decoded once, at a line boundary. */ function* locatedRecords( file: string, + fd: number, ): Generator<{ entry: TranscriptEntry; location: RecordLocation }> { - const fd = fs.openSync(file, "r"); - try { - const stat = fs.fstatSync(fd); - // Opening a directory succeeds on some platforms; it must never look like empty history. - if (stat.isDirectory()) { - throw Object.assign(new Error(`EISDIR: illegal operation on a directory, read '${file}'`), { - code: "EISDIR", - syscall: "read", - path: file, - }); - } - let remaining = stat.size; - let offset = 0; - let pieces: Buffer[] = []; - let length = 0; - while (remaining > 0) { - const buffer = Buffer.allocUnsafe(Math.min(64 * 1024, remaining)); - const count = fs.readSync(fd, buffer, 0, buffer.length, null); - if (count === 0) break; - remaining -= count; - let start = 0; - let end: number; - while ((end = buffer.indexOf(10, start)) !== -1 && end < count) { - pieces.push(buffer.subarray(start, end)); - length += end - start; - const entry = parseRecord(Buffer.concat(pieces, length).toString("utf8")); - const location = { - file, - offset, - length, - timestamp: normalizeProviderReplayTimestamp(entry?.timestamp), - }; - offset += length + 1; - pieces = []; - length = 0; - if (entry) yield { entry, location }; - start = end + 1; - } - if (start < count) { - pieces.push(buffer.subarray(start, count)); - length += count - start; - } - } - const entry = parseRecord(Buffer.concat(pieces, length).toString("utf8")); - if (entry) - yield { - entry, - location: { - file, - offset, - length, - timestamp: normalizeProviderReplayTimestamp(entry.timestamp), - }, - }; - } finally { - fs.closeSync(fd); + const stat = fs.fstatSync(fd); + // Opening a directory succeeds on some platforms; it must never look like empty history. + if (stat.isDirectory()) { + throw Object.assign(new Error(`EISDIR: illegal operation on a directory, read '${file}'`), { + code: "EISDIR", + syscall: "read", + path: file, + }); } + let remaining = stat.size; + let offset = 0; + let pieces: Buffer[] = []; + let length = 0; + while (remaining > 0) { + const buffer = Buffer.allocUnsafe(Math.min(64 * 1024, remaining)); + const count = fs.readSync(fd, buffer, 0, buffer.length, null); + if (count === 0) break; + remaining -= count; + let start = 0; + let end: number; + while ((end = buffer.indexOf(10, start)) !== -1 && end < count) { + pieces.push(buffer.subarray(start, end)); + length += end - start; + const entry = parseRecord(Buffer.concat(pieces, length).toString("utf8")); + const location = { + file, + offset, + length, + timestamp: normalizeProviderReplayTimestamp(entry?.timestamp), + }; + offset += length + 1; + pieces = []; + length = 0; + if (entry) yield { entry, location }; + start = end + 1; + } + if (start < count) { + pieces.push(buffer.subarray(start, count)); + length += count - start; + } + } + const entry = parseRecord(Buffer.concat(pieces, length).toString("utf8")); + if (entry) + yield { + entry, + location: { + file, + offset, + length, + timestamp: normalizeProviderReplayTimestamp(entry.timestamp), + }, + }; } -/** Indexed sources parse only their own records from the replay-owned snapshots. */ -function indexedRecords(locations: readonly RecordLocation[]): Iterable { +/** Indexed sources parse only their own records using descriptors owned by this replay. */ +function indexedRecords( + locations: readonly RecordLocation[], + descriptors: ReadonlyMap, +): Iterable { return { *[Symbol.iterator]() { - let fd: number | null = null; - let currentFile: string | null = null; - try { - for (const location of locations) { - if (currentFile !== location.file) { - if (fd !== null) fs.closeSync(fd); - fd = null; - fd = fs.openSync(location.file, "r"); - currentFile = location.file; - } - const buffer = Buffer.allocUnsafe(location.length); - let read = 0; - while (read < buffer.length) { - const count = fs.readSync( - fd!, - buffer, - read, - buffer.length - read, - location.offset + read, - ); - if (count === 0) - throw new Error(`Claude transcript was truncated during replay: ${location.file}`); - read += count; - } - const entry = parseRecord(buffer.toString("utf8")); - if (entry) yield entry; + for (const location of locations) { + const fd = descriptors.get(location.file)!; + const buffer = Buffer.allocUnsafe(location.length); + let read = 0; + while (read < buffer.length) { + const count = fs.readSync(fd, buffer, read, buffer.length - read, location.offset + read); + if (count === 0) + throw new Error(`Claude transcript was truncated during replay: ${location.file}`); + read += count; } - } finally { - if (fd !== null) fs.closeSync(fd); + const entry = parseRecord(buffer.toString("utf8")); + if (entry) yield entry; } }, }; @@ -163,21 +144,21 @@ export interface ClaudeReplayHistory { workflowEntriesByRunId: Map>; } -/** Own snapshots through every pass so unlink, rename or truncation of provider files cannot - * invalidate discovered history. Sources must be consumed before the callback returns. */ +/** Own descriptors through every pass so unlink or rename cannot invalidate discovered history. + * Sources must be consumed before the callback returns. No transcript copies are needed. */ export function withClaudeReplayHistory( historyPath: string, consume: (history: ClaudeReplayHistory) => T, ): T { - const snapshotDirectory = fs.mkdtempSync(path.join(os.tmpdir(), "claude-replay-")); + const descriptors = new Map(); try { - return consume(indexHistory(historyPath, snapshotDirectory)); + return consume(indexHistory(historyPath, descriptors)); } finally { - fs.rmSync(snapshotDirectory, { recursive: true, force: true }); + for (const fd of descriptors.values()) fs.closeSync(fd); } } -function indexHistory(historyPath: string, snapshotDirectory: string): ClaudeReplayHistory { +function indexHistory(historyPath: string, descriptors: Map): ClaudeReplayHistory { const sessionDirectory = path.join( path.dirname(historyPath), path.basename(historyPath, ".jsonl"), @@ -216,15 +197,13 @@ function indexHistory(historyPath: string, snapshotDirectory: string): ClaudeRep } const parentLocations: RecordLocation[] = []; const locationsByAgentId = new Map(); - let snapshotIndex = 0; function scan( file: string, visit: (entry: TranscriptEntry, location: RecordLocation) => void, ): void { - const snapshot = path.join(snapshotDirectory, `${snapshotIndex++}.jsonl`); - // Reflink where available; otherwise copy on disk, never into the daemon heap. - fs.copyFileSync(file, snapshot, fs.constants.COPYFILE_FICLONE); - for (const { entry, location } of locatedRecords(snapshot)) visit(entry, location); + const fd = fs.openSync(file, "r"); + descriptors.set(file, fd); + for (const { entry, location } of locatedRecords(file, fd)) visit(entry, location); } for (const file of sidechainFiles) { scan(file, (entry, location) => { @@ -248,14 +227,14 @@ function indexHistory(historyPath: string, snapshotDirectory: string): ClaudeRep if (!b.timestamp) return -1; return Date.parse(a.timestamp) - Date.parse(b.timestamp); }); - workflowEntriesByRunId.set(runId, indexedRecords(locations)); + workflowEntriesByRunId.set(runId, indexedRecords(locations, descriptors)); } return { - parentEntries: indexedRecords(parentLocations), + parentEntries: indexedRecords(parentLocations, descriptors), subagents: [...locationsByAgentId].map(([agentId, locations]) => ({ agentId, meta: metaByAgentId.get(agentId) ?? null, - entries: indexedRecords(locations), + entries: indexedRecords(locations, descriptors), })), workflows: readWorkflows(sessionDirectory), workflowEntriesByRunId,