diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index 8662ee3287..eb6b978bb3 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. +- **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). - **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 8701ab87ec..65aefd294b 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,7 @@ 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 { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { createTestLogger } from "../../../../test-utils/test-logger.js"; @@ -432,6 +433,221 @@ 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], + }); + + 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("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[] = []; + 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"), + ); + + 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); + }); + + 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( + claudeProjectDirSync(cwd, { configDir }), + "replay-session", + "subagents", + ); + writeSubagent({ + subagentDir: subagentDirectory, + agentId: "wf-bulk-agent", + sidechainLines: [bulkEntry], + }); + + 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); + }); + 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 aa60e40462..892c7acae7 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -4937,10 +4937,14 @@ class ClaudeAgentSession implements AgentSession { return; } const content = fs.readFileSync(historyPath, "utf8"); - const replay = this.ingestPersistedSidechains( - content, - readClaudeSidechainHistory(historyPath), - ); + 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); } catch (error) { this.logger.warn( @@ -4970,11 +4974,12 @@ class ClaudeAgentSession implements AgentSession { parentContent: string, sidechains: ClaudeSidechainHistory, ): ClaudeReplayOwnership { - const parentEntries = parseClaudeHistoryRecords(parentContent).filter( - (entry) => entry.isSidechain !== true, - ); - const sidechainEntries = [parentContent, ...sidechains.contents] - .flatMap(parseClaudeHistoryRecords) + // 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); @@ -5906,37 +5911,100 @@ interface ClaudeSidechainHistory { 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$/; -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; +// 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.parseInt(process.env.PASEO_CLAUDE_REPLAY_BUDGET_MB ?? "", 10); + 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(); @@ -5949,41 +6017,90 @@ function readClaudeSidechainHistory(historyPath: string): ClaudeSidechainHistory } if (!entry.isFile()) continue; if (entry.name.endsWith(".jsonl")) { - recordClaudeSidechainContents(history, sidechainDirectory, entryPath); + plan.bytes += recordClaudeSidechainFile(plan, 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. - } + const metaBytes = claudeSidechainFileSize(entryPath); + if (metaBytes === null) continue; + plan.bytes += metaBytes; + plan.metaFiles.push({ entryPath, agentId: metaMatch[1] }); } } - return history; + return plan; } -function recordClaudeSidechainContents( +function storeClaudeSubagentMeta( history: ClaudeSidechainHistory, - sidechainDirectory: string, - entryPath: string, + file: { entryPath: string; agentId: 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) { + // 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(workflowRunId) ?? []; + const workflowContents = history.workflowSidechainContentsByRunId.get(file.workflowRunId) ?? []; workflowContents.push(contents); - history.workflowSidechainContentsByRunId.set(workflowRunId, workflowContents); + 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 {