From 5991d2f7892a1dfe434b5129bbb9576f29f9686e Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 5 Oct 2026 18:34:30 +0200 Subject: [PATCH] Stream Claude transcript replay without discarding subagent history Read JSONL incrementally, retain ownership facts instead of raw records, and replay workflow payloads through a chronological byte-location index. Preserve all subagent history and remove the input byte budget. Co-authored-by: ZouYusheng --- docs/agent-lifecycle.md | 2 +- .../claude/agent.sub-agent-replay.test.ts | 337 +++++--------- .../server/agent/providers/claude/agent.ts | 422 +++--------------- .../providers/claude/replay-memory.fixture.ts | 186 ++++++++ .../claude/subagents/replay-source.ts | 122 +++-- .../subagents/workflow-replay-source.ts | 21 +- .../providers/claude/transcript-history.ts | 274 ++++++++++++ 7 files changed, 690 insertions(+), 674 deletions(-) create mode 100644 packages/server/src/server/agent/providers/claude/replay-memory.fixture.ts create mode 100644 packages/server/src/server/agent/providers/claude/transcript-history.ts diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index eb6b978bb3..8662ee3287 100644 --- a/docs/agent-lifecycle.md +++ b/docs/agent-lifecycle.md @@ -202,7 +202,7 @@ Claude Code announces subagent lifecycle on the SDK stream (`task_started` / `ta - **Effort is only reachable through hooks.** It appears nowhere on the message stream at any depth, and the level Paseo requests is not necessarily the level that runs — a model that does not support it is silently downgraded. A hook firing inside a subagent reports the active post-downgrade level next to its `agent_id`, which is the same id `task_started` calls `task_id`. - **Backgrounded subagents emit no frames carrying `parent_tool_use_id` at all.** Everything keyed off that field sees nothing for one; they are visible only because the task protocol announces them. - **Nested ownership comes from the launching sidechain, not `spawn_depth` alone.** A sidechain's Agent or Bash tool call records the direct owner of that `tool_use_id`; the following `task_started` inherits it. This routes a grandchild descriptor and child-owned background notifications without relying on labels or flattening them into the managed parent. -- **On replay, `/subagents/` holds every descendant beside the root.** Resolve the tree one proven generation at a time: the root transcript admits direct children, then each admitted sidechain transcript admits its children by `toolUseId`. `spawnDepth` orders candidates but does not establish ownership. Unresolved sidecars remain excluded as ambient or unrelated work. When the parent plus sidechain transcripts exceed the replay budget (256 MB by default, `PASEO_CLAUDE_REPLAY_BUDGET_MB` to override), the sidechain files are left unread and only the parent replays — loading every subagent transcript at once OOMs the daemon on large sessions (issue #5820). +- **On replay, `/subagents/` holds every descendant beside the root.** Resolve the tree one proven generation at a time: the root transcript admits direct children, then each admitted sidechain transcript admits its children by `toolUseId`. `spawnDepth` orders candidates but does not establish ownership. Unresolved sidecars remain excluded as ambient or unrelated work. - **Replay `totalTokens` is a context-size reading, not cumulative spend.** Claude Code finalizes a subagent by summing the _last_ assistant message's usage block and shipping that as `usage.total_tokens`. Summing per-entry usage instead multiplies the cached prefix by the turn count and reports a number several times larger than the live path. Archived Paseo subagents disappear from the track, by design. To remove one from the track without closing its tab, use the **archive button** on the row — it opens a confirm dialog and archives the subagent on confirm. Provider-owned rows have no individual Paseo lifecycle controls. 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 1516358ce4..3290e819ce 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,7 +1,9 @@ import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import * as os from "node:os"; import * as path from "node:path"; -import pino from "pino"; +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"; @@ -433,241 +435,132 @@ describe("ClaudeAgentSession persisted subagent replay", () => { expect(upserts(await replayDescriptors())).toEqual([]); }); - test("skips subagent history replay when the transcript exceeds the replay budget", async () => { - // Issue #5820: loading every sidechain transcript at once OOMs the daemon worker on large - // sessions. Over budget, the sidechain files stay unread and the parent history still loads. - vi.stubEnv("PASEO_CLAUDE_REPLAY_BUDGET_MB", "1"); - const bulkEntry = JSON.stringify({ - type: "user", - isSidechain: true, - agentId: AGENT_ID, - sessionId: "replay-session", - timestamp: "2026-07-26T06:27:55.000Z", - message: { role: "user", content: "x".repeat(1024 * 1024 + 8192) }, - }); - writeSession({ - parentLines: [taskToolUse(), taskToolResult()], - meta: JSON.stringify({ toolUseId: TOOL_USE_ID }), - sidechainLines: [sidechainEntry(), bulkEntry], - }); + test.each([ + { + label: "395 MiB parent and 269 sidechains", + mode: "children", + generate: generateReplayCorpus, + expected: { parentItems: 539, completed: 269, histories: 269 }, + }, + { + label: "five 320 MiB workflow sidechains", + mode: "workflow", + generate: generateWorkflowReplayCorpus, + expected: { parentItems: 3, completed: 1, histories: 5 }, + }, + ])( + "opens $label with all history under a 512 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-")); + try { + generate(root); + const worker = spawnSync( + process.execPath, + [ + "--max-old-space-size=512", + "--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, + ); - const events = await replayEvents(); - - expect(events.filter((event) => event.type === "provider_subagent")).toEqual([]); - - const parentTaskCalls = events.filter( - (event) => - event.type === "timeline" && - event.item.type === "tool_call" && - event.item.callId === TOOL_USE_ID, - ); - expect(parentTaskCalls.length).toBeGreaterThanOrEqual(1); - }); - - test("still replays subagent history when the transcript is under the replay budget", async () => { - vi.stubEnv("PASEO_CLAUDE_REPLAY_BUDGET_MB", "16"); - writeSession({ - parentLines: [taskToolUse(), taskToolResult()], - meta: JSON.stringify({ toolUseId: TOOL_USE_ID }), - sidechainLines: [sidechainEntry({ stopReason: "end_turn" })], - }); - - const descriptors = upserts(await replayDescriptors()); - expect(descriptors.map((descriptor) => descriptor.id)).toContain(TOOL_USE_ID); - expect(descriptors.at(-1)).toMatchObject({ id: TOOL_USE_ID, status: "completed" }); - }); - - test("falls back to the default budget when the env override is not a positive number", async () => { - vi.stubEnv("PASEO_CLAUDE_REPLAY_BUDGET_MB", "0"); - writeSession({ - parentLines: [taskToolUse(), taskToolResult()], - meta: JSON.stringify({ toolUseId: TOOL_USE_ID }), - sidechainLines: [sidechainEntry({ stopReason: "end_turn" })], - }); - - const descriptors = upserts(await replayDescriptors()); - expect(descriptors.map((descriptor) => descriptor.id)).toContain(TOOL_USE_ID); - }); - - test("honors a fractional replay budget instead of rounding it up to the default", async () => { - // 0.5 MB is 524,288 bytes; parseInt would read it as 0 and silently fall back to 256 MB. - vi.stubEnv("PASEO_CLAUDE_REPLAY_BUDGET_MB", "0.5"); - const bulkEntry = JSON.stringify({ - type: "user", - isSidechain: true, - agentId: AGENT_ID, - sessionId: "replay-session", - timestamp: "2026-07-26T06:27:55.000Z", - message: { role: "user", content: "x".repeat(600 * 1024) }, - }); - writeSession({ - parentLines: [taskToolUse(), taskToolResult()], - meta: JSON.stringify({ toolUseId: TOOL_USE_ID }), - sidechainLines: [sidechainEntry(), bulkEntry], - }); - - const events = await replayEvents(); - - expect(events.filter((event) => event.type === "provider_subagent")).toEqual([]); - }); - - test("replays the parent unchanged, matching the same session without subagent transcripts", async () => { - // The #5820 reporter's own technique: moving the subagent transcripts away leaves the parent - // loading fine. The over-budget skip has to be indistinguishable from that state. - vi.stubEnv("PASEO_CLAUDE_REPLAY_BUDGET_MB", "1"); - const bulkEntry = JSON.stringify({ - type: "user", - isSidechain: true, - agentId: AGENT_ID, - sessionId: "replay-session", - timestamp: "2026-07-26T06:27:55.000Z", - message: { role: "user", content: "x".repeat(1024 * 1024 + 8192) }, - }); - writeSession({ - parentLines: [taskToolUse(), taskToolResult()], - meta: JSON.stringify({ toolUseId: TOOL_USE_ID }), - sidechainLines: [sidechainEntry(), bulkEntry], - }); - const withSidechainsSkipped = await replayEvents(); - - rmSync(path.join(claudeProjectDirSync(cwd, { configDir }), "replay-session"), { - recursive: true, - force: true, - }); - const withoutSidechains = await replayEvents(); - - expect(withSidechainsSkipped).toEqual(withoutSidechains); - }); - - test("does not warn about skipped subagent history when the session has none", async () => { - vi.stubEnv("PASEO_CLAUDE_REPLAY_BUDGET_MB", "1"); - const records: unknown[] = []; + 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"), - [ - taskToolUse(), - taskToolResult(), - parentEntry([{ type: "text", text: "x".repeat(1024 * 1024 + 8192) }]), - ].join("\n"), + parentEntry([{ type: "text", text }]) + + "\r\nnot json\r\n" + + parentEntry([{ type: "text", text: "final row" }]), ); - - const client = new ClaudeAgentClient({ - logger: pino({ level: "info" }, { write: (line: string) => records.push(JSON.parse(line)) }), - queryFactory, - resolveVersion: async () => "2.1.220", - }); - const session = await client.resumeSession( - { provider: "claude", sessionId: "replay-session" }, - { cwd }, - ); - const events: AgentStreamEvent[] = []; - for await (const event of session.streamHistory()) { - events.push(event); - } - await session.close(); - - expect( - records.filter( - (record) => - (record as { msg?: string }).msg === - "Skipped Claude subagent history replay for oversized transcript", - ), - ).toEqual([]); - expect(events.filter((event) => event.type === "timeline").length).toBeGreaterThan(0); + 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("warns with the byte accounting when subagent history is skipped", async () => { - vi.stubEnv("PASEO_CLAUDE_REPLAY_BUDGET_MB", "1"); - const records: unknown[] = []; - const bulkEntry = JSON.stringify({ - type: "user", - isSidechain: true, - agentId: AGENT_ID, - sessionId: "replay-session", - timestamp: "2026-07-26T06:27:55.000Z", - message: { role: "user", content: "x".repeat(1024 * 1024 + 8192) }, - }); - writeSession({ - parentLines: [taskToolUse(), taskToolResult()], - meta: JSON.stringify({ toolUseId: TOOL_USE_ID }), - sidechainLines: [sidechainEntry(), bulkEntry], - }); - - const client = new ClaudeAgentClient({ - logger: pino({ level: "info" }, { write: (line: string) => records.push(JSON.parse(line)) }), - queryFactory, - resolveVersion: async () => "2.1.220", - }); - const session = await client.resumeSession( - { provider: "claude", sessionId: "replay-session" }, - { cwd }, - ); - for await (const _event of session.streamHistory()) { - // Drain the replay so the skip decision has run. - } - await session.close(); - - expect(records).toContainEqual( - expect.objectContaining({ - level: pino.levels.values.warn, - msg: "Skipped Claude subagent history replay for oversized transcript", - sessionId: "replay-session", - parentBytes: expect.any(Number), - sidechainBytes: expect.any(Number), - budgetBytes: 1024 * 1024, - }), - ); - const warned = records.find( - (record) => - (record as { msg?: string }).msg === - "Skipped Claude subagent history replay for oversized transcript", - ) as { parentBytes: number; sidechainBytes: number }; - expect(warned.parentBytes).toBeGreaterThan(0); - expect(warned.sidechainBytes).toBeGreaterThan(1024 * 1024); - }); - - test("skips workflow replay too when the transcript exceeds the replay budget", async () => { - vi.stubEnv("PASEO_CLAUDE_REPLAY_BUDGET_MB", "1"); - const bulkEntry = JSON.stringify({ - type: "user", - isSidechain: true, - agentId: "wf-bulk-agent", - sessionId: "replay-session", - timestamp: "2026-07-26T06:27:55.000Z", - message: { role: "user", content: "x".repeat(1024 * 1024 + 8192) }, - }); - writeWorkflowSession("completed", { - children: [ - { - agentId: "wf-bulk-agent", - output: "child output", - timestamp: "2026-07-26T06:29:00.000Z", - }, - ], - }); - const subagentDirectory = path.join( + 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: subagentDirectory, - agentId: "wf-bulk-agent", - sidechainLines: [bulkEntry], + 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" }], + }, + }), + ], }); - - const events = await replayEvents(); - - expect(events.filter((event) => event.type === "provider_subagent")).toEqual([]); - const notifications = events.filter( - (event) => - event.type === "timeline" && - event.item.type === "tool_call" && - event.item.name === "task_notification", - ); - expect(notifications.length).toBeGreaterThanOrEqual(1); + 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 () => { diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 1e998f98ea..0a2f8df5d8 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -43,23 +43,16 @@ import { resolveClaudeDisabledThinkingForModel, } from "./model-manifest.js"; import { parsePartialJsonObject } from "./partial-json.js"; +import { readClaudeReplayHistory } from "./transcript-history.js"; import { ClaudeSidechainTracker } from "./sidechain-tracker.js"; import { ClaudeTaskState } from "./task-state.js"; 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 { readClaudeWorkflowResultFile } from "./subagents/workflow-output.js"; import { buildClaudeFeatures, claudeModelSupportsFastMode } from "./feature-definitions.js"; import { @@ -4936,16 +4929,9 @@ class ClaudeAgentSession implements AgentSession { ); return; } - const content = fs.readFileSync(historyPath, "utf8"); - const sidechains = readClaudeSidechainHistory(historyPath, fs.statSync(historyPath).size); - if (sidechains.skippedForBudget) { - this.logger.warn( - { sessionId, cwd: this.config.cwd, historyPath, ...sidechains.skippedForBudget }, - "Skipped Claude subagent history replay for oversized transcript", - ); - } - const replay = this.ingestPersistedSidechains(content, sidechains); - this.ingestPersistedHistory(content, replay); + const history = readClaudeReplayHistory(historyPath); + const replay = this.ingestPersistedSidechains(history); + this.ingestPersistedHistory(history.parentEntries, replay); } catch (error) { this.logger.warn( { err: error, sessionId, historyPath }, @@ -4954,14 +4940,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) { @@ -4971,46 +4956,27 @@ class ClaudeAgentSession implements AgentSession { } private ingestPersistedSidechains( - parentContent: string, - sidechains: ClaudeSidechainHistory, + history: ReturnType, ): ClaudeReplayOwnership { - // Parse the parent transcript once and split it, instead of re-parsing the whole string to - // recover the sidechain records it already contained. - const parentRecords = parseClaudeHistoryRecords(parentContent); - const parentEntries = parentRecords.filter((entry) => entry.isSidechain !== true); - const sidechainEntries = parentRecords - .concat(sidechains.contents.flatMap(parseClaudeHistoryRecords)) - .filter((entry) => entry.isSidechain === true && typeof entry.agentId === "string"); - const parentFacts = readClaudeReplayParentFacts(parentEntries); + 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( @@ -5038,28 +5004,11 @@ class ClaudeAgentSession implements AgentSession { 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; } @@ -5828,21 +5777,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, @@ -5875,299 +5809,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; - /** - * Set when the sidechain files (subagent transcripts, workflow summaries, sidecars) were left - * unread because the parent plus sidechain transcripts exceed the replay budget, so only - * parent history is replayed (issue #5820: loading every subagent transcript at once OOMs the - * daemon worker on large sessions). - */ - skippedForBudget?: { - parentBytes: number; - sidechainBytes: number; - budgetBytes: number; - }; -} - -const CLAUDE_SUBAGENT_META_FILE = /^agent-(.+)\.meta\.json$/; - -// Issue #5820 measured a 757 MB transcript set peaking near ~2.1 GB in the daemon worker before -// the heap gave out. A 256 MB replay-input budget keeps that peak an order of magnitude lower -// while leaving ordinary sessions (kilobytes to a few megabytes) untouched. -const DEFAULT_CLAUDE_REPLAY_BUDGET_BYTES = 256 * 1024 * 1024; - -function claudeReplayBudgetBytes(): number { - const megabytes = Number.parseFloat(process.env.PASEO_CLAUDE_REPLAY_BUDGET_MB ?? ""); - return Number.isFinite(megabytes) && megabytes > 0 - ? megabytes * 1024 * 1024 - : DEFAULT_CLAUDE_REPLAY_BUDGET_BYTES; -} - -/** - * Returns null instead of throwing, so a file that disappears or becomes unreadable between the - * stat and the read is skipped rather than failing the session's history load. - */ -function readClaudeSidechainFile(entryPath: string): string | null { - try { - return fs.readFileSync(entryPath, "utf8"); - } catch { - return null; - } -} - -/** Byte size for the replay budget, or null when the file cannot be stat-ed. */ -function claudeSidechainFileSize(entryPath: string): number | null { - try { - return fs.statSync(entryPath).size; - } catch { - return null; - } -} - -interface ClaudeSidechainFile { - entryPath: string; - workflowRunId?: string; -} - -interface ClaudeSidechainFilePlan { - sidechainFiles: ClaudeSidechainFile[]; - metaFiles: { entryPath: string; agentId: string }[]; - bytes: number; -} - -function collectClaudeWorkflowSummaryPaths(sessionDirectory: string): { - paths: string[]; - bytes: number; -} { - const workflowDirectory = path.join(sessionDirectory, "workflows"); - const result = { paths: [] as string[], bytes: 0 }; - if (!fs.existsSync(workflowDirectory)) return result; - for (const entry of fs.readdirSync(workflowDirectory, { withFileTypes: true })) { - if (!entry.isFile() || !entry.name.endsWith(".json")) continue; - const summaryPath = path.join(workflowDirectory, entry.name); - const summaryBytes = claudeSidechainFileSize(summaryPath); - if (summaryBytes === null) continue; - result.bytes += summaryBytes; - result.paths.push(summaryPath); - } - return result; -} - -function recordClaudeSidechainFile( - plan: ClaudeSidechainFilePlan, - sidechainDirectory: string, - entryPath: string, -): number { - const relativeParts = path.relative(sidechainDirectory, entryPath).split(path.sep); - const workflowRunId = - relativeParts[0] === "workflows" && relativeParts.length >= 3 ? relativeParts[1] : undefined; - const entryBytes = claudeSidechainFileSize(entryPath); - if (entryBytes === null) return 0; - plan.sidechainFiles.push({ entryPath, workflowRunId }); - return entryBytes; -} - -function collectClaudeSidechainFilePlan(sidechainDirectory: string): ClaudeSidechainFilePlan { - const plan: ClaudeSidechainFilePlan = { sidechainFiles: [], metaFiles: [], bytes: 0 }; - if (!fs.existsSync(sidechainDirectory)) return plan; - 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")) { - plan.bytes += recordClaudeSidechainFile(plan, sidechainDirectory, entryPath); - continue; - } - const metaMatch = CLAUDE_SUBAGENT_META_FILE.exec(entry.name); - if (!metaMatch?.[1]) continue; - const metaBytes = claudeSidechainFileSize(entryPath); - if (metaBytes === null) continue; - plan.bytes += metaBytes; - plan.metaFiles.push({ entryPath, agentId: metaMatch[1] }); - } - } - return plan; -} - -function storeClaudeSubagentMeta( - history: ClaudeSidechainHistory, - file: { entryPath: string; agentId: string }, -): void { - // 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 contents = readClaudeSidechainFile(file.entryPath); - if (contents === null) return; - try { - const meta = parseClaudeSubagentMeta(contents); - if (meta) history.metaByAgentId.set(file.agentId, meta); - } catch { - // Undocumented internals: a malformed sidecar must never fail ingestion. - } -} - -function storeClaudeSidechainContents( - history: ClaudeSidechainHistory, - file: ClaudeSidechainFile, -): void { - const contents = readClaudeSidechainFile(file.entryPath); - if (contents === null) return; - if (!file.workflowRunId) { - history.contents.push(contents); - return; - } - const workflowContents = history.workflowSidechainContentsByRunId.get(file.workflowRunId) ?? []; - workflowContents.push(contents); - history.workflowSidechainContentsByRunId.set(file.workflowRunId, workflowContents); -} - -function readClaudeSidechainHistory( - historyPath: string, - parentBytes: number, -): 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(), - }; - - // Stat every file before reading any, so an over-budget session is decided from directory - // metadata alone instead of partial contents held in memory while the rest is still loading. - const summaries = collectClaudeWorkflowSummaryPaths(sessionDirectory); - const plan = collectClaudeSidechainFilePlan(sidechainDirectory); - const sidechainBytes = summaries.bytes + plan.bytes; - const budgetBytes = claudeReplayBudgetBytes(); - // A session with no sidechain files at all has nothing to skip, however large the parent is. - if (sidechainBytes > 0 && parentBytes + sidechainBytes > budgetBytes) { - history.skippedForBudget = { parentBytes, sidechainBytes, budgetBytes }; - return history; - } - - for (const summaryPath of summaries.paths) { - const contents = readClaudeSidechainFile(summaryPath); - if (contents !== null) history.workflowContents.push(contents); - } - for (const metaFile of plan.metaFiles) { - storeClaudeSubagentMeta(history, metaFile); - } - for (const file of plan.sidechainFiles) { - storeClaudeSidechainContents(history, file); - } - return history; -} - -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..e37154f134 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/replay-memory.fixture.ts @@ -0,0 +1,186 @@ +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"; + +export function generateReplayCorpus(root: string): { 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((395 * 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++) { + 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((362 * 1024 * 1024) / 269 / 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]; + 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): { 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((320 * 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..ff1f91ddb7 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); + observations.push(...timeline); + 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,9 @@ 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)); + observations.push( + ...observeSubagent(subagent, parent, input.convertEntry, toolOwners, ownerId), + ); 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..0faa24059a 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); @@ -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..c342e4aa58 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/transcript-history.ts @@ -0,0 +1,274 @@ +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, +): Generator<{ entry: TranscriptEntry; location: RecordLocation }> { + let fd: number; + try { + fd = fs.openSync(file, "r"); + } catch { + return; + } + try { + let remaining = fs.fstatSync(fd).size; + let offset = 0; + let pieces: Buffer[] = []; + let length = 0; + while (remaining > 0) { + const buffer = Buffer.allocUnsafe(Math.min(64 * 1024, remaining)); + const count = fs.readSync(fd, buffer, 0, buffer.length, null); + if (count === 0) break; + remaining -= count; + let start = 0; + let end: number; + while ((end = buffer.indexOf(10, start)) !== -1 && end < count) { + pieces.push(buffer.subarray(start, end)); + length += end - start; + const entry = parseRecord(Buffer.concat(pieces, length).toString("utf8")); + const location = { + file, + offset, + length, + timestamp: normalizeProviderReplayTimestamp(entry?.timestamp), + }; + offset += length + 1; + pieces = []; + length = 0; + if (entry) yield { entry, location }; + start = end + 1; + } + if (start < count) { + pieces.push(buffer.subarray(start, count)); + length += count - start; + } + } + const entry = parseRecord(Buffer.concat(pieces, length).toString("utf8")); + if (entry) + yield { + entry, + location: { + file, + offset, + length, + timestamp: normalizeProviderReplayTimestamp(entry.timestamp), + }, + }; + } finally { + fs.closeSync(fd); + } +} + +/** Repeatable sources let ownership be resolved before replay without retaining raw records. */ +function records(file: string): 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; + try { + for (const location of locations) { + if (currentFile !== location.file) { + if (fd !== null) fs.closeSync(fd); + fd = null; + currentFile = location.file; + try { + fd = fs.openSync(currentFile, "r"); + } catch { + continue; + } + } + if (fd === null) continue; + 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) break; + read += count; + } + const entry = parseRecord(buffer.subarray(0, read).toString("utf8")); + if (entry) yield entry; + } + } finally { + if (fd !== null) fs.closeSync(fd); + } + }, + }; +} + +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 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"); + } 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 function readClaudeReplayHistory(historyPath: string): { + parentEntries: Iterable; + subagents: { + agentId: string; + meta: ClaudeSubagentMeta | null; + entries: Iterable; + }[]; + workflows: ClaudeWorkflowRun[]; + workflowEntriesByRunId: Map>; +} { + 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 filesByAgentId = new Map(); + 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); + } + } + return { + parentEntries: selectRecords([historyPath], (entry) => entry.isSidechain !== true), + subagents: [...filesByAgentId].map(([agentId, files]) => ({ + agentId, + meta: metaByAgentId.get(agentId) ?? null, + entries: selectRecords( + files, + (entry) => entry.isSidechain === true && entry.agentId === agentId, + ), + })), + workflows: readWorkflows(sessionDirectory), + workflowEntriesByRunId: new Map( + [...workflowFiles].map(([runId, files]) => [runId, workflowRecords(files)]), + ), + }; +} + +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; +}