From 98e06d7ba088965c938aa76d336927e181db3594 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 5 Oct 2026 23:03:20 +0200 Subject: [PATCH] fix(server): bound Claude replay output and index legacy history Apply the existing tool content limit before replay retains events, copying truncated prefixes so sliced strings do not pin the original output. Index each child once and own transcript snapshots through ingestion to preserve history across provider file deletion. Keep full-scale memory reproductions opt-in. --- .../agent/agent-timeline-content.test.ts | 23 ++++ .../server/agent/agent-timeline-content.ts | 13 +- .../claude/agent.history-load.test.ts | 45 ++++-- .../claude/agent.sub-agent-replay.test.ts | 101 ++++++++++++-- .../server/agent/providers/claude/agent.ts | 18 +-- .../providers/claude/replay-memory.fixture.ts | 28 +++- .../providers/claude/transcript-history.ts | 128 +++++++++--------- 7 files changed, 257 insertions(+), 99 deletions(-) diff --git a/packages/server/src/server/agent/agent-timeline-content.test.ts b/packages/server/src/server/agent/agent-timeline-content.test.ts index c40b712fad..1ccb78c67a 100644 --- a/packages/server/src/server/agent/agent-timeline-content.test.ts +++ b/packages/server/src/server/agent/agent-timeline-content.test.ts @@ -3,6 +3,29 @@ import { describe, expect, test } from "vitest"; import { limitAgentTimelineItemContent } from "./agent-timeline-content.js"; describe("agent timeline content", () => { + test.each([false, true])( + "preserves the exact UTF-16 prefix for oversized shell output (failed=%s)", + (failed) => { + const text = "x".repeat(64 * 1024 - 1) + "🙂"; + const item = limitAgentTimelineItemContent({ + type: "tool_call", + callId: "large-shell", + name: "Bash", + status: failed ? "failed" : "completed", + detail: { type: "shell", command: "echo large", output: text }, + error: failed ? { content: text } : null, + }); + expect(item).toEqual({ + type: "tool_call", + callId: "large-shell", + name: "Bash", + status: failed ? "failed" : "completed", + detail: { type: "shell", command: "echo large", output: text.slice(0, 64 * 1024) }, + error: failed ? { content: text.slice(0, 64 * 1024) } : null, + }); + }, + ); + test("marks truncated plain-text tool content within the content budget", () => { const oversizedInput = "x".repeat(64 * 1024 + 1); diff --git a/packages/server/src/server/agent/agent-timeline-content.ts b/packages/server/src/server/agent/agent-timeline-content.ts index d9bb08c0bc..003239b283 100644 --- a/packages/server/src/server/agent/agent-timeline-content.ts +++ b/packages/server/src/server/agent/agent-timeline-content.ts @@ -11,6 +11,12 @@ export function assertPluginTimelineDataSize(data: JsonValue): void { } } +/** A sliced string can keep the entire provider output alive. Copy only the retained UTF-16 + * units, including an unmatched surrogate at the existing character limit. */ +function retainTextPrefix(text: string, length: number): string { + return Buffer.from(text.slice(0, length), "utf16le").toString("utf16le"); +} + function limitFailedShellError(item: AgentTimelineItem): AgentTimelineItem { if ( item.type !== "tool_call" || @@ -28,7 +34,7 @@ function limitFailedShellError(item: AgentTimelineItem): AgentTimelineItem { ...item, error: { ...item.error, - content: item.error.content.slice(0, TOOL_CALL_CONTENT_MAX_LENGTH), + content: retainTextPrefix(item.error.content, TOOL_CALL_CONTENT_MAX_LENGTH), }, }; } @@ -47,7 +53,8 @@ function limitPlainText(item: AgentTimelineItem): AgentTimelineItem { ...item, detail: { ...item.detail, - text: item.detail.text.slice(0, TOOL_CALL_CONTENT_MAX_LENGTH - notice.length) + notice, + text: + retainTextPrefix(item.detail.text, TOOL_CALL_CONTENT_MAX_LENGTH - notice.length) + notice, }, }; } @@ -69,7 +76,7 @@ export function limitAgentTimelineItemContent(item: AgentTimelineItem): AgentTim ...item, detail: { ...item.detail, - output: item.detail.output.slice(0, TOOL_CALL_CONTENT_MAX_LENGTH), + output: retainTextPrefix(item.detail.output, TOOL_CALL_CONTENT_MAX_LENGTH), }, }; } 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 1253610350..01f743cb71 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 @@ -6,7 +6,7 @@ import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { ClaudeAgentClient } from "./agent.js"; import { claudeProjectDirSync } from "./project-dir.js"; -import { readClaudeReplayHistory } from "./transcript-history.js"; +import { withClaudeReplayHistory } from "./transcript-history.js"; /** * A resumed Claude agent whose transcript is missing or unreadable opens with an empty timeline. @@ -92,13 +92,38 @@ describe("ClaudeAgentSession persisted history load", () => { ); }); - test("propagates read errors when a transcript disappears before replay", () => { - const file = transcriptPath(); - mkdirSync(path.dirname(file), { recursive: true }); - writeFileSync(file, ""); - const history = readClaudeReplayHistory(file); - rmSync(file); - - expect(() => [...history.parentEntries]).toThrow("ENOENT"); - }); + test.each(["parent", "child", "workflow"])( + "preserves indexed %s history when its transcript disappears between passes", + (target) => { + const parent = transcriptPath(); + const sessionDirectory = parent.slice(0, -".jsonl".length); + const child = path.join(sessionDirectory, "subagents", "agent-child.jsonl"); + const workflow = path.join( + sessionDirectory, + "subagents", + "workflows", + "wf_test", + "agent-child.jsonl", + ); + const parentEntry = { + type: "assistant", + message: { content: [{ type: "text", text: "parent" }] }, + }; + const childEntry = { ...parentEntry, isSidechain: true, agentId: "child" }; + mkdirSync(path.dirname(workflow), { recursive: true }); + writeFileSync(parent, JSON.stringify(parentEntry)); + writeFileSync(child, JSON.stringify(childEntry)); + writeFileSync(workflow, JSON.stringify(childEntry)); + withClaudeReplayHistory(parent, (history) => { + rmSync({ parent, child, workflow }[target]); + // Ownership facts and visible payloads are separate passes over the same source. + for (let pass = 0; pass < 2; pass++) { + expect([...history.parentEntries]).toEqual([parentEntry]); + expect(history.subagents).toHaveLength(1); + expect([...history.subagents[0]!.entries]).toEqual([childEntry]); + expect([...history.workflowEntriesByRunId.get("wf_test")!]).toEqual([childEntry]); + } + }); + }, + ); }); diff --git a/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts b/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts index 3290e819ce..db6690beb1 100644 --- a/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts @@ -435,31 +435,112 @@ describe("ClaudeAgentSession persisted subagent replay", () => { expect(upserts(await replayDescriptors())).toEqual([]); }); + test("replays several legacy children interleaved in the parent transcript", async () => { + const lines: string[] = []; + for (let index = 0; index < 8; index++) { + const agentId = `legacy-${index}`; + const toolUseId = `task-${index}`; + lines.push(parentEntry([{ type: "tool_use", id: toolUseId, name: "Task", input: {} }])); + lines.push( + JSON.stringify({ + type: "user", + message: { + content: [ + { type: "tool_result", tool_use_id: toolUseId, content: `agentId: ${agentId}\ndone` }, + ], + }, + }), + ); + lines.push(sidechainEntry({ agentId, stopReason: "end_turn" })); + } + writeParentSession(lines); + const events = await replayDescriptors(); + const histories = events + .map((event) => event.event) + .filter((event) => event.type === "timeline"); + expect(histories.map((event) => event.id).sort()).toEqual( + Array.from({ length: 8 }, (_, index) => `task-${index}`), + ); + expect(histories.map((event) => event.item)).toEqual( + Array.from({ length: 8 }, () => ({ type: "assistant_message", text: "summary of the docs" })), + ); + }); + + test("limits retained child shell output to the timeline's existing content limit", async () => { + const output = "x".repeat(128 * 1024); + writeSession({ + parentLines: [taskToolUse(), taskToolResult()], + meta: JSON.stringify({ toolUseId: TOOL_USE_ID }), + sidechainLines: [ + JSON.stringify({ + type: "assistant", + isSidechain: true, + agentId: AGENT_ID, + message: { + content: [ + { + type: "tool_use", + id: "bash-large", + name: "Bash", + input: { command: "echo large" }, + }, + ], + }, + }), + JSON.stringify({ + type: "user", + isSidechain: true, + agentId: AGENT_ID, + message: { + content: [{ type: "tool_result", tool_use_id: "bash-large", content: output }], + }, + }), + ], + }); + const events = await replayDescriptors(); + const completed = events + .map((event) => event.event) + .filter( + (event) => + event.type === "timeline" && + event.item.type === "tool_call" && + event.item.status === "completed", + ); + expect(completed).toEqual([ + expect.objectContaining({ + type: "timeline", + id: TOOL_USE_ID, + item: expect.objectContaining({ + detail: expect.objectContaining({ type: "shell", output: output.slice(0, 64 * 1024) }), + }), + }), + ]); + }); + test.each([ { - label: "395 MiB parent and 269 sidechains", + label: "4 MiB parent and 20 sidechains", mode: "children", - generate: generateReplayCorpus, - expected: { parentItems: 539, completed: 269, histories: 269 }, + generate: (root: string) => + generateReplayCorpus(root, { parentMiB: 4, childrenMiB: 4, children: 20 }), + expected: { parentItems: 41, completed: 20, histories: 20 }, }, { - label: "five 320 MiB workflow sidechains", + label: "five 2 MiB workflow sidechains", mode: "workflow", - generate: generateWorkflowReplayCorpus, + generate: (root: string) => generateWorkflowReplayCorpus(root, 2), expected: { parentItems: 3, completed: 1, histories: 5 }, }, ])( - "opens $label with all history under a 512 MB heap", + "opens $label with all history under a 256 MB heap", ({ mode, generate, expected }) => { - const cache = path.join(os.homedir(), ".cache", "paseo-5820"); - mkdirSync(cache, { recursive: true }); - const root = mkdtempSync(path.join(cache, "repro-")); + const root = mkdtempSync(path.join(os.tmpdir(), "claude-replay-")); try { generate(root); const worker = spawnSync( process.execPath, [ - "--max-old-space-size=512", + "--max-old-space-size=256", "--import", "tsx", fileURLToPath(new URL("./replay-memory.fixture.ts", import.meta.url)), diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 0a2f8df5d8..3b1b0b7589 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -43,7 +43,7 @@ import { resolveClaudeDisabledThinkingForModel, } from "./model-manifest.js"; import { parsePartialJsonObject } from "./partial-json.js"; -import { readClaudeReplayHistory } from "./transcript-history.js"; +import { withClaudeReplayHistory, type ClaudeReplayHistory } from "./transcript-history.js"; import { ClaudeSidechainTracker } from "./sidechain-tracker.js"; import { ClaudeTaskState } from "./task-state.js"; import { @@ -53,6 +53,7 @@ import { import { observeReplaySubagents, type ClaudeReplayParentFacts } from "./subagents/replay-source.js"; import { foldSubagentObservations, type SubagentObservation } from "./subagents/observation.js"; import { observeReplayWorkflows } from "./subagents/workflow-replay-source.js"; +import { limitAgentTimelineItemContent } from "../../agent-timeline-content.js"; import { readClaudeWorkflowResultFile } from "./subagents/workflow-output.js"; import { buildClaudeFeatures, claudeModelSupportsFastMode } from "./feature-definitions.js"; import { @@ -4929,9 +4930,10 @@ class ClaudeAgentSession implements AgentSession { ); return; } - const history = readClaudeReplayHistory(historyPath); - const replay = this.ingestPersistedSidechains(history); - this.ingestPersistedHistory(history.parentEntries, replay); + withClaudeReplayHistory(historyPath, (history) => { + const replay = this.ingestPersistedSidechains(history); + this.ingestPersistedHistory(history.parentEntries, replay); + }); } catch (error) { this.logger.warn( { err: error, sessionId, historyPath }, @@ -4955,9 +4957,7 @@ class ClaudeAgentSession implements AgentSession { } } - private ingestPersistedSidechains( - history: ReturnType, - ): ClaudeReplayOwnership { + private ingestPersistedSidechains(history: ClaudeReplayHistory): ClaudeReplayOwnership { const parentFacts = readClaudeReplayParentFacts(history.parentEntries); // Replay produces the same observations the live task protocol produces, then folds them @@ -5092,7 +5092,9 @@ class ClaudeAgentSession implements AgentSession { } private convertHistoryEntry(entry: ClaudeHistoryEntry): AgentTimelineItem[] { - return convertClaudeHistoryEntry(entry, (content) => this.mapBlocksToTimeline(content)); + return convertClaudeHistoryEntry(entry, (content) => this.mapBlocksToTimeline(content)).map( + limitAgentTimelineItemContent, + ); } // Maps Claude content blocks into AgentTimelineItems. diff --git a/packages/server/src/server/agent/providers/claude/replay-memory.fixture.ts b/packages/server/src/server/agent/providers/claude/replay-memory.fixture.ts index e37154f134..cab987e473 100644 --- a/packages/server/src/server/agent/providers/claude/replay-memory.fixture.ts +++ b/packages/server/src/server/agent/providers/claude/replay-memory.fixture.ts @@ -5,7 +5,11 @@ import pino from "pino"; import { ClaudeAgentClient } from "./agent.js"; import { claudeProjectDirSync } from "./project-dir.js"; -export function generateReplayCorpus(root: string): { cwd: string; configDir: string } { +/** Full-scale reproduction is opt-in: run this file with ROOT children --generate. */ +export function generateReplayCorpus( + root: string, + size = { parentMiB: 395, childrenMiB: 362, children: 269 }, +): { cwd: string; configDir: string } { const cwd = path.join(root, "repo"); const configDir = path.join(root, "claude"); mkdirSync(cwd, { recursive: true }); @@ -18,11 +22,11 @@ export function generateReplayCorpus(root: string): { cwd: string; configDir: st // Generate those incrementally: the fixture generator itself must not hold the corpus. const padding = row({ type: "progress", data: "x".repeat(64 * 1024) }); writeFileSync(parent, ""); - for (let i = 0; i < Math.ceil((395 * 1024 * 1024) / padding.length); i++) { + for (let i = 0; i < Math.ceil((size.parentMiB * 1024 * 1024) / padding.length); i++) { appendFileSync(parent, padding); } appendFileSync(parent, row({ type: "user", message: { content: "Open the large session" } })); - for (let i = 0; i < 269; i++) { + for (let i = 0; i < size.children; i++) { const agentId = `child-${i}`; const id = `task-${i}`; appendFileSync( @@ -50,7 +54,11 @@ export function generateReplayCorpus(root: string): { cwd: string; configDir: st agentId, message: { content: "x".repeat(64 * 1024) }, }); - for (let j = 0; j < Math.ceil((362 * 1024 * 1024) / 269 / summary.length); j++) + for ( + let j = 0; + j < Math.ceil((size.childrenMiB * 1024 * 1024) / size.children / summary.length); + j++ + ) appendFileSync(file, summary); appendFileSync( file, @@ -75,6 +83,11 @@ export function generateReplayCorpus(root: string): { cwd: string; configDir: st if (process.argv[1] && path.resolve(process.argv[1]) === fileURLToPath(import.meta.url)) { const root = process.argv[2]; + if (!root) throw new Error("Usage: replay-memory.fixture.ts ROOT children|workflow [--generate]"); + if (process.argv.includes("--generate")) { + if (process.argv[3] === "workflow") generateWorkflowReplayCorpus(root); + else generateReplayCorpus(root); + } process.env.CLAUDE_CONFIG_DIR = path.join(root, "claude"); const client = new ClaudeAgentClient({ logger: pino({ level: "silent" }), @@ -116,7 +129,10 @@ if (process.argv[1] && path.resolve(process.argv[1]) === fileURLToPath(import.me ); } -export function generateWorkflowReplayCorpus(root: string): { cwd: string; configDir: string } { +export function generateWorkflowReplayCorpus( + root: string, + childMiB = 320, +): { cwd: string; configDir: string } { const cwd = path.join(root, "repo"); const configDir = path.join(root, "claude"); mkdirSync(cwd, { recursive: true }); @@ -166,7 +182,7 @@ export function generateWorkflowReplayCorpus(root: string): { cwd: string; confi agentId: `child-${i}`, message: { content: "x".repeat(64 * 1024) }, }) + "\n"; - for (let j = 0; j < Math.ceil((320 * 1024 * 1024) / prompt.length); j++) + for (let j = 0; j < Math.ceil((childMiB * 1024 * 1024) / prompt.length); j++) appendFileSync(file, prompt); appendFileSync( file, 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 8cddab5ddd..9652ab6d32 100644 --- a/packages/server/src/server/agent/providers/claude/transcript-history.ts +++ b/packages/server/src/server/agent/providers/claude/transcript-history.ts @@ -1,5 +1,6 @@ import fs from "node:fs"; import path from "node:path"; +import os from "node:os"; import { normalizeProviderReplayTimestamp } from "../../provider-history-timestamps.js"; import { parseClaudeSubagentMeta, @@ -85,55 +86,35 @@ function* locatedRecords( } } -/** Repeatable sources let ownership be resolved before replay without retaining raw records. */ -function records(file: string): Iterable { +/** Indexed sources parse only their own records from the replay-owned snapshots. */ +function indexedRecords(locations: readonly RecordLocation[]): Iterable { return { *[Symbol.iterator]() { - for (const { entry } of locatedRecords(file)) yield entry; - }, - }; -} - -/** Workflow children share a timeline. Sort byte locations, then parse in that order so tool - * results see their declarations even when directory order differs from chronological order. */ -function workflowRecords(files: readonly string[]): Iterable { - return { - *[Symbol.iterator]() { - const locations: RecordLocation[] = []; - for (const file of files) - for (const { entry, location } of locatedRecords(file)) { - if (entry.type !== "user" || isToolResult(entry)) locations.push(location); - } - locations.sort((a, b) => { - if (!a.timestamp && !b.timestamp) return 0; - if (!a.timestamp) return 1; - if (!b.timestamp) return -1; - return Date.parse(a.timestamp) - Date.parse(b.timestamp); - }); - let currentFile: string | null = null; let fd: number | null = null; + let currentFile: string | null = null; try { for (const location of locations) { - if (fd === null || currentFile !== location.file) { + if (currentFile !== location.file) { if (fd !== null) fs.closeSync(fd); fd = null; + fd = fs.openSync(location.file, "r"); currentFile = location.file; - fd = fs.openSync(currentFile, "r"); } const buffer = Buffer.allocUnsafe(location.length); let read = 0; while (read < buffer.length) { const count = fs.readSync( - fd, + fd!, buffer, read, buffer.length - read, location.offset + read, ); - if (count === 0) break; + if (count === 0) + throw new Error(`Claude transcript was truncated during replay: ${location.file}`); read += count; } - const entry = parseRecord(buffer.subarray(0, read).toString("utf8")); + const entry = parseRecord(buffer.toString("utf8")); if (entry) yield entry; } } finally { @@ -161,17 +142,6 @@ function parseRecord(line: string): TranscriptEntry | null { } } -function selectRecords( - files: readonly string[], - accept: (entry: TranscriptEntry) => boolean, -): Iterable { - return { - *[Symbol.iterator]() { - for (const file of files) for (const entry of records(file)) if (accept(entry)) yield entry; - }, - }; -} - function readOptionalFile(file: string): string | null { try { return fs.readFileSync(file, "utf8"); @@ -182,7 +152,7 @@ function readOptionalFile(file: string): string | null { /** Discover sources without retaining transcript contents. Unrelated sidechains stay excluded * by the replay ownership resolver; workflow children retain their separate run ownership. */ -export function readClaudeReplayHistory(historyPath: string): { +export interface ClaudeReplayHistory { parentEntries: Iterable; subagents: { agentId: string; @@ -191,7 +161,23 @@ export function readClaudeReplayHistory(historyPath: string): { }[]; workflows: ClaudeWorkflowRun[]; 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. */ +export function withClaudeReplayHistory( + historyPath: string, + consume: (history: ClaudeReplayHistory) => T, +): T { + const snapshotDirectory = fs.mkdtempSync(path.join(os.tmpdir(), "claude-replay-")); + try { + return consume(indexHistory(historyPath, snapshotDirectory)); + } finally { + fs.rmSync(snapshotDirectory, { recursive: true, force: true }); + } +} + +function indexHistory(historyPath: string, snapshotDirectory: string): ClaudeReplayHistory { const sessionDirectory = path.join( path.dirname(historyPath), path.basename(historyPath, ".jsonl"), @@ -228,33 +214,51 @@ export function readClaudeReplayHistory(historyPath: string): { if (meta) metaByAgentId.set(agentId, meta); } } - const filesByAgentId = new Map(); + 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); + } for (const file of sidechainFiles) { - const agentsInFile = new Set(); - for (const entry of records(file)) { - if (entry.isSidechain === true && typeof entry.agentId === "string") - agentsInFile.add(entry.agentId); - } - for (const agentId of agentsInFile) { - const files = filesByAgentId.get(agentId) ?? []; - files.push(file); - filesByAgentId.set(agentId, files); - } + scan(file, (entry, location) => { + if (file === historyPath && entry.isSidechain !== true) parentLocations.push(location); + if (entry.isSidechain !== true || typeof entry.agentId !== "string") return; + const locations = locationsByAgentId.get(entry.agentId) ?? []; + locations.push(location); + locationsByAgentId.set(entry.agentId, locations); + }); + } + const workflowEntriesByRunId = new Map>(); + for (const [runId, files] of workflowFiles) { + const locations: RecordLocation[] = []; + for (const file of files) + scan(file, (entry, location) => { + if (entry.type !== "user" || isToolResult(entry)) locations.push(location); + }); + locations.sort((a, b) => { + if (!a.timestamp && !b.timestamp) return 0; + if (!a.timestamp) return 1; + if (!b.timestamp) return -1; + return Date.parse(a.timestamp) - Date.parse(b.timestamp); + }); + workflowEntriesByRunId.set(runId, indexedRecords(locations)); } return { - parentEntries: selectRecords([historyPath], (entry) => entry.isSidechain !== true), - subagents: [...filesByAgentId].map(([agentId, files]) => ({ + parentEntries: indexedRecords(parentLocations), + subagents: [...locationsByAgentId].map(([agentId, locations]) => ({ agentId, meta: metaByAgentId.get(agentId) ?? null, - entries: selectRecords( - files, - (entry) => entry.isSidechain === true && entry.agentId === agentId, - ), + entries: indexedRecords(locations), })), workflows: readWorkflows(sessionDirectory), - workflowEntriesByRunId: new Map( - [...workflowFiles].map(([runId, files]) => [runId, workflowRecords(files)]), - ), + workflowEntriesByRunId, }; }