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, }; }