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 50d9c56ab5..a8237d87d0 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 @@ -1,4 +1,4 @@ -import { chmodSync, mkdirSync, mkdtempSync, rmSync } from "node:fs"; +import { chmodSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; import * as os from "node:os"; import * as path from "node:path"; import pino from "pino"; @@ -6,6 +6,7 @@ import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { ClaudeAgentClient } from "./agent.js"; import { claudeProjectDirSync } from "./project-dir.js"; +import { withClaudeReplayHistory } from "./transcript-history.js"; /** * A resumed Claude agent whose transcript is missing or unreadable opens with an empty timeline. @@ -91,6 +92,53 @@ 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) => { + 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]); + } + }); + }, + ); + // chmod does not remove read access on Windows, or for root. test.skipIf(process.platform === "win32" || process.getuid?.() === 0)( "logs a warning with the error when Claude's project folders cannot be listed", 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 48ddab5973..8f2ba60907 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 @@ -1,6 +1,9 @@ import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import * as os from "node:os"; import * as path from "node:path"; +import { spawnSync } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { generateReplayCorpus, generateWorkflowReplayCorpus } from "./replay-memory.fixture.js"; import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { createTestLogger } from "../../../../test-utils/test-logger.js"; @@ -432,6 +435,267 @@ 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("replays more child events than V8 permits in one spread call", async () => { + const childCount = 16; + const entriesPerChild = 10_000; + const children = Array.from({ length: childCount }, (_, index) => ({ + agentId: `large-child-${index}`, + toolUseId: `large-task-${index}`, + })); + const parentLines = children.flatMap(({ agentId, toolUseId }) => [ + parentEntry([{ type: "tool_use", id: toolUseId, name: "Task", input: {} }]), + JSON.stringify({ + type: "user", + message: { + content: [ + { + type: "tool_result", + tool_use_id: toolUseId, + content: `agentId: ${agentId}\ndone`, + }, + ], + }, + }), + ]); + const subagentDir = writeParentSession(parentLines); + for (const { agentId, toolUseId } of children) { + writeSubagent({ + subagentDir, + agentId, + meta: JSON.stringify({ toolUseId }), + sidechainLines: Array.from({ length: entriesPerChild }, () => sidechainEntry({ agentId })), + }); + } + + const events = await replayDescriptors(); + const timelineCount = events.filter((event) => event.event.type === "timeline").length; + expect(timelineCount).toBe(childCount * entriesPerChild); + expect(upserts(events).filter((event) => event.status === "completed")).toHaveLength( + childCount, + ); + }, 120_000); + + 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: "a parent transcript without sidechains", + mode: "children", + // Cross V8's whole-string limit only in the opt-in reproduction, keeping routine CI small. + generate: (root: string) => + generateReplayCorpus(root, { + parentMiB: process.env.PASEO_TEST_LARGE_CLAUDE_REPLAY === "1" ? 550 : 2, + childrenMiB: 0, + children: 0, + }), + expected: { parentItems: 1, completed: 0, histories: 0 }, + }, + { + label: "4 MiB parent and 20 sidechains", + mode: "children", + generate: (root: string) => + generateReplayCorpus(root, { parentMiB: 4, childrenMiB: 4, children: 20 }), + expected: { parentItems: 41, completed: 20, histories: 20 }, + }, + { + label: "five 2 MiB workflow sidechains", + mode: "workflow", + generate: (root: string) => generateWorkflowReplayCorpus(root, 2), + expected: { parentItems: 3, completed: 1, histories: 5 }, + }, + ])( + "opens $label with all history under a 256 MB heap", + ({ mode, generate, expected }) => { + const root = mkdtempSync(path.join(os.tmpdir(), "claude-replay-")); + try { + generate(root); + const worker = spawnSync( + process.execPath, + [ + "--max-old-space-size=256", + "--import", + "tsx", + fileURLToPath(new URL("./replay-memory.fixture.ts", import.meta.url)), + root, + mode, + ], + { encoding: "utf8", env: { ...process.env, NODE_NO_WARNINGS: "1" }, timeout: 120_000 }, + ); + expect({ status: worker.status, signal: worker.signal, stderr: worker.stderr }).toEqual({ + status: 0, + signal: null, + stderr: "", + }); + const result = JSON.parse(worker.stdout); + expect(result).toMatchObject(expected); + console.log("Claude replay memory:", result); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }, + 150_000, + ); + + test("replays UTF-8 across read chunks, corrupt rows, CRLF and an unterminated final row", async () => { + const text = "é🙂".repeat(30_000); + const historyDir = claudeProjectDirSync(cwd, { configDir }); + mkdirSync(historyDir, { recursive: true }); + writeFileSync( + path.join(historyDir, "replay-session.jsonl"), + parentEntry([{ type: "text", text }]) + + "\r\nnot json\r\n" + + parentEntry([{ type: "text", text: "final row" }]), + ); + const events = await replayEvents(); + expect(events.filter((event) => event.type === "timeline").map((event) => event.item)).toEqual([ + { type: "assistant_message", text }, + { type: "assistant_message", text: "final row" }, + ]); + }); + + test("converts workflow tools in timestamp order before correlating their results", async () => { + writeWorkflowSession("completed"); + const directory = path.join( + claudeProjectDirSync(cwd, { configDir }), + "replay-session", + "subagents", + "workflows", + WORKFLOW_RUN_ID, + ); + writeSubagent({ + subagentDir: directory, + agentId: "a-result", + sidechainLines: [ + JSON.stringify({ + type: "user", + timestamp: "2026-07-26T06:28:01.000Z", + message: { + content: [{ type: "tool_result", tool_use_id: "workflow-bash", content: "finished" }], + }, + }), + ], + }); + writeSubagent({ + subagentDir: directory, + agentId: "z-call", + sidechainLines: [ + JSON.stringify({ + type: "assistant", + timestamp: "2026-07-26T06:28:00.000Z", + message: { + content: [ + { + type: "tool_use", + id: "workflow-bash", + name: "Bash", + input: { command: "echo finished" }, + }, + ], + }, + }), + ], + }); + const events = await replayDescriptors(); + const calls = events + .map((event) => event.event) + .filter((event) => event.type === "timeline" && event.item.type === "tool_call"); + expect(calls).toEqual([ + expect.objectContaining({ + id: WORKFLOW_TOOL_USE_ID, + item: expect.objectContaining({ callId: "workflow-bash", name: "Bash", status: "running" }), + }), + expect.objectContaining({ + id: WORKFLOW_TOOL_USE_ID, + item: expect.objectContaining({ + callId: "workflow-bash", + name: "Bash", + status: "completed", + detail: expect.objectContaining({ type: "shell", output: "finished" }), + }), + }), + ]); + }); + test("does not accumulate internal workflow agents as replay-only running rows", async () => { const subagentRoot = writeParentSession([ parentEntry([ diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 681988a813..859e2d5f2e 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -45,6 +45,7 @@ import { resolveClaudeDisabledThinkingForModel, } from "./model-manifest.js"; import { parsePartialJsonObject } from "./partial-json.js"; +import { withClaudeReplayHistory, type ClaudeReplayHistory } from "./transcript-history.js"; import { ClaudeSidechainTracker } from "./sidechain-tracker.js"; import { readClaudeSubagentHandback } from "./subagent-handback.js"; import { ClaudeTaskState } from "./task-state.js"; @@ -52,17 +53,10 @@ import { ClaudeTaskProtocolSource, type ClaudeHookObservationInput, } from "./subagents/live-source.js"; -import { - observeReplaySubagents, - parseClaudeSubagentMeta, - type ClaudeReplayParentFacts, - type ClaudeSubagentMeta, -} from "./subagents/replay-source.js"; +import { observeReplaySubagents, type ClaudeReplayParentFacts } from "./subagents/replay-source.js"; import { foldSubagentObservations, type SubagentObservation } from "./subagents/observation.js"; -import { - observeReplayWorkflows, - parseClaudeWorkflowRun, -} from "./subagents/workflow-replay-source.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 { @@ -5269,12 +5263,10 @@ class ClaudeAgentSession implements AgentSession { ); return; } - const content = fs.readFileSync(historyPath, "utf8"); - const replay = this.ingestPersistedSidechains( - content, - readClaudeSidechainHistory(historyPath), - ); - this.ingestPersistedHistory(content, replay); + withClaudeReplayHistory(historyPath, (history) => { + const replay = this.ingestPersistedSidechains(history); + this.ingestPersistedHistory(history.parentEntries, replay); + }); } catch (error) { this.logger.warn( { err: error, sessionId, historyPath }, @@ -5283,14 +5275,13 @@ class ClaudeAgentSession implements AgentSession { } } - private ingestPersistedHistory(content: string, replay: ClaudeReplayOwnership): void { - if (!content) { - return; - } - + private ingestPersistedHistory( + entries: Iterable, + replay: ClaudeReplayOwnership, + ): void { const timeline: PersistedTimelineEntry[] = []; - for (const line of content.split(/\r?\n/)) { - this.ingestPersistedHistoryLine(line, timeline, replay); + for (const entry of entries) { + this.ingestPersistedHistoryEntry(entry, timeline, replay); } if (timeline.length > 0) { @@ -5299,46 +5290,26 @@ class ClaudeAgentSession implements AgentSession { } } - private ingestPersistedSidechains( - parentContent: string, - sidechains: ClaudeSidechainHistory, - ): ClaudeReplayOwnership { - const parentEntries = parseClaudeHistoryRecords(parentContent).filter( - (entry) => entry.isSidechain !== true, - ); - const sidechainEntries = [parentContent, ...sidechains.contents] - .flatMap(parseClaudeHistoryRecords) - .filter((entry) => entry.isSidechain === true && typeof entry.agentId === "string"); - const parentFacts = readClaudeReplayParentFacts(parentEntries); + private ingestPersistedSidechains(history: ClaudeReplayHistory): ClaudeReplayOwnership { + const parentFacts = readClaudeReplayParentFacts(history.parentEntries); // Replay produces the same observations the live task protocol produces, then folds them // with the same function, so identity and status are derived once for both paths. const subagentReplay = observeReplaySubagents({ - subagents: [...groupClaudeSidechainEntries(sidechainEntries)].map(([agentId, entries]) => ({ - agentId, - meta: sidechains.metaByAgentId.get(agentId) ?? null, - entries, - parentFacts: readClaudeReplayParentFacts(entries as ClaudeHistoryEntry[]), + subagents: history.subagents.map((subagent) => ({ + ...subagent, + parentFacts: readClaudeReplayParentFacts(subagent.entries), })), parent: parentFacts, - convertEntry: (entry) => this.convertHistoryEntry(entry as ClaudeHistoryEntry), + convertEntry: (entry) => this.convertHistoryEntry(entry), }); const observations = [ ...subagentReplay.observations, ...observeReplayWorkflows({ - workflows: sidechains.workflowContents - .map(parseClaudeWorkflowRun) - .filter((workflow) => workflow !== null), - parentEntries, - entriesByRunId: new Map( - [...sidechains.workflowSidechainContentsByRunId].map(([runId, contents]) => [ - runId, - contents - .flatMap(parseClaudeHistoryRecords) - .filter((entry) => entry.type !== "user" || isToolResultUserEntry(entry)), - ]), - ), - convertEntry: (entry) => this.convertHistoryEntry(entry as ClaudeHistoryEntry), + workflows: history.workflows, + parentEntries: history.parentEntries, + entriesByRunId: history.workflowEntriesByRunId, + convertEntry: (entry) => this.convertHistoryEntry(entry), }), ]; const restoredProviderSubagentIds = new Set( @@ -5353,41 +5324,22 @@ class ClaudeAgentSession implements AgentSession { }; if (observations.length === 0) return replay; - this.persistedProviderSubagentEvents.push( - ...foldSubagentObservations(observations).map( - (event): Extract => ({ - type: "provider_subagent", - provider: "claude", - event, - }), - ), - ); + for (const event of foldSubagentObservations(observations)) { + this.persistedProviderSubagentEvents.push({ + type: "provider_subagent", + provider: "claude", + event, + }); + } this.historyPending = true; return replay; } - private ingestPersistedHistoryLine( - line: string, + private ingestPersistedHistoryEntry( + entry: ClaudeHistoryEntry, timeline: PersistedTimelineEntry[], replay: ClaudeReplayOwnership, ): void { - const trimmed = line.trim(); - if (!trimmed) { - return; - } - - let entry: Record; - try { - const parsed: unknown = JSON.parse(trimmed); - const record = toObjectRecord(parsed); - if (!record) { - return; - } - entry = record; - } catch { - return; - } - if (entry.isSidechain) { return; } @@ -5436,12 +5388,12 @@ class ClaudeAgentSession implements AgentSession { } if (items.length > 0) { - timeline.push( - ...items.map((item) => ({ + for (const item of items) { + timeline.push({ item, timestamp: historyTimestamp ?? undefined, - })), - ); + }); + } } } @@ -5456,7 +5408,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. @@ -6176,21 +6130,6 @@ function normalizeHistoryBlocks(content: unknown): ClaudeContentChunk[] | null { return null; } -function parseClaudeHistoryRecords(content: string): ClaudeHistoryEntry[] { - const entries: ClaudeHistoryEntry[] = []; - for (const line of content.split(/\r?\n/)) { - const trimmed = line.trim(); - if (!trimmed) continue; - try { - const entry = toObjectRecord(JSON.parse(trimmed)); - if (entry) entries.push(entry); - } catch { - // Ignore individual corrupt history rows, matching the parent history replay behavior. - } - } - return entries; -} - /** The parent's card for a subagent, labeled with its type and task. */ function buildClaudeSubagentCardDetail( facts: ClaudeSubagentCardFacts, @@ -6223,187 +6162,49 @@ function labelReplayedSubagentCard( * source consumes. The agentId scrape is kept only as a fallback for sessions recorded before * Claude Code wrote the meta sidecar. */ -function readClaudeReplayParentFacts(parentEntries: ClaudeHistoryEntry[]): ClaudeReplayParentFacts { +function readClaudeReplayParentFacts( + entries: Iterable, +): ClaudeReplayParentFacts { const toolCalls = new Map(); - for (const [id, call] of readClaudeHistoricalSubagentToolCalls(parentEntries)) { - toolCalls.set(id, { - ...((call.name ?? call.subagentType) ? { title: call.name ?? call.subagentType } : {}), - ...(call.description ? { description: call.description } : {}), - }); - } - - // Read outcomes straight off the tool_result blocks rather than reusing the agentId scrape, - // so a subagent linked through its meta sidecar still gets a status when the scrape missed it. - const outcomesByToolCallId = new Map(); - for (const entry of parentEntries) { - const content = toObjectRecord(entry.message)?.content; - if (!Array.isArray(content)) continue; - for (const value of content) { - const block = toObjectRecord(value); - if (block?.type !== "tool_result" || typeof block.tool_use_id !== "string") continue; - if (!toolCalls.has(block.tool_use_id)) continue; - outcomesByToolCallId.set(block.tool_use_id, { failed: block.is_error === true }); - } - } - - return { - toolCalls, - linksByAgentId: readClaudeHistoricalSubagentToolResults(parentEntries), - outcomesByToolCallId, - }; -} - -interface ClaudeSidechainHistory { - contents: string[]; - workflowContents: string[]; - workflowSidechainContentsByRunId: Map; - /** agentId -> sidecar metadata, when Claude Code wrote one next to the transcript. */ - metaByAgentId: Map; -} - -const CLAUDE_SUBAGENT_META_FILE = /^agent-(.+)\.meta\.json$/; - -function readClaudeSidechainHistory(historyPath: string): ClaudeSidechainHistory { - const sessionDirectory = path.join( - path.dirname(historyPath), - path.basename(historyPath, ".jsonl"), - ); - const sidechainDirectory = path.join(sessionDirectory, "subagents"); - const history: ClaudeSidechainHistory = { - contents: [], - workflowContents: [], - workflowSidechainContentsByRunId: new Map(), - metaByAgentId: new Map(), - }; - const workflowDirectory = path.join(sessionDirectory, "workflows"); - if (fs.existsSync(workflowDirectory)) { - for (const entry of fs.readdirSync(workflowDirectory, { withFileTypes: true })) { - if (!entry.isFile() || !entry.name.endsWith(".json")) continue; - try { - history.workflowContents.push( - fs.readFileSync(path.join(workflowDirectory, entry.name), "utf8"), - ); - } catch { - // A partial or unreadable run summary must not fail the rest of history ingestion. - } - } - } - if (!fs.existsSync(sidechainDirectory)) return history; - - const directories = [sidechainDirectory]; - while (directories.length > 0) { - const directory = directories.pop(); - if (!directory) continue; - for (const entry of fs.readdirSync(directory, { withFileTypes: true })) { - const entryPath = path.join(directory, entry.name); - if (entry.isDirectory()) { - directories.push(entryPath); - continue; - } - if (!entry.isFile()) continue; - if (entry.name.endsWith(".jsonl")) { - recordClaudeSidechainContents(history, sidechainDirectory, entryPath); - continue; - } - // The sidecar carries the Task tool_use id, which is the same id the live stream keys on. - // Reading it is what lets replay and live agree instead of each deriving its own link. - const metaMatch = CLAUDE_SUBAGENT_META_FILE.exec(entry.name); - if (!metaMatch?.[1]) continue; - try { - const meta = parseClaudeSubagentMeta(fs.readFileSync(entryPath, "utf8")); - if (meta) history.metaByAgentId.set(metaMatch[1], meta); - } catch { - // Undocumented internals: a missing or unreadable sidecar must never fail ingestion. - } - } - } - return history; -} - -function recordClaudeSidechainContents( - history: ClaudeSidechainHistory, - sidechainDirectory: string, - entryPath: string, -): void { - const contents = fs.readFileSync(entryPath, "utf8"); - const relativeParts = path.relative(sidechainDirectory, entryPath).split(path.sep); - const workflowRunId = - relativeParts[0] === "workflows" && relativeParts.length >= 3 ? relativeParts[1] : undefined; - if (!workflowRunId) { - history.contents.push(contents); - return; - } - - const workflowContents = history.workflowSidechainContentsByRunId.get(workflowRunId) ?? []; - workflowContents.push(contents); - history.workflowSidechainContentsByRunId.set(workflowRunId, workflowContents); -} - -interface ClaudeHistoricalSubagentToolCall { - name?: string; - subagentType?: string; - description?: string; -} - -function readClaudeHistoricalSubagentToolCalls( - entries: ClaudeHistoryEntry[], -): Map { - const toolCalls = new Map(); + const linksByAgentId = new Map(); + const outcomes = new Map(); for (const entry of entries) { const content = toObjectRecord(entry.message)?.content; if (!Array.isArray(content)) continue; for (const value of content) { const block = toObjectRecord(value); if ( - block?.type !== "tool_use" || - (block.name !== "Task" && block.name !== "Agent") || - typeof block.id !== "string" + block?.type === "tool_use" && + (block.name === "Task" || block.name === "Agent") && + typeof block.id === "string" ) { - continue; + const input = toObjectRecord(block.input); + const title = readNonEmptyString(input?.name) ?? readNonEmptyString(input?.subagent_type); + const description = readNonEmptyString(input?.description); + toolCalls.set(block.id, { + ...(title ? { title } : {}), + ...(description ? { description } : {}), + }); } - const input = toObjectRecord(block.input); - const name = readNonEmptyString(input?.name); - const subagentType = readNonEmptyString(input?.subagent_type); - const description = readNonEmptyString(input?.description); - toolCalls.set(block.id, { - ...(name ? { name } : {}), - ...(subagentType ? { subagentType } : {}), - ...(description ? { description } : {}), - }); - } - } - return toolCalls; -} - -function readClaudeHistoricalSubagentToolResults( - entries: ClaudeHistoryEntry[], -): Map { - const results = new Map(); - for (const entry of entries) { - const content = toObjectRecord(entry.message)?.content; - if (!Array.isArray(content)) continue; - for (const value of content) { - const block = toObjectRecord(value); if (block?.type !== "tool_result" || typeof block.tool_use_id !== "string") continue; - const match = /agentId:\s*([\w-]+)/.exec(JSON.stringify(block.content)); - if (!match?.[1]) continue; - results.set(match[1], { toolCallId: block.tool_use_id, failed: block.is_error === true }); + outcomes.set(block.tool_use_id, { failed: block.is_error === true }); + const agentId = readLegacyClaudeSubagentId(block.content); + if (agentId) + linksByAgentId.set(agentId, { + toolCallId: block.tool_use_id, + failed: block.is_error === true, + }); } } - return results; + return { + toolCalls, + linksByAgentId, + outcomesByToolCallId: new Map([...outcomes].filter(([id]) => toolCalls.has(id))), + }; } -function groupClaudeSidechainEntries( - entries: ClaudeHistoryEntry[], -): Map { - const entriesByAgentId = new Map(); - for (const entry of entries) { - if (typeof entry.agentId !== "string") continue; - const grouped = entriesByAgentId.get(entry.agentId) ?? []; - grouped.push(entry); - entriesByAgentId.set(entry.agentId, grouped); - } - return entriesByAgentId; +function readLegacyClaudeSubagentId(content: unknown): string | undefined { + return /agentId:\s*([\w-]+)/.exec(JSON.stringify(content))?.[1]; } interface ClaudeHistoryEntry { 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 new file mode 100644 index 0000000000..cab987e473 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/replay-memory.fixture.ts @@ -0,0 +1,202 @@ +import { appendFileSync, mkdirSync, writeFileSync } from "node:fs"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import pino from "pino"; +import { ClaudeAgentClient } from "./agent.js"; +import { claudeProjectDirSync } from "./project-dir.js"; + +/** 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 }); + const project = claudeProjectDirSync(cwd, { configDir }); + const subagents = path.join(project, "large-session", "subagents"); + mkdirSync(subagents, { recursive: true }); + const parent = path.join(project, "large-session.jsonl"); + const row = (value: unknown) => JSON.stringify(value) + "\n"; + // Large transcripts include records and context summaries that never become timeline items. + // 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((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 < size.children; i++) { + const agentId = `child-${i}`; + const id = `task-${i}`; + appendFileSync( + parent, + row({ + type: "assistant", + message: { + content: [{ type: "tool_use", id, name: "Task", input: { description: agentId } }], + }, + }), + ); + appendFileSync( + parent, + row({ + type: "user", + message: { content: [{ type: "tool_result", tool_use_id: id, content: "done" }] }, + }), + ); + const file = path.join(subagents, `agent-${agentId}.jsonl`); + writeFileSync(file, ""); + const summary = row({ + type: "user", + isCompactSummary: true, + isSidechain: true, + agentId, + message: { content: "x".repeat(64 * 1024) }, + }); + for ( + let j = 0; + j < Math.ceil((size.childrenMiB * 1024 * 1024) / size.children / summary.length); + j++ + ) + appendFileSync(file, summary); + appendFileSync( + file, + row({ + type: "assistant", + isSidechain: true, + agentId, + timestamp: "2026-07-26T06:28:00.000Z", + message: { + content: [{ type: "text", text: `History of ${agentId}` }], + stop_reason: "end_turn", + }, + }), + ); + writeFileSync( + path.join(subagents, `agent-${agentId}.meta.json`), + JSON.stringify({ toolUseId: id }), + ); + } + return { cwd, configDir }; +} + +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" }), + queryFactory: () => { + throw new Error("Replay must not query Claude"); + }, + resolveVersion: async () => "2.1.220", + }); + const session = await client.resumeSession( + { provider: "claude", sessionId: "large-session" }, + { cwd: path.join(root, "repo") }, + ); + let parentItems = 0; + const completed = new Set(); + const histories = new Set(); + for await (const event of session.streamHistory()) { + if (event.type === "timeline") parentItems++; + if ( + event.type === "provider_subagent" && + event.event.type === "upsert" && + event.event.status === "completed" + ) + completed.add(event.event.id); + if ( + event.type === "provider_subagent" && + event.event.type === "timeline" && + event.event.item.type === "assistant_message" + ) + histories.add(event.event.item.text); + } + await session.close(); + console.log( + JSON.stringify({ + parentItems, + completed: completed.size, + histories: histories.size, + maxRSSKiB: process.resourceUsage().maxRSS, + }), + ); +} + +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 }); + const project = claudeProjectDirSync(cwd, { configDir }); + const runId = "wf_large"; + const sessionDirectory = path.join(project, "large-session"); + const childrenDirectory = path.join(sessionDirectory, "subagents", "workflows", runId); + mkdirSync(childrenDirectory, { recursive: true }); + mkdirSync(path.join(sessionDirectory, "workflows"), { recursive: true }); + writeFileSync( + path.join(project, "large-session.jsonl"), + [ + { type: "user", message: { content: "Open the large session" } }, + { + type: "assistant", + message: { + content: [{ type: "tool_use", id: "workflow-task", name: "Workflow", input: {} }], + }, + }, + { + type: "user", + message: { + content: [ + { + type: "tool_result", + tool_use_id: "workflow-task", + content: `Workflow launched in background.\nRun ID: ${runId}`, + }, + ], + }, + }, + ] + .map((entry) => JSON.stringify(entry)) + .join("\n"), + ); + writeFileSync( + path.join(sessionDirectory, "workflows", `${runId}.json`), + JSON.stringify({ runId, status: "completed", summary: "Large workflow" }), + ); + for (let i = 0; i < 5; i++) { + const file = path.join(childrenDirectory, `agent-child-${i}.jsonl`); + writeFileSync(file, ""); + const prompt = + JSON.stringify({ + type: "user", + isSidechain: true, + agentId: `child-${i}`, + message: { content: "x".repeat(64 * 1024) }, + }) + "\n"; + for (let j = 0; j < Math.ceil((childMiB * 1024 * 1024) / prompt.length); j++) + appendFileSync(file, prompt); + appendFileSync( + file, + JSON.stringify({ + type: "assistant", + isSidechain: true, + agentId: `child-${i}`, + timestamp: `2026-07-26T06:28:0${4 - i}.000Z`, + message: { + content: [{ type: "text", text: `History of child-${i}` }], + stop_reason: "end_turn", + }, + }) + "\n", + ); + } + return { cwd, configDir }; +} diff --git a/packages/server/src/server/agent/providers/claude/subagents/replay-source.ts b/packages/server/src/server/agent/providers/claude/subagents/replay-source.ts index 372cbc01b2..23ba35c0fe 100644 --- a/packages/server/src/server/agent/providers/claude/subagents/replay-source.ts +++ b/packages/server/src/server/agent/providers/claude/subagents/replay-source.ts @@ -65,17 +65,12 @@ export interface ClaudeReplayEntry { * Effort is replay-only: the SDK's live assistant message does not carry it, and it can be * silently downgraded for the selected model, so any live value would be a guess. */ -function readRuntime(entries: readonly ClaudeReplayEntry[]): { model?: string; effort?: string } { - const runtime: { model?: string; effort?: string } = {}; - for (const entry of entries) { - if (entry.type !== "assistant") continue; - const effort = typeof entry.effort === "string" ? entry.effort.trim() : ""; - if (effort) runtime.effort = effort; - const rawModel = (entry.message as { model?: unknown } | undefined)?.model; - const model = resolveObservedClaudeModelId(typeof rawModel === "string" ? rawModel : undefined); - if (model) runtime.model = model; - } - return runtime; +function readRuntime(entry: ClaudeReplayEntry): { model?: string; effort?: string } { + if (entry.type !== "assistant") return {}; + const effort = typeof entry.effort === "string" ? entry.effort.trim() : ""; + const rawModel = entry.message?.model; + const model = resolveObservedClaudeModelId(typeof rawModel === "string" ? rawModel : undefined); + return { ...(effort ? { effort } : {}), ...(model ? { model } : {}) }; } /** @@ -126,22 +121,16 @@ function readTotalTokens(raw: unknown): number | undefined { * * The last assistant entry's summed usage matches Claude Code's live `total_tokens` definition. */ -function readUsage(entries: readonly ClaudeReplayEntry[]): ClaudeSubagentUsage | null { - let totalTokens: number | undefined; - - for (const entry of entries) { - if (entry.type !== "assistant") continue; - const observed = readTotalTokens(entry.message?.usage); - if (observed !== undefined) totalTokens = observed; - } - +function readUsage(entry: ClaudeReplayEntry): ClaudeSubagentUsage | null { + if (entry.type !== "assistant") return null; + const totalTokens = readTotalTokens(entry.message?.usage); return totalTokens === undefined ? null : { totalTokens }; } export interface ClaudeReplaySubagentInput { agentId: string; meta: ClaudeSubagentMeta | null; - entries: ClaudeReplayEntry[]; + entries: Iterable; /** Task declarations made by this subagent, used to attach its direct children. */ parentFacts?: ClaudeReplayParentFacts; } @@ -211,18 +200,6 @@ function resolveParentLink( return null; } -/** A final `end_turn` is the child's own completion signal when the parent result is absent. */ -function readChildTerminalStatus( - entries: readonly ClaudeReplayEntry[], -): ProviderSubagentStatus | null { - for (let index = entries.length - 1; index >= 0; index--) { - const entry = entries[index]; - if (entry?.type !== "assistant") continue; - return entry.message?.stop_reason === "end_turn" ? "completed" : null; - } - return null; -} - /** * The parent's tool input is the same identity source the live path reads; meta.json contributes * provider details only after that declaration has admitted the child. @@ -231,11 +208,11 @@ function declareSubagent( subagent: ClaudeReplaySubagentInput, link: ParentLink, toolCall: { title?: string; description?: string } | undefined, - parentSubagentId?: string, + parentSubagentId: string | undefined, + firstTimestamp: string | null, ): SubagentObservation { const title = toolCall?.title ?? subagent.meta?.agentType; const description = toolCall?.description ?? subagent.meta?.description; - const firstTimestamp = normalizeProviderReplayTimestamp(subagent.entries[0]?.timestamp); return { kind: "declared", id: link.id, @@ -248,12 +225,12 @@ function declareSubagent( } function observeSubtitle( - subagent: ClaudeReplaySubagentInput, link: ParentLink, title: string | undefined, + runtime: { model?: string; effort?: string }, + usage: ClaudeSubagentUsage | null, + timestamp: string | null, ): SubagentObservation | null { - const runtime = readRuntime(subagent.entries); - const usage = readUsage(subagent.entries); const hasDetails = runtime.model !== undefined || runtime.effort !== undefined || @@ -265,7 +242,6 @@ function observeSubtitle( ...(usage ? { usage } : {}), }); if (!subtitle) return null; - const timestamp = normalizeProviderReplayTimestamp(subagent.entries.at(-1)?.timestamp); return { kind: "subtitle", id: link.id, @@ -278,35 +254,50 @@ function observeSubagent( subagent: ClaudeReplaySubagentInput, parent: ClaudeReplayParentFacts, convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[], + toolOwners: Map, parentSubagentId?: string, ): SubagentObservation[] { const link = resolveParentLink(subagent, parent); if (!link) return []; const toolCall = parent.toolCalls.get(link.toolCallId); + const timeline: SubagentObservation[] = []; + const runtime: { model?: string; effort?: string } = {}; + let usage: ClaudeSubagentUsage | null = null; + let firstTimestamp: string | null = null; + let lastTimestamp: string | null = null; + let seenEntry = false; + let childTerminalStatus: ProviderSubagentStatus | null = null; - const observations: SubagentObservation[] = [ - declareSubagent(subagent, link, toolCall, parentSubagentId), - ]; - const subtitle = observeSubtitle(subagent, link, toolCall?.title ?? subagent.meta?.agentType); - if (subtitle) observations.push(subtitle); - + // Consume each source once for presentation and timeline; keep no raw transcript records. for (const entry of subagent.entries) { const timestamp = normalizeProviderReplayTimestamp(entry.timestamp); + if (!seenEntry) firstTimestamp = timestamp; + seenEntry = true; + lastTimestamp = timestamp; + Object.assign(runtime, readRuntime(entry)); + usage = readUsage(entry) ?? usage; + if (entry.type === "assistant") { + childTerminalStatus = entry.message?.stop_reason === "end_turn" ? "completed" : null; + } + recordReplayToolOwners(toolOwners, entry, link.id); for (const item of convertEntry(entry)) { - observations.push({ - kind: "timeline", - id: link.id, - item, - ...(timestamp ? { timestamp } : {}), - }); + timeline.push({ kind: "timeline", id: link.id, item, ...(timestamp ? { timestamp } : {}) }); } } - - // Prefer the parent's result because it carries failure. If that result is missing, the child's - // final end_turn still proves completion. Any other shape remains running. - const terminalStatus = link.status ?? readChildTerminalStatus(subagent.entries); + const observations = [ + declareSubagent(subagent, link, toolCall, parentSubagentId, firstTimestamp), + ]; + const subtitle = observeSubtitle( + link, + toolCall?.title ?? subagent.meta?.agentType, + runtime, + usage, + lastTimestamp, + ); + if (subtitle) observations.push(subtitle); + for (const observation of timeline) observations.push(observation); + const terminalStatus = link.status ?? childTerminalStatus; if (terminalStatus) { - const lastTimestamp = normalizeProviderReplayTimestamp(subagent.entries.at(-1)?.timestamp); observations.push({ kind: "status", id: link.id, @@ -314,22 +305,18 @@ function observeSubagent( ...(lastTimestamp ? { timestamp: lastTimestamp } : {}), }); } - return observations; } function recordReplayToolOwners( owners: Map, - entries: readonly ClaudeReplayEntry[], + entry: ClaudeReplayEntry, subagentId: string, ): void { - for (const entry of entries) { - if (entry.type !== "assistant" || !Array.isArray(entry.message?.content)) continue; - for (const block of entry.message.content) { - if (block?.type === "tool_use" && typeof block.id === "string") { - owners.set(block.id, subagentId); - } - } + if (entry.type !== "assistant" || !Array.isArray(entry.message?.content)) return; + for (const block of entry.message.content) { + if (block?.type === "tool_use" && typeof block.id === "string") + owners.set(block.id, subagentId); } } @@ -359,8 +346,15 @@ export function observeReplaySubagents(input: { const { ownerId, parent, link } = resolved; // Only proven descendants may own notifications; ambient sidecars cannot claim them. - recordReplayToolOwners(toolOwners, subagent.entries, link.id); - observations.push(...observeSubagent(subagent, parent, input.convertEntry, ownerId)); + for (const observation of observeSubagent( + subagent, + parent, + input.convertEntry, + toolOwners, + ownerId, + )) { + observations.push(observation); + } if (subagent.parentFacts) resolvedParents.set(link.id, subagent.parentFacts); unresolved.splice(index, 1); madeProgress = true; diff --git a/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts b/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts index ede48a5f48..1d7ff2b1a8 100644 --- a/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts +++ b/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts @@ -47,8 +47,9 @@ export function parseClaudeWorkflowRun(contents: string): ClaudeWorkflowRun | nu */ export function observeReplayWorkflows(input: { workflows: readonly ClaudeWorkflowRun[]; - parentEntries: readonly ClaudeWorkflowParentEntry[]; - entriesByRunId?: ReadonlyMap; + parentEntries: Iterable; + /** Repeatable entries in chronological order, supplied by the transcript reader. */ + entriesByRunId?: ReadonlyMap>; convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[]; }): SubagentObservation[] { const toolCallIdByRunId = readWorkflowLinks(input.parentEntries); @@ -78,15 +79,15 @@ export function observeReplayWorkflows(input: { }); } - observations.push( - ...observeWorkflowTimeline({ - id, - result: workflow.result, - entries: input.entriesByRunId?.get(workflow.runId) ?? [], - convertEntry: input.convertEntry, - finishedAt, - }), - ); + for (const observation of observeWorkflowTimeline({ + id, + result: workflow.result, + entries: input.entriesByRunId?.get(workflow.runId) ?? [], + convertEntry: input.convertEntry, + finishedAt, + })) { + observations.push(observation); + } const subtitle = buildClaudeSubagentSubtitle({ title: "Workflow", @@ -117,15 +118,14 @@ export function observeReplayWorkflows(input: { function observeWorkflowTimeline(input: { id: string; result: unknown; - entries: readonly ClaudeReplayEntry[]; + entries: Iterable; convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[]; finishedAt?: string; }): SubagentObservation[] { const observations: SubagentObservation[] = []; const replayedAssistantText = new Set(); - const entries = [...input.entries].sort(compareReplayTimestamps); - for (const entry of entries) { + for (const entry of input.entries) { const timestamp = normalizeProviderReplayTimestamp(entry.timestamp); const items = input.convertEntry?.(entry) ?? []; for (const item of items) { @@ -150,16 +150,7 @@ function observeWorkflowTimeline(input: { return observations; } -function compareReplayTimestamps(a: ClaudeReplayEntry, b: ClaudeReplayEntry): number { - const aTimestamp = normalizeProviderReplayTimestamp(a.timestamp); - const bTimestamp = normalizeProviderReplayTimestamp(b.timestamp); - if (!aTimestamp && !bTimestamp) return 0; - if (!aTimestamp) return 1; - if (!bTimestamp) return -1; - return Date.parse(aTimestamp) - Date.parse(bTimestamp); -} - -function readWorkflowLinks(entries: readonly ClaudeWorkflowParentEntry[]): Map { +function readWorkflowLinks(entries: Iterable): Map { const workflowToolCallIds = new Set(); const toolCallIdByRunId = new Map(); diff --git a/packages/server/src/server/agent/providers/claude/transcript-history.ts b/packages/server/src/server/agent/providers/claude/transcript-history.ts new file mode 100644 index 0000000000..cba606b0e4 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/transcript-history.ts @@ -0,0 +1,256 @@ +import fs from "node:fs"; +import path from "node:path"; +import { normalizeProviderReplayTimestamp } from "../../provider-history-timestamps.js"; +import { + parseClaudeSubagentMeta, + type ClaudeReplayEntry, + type ClaudeSubagentMeta, +} from "./subagents/replay-source.js"; +import { + parseClaudeWorkflowRun, + type ClaudeWorkflowRun, +} from "./subagents/workflow-replay-source.js"; + +interface TranscriptEntry extends ClaudeReplayEntry { + isSidechain?: unknown; +} + +interface RecordLocation { + file: string; + offset: number; + length: number; + timestamp: string | null; +} + +/** 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 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 using descriptors owned by this replay. */ +function indexedRecords( + locations: readonly RecordLocation[], + descriptors: ReadonlyMap, +): Iterable { + return { + *[Symbol.iterator]() { + 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; + } + const entry = parseRecord(buffer.toString("utf8")); + if (entry) yield entry; + } + }, + }; +} + +function isToolResult(entry: TranscriptEntry): boolean { + return ( + Array.isArray(entry.message?.content) && + entry.message.content.some((block) => block?.type === "tool_result") + ); +} + +function parseRecord(line: string): TranscriptEntry | null { + try { + const value: unknown = JSON.parse(line); + return value !== null && typeof value === "object" && !Array.isArray(value) + ? (value as TranscriptEntry) + : null; + } catch { + return null; + } +} + +function readOptionalFile(file: string): string | null { + try { + return fs.readFileSync(file, "utf8"); + } catch { + return null; + } +} + +/** Discover sources without retaining transcript contents. Unrelated sidechains stay excluded + * by the replay ownership resolver; workflow children retain their separate run ownership. */ +export interface ClaudeReplayHistory { + parentEntries: Iterable; + subagents: { + agentId: string; + meta: ClaudeSubagentMeta | null; + entries: Iterable; + }[]; + workflows: ClaudeWorkflowRun[]; + workflowEntriesByRunId: Map>; +} + +/** 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 descriptors = new Map(); + try { + return consume(indexHistory(historyPath, descriptors)); + } finally { + for (const fd of descriptors.values()) fs.closeSync(fd); + } +} + +function indexHistory(historyPath: string, descriptors: Map): ClaudeReplayHistory { + const sessionDirectory = path.join( + path.dirname(historyPath), + path.basename(historyPath, ".jsonl"), + ); + const sidechainDirectory = path.join(sessionDirectory, "subagents"); + const sidechainFiles = [historyPath]; + const workflowFiles = new Map(); + const metaByAgentId = new Map(); + const directories = [sidechainDirectory]; + while (directories.length) { + const directory = directories.pop()!; + if (!fs.existsSync(directory)) continue; + for (const entry of fs.readdirSync(directory, { withFileTypes: true })) { + const file = path.join(directory, entry.name); + if (entry.isDirectory()) { + directories.push(file); + continue; + } + if (!entry.isFile()) continue; + if (entry.name.endsWith(".jsonl")) { + const parts = path.relative(sidechainDirectory, file).split(path.sep); + if (parts[0] === "workflows" && parts.length >= 3) { + const runId = parts[1]!; + const files = workflowFiles.get(runId) ?? []; + files.push(file); + workflowFiles.set(runId, files); + } else sidechainFiles.push(file); + continue; + } + const agentId = /^agent-(.+)\.meta\.json$/.exec(entry.name)?.[1]; + if (!agentId) continue; + const contents = readOptionalFile(file); + const meta = contents === null ? null : parseClaudeSubagentMeta(contents); + if (meta) metaByAgentId.set(agentId, meta); + } + } + const parentLocations: RecordLocation[] = []; + const locationsByAgentId = new Map(); + function scan( + file: string, + visit: (entry: TranscriptEntry, location: RecordLocation) => void, + ): void { + 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) => { + 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, descriptors)); + } + return { + parentEntries: indexedRecords(parentLocations, descriptors), + subagents: [...locationsByAgentId].map(([agentId, locations]) => ({ + agentId, + meta: metaByAgentId.get(agentId) ?? null, + entries: indexedRecords(locations, descriptors), + })), + workflows: readWorkflows(sessionDirectory), + workflowEntriesByRunId, + }; +} + +function readWorkflows(sessionDirectory: string): ClaudeWorkflowRun[] { + const workflows: ClaudeWorkflowRun[] = []; + const workflowDirectory = path.join(sessionDirectory, "workflows"); + if (fs.existsSync(workflowDirectory)) { + for (const entry of fs.readdirSync(workflowDirectory, { withFileTypes: true })) { + if (!entry.isFile() || !entry.name.endsWith(".json")) continue; + const contents = readOptionalFile(path.join(workflowDirectory, entry.name)); + const workflow = contents === null ? null : parseClaudeWorkflowRun(contents); + if (workflow) workflows.push(workflow); + } + } + return workflows; +}