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 <admin@tj7mun.org>
This commit is contained in:
Mohamed Boudra
2026-10-06 13:32:20 +02:00
co-authored by ZouYusheng
parent 448b1e44b0
commit 5991d2f789
7 changed files with 690 additions and 674 deletions
+1 -1
View File
@@ -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, `<session>/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, `<session>/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.
@@ -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 () => {
@@ -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<ClaudeHistoryEntry>,
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<typeof readClaudeReplayHistory>,
): 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<string, unknown>;
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<ClaudeHistoryEntry>,
): ClaudeReplayParentFacts {
const toolCalls = new Map<string, { title?: string; description?: string }>();
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<string, { failed: boolean }>();
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<string, string[]>;
/** agentId -> sidecar metadata, when Claude Code wrote one next to the transcript. */
metaByAgentId: Map<string, ClaudeSubagentMeta>;
/**
* 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<string, ClaudeHistoricalSubagentToolCall> {
const toolCalls = new Map<string, ClaudeHistoricalSubagentToolCall>();
const linksByAgentId = new Map<string, { toolCallId: string; failed: boolean }>();
const outcomes = new Map<string, { failed: boolean }>();
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<string, { toolCallId: string; failed: boolean }> {
const results = new Map<string, { toolCallId: string; failed: boolean }>();
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<string, ClaudeHistoryEntry[]> {
const entriesByAgentId = new Map<string, ClaudeHistoryEntry[]>();
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 {
@@ -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<string>();
const histories = new Set<string>();
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 };
}
@@ -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<ClaudeReplayEntry>;
/** 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<string, string>,
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<string, string>,
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;
@@ -47,8 +47,9 @@ export function parseClaudeWorkflowRun(contents: string): ClaudeWorkflowRun | nu
*/
export function observeReplayWorkflows(input: {
workflows: readonly ClaudeWorkflowRun[];
parentEntries: readonly ClaudeWorkflowParentEntry[];
entriesByRunId?: ReadonlyMap<string, readonly ClaudeReplayEntry[]>;
parentEntries: Iterable<ClaudeWorkflowParentEntry>;
/** Repeatable entries in chronological order, supplied by the transcript reader. */
entriesByRunId?: ReadonlyMap<string, Iterable<ClaudeReplayEntry>>;
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<ClaudeReplayEntry>;
convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[];
finishedAt?: string;
}): SubagentObservation[] {
const observations: SubagentObservation[] = [];
const replayedAssistantText = new Set<string>();
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<string, string> {
function readWorkflowLinks(entries: Iterable<ClaudeWorkflowParentEntry>): Map<string, string> {
const workflowToolCallIds = new Set<string>();
const toolCallIdByRunId = new Map<string, string>();
@@ -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<TranscriptEntry> {
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<TranscriptEntry> {
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<TranscriptEntry> {
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<TranscriptEntry>;
subagents: {
agentId: string;
meta: ClaudeSubagentMeta | null;
entries: Iterable<TranscriptEntry>;
}[];
workflows: ClaudeWorkflowRun[];
workflowEntriesByRunId: Map<string, Iterable<TranscriptEntry>>;
} {
const sessionDirectory = path.join(
path.dirname(historyPath),
path.basename(historyPath, ".jsonl"),
);
const sidechainDirectory = path.join(sessionDirectory, "subagents");
const sidechainFiles = [historyPath];
const workflowFiles = new Map<string, string[]>();
const metaByAgentId = new Map<string, ClaudeSubagentMeta>();
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<string, string[]>();
for (const file of sidechainFiles) {
const agentsInFile = new Set<string>();
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;
}