puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-10-27 11:42:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Keep the daemon responsive during Claude history replay (#7940)

Browse Source 
* fix(server): keep Claude history replay responsive

Load persisted history through session readiness rather than the constructor. Async descriptor-owned replay preserves ordering while serving other requests, and rebind retires the old query before rebuilding history and rewind state.

* fix(server): recover Claude history readiness after rebind errors

* test(server): calibrate Claude replay fixture for CI

* fix(server): report Claude rewind rebind failures to callers
This commit is contained in:
Mohamed Boudra
2026-10-27 10:23:59 +07:00
committed by GitHub
parent 7f6a805e0e
commit 173fca3c0b
10 changed files with 920 additions and 457 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/providers/claude/agent.history-load.test.ts
+326 -9
View File
@@ -1,9 +1,13 @@
import fs, { chmodSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { open } from "node:fs/promises";
import { performance } from "node:perf_hooks";
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 { createTestPaseoDaemon } from "../../../test-utils/paseo-daemon.js";
import type { ClaudeRewindSdk } from "./rewind.js";
import { ClaudeAgentClient } from "./agent.js";
import { claudeProjectDirSync } from "./project-dir.js";
import { withClaudeReplayHistory } from "./transcript-history.js";
@@ -32,7 +36,10 @@ describe("ClaudeAgentSession persisted history load", () => {
let configDir: string;
let records: LogRecord[];
async function resume(queryFactory?: ClaudeQueryFactory): Promise<AgentSession> {
async function resume(
queryFactory?: ClaudeQueryFactory,
rewindSdk?: ClaudeRewindSdk,
): Promise<AgentSession> {
const client = new ClaudeAgentClient({
logger: pino({ level: "info" }, { write: (line: string) => records.push(JSON.parse(line)) }),
queryFactory:
@@ -42,10 +49,36 @@ describe("ClaudeAgentSession persisted history load", () => {
}),
resolveBinary: async () => process.execPath,
resolveVersion: async () => "2.1.220",
rewindSdk,
});
return client.resumeSession({ provider: "claude", sessionId: SESSION_ID }, { cwd });
}
function replyFactory(text: string, messageId = "live-reply"): ClaudeQueryFactory {
return (() => {
const messages = [
{ type: "assistant", uuid: messageId, message: { content: text } },
{ type: "result", subtype: "success", total_cost_usd: 0 },
];
return {
async next() {
const value = messages.shift();
return value ? { done: false, value } : { done: true, value: undefined };
},
close() {},
async return() {
return { done: true, value: undefined };
},
async supportedCommands() {
return [];
},
[Symbol.asyncIterator]() {
return this;
},
};
}) as ClaudeQueryFactory;
}
function transcriptPath(): string {
return path.join(claudeProjectDirSync(cwd, { configDir }), `${SESSION_ID}.jsonl`);
}
@@ -65,6 +98,288 @@ describe("ClaudeAgentSession persisted history load", () => {
rmSync(tempRoot, { recursive: true, force: true });
});
test("serves unrelated health requests while resuming 32 MiB of ordered history", async () => {
const startedAt = performance.now();
mkdirSync(path.dirname(transcriptPath()), { recursive: true });
const file = await open(transcriptPath(), "w");
const text = "x".repeat(16 * 1024);
let bytes = 0;
let count = 0;
try {
while (bytes < 32 * 1024 * 1024) {
const line =
JSON.stringify({ type: "assistant", message: { content: `${count}:${text}` } }) + "\n";
bytes += Buffer.byteLength(line);
await file.write(line);
count++;
}
} finally {
await file.close();
}
const daemon = await createTestPaseoDaemon();
let lastTick = performance.now();
let longestGap = 0;
const probe = setInterval(() => {
const now = performance.now();
longestGap = Math.max(longestGap, now - lastTick);
lastTick = now;
}, 2);
let loading = true;
let answeredDuringLoad = false;
const health = fetch(`http://127.0.0.1:${daemon.port}/api/health`).then(async (response) => {
expect(response.status).toBe(200);
await response.text();
answeredDuringLoad = loading;
return undefined;
});
let session: AgentSession | undefined;
try {
session = await resume(replyFactory("live after large resume"));
let index = 0;
for await (const event of session.streamHistory()) {
expect(event).toEqual({
type: "timeline",
provider: "claude",
timestamp: undefined,
item: { type: "assistant_message", text: `${index}:${text}` },
});
index++;
}
loading = false;
await health;
await new Promise((resolve) => setTimeout(resolve, 5));
console.info(
JSON.stringify({
bytes,
count,
longestGap,
answeredDuringLoad,
elapsedMs: performance.now() - startedAt,
}),
);
expect(index).toBe(count);
expect(longestGap).toBeLessThan(50);
expect(answeredDuringLoad).toBe(true);
const live = await collectSessionTurnEvents(session, "follow up");
expect(
live
.filter((event) => event.type === "timeline" && event.item.type === "assistant_message")
.map((event) => (event.type === "timeline" ? event.item : null)),
).toEqual([
{ type: "assistant_message", messageId: "live-reply", text: "live after large resume" },
]);
} finally {
clearInterval(probe);
await session?.close();
await daemon.close();
}
}, 60_000);
test("rebinds ordered history before returning from rewind and accepts a live turn", async () => {
const entries = [
{ type: "user", uuid: "u1", message: { content: "first" } },
{ type: "assistant", uuid: "a1", message: { content: "first reply" } },
{ type: "user", uuid: "u2", message: { content: "second" } },
{ type: "assistant", uuid: "a2", message: { content: "second reply" } },
];
mkdirSync(path.dirname(transcriptPath()), { recursive: true });
writeFileSync(transcriptPath(), entries.map((entry) => JSON.stringify(entry)).join("\n"));
const forkId = "rebound-history";
const forkFile = path.join(path.dirname(transcriptPath()), `${forkId}.jsonl`);
writeFileSync(
forkFile,
entries
.slice(0, 2)
.map((entry) => JSON.stringify(entry))
.join("\n"),
);
const forks: unknown[] = [];
const late = {
type: "assistant",
uuid: "retired-reply",
message: { content: "retired query" },
};
let pendingRead: ((value: unknown) => void) | undefined;
let releaseRetirement!: () => void;
const retirementGate = new Promise<void>((resolve) => {
releaseRetirement = resolve;
});
let retirementStarted!: () => void;
const retiring = new Promise<void>((resolve) => {
retirementStarted = resolve;
});
let firstQuery = true;
let retired = false;
const queryFactory = (() => {
if (!firstQuery) return replyFactory("live after rebind")({} as never);
firstQuery = false;
const messages = [
{ type: "assistant", uuid: "before-rebind", message: { content: "before rebind" } },
{ type: "result", subtype: "success", total_cost_usd: 0 },
];
return {
async next() {
const value = messages.shift();
if (value) return { done: false, value };
if (retired) return { done: true, value: undefined };
return new Promise((resolve) => {
pendingRead = resolve;
});
},
close() {
retired = true;
pendingRead?.({ done: false, value: late });
},
async return() {
retirementStarted();
await retirementGate;
return { done: true, value: undefined };
},
async supportedCommands() {
return [];
},
[Symbol.asyncIterator]() {
return this;
},
};
}) as ClaudeQueryFactory;
const session = await resume(queryFactory, {
async forkSession(sessionId, options) {
forks.push({ sessionId, ...options });
return { sessionId: forkId };
},
});
try {
const original = await Array.fromAsync(session.streamHistory());
expect(original).toHaveLength(4);
await collectSessionTurnEvents(session, "before rewind");
const emitted: AgentStreamEvent[] = [];
const subscription = session.subscribe((event) => emitted.push(event));
let rewindDone = false;
const rewind = (async () => {
await session.revertConversation!({ messageId: "u2" });
rewindDone = true;
})();
await retiring;
let historyDone = false;
const reading = (async () => {
const history = await Array.fromAsync(session.streamHistory());
historyDone = true;
return history;
})();
const readingAgain = Array.fromAsync(session.streamHistory());
await Promise.resolve();
expect(rewindDone).toBe(false);
expect(historyDone).toBe(false);
releaseRetirement();
await rewind;
subscription();
expect(retired).toBe(true);
expect(
emitted.filter(
(event) => event.type === "timeline" && event.item.type === "assistant_message",
),
).toEqual([
{
type: "timeline",
provider: "claude",
item: {
type: "assistant_message",
text: `Claude switched to a new session: ${SESSION_ID} -> ${forkId}`,
},
},
]);
expect(forks).toEqual([
{ sessionId: SESSION_ID, upToMessageId: "a1", transcriptPath: transcriptPath() },
]);
expect(session.id).toBe(forkId);
const [history, secondRead] = await Promise.all([reading, readingAgain]);
expect(history).toEqual(original.slice(0, 2));
expect(secondRead).toEqual([]);
const live = await collectSessionTurnEvents(session, "continue the fork");
expect(
live
.filter((event) => event.type === "timeline" && event.item.type === "assistant_message")
.map((event) => (event.type === "timeline" ? event.item : null)),
).toEqual([
{ type: "assistant_message", messageId: "live-reply", text: "live after rebind" },
]);
} finally {
releaseRetirement();
await session.close();
}
});
test("reports a failed rewind without poisoning readiness when query retirement throws", async () => {
mkdirSync(path.dirname(transcriptPath()), { recursive: true });
writeFileSync(
transcriptPath(),
[
{ type: "user", uuid: "u1", message: { content: "first" } },
{ type: "assistant", uuid: "a1", message: { content: "reply" } },
{ type: "user", uuid: "u2", message: { content: "second" } },
]
.map((entry) => JSON.stringify(entry))
.join("\n"),
);
let firstQuery = true;
let finishRead: ((value: unknown) => void) | undefined;
const queryFactory = ((input) => {
const query = replyFactory(
firstQuery ? "before retirement" : "still available",
firstQuery ? "before-retirement" : "live-reply",
)(input);
if (!firstQuery) return query;
firstQuery = false;
let reads = 0;
return {
...query,
async next() {
if (++reads <= 2) return query.next();
return new Promise((resolve) => {
finishRead = resolve;
});
},
close() {
finishRead?.({ done: true, value: undefined });
throw new Error("retirement failed");
},
[Symbol.asyncIterator]() {
return this;
},
};
}) as ClaudeQueryFactory;
const session = await resume(queryFactory, {
async forkSession() {
return { sessionId: "failed-rebind" };
},
});
try {
await Array.fromAsync(session.streamHistory());
await collectSessionTurnEvents(session, "before rewind");
await expect(session.revertConversation!({ messageId: "u2" })).rejects.toThrow(
"retirement failed",
);
expect(records).toContainEqual(
expect.objectContaining({
level: pino.levels.values.warn,
msg: "Failed to rebind Claude session",
sessionId: "failed-rebind",
err: expect.objectContaining({ message: "retirement failed" }),
}),
);
expect(session.id).toBe(SESSION_ID);
const live = await collectSessionTurnEvents(session, "after failed rebind");
expect(
live
.filter((event) => event.type === "timeline" && event.item.type === "assistant_message")
.map((event) => (event.type === "timeline" ? event.item : null)),
).toEqual([{ type: "assistant_message", messageId: "live-reply", text: "still available" }]);
} finally {
await session.close();
}
});
test.each(["file_path", "files"])(
"replays %s without probing targets and preserves live existence checks",
async (field) => {
@@ -300,21 +615,21 @@ describe("ClaudeAgentSession persisted history load", () => {
);
});
test("loads readable history when the temporary directory is unavailable", () => {
test("loads readable history when the temporary directory is unavailable", async () => {
const file = transcriptPath();
const entry = { type: "assistant", message: { content: [{ type: "text", text: "saved" }] } };
mkdirSync(path.dirname(file), { recursive: true });
writeFileSync(file, JSON.stringify(entry));
const unavailable = path.join(tempRoot, "unavailable-tmp");
for (const variable of ["TMPDIR", "TEMP", "TMP"]) vi.stubEnv(variable, unavailable);
withClaudeReplayHistory(file, (history) => {
expect([...history.parentEntries]).toEqual([entry]);
await withClaudeReplayHistory(file, async (history) => {
expect(await Array.fromAsync(history.parentEntries)).toEqual([entry]);
});
});
test.each(["parent", "child", "workflow"])(
"preserves indexed %s history when its transcript disappears between passes",
(target) => {
async (target) => {
const parent = transcriptPath();
const sessionDirectory = parent.slice(0, -".jsonl".length);
const child = path.join(sessionDirectory, "subagents", "agent-child.jsonl");
@@ -334,14 +649,16 @@ describe("ClaudeAgentSession persisted history load", () => {
writeFileSync(parent, JSON.stringify(parentEntry));
writeFileSync(child, JSON.stringify(childEntry));
writeFileSync(workflow, JSON.stringify(childEntry));
withClaudeReplayHistory(parent, (history) => {
await withClaudeReplayHistory(parent, async (history) => {
rmSync({ parent, child, workflow }[target]);
// Ownership facts and visible payloads are separate passes over the same source.
for (let pass = 0; pass < 2; pass++) {
expect([...history.parentEntries]).toEqual([parentEntry]);
expect(await Array.fromAsync(history.parentEntries)).toEqual([parentEntry]);
expect(history.subagents).toHaveLength(1);
expect([...history.subagents[0]!.entries]).toEqual([childEntry]);
expect([...history.workflowEntriesByRunId.get("wf_test")!]).toEqual([childEntry]);
expect(await Array.fromAsync(history.subagents[0]!.entries)).toEqual([childEntry]);
expect(await Array.fromAsync(history.workflowEntriesByRunId.get("wf_test")!)).toEqual([
childEntry,
]);
}
});
},
packages/server/src/server/agent/providers/claude/agent.ts
+160 -98
View File
@@ -1,3 +1,4 @@
import { setImmediate } from "node:timers/promises";
import { ClaudeRuntimeSafety } from "./runtime-safety.js";
import { validateProviderOptions } from "../../provider-options.js";
import type { ChildProcess } from "node:child_process";
@@ -82,7 +83,7 @@ import {
type ClaudeRewindSdk,
} from "./rewind.js";
import { normalizeProviderReplayTimestamp } from "../../provider-history-timestamps.js";
import { claudeConfigDir, claudeProjectDirSync, claudeTranscriptPathSync } from "./project-dir.js";
import { claudeConfigDir, claudeProjectDirSync, claudeTranscriptPath } from "./project-dir.js";
import { THINKING_APPLIES_NEXT_TURN_NOTICE } from "../../provider-notices.js";
import {
isProviderImageMarkdown,
@@ -1634,7 +1635,7 @@ export class ClaudeAgentClient implements AgentClient {
cwd: merged.cwd,
};
const claudeConfig = this.assertConfig(mergedConfig);
return new ClaudeAgentSession(claudeConfig, {
const session = new ClaudeAgentSession(claudeConfig, {
defaults: this.defaults,
runtimeSettings: this.runtimeSettings,
handle,
@@ -1645,6 +1646,8 @@ export class ClaudeAgentClient implements AgentClient {
resolveBinary: this.resolveBinary,
rewindSdk: this.rewindSdk,
});
await session.ensureHistoryReady();
return session;
}
async getCatalogCacheKey(_options: FetchCatalogOptions): Promise<string> {
@@ -2329,6 +2332,9 @@ class ClaudeAgentSession implements AgentSession {
{ type: "provider_subagent" }
>[] = [];
private historyPending = false;
// Readiness never rejects; history/rebind failures are logged at their owning boundary.
private historyReady: Promise<void> | null = null;
private historySessionId: string | null = null;
private replayingHistory = false;
private turnState: TurnState = "idle";
private nextTurnOrdinal = 1;
@@ -2382,7 +2388,7 @@ class ClaudeAgentSession implements AgentSession {
}
this.claudeSessionId = handle.sessionId;
this.persistence = handle;
this.loadPersistedHistory(handle.sessionId);
this.historySessionId = handle.sessionId;
} else {
this.claudeSessionId = null;
this.persistence = null;
@@ -2462,6 +2468,7 @@ class ClaudeAgentSession implements AgentSession {
prompt: AgentPromptInput,
options?: AgentRunOptions,
): Promise<{ turnId: string }> {
await this.ensureHistoryReady();
if (this.closed) {
throw new Error("Claude session is closed");
}
@@ -2632,6 +2639,7 @@ class ClaudeAgentSession implements AgentSession {
}
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
await this.ensureHistoryReady();
if (
!this.historyPending ||
(this.persistedHistory.length === 0 && this.persistedProviderSubagentEvents.length === 0)
@@ -2643,7 +2651,9 @@ class ClaudeAgentSession implements AgentSession {
this.persistedHistory = [];
this.persistedProviderSubagentEvents = [];
this.historyPending = false;
let replayed = 0;
for (const entry of history) {
if (++replayed % 32 === 0) await setImmediate();
yield {
type: "timeline",
item: entry.item,
@@ -2961,6 +2971,7 @@ class ClaudeAgentSession implements AgentSession {
"provider.claude.session_close.start",
);
this.closed = true;
await this.ensureHistoryReady();
this.rejectAllPendingPermissions(new Error("Claude session closed"));
this.cancelCurrentTurn?.();
this.subscribers.clear();
@@ -3001,7 +3012,7 @@ class ClaudeAgentSession implements AgentSession {
// (see `claude --help`), so the SDK's persistSession=false is silently dropped
// in stream-json mode. Sweep the transcript ourselves so ephemeral runs
// (metadata generator, branch-name generator) don't show up as resumable.
const historyPath = this.resolveHistoryPath(this.claudeSessionId);
const historyPath = await this.resolveHistoryPath(this.claudeSessionId);
if (historyPath) {
try {
await promises.rm(historyPath, { force: true });
@@ -3045,6 +3056,7 @@ class ClaudeAgentSession implements AgentSession {
}
async revertConversation(input: { messageId: string }): Promise<void> {
await this.ensureHistoryReady();
const target = this.resolveConversationRewindTarget(input.messageId);
if (target.kind === "fresh-session") {
this.startFreshConversationSession();
@@ -3054,16 +3066,15 @@ class ClaudeAgentSession implements AgentSession {
await revertClaudeConversation({
sdk: this.rewindSdk,
sessionId: currentSessionId,
transcriptPath: currentSessionId ? this.resolveHistoryPath(currentSessionId) : null,
transcriptPath: currentSessionId ? await this.resolveHistoryPath(currentSessionId) : null,
messageId: target.messageId,
resolveMessageId: (messageId) => this.resolveClaudeMessageId(messageId),
setSessionId: (sessionId) => {
this.rebindConversationSession(sessionId);
},
setSessionId: (sessionId) => this.rebindConversationSession(sessionId),
});
}
async revertFiles(input: { messageId: string }): Promise<void> {
await this.ensureHistoryReady();
const messageId = await this.resolveClaudeMessageId(input.messageId);
await revertClaudeFiles({
query: await this.ensureQuery(),
@@ -3238,38 +3249,52 @@ class ClaudeAgentSession implements AgentSession {
return candidates;
}
private rebindConversationSession(sessionId: string): void {
const oldSessionId = this.claudeSessionId;
this.claudeSessionId = sessionId;
this.pendingFreshSessionId = null;
this.persistence = null;
this.cachedRuntimeInfo = null;
this.queryRestartNeeded = true;
this.persistedHistory = [];
this.persistedProviderSubagentEvents = [];
this.historyPending = false;
this.userMessageIds = [];
this.emittedUserMessageIds.clear();
this.rewindTurnAnchors.length = 0;
this.taskState.reset();
this.loadPersistedHistory(sessionId);
if (oldSessionId && oldSessionId !== sessionId) {
this.dispatchEvents([
{
type: "timeline",
provider: "claude",
item: this.createClaudeSessionChangedNotice(oldSessionId, sessionId),
},
{
type: "thread_started",
provider: "claude",
sessionId,
},
]);
}
private rebindConversationSession(sessionId: string): Promise<void> {
const previousReady = this.ensureHistoryReady();
const rebind = (async () => {
await previousReady;
await this.retireQuery();
if (this.closed) return;
const oldSessionId = this.claudeSessionId;
this.claudeSessionId = sessionId;
this.pendingFreshSessionId = null;
this.persistence = null;
this.cachedRuntimeInfo = null;
this.queryRestartNeeded = true;
this.persistedHistory = [];
this.persistedProviderSubagentEvents = [];
this.historyPending = false;
this.userMessageIds = [];
this.emittedUserMessageIds.clear();
this.rewindTurnAnchors.length = 0;
this.taskState.reset();
this.historySessionId = sessionId;
await this.loadPersistedHistory(sessionId);
if (this.closed) return;
if (oldSessionId && oldSessionId !== sessionId) {
this.dispatchEvents([
{
type: "timeline",
provider: "claude",
item: this.createClaudeSessionChangedNotice(oldSessionId, sessionId),
},
{
type: "thread_started",
provider: "claude",
sessionId,
},
]);
}
})();
this.historyReady = rebind.catch((error) => {
this.logger.warn({ err: error, sessionId }, "Failed to rebind Claude session");
});
return rebind;
}
private startFreshConversationSession(): void {
this.historySessionId = null;
this.historyReady = Promise.resolve();
const sessionId = randomUUID();
this.claudeSessionId = sessionId;
this.pendingFreshSessionId = sessionId;
@@ -3390,7 +3415,43 @@ class ClaudeAgentSession implements AgentSession {
return { kind: "fresh-session" };
}
private async retireQuery(): Promise<void> {
const oldQuery = this.query;
if (!oldQuery) return;
const oldInput = this.input;
// Null out query/input BEFORE awaiting the old iterator's return so the
// old pump sees this.query !== activeQuery and skips failActiveTurns.
this.query = null;
this.input = null;
this.queryPumpPromise = null;
this.queryRestartNeeded = false;
// Ending the input retires the process on purpose. Detach first so its
// exit is not reported as a crash.
const retiredChild = this.childProcess;
this.childProcess = null;
if (retiredChild) this.failRunningRuntimeTasks();
oldInput?.end();
oldQuery.close?.();
try {
await oldQuery.return?.();
} catch {
/* ignore */
}
// Tree-kill the old process tree now that the SDK has cleaned up.
// If we skip this, MCP children of the previous claude process can
// survive as orphans when the session spawns a replacement query.
if (retiredChild) {
await terminateWithTreeKill(retiredChild, {
gracefulTimeoutMs: 2_000,
forceTimeoutMs: 2_000,
}).catch(() => {
/* process may already be dead */
});
}
}
private async ensureQuery(launchMode: PermissionMode = this.currentMode): Promise<Query> {
await this.ensureHistoryReady();
// Controls use the running turn's query. Pending launch settings take effect
// after that turn finishes, when its active query handle has been cleared.
if (this.query && (!this.queryRestartNeeded || this.activeForegroundQuery === this.query)) {
@@ -3398,37 +3459,7 @@ class ClaudeAgentSession implements AgentSession {
}
if (this.queryRestartNeeded && this.query) {
const oldQuery = this.query;
const oldInput = this.input;
// Null out query/input BEFORE awaiting the old iterator's return so the
// old pump sees this.query !== activeQuery and skips failActiveTurns.
this.query = null;
this.input = null;
this.queryPumpPromise = null;
this.queryRestartNeeded = false;
// Ending the input retires the process on purpose. Detach first so its
// exit is not reported as a crash.
const retiredChild = this.childProcess;
this.childProcess = null;
if (retiredChild) this.failRunningRuntimeTasks();
oldInput?.end();
oldQuery.close?.();
try {
await oldQuery.return?.();
} catch {
/* ignore */
}
// Tree-kill the old process tree now that the SDK has cleaned up.
// If we skip this, MCP children of the previous claude process can
// survive as orphans when the session spawns a replacement query.
if (retiredChild) {
await terminateWithTreeKill(retiredChild, {
gracefulTimeoutMs: 2_000,
forceTimeoutMs: 2_000,
}).catch(() => {
/* process may already be dead */
});
}
await this.retireQuery();
}
// Preserve claudeSessionId across query recreation so buildOptions() passes
@@ -4095,13 +4126,15 @@ class ClaudeAgentSession implements AgentSession {
);
};
const handlePumpedMessage = async (message: SDKMessage): Promise<boolean> => {
if (!this.ownsQuery(activeQuery)) return true;
this.processingSdkMessages++;
try {
this.runtimeSafety.observe(message);
logRawMessage(message);
consecutiveInterruptAbortRecoveries = 0;
if (await this.handleMissingResumedConversation(message, activeQuery)) return true;
await this.routeSdkMessageFromPump(message);
if (!this.ownsQuery(activeQuery)) return true;
this.routeSdkMessageFromPump(message);
return false;
} finally {
this.processingSdkMessages--;
@@ -4221,7 +4254,11 @@ class ClaudeAgentSession implements AgentSession {
return isClaudeTurnInit(message) && !this.activeForegroundTurnId;
}
private async routeSdkMessageFromPump(message: SDKMessage): Promise<void> {
private ownsQuery(query: Query): boolean {
return !this.closed && this.query === query;
}
private routeSdkMessageFromPump(message: SDKMessage): void {
if (this.consumeSessionStateChange(message)) {
return;
}
@@ -4255,7 +4292,7 @@ class ClaudeAgentSession implements AgentSession {
"provider.claude.parsed_event",
);
const events = await this.buildPumpedMessageEvents(message, identifiers.messageId, turnId);
const events = this.buildPumpedMessageEvents(message, identifiers.messageId, turnId);
if (events.length === 0) {
return;
@@ -4290,11 +4327,11 @@ class ClaudeAgentSession implements AgentSession {
this.dispatchEvents(events);
}
private async buildPumpedMessageEvents(
private buildPumpedMessageEvents(
message: SDKMessage,
messageIdHint: string | null,
turnId: string | null,
): Promise<AgentStreamEvent[]> {
): AgentStreamEvent[] {
const messageEvents = this.translateMessageToEvents(message, {
suppressAssistantText: true,
suppressReasoning: true,
@@ -5272,22 +5309,31 @@ class ClaudeAgentSession implements AgentSession {
}
}
private loadPersistedHistory(sessionId: string): void {
ensureHistoryReady(): Promise<void> {
if (!this.historyReady) {
this.historyReady = this.historySessionId
? this.loadPersistedHistory(this.historySessionId)
: Promise.resolve();
}
return this.historyReady;
}
private async loadPersistedHistory(sessionId: string): Promise<void> {
let historyPath: string | null = null;
this.replayingHistory = true;
try {
this.taskState.reset();
historyPath = this.resolveHistoryPath(sessionId);
if (!historyPath || !fs.existsSync(historyPath)) {
historyPath = await this.resolveHistoryPath(sessionId);
if (!historyPath || !(await transcriptExists(historyPath))) {
this.logger.info(
{ sessionId, cwd: this.config.cwd, historyPath },
"No Claude transcript to load history from",
);
return;
}
withClaudeReplayHistory(historyPath, (history) => {
const replay = this.ingestPersistedSidechains(history);
this.ingestPersistedHistory(history.parentEntries, replay);
await withClaudeReplayHistory(historyPath, async (history) => {
const replay = await this.ingestPersistedSidechains(history);
await this.ingestPersistedHistory(history.parentEntries, replay);
});
} catch (error) {
this.logger.warn(
@@ -5299,12 +5345,12 @@ class ClaudeAgentSession implements AgentSession {
}
}
private ingestPersistedHistory(
entries: Iterable<ClaudeHistoryEntry>,
private async ingestPersistedHistory(
entries: Iterable<ClaudeHistoryEntry> | AsyncIterable<ClaudeHistoryEntry>,
replay: ClaudeReplayOwnership,
): void {
): Promise<void> {
const timeline: PersistedTimelineEntry[] = [];
for (const entry of entries) {
for await (const entry of entries) {
this.ingestPersistedHistoryEntry(entry, timeline, replay);
}
@@ -5314,27 +5360,33 @@ class ClaudeAgentSession implements AgentSession {
}
}
private ingestPersistedSidechains(history: ClaudeReplayHistory): ClaudeReplayOwnership {
const parentFacts = readClaudeReplayParentFacts(history.parentEntries);
private async ingestPersistedSidechains(
history: ClaudeReplayHistory,
): Promise<ClaudeReplayOwnership> {
const parentFacts = await 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: history.subagents.map((subagent) => ({
const subagents = [];
for (const subagent of history.subagents) {
subagents.push({
...subagent,
parentFacts: readClaudeReplayParentFacts(subagent.entries),
})),
parentFacts: await readClaudeReplayParentFacts(subagent.entries),
});
}
const subagentReplay = await observeReplaySubagents({
subagents,
parent: parentFacts,
convertEntry: (entry) => this.convertHistoryEntry(entry),
});
const observations = [
...subagentReplay.observations,
...observeReplayWorkflows({
...(await observeReplayWorkflows({
workflows: history.workflows,
parentEntries: history.parentEntries,
entriesByRunId: history.workflowEntriesByRunId,
convertEntry: (entry) => this.convertHistoryEntry(entry),
}),
})),
];
const restoredProviderSubagentIds = new Set(
observations
@@ -5421,10 +5473,10 @@ class ClaudeAgentSession implements AgentSession {
}
}
private resolveHistoryPath(sessionId: string): string | null {
private async resolveHistoryPath(sessionId: string): Promise<string | null> {
const cwd = this.config.cwd;
if (!cwd) return null;
return claudeTranscriptPathSync({
return claudeTranscriptPath({
cwd,
sessionId,
configDir: claudeConfigDir(this.harnessEnvironment),
@@ -6191,13 +6243,13 @@ 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(
entries: Iterable<ClaudeHistoryEntry>,
): ClaudeReplayParentFacts {
async function readClaudeReplayParentFacts(
entries: Iterable<ClaudeHistoryEntry> | AsyncIterable<ClaudeHistoryEntry>,
): Promise<ClaudeReplayParentFacts> {
const toolCalls = new Map<string, { title?: string; description?: string }>();
const linksByAgentId = new Map<string, { toolCallId: string; failed: boolean }>();
const outcomes = new Map<string, { failed: boolean }>();
for (const entry of entries) {
for await (const entry of entries) {
const content = toObjectRecord(entry.message)?.content;
if (!Array.isArray(content)) continue;
for (const value of content) {
@@ -6729,7 +6781,7 @@ async function readClaudeRenamedSessionTitle(input: {
}): Promise<string | null> {
let content: string;
try {
content = await fsPromises.readFile(claudeTranscriptPathSync(input), "utf8");
content = await fsPromises.readFile(await claudeTranscriptPath(input), "utf8");
} catch (error) {
// A session with no transcript on disk has no rename to carry over. Every
// other read failure is a real fault and stays explicit.
@@ -6829,3 +6881,13 @@ function readClaudeCommandLifecycle(message: unknown): ClaudeCommandLifecycle |
state: record.state as ClaudeCommandLifecycle["state"],
};
}
async function transcriptExists(file: string): Promise<boolean> {
try {
await fs.promises.access(file);
return true;
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") return false;
throw error;
}
}
packages/server/src/server/agent/providers/claude/project-dir.ts
+18 -8
View File
@@ -1,5 +1,5 @@
import { existsSync, readdirSync, realpathSync } from "node:fs";
import { realpath } from "node:fs/promises";
import { realpathSync } from "node:fs";
import { realpath, access, readdir } from "node:fs/promises";
import { homedir } from "node:os";
import { join } from "node:path";
@@ -35,21 +35,21 @@ export function claudeProjectDirSync(cwd: string, options?: ClaudeProjectDirOpti
// one `cwd` encodes to when the project moved, the daemon migrated, or Claude Code sees the
// cwd under another name (a Windows binary under WSL). Returns the expected path when no
// transcript exists anywhere.
export function claudeTranscriptPathSync(input: {
export async function claudeTranscriptPath(input: {
cwd: string;
sessionId: string;
configDir?: string;
}): string {
}): Promise<string> {
const options = { configDir: input.configDir };
const fileName = `${input.sessionId}.jsonl`;
const expected = join(claudeProjectDirSync(input.cwd, options), fileName);
if (existsSync(expected)) {
const expected = join(await claudeProjectDir(input.cwd, options), fileName);
if (await exists(expected)) {
return expected;
}
const projectsRoot = join(resolveConfigDir(options), "projects");
let projectDirs: string[];
try {
projectDirs = readdirSync(projectsRoot);
projectDirs = await readdir(projectsRoot);
} catch (error) {
if (isMissingDirectoryError(error)) {
return expected;
@@ -58,13 +58,23 @@ export function claudeTranscriptPathSync(input: {
}
for (const projectDir of projectDirs) {
const candidate = join(projectsRoot, projectDir, fileName);
if (existsSync(candidate)) {
if (await exists(candidate)) {
return candidate;
}
}
return expected;
}
async function exists(file: string): Promise<boolean> {
try {
await access(file);
return true;
} catch (error) {
if (isMissingDirectoryError(error)) return false;
throw error;
}
}
function isMissingDirectoryError(error: unknown): boolean {
const code = (error as NodeJS.ErrnoException | null)?.code;
return code === "ENOENT" || code === "ENOTDIR";
packages/server/src/server/agent/providers/claude/rewind.ts
+3 -3
View File
@@ -76,7 +76,7 @@ export async function revertClaudeConversation(input: {
transcriptPath: string | null;
messageId: string;
resolveMessageId?: (messageId: string) => string | Promise<string>;
setSessionId: (sessionId: string) => void;
setSessionId: (sessionId: string) => void | Promise<void>;
}): Promise<void> {
if (!input.sessionId || !input.transcriptPath) {
throw new Error("Claude session is not ready for rewind");
@@ -86,7 +86,7 @@ export async function revertClaudeConversation(input: {
upToMessageId: messageId,
transcriptPath: input.transcriptPath,
});
input.setSessionId(fork.sessionId);
await input.setSessionId(fork.sessionId);
}
export async function revertClaudeFiles(input: {
@@ -108,7 +108,7 @@ export async function revertClaudeConversationAndFiles(input: {
transcriptPath: string | null;
messageId: string;
resolveMessageId?: (messageId: string) => string | Promise<string>;
setSessionId: (sessionId: string) => void;
setSessionId: (sessionId: string) => void | Promise<void>;
}): Promise<void> {
await revertClaudeFiles({
query: input.query,
packages/server/src/server/agent/providers/claude/subagents/replay-source.test.ts
+306 -262
View File
@@ -45,7 +45,7 @@ function applyToStore(observations: SubagentObservation[]): ProviderSubagentDesc
}
describe("parseClaudeSubagentMeta", () => {
it("reads the real sidecar shape", () => {
it("reads the real sidecar shape", async () => {
expect(
parseClaudeSubagentMeta(
'{"agentType":"general-purpose","description":"Reply banana","toolUseId":"toolu_013i","spawnDepth":1}',
@@ -68,7 +68,7 @@ describe("parseClaudeSubagentMeta", () => {
expect(parseClaudeSubagentMeta(contents)).toBeNull();
});
it("keeps the good fields when one field has the wrong type", () => {
it("keeps the good fields when one field has the wrong type", async () => {
expect(parseClaudeSubagentMeta('{"agentType":"Explore","toolUseId":42}')).toEqual({
agentType: "Explore",
description: undefined,
@@ -79,12 +79,14 @@ describe("parseClaudeSubagentMeta", () => {
});
describe("observeReplaySubagents", () => {
it("links a subagent to its Task call through meta.toolUseId", () => {
const observations = observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
parent: parentWithTaskCall(),
convertEntry: () => [],
}).observations;
it("links a subagent to its Task call through meta.toolUseId", async () => {
const observations = (
await observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
parent: parentWithTaskCall(),
convertEntry: () => [],
})
).observations;
expect(observations[0]).toMatchObject({
kind: "declared",
@@ -95,31 +97,33 @@ describe("observeReplaySubagents", () => {
});
});
it("replays descendants beneath the sidechain that declared their Task", () => {
it("replays descendants beneath the sidechain that declared their Task", async () => {
const nestedToolUseId = "toolu_012rzYnFZA";
const observations = observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID, spawnDepth: 1 },
entries: [],
parentFacts: {
toolCalls: new Map([
[nestedToolUseId, { title: "Explore", description: "Nested audit" }],
]),
linksByAgentId: new Map(),
outcomesByToolCallId: new Map([[nestedToolUseId, { failed: false }]]),
const observations = (
await observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID, spawnDepth: 1 },
entries: [],
parentFacts: {
toolCalls: new Map([
[nestedToolUseId, { title: "Explore", description: "Nested audit" }],
]),
linksByAgentId: new Map(),
outcomesByToolCallId: new Map([[nestedToolUseId, { failed: false }]]),
},
},
},
{
agentId: "a6acb4b898",
meta: { toolUseId: nestedToolUseId, agentType: "Explore", spawnDepth: 2 },
entries: [],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [],
}).observations;
{
agentId: "a6acb4b898",
meta: { toolUseId: nestedToolUseId, agentType: "Explore", spawnDepth: 2 },
entries: [],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [],
})
).observations;
expect(observations).toContainEqual(
expect.objectContaining({
@@ -130,89 +134,143 @@ describe("observeReplaySubagents", () => {
);
});
it("keeps a subagent recorded before spawnDepth existed", () => {
const observations = observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
parent: parentWithTaskCall(),
convertEntry: () => [],
}).observations;
it("keeps a subagent recorded before spawnDepth existed", async () => {
const observations = (
await observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
parent: parentWithTaskCall(),
convertEntry: () => [],
})
).observations;
expect(observations[0]).toMatchObject({ kind: "declared", id: TOOL_USE_ID });
});
it("drops a meta-linked subagent when the parent never declared its Task", () => {
const observations = observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID, agentType: "Explore", description: "Find the code" },
entries: [],
},
],
parent: emptyParent(),
convertEntry: () => [],
}).observations;
it("drops a meta-linked subagent when the parent never declared its Task", async () => {
const observations = (
await observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID, agentType: "Explore", description: "Find the code" },
entries: [],
},
],
parent: emptyParent(),
convertEntry: () => [],
})
).observations;
expect(observations).toEqual([]);
});
it("falls back to the scraped link when there is no meta file", () => {
it("falls back to the scraped link when there is no meta file", async () => {
const parent: ClaudeReplayParentFacts = {
toolCalls: new Map([[TOOL_USE_ID, { title: "Explore" }]]),
linksByAgentId: new Map([[AGENT_ID, { toolCallId: TOOL_USE_ID, failed: false }]]),
outcomesByToolCallId: new Map(),
};
const [declared, ...rest] = observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
parent,
convertEntry: () => [],
}).observations;
const [declared, ...rest] = (
await observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
parent,
convertEntry: () => [],
})
).observations;
expect(declared).toMatchObject({ id: TOOL_USE_ID, toolCallId: TOOL_USE_ID, title: "Explore" });
expect(rest.at(-1)).toMatchObject({ kind: "status", status: "completed" });
});
it("drops an unlinkable subagent", () => {
const observations = observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
parent: emptyParent(),
convertEntry: () => [],
}).observations;
it("drops an unlinkable subagent", async () => {
const observations = (
await observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
parent: emptyParent(),
convertEntry: () => [],
})
).observations;
expect(observations).toEqual([]);
});
it("drops a terminal subagent the parent has no Task call for", () => {
it("drops a terminal subagent the parent has no Task call for", async () => {
// A legacy grandchild: its toolUseId names a Task call made inside a sibling's transcript, so
// no tool_result for it can ever reach the parent. Left running, it never stops.
const descriptor = applyToStore(
observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [
{
type: "assistant",
timestamp: "2026-07-31T03:41:00.000Z",
message: { stop_reason: "end_turn" },
},
],
},
],
parent: emptyParent(),
convertEntry: () => [],
}).observations,
(
await observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [
{
type: "assistant",
timestamp: "2026-07-31T03:41:00.000Z",
message: { stop_reason: "end_turn" },
},
],
},
],
parent: emptyParent(),
convertEntry: () => [],
})
).observations,
);
expect(descriptor).toBeNull();
});
it("drops a terminal skill-spawned subagent that has no link at all", () => {
it("drops a terminal skill-spawned subagent that has no link at all", async () => {
// A /code-review run: no Task tool_use exists to name it, so its sidecar carries only
// agentType and its id appears nowhere in the parent transcript.
const descriptor = applyToStore(
observeReplaySubagents({
(
await observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { agentType: "general-purpose" },
entries: [
{
type: "assistant",
timestamp: "2026-07-31T03:41:00.000Z",
message: { stop_reason: "end_turn" },
},
],
},
],
parent: emptyParent(),
convertEntry: () => [],
})
).observations,
);
expect(descriptor).toBeNull();
});
it("reports a scraped link's failure as failed", async () => {
const descriptor = applyToStore(
(
await observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
parent: {
toolCalls: new Map([[TOOL_USE_ID, { title: "Explore" }]]),
linksByAgentId: new Map([[AGENT_ID, { toolCallId: TOOL_USE_ID, failed: true }]]),
outcomesByToolCallId: new Map(),
},
convertEntry: () => [],
})
).observations,
);
expect(descriptor).toMatchObject({ id: TOOL_USE_ID, status: "failed" });
});
it("drops a non-terminal linkless subagent", async () => {
const observations = (
await observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
@@ -221,111 +279,81 @@ describe("observeReplaySubagents", () => {
{
type: "assistant",
timestamp: "2026-07-31T03:41:00.000Z",
message: { stop_reason: "end_turn" },
message: { stop_reason: null },
},
],
},
],
parent: emptyParent(),
convertEntry: () => [],
}).observations,
);
expect(descriptor).toBeNull();
});
it("reports a scraped link's failure as failed", () => {
const descriptor = applyToStore(
observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
parent: {
toolCalls: new Map([[TOOL_USE_ID, { title: "Explore" }]]),
linksByAgentId: new Map([[AGENT_ID, { toolCallId: TOOL_USE_ID, failed: true }]]),
outcomesByToolCallId: new Map(),
},
convertEntry: () => [],
}).observations,
);
expect(descriptor).toMatchObject({ id: TOOL_USE_ID, status: "failed" });
});
it("drops a non-terminal linkless subagent", () => {
const observations = observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { agentType: "general-purpose" },
entries: [
{
type: "assistant",
timestamp: "2026-07-31T03:41:00.000Z",
message: { stop_reason: null },
},
],
},
],
parent: emptyParent(),
convertEntry: () => [],
}).observations;
})
).observations;
expect(observations).toEqual([]);
});
it("finishes a terminal Task child when the parent has not recorded its outcome", () => {
it("finishes a terminal Task child when the parent has not recorded its outcome", async () => {
const descriptor = applyToStore(
observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [
{
type: "assistant",
timestamp: "2026-07-31T03:41:00.000Z",
message: { stop_reason: "end_turn" },
},
],
},
],
parent: { ...parentWithTaskCall(), outcomesByToolCallId: new Map() },
convertEntry: () => [],
}).observations,
(
await observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [
{
type: "assistant",
timestamp: "2026-07-31T03:41:00.000Z",
message: { stop_reason: "end_turn" },
},
],
},
],
parent: { ...parentWithTaskCall(), outcomesByToolCallId: new Map() },
convertEntry: () => [],
})
).observations,
);
expect(descriptor).toMatchObject({ status: "completed" });
});
it("leaves a subagent with no recorded outcome running", () => {
const observations = observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
parent: { ...parentWithTaskCall(), outcomesByToolCallId: new Map() },
convertEntry: () => [],
}).observations;
it("leaves a subagent with no recorded outcome running", async () => {
const observations = (
await observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
parent: { ...parentWithTaskCall(), outcomesByToolCallId: new Map() },
convertEntry: () => [],
})
).observations;
expect(observations.some((o) => o.kind === "status")).toBe(false);
});
it("reports a failed outcome as failed", () => {
const observations = observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
parent: parentWithTaskCall(true),
convertEntry: () => [],
}).observations;
it("reports a failed outcome as failed", async () => {
const observations = (
await observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
parent: parentWithTaskCall(true),
convertEntry: () => [],
})
).observations;
expect(observations.at(-1)).toMatchObject({ kind: "status", status: "failed" });
});
it("carries entry timestamps onto timeline observations", () => {
const observations = observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [{ type: "assistant", timestamp: "2026-07-26T06:27:47.034Z" }],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [{ type: "reasoning", text: "thinking" }],
}).observations;
it("carries entry timestamps onto timeline observations", async () => {
const observations = (
await observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [{ type: "assistant", timestamp: "2026-07-26T06:27:47.034Z" }],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [{ type: "reasoning", text: "thinking" }],
})
).observations;
expect(observations.find((observation) => observation.kind === "timeline")).toMatchObject({
kind: "timeline",
@@ -336,18 +364,20 @@ describe("observeReplaySubagents", () => {
});
describe("replay runtime", () => {
it("reads the model and effort off the subagent's own transcript", () => {
const observations = observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [{ type: "assistant", effort: "high", message: { model: "claude-opus-5" } }],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [],
}).observations;
it("reads the model and effort off the subagent's own transcript", async () => {
const observations = (
await observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [{ type: "assistant", effort: "high", message: { model: "claude-opus-5" } }],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [],
})
).observations;
expect(observations.find((o) => o.kind === "subtitle")).toEqual({
kind: "subtitle",
@@ -356,49 +386,55 @@ describe("replay runtime", () => {
});
});
it("takes the last observed values when the model changes mid-transcript", () => {
const observations = observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [
{ type: "assistant", effort: "high", message: { model: "claude-opus-5" } },
{ type: "assistant", effort: "low", message: { model: "claude-sonnet-5" } },
],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [],
}).observations;
it("takes the last observed values when the model changes mid-transcript", async () => {
const observations = (
await observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [
{ type: "assistant", effort: "high", message: { model: "claude-opus-5" } },
{ type: "assistant", effort: "low", message: { model: "claude-sonnet-5" } },
],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [],
})
).observations;
expect(observations.find((o) => o.kind === "subtitle")).toMatchObject({
subtitle: "general-purpose · Sonnet 5 · Low",
});
});
it("reports nothing when the transcript never names a model or effort", () => {
const observations = observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
parent: parentWithTaskCall(),
convertEntry: () => [],
}).observations;
it("reports nothing when the transcript never names a model or effort", async () => {
const observations = (
await observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
parent: parentWithTaskCall(),
convertEntry: () => [],
})
).observations;
expect(observations.some((o) => o.kind === "subtitle")).toBe(false);
});
it("passes through a model the manifest does not know", () => {
it("passes through a model the manifest does not know", async () => {
// Claude Code is an Anthropic-compatible client; subagents legitimately report other models.
const observations = observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [{ type: "assistant", message: { model: "glm-5.1" } }],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [],
}).observations;
const observations = (
await observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [{ type: "assistant", message: { model: "glm-5.1" } }],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [],
})
).observations;
expect(observations.find((o) => o.kind === "subtitle")).toMatchObject({
subtitle: "general-purpose · glm-5.1",
});
@@ -406,20 +442,24 @@ describe("replay runtime", () => {
});
describe("replay usage", () => {
function subtitleOf(entries: ClaudeReplayEntry[]): SubagentObservation | undefined {
return observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries }],
parent: parentWithTaskCall(),
convertEntry: () => [],
}).observations.find((observation) => observation.kind === "subtitle");
async function subtitleOf(
entries: ClaudeReplayEntry[],
): Promise<SubagentObservation | undefined> {
return (
await observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries }],
parent: parentWithTaskCall(),
convertEntry: () => [],
})
).observations.find((observation) => observation.kind === "subtitle");
}
it("derives the counters Claude Code itself reports, from a real transcript's shape", () => {
it("derives the counters Claude Code itself reports, from a real transcript's shape", async () => {
// Shape and numbers taken from an on-disk subagent transcript. Claude Code finalizes a
// subagent from the LAST assistant entry's usage, so the earlier, smaller turn must not win
// and must not be added in.
expect(
subtitleOf([
await subtitleOf([
{
type: "assistant",
timestamp: "2026-07-23T23:18:43.023Z",
@@ -462,9 +502,9 @@ describe("replay usage", () => {
});
});
it("formats tokens for a child that only answered", () => {
it("formats tokens for a child that only answered", async () => {
expect(
subtitleOf([
await subtitleOf([
{
type: "assistant",
timestamp: "2026-07-27T16:42:14.698Z",
@@ -479,12 +519,12 @@ describe("replay usage", () => {
).toMatchObject({ subtitle: "general-purpose · 16.4k tokens" });
});
it("emits nothing for a transcript with no entries", () => {
expect(subtitleOf([])).toBeUndefined();
it("emits nothing for a transcript with no entries", async () => {
expect(await subtitleOf([])).toBeUndefined();
});
it("keeps duration out of the compact subtitle", () => {
const observation = subtitleOf([
it("keeps duration out of the compact subtitle", async () => {
const observation = await subtitleOf([
{
type: "assistant",
timestamp: "2026-07-27T16:42:14.698Z",
@@ -494,8 +534,8 @@ describe("replay usage", () => {
expect(observation).toMatchObject({ subtitle: "general-purpose · 10 tokens" });
});
it("omits totalTokens when no entry carries a usage block", () => {
const observation = subtitleOf([
it("omits totalTokens when no entry carries a usage block", async () => {
const observation = await subtitleOf([
{ type: "assistant", timestamp: "2026-07-27T16:42:14.698Z", message: { content: [] } },
{
type: "assistant",
@@ -506,9 +546,9 @@ describe("replay usage", () => {
expect(observation).toBeUndefined();
});
it("keeps the last usage-bearing entry when the final assistant entry has none", () => {
it("keeps the last usage-bearing entry when the final assistant entry has none", async () => {
expect(
subtitleOf([
await subtitleOf([
{
type: "assistant",
timestamp: "2026-07-27T16:42:14.698Z",
@@ -519,34 +559,36 @@ describe("replay usage", () => {
).toMatchObject({ subtitle: "general-purpose · 1k tokens" });
});
it("only reports counters the transcript actually shows", () => {
it("only reports counters the transcript actually shows", async () => {
// A user-only transcript says nothing about cost: no assistant turn, so no tool count either.
const observation = subtitleOf([
const observation = await subtitleOf([
{ type: "user", timestamp: "2026-07-27T16:42:14.698Z", message: { content: "prompt" } },
{ type: "user", timestamp: "2026-07-27T16:42:20.000Z", message: { content: "more" } },
]);
expect(observation).toBeUndefined();
});
it("lands the complete provider-owned subtitle on the descriptor", () => {
it("lands the complete provider-owned subtitle on the descriptor", async () => {
const store = new ProviderSubagentStore();
const observations = observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [
{
type: "assistant",
timestamp: "2026-07-27T16:42:14.698Z",
message: { content: [], usage: { input_tokens: 7 } },
},
],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [],
}).observations;
const observations = (
await observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: { toolUseId: TOOL_USE_ID },
entries: [
{
type: "assistant",
timestamp: "2026-07-27T16:42:14.698Z",
message: { content: [], usage: { input_tokens: 7 } },
},
],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [],
})
).observations;
for (const event of foldSubagentObservations(observations)) {
store.apply("parent", "claude", event);
}
@@ -559,7 +601,7 @@ describe("live and replay agree", () => {
// The property the design exists to guarantee: one session, two sources, one descriptor.
// This is inexpressible against the inference implementation, where the paths share no
// vocabulary and each derives identity and status its own way.
it("produces identical descriptor state from the wire and from disk", () => {
it("produces identical descriptor state from the wire and from disk", async () => {
const live = new ClaudeTaskProtocolSource();
const liveObservations = [
...live.observe({
@@ -579,21 +621,23 @@ describe("live and replay agree", () => {
} as unknown as SDKMessage),
];
const replayObservations = observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: {
toolUseId: TOOL_USE_ID,
agentType: "general-purpose",
description: "Summarize hover and unistyles docs",
const replayObservations = (
await observeReplaySubagents({
subagents: [
{
agentId: AGENT_ID,
meta: {
toolUseId: TOOL_USE_ID,
agentType: "general-purpose",
description: "Summarize hover and unistyles docs",
},
entries: [],
},
entries: [],
},
],
parent: parentWithTaskCall(),
convertEntry: () => [],
}).observations;
],
parent: parentWithTaskCall(),
convertEntry: () => [],
})
).observations;
const fromWire = applyToStore(liveObservations);
const fromDisk = applyToStore(replayObservations);
packages/server/src/server/agent/providers/claude/subagents/replay-source.ts
+7 -7
View File
@@ -130,7 +130,7 @@ function readUsage(entry: ClaudeReplayEntry): ClaudeSubagentUsage | null {
export interface ClaudeReplaySubagentInput {
agentId: string;
meta: ClaudeSubagentMeta | null;
entries: Iterable<ClaudeReplayEntry>;
entries: Iterable<ClaudeReplayEntry> | AsyncIterable<ClaudeReplayEntry>;
/** Task declarations made by this subagent, used to attach its direct children. */
parentFacts?: ClaudeReplayParentFacts;
}
@@ -250,13 +250,13 @@ function observeSubtitle(
};
}
function observeSubagent(
async function observeSubagent(
subagent: ClaudeReplaySubagentInput,
parent: ClaudeReplayParentFacts,
convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[],
toolOwners: Map<string, string>,
parentSubagentId?: string,
): SubagentObservation[] {
): Promise<SubagentObservation[]> {
const link = resolveParentLink(subagent, parent);
if (!link) return [];
const toolCall = parent.toolCalls.get(link.toolCallId);
@@ -269,7 +269,7 @@ function observeSubagent(
let childTerminalStatus: ProviderSubagentStatus | null = null;
// Consume each source once for presentation and timeline; keep no raw transcript records.
for (const entry of subagent.entries) {
for await (const entry of subagent.entries) {
const timestamp = normalizeProviderReplayTimestamp(entry.timestamp);
if (!seenEntry) firstTimestamp = timestamp;
seenEntry = true;
@@ -320,11 +320,11 @@ function recordReplayToolOwners(
}
}
export function observeReplaySubagents(input: {
export async function observeReplaySubagents(input: {
subagents: readonly ClaudeReplaySubagentInput[];
parent: ClaudeReplayParentFacts;
convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[];
}): { observations: SubagentObservation[]; toolOwners: ReadonlyMap<string, string> } {
}): Promise<{ observations: SubagentObservation[]; toolOwners: ReadonlyMap<string, string> }> {
const observations: SubagentObservation[] = [];
const toolOwners = new Map<string, string>();
const unresolved = [...input.subagents].sort(
@@ -346,7 +346,7 @@ export function observeReplaySubagents(input: {
const { ownerId, parent, link } = resolved;
// Only proven descendants may own notifications; ambient sidecars cannot claim them.
for (const observation of observeSubagent(
for (const observation of await observeSubagent(
subagent,
parent,
input.convertEntry,
packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.test.ts
+13 -13
View File
@@ -35,7 +35,7 @@ function parentEntries(
}
describe("Claude workflow replay", () => {
it("parses the persisted run summary observed from Claude Code 2.1.220", () => {
it("parses the persisted run summary observed from Claude Code 2.1.220", async () => {
expect(
parseClaudeWorkflowRun(
JSON.stringify({
@@ -62,8 +62,8 @@ describe("Claude workflow replay", () => {
});
});
it("rebuilds the same generic completed descriptor through the parent Workflow call", () => {
const observations = observeReplayWorkflows({
it("rebuilds the same generic completed descriptor through the parent Workflow call", async () => {
const observations = await observeReplayWorkflows({
workflows: [
{
runId: RUN_ID,
@@ -91,8 +91,8 @@ describe("Claude workflow replay", () => {
});
});
it("rebuilds a workflow when Claude persists its result as structured text blocks", () => {
const observations = observeReplayWorkflows({
it("rebuilds a workflow when Claude persists its result as structured text blocks", async () => {
const observations = await observeReplayWorkflows({
workflows: [{ runId: RUN_ID, summary: "Structured result workflow", status: "completed" }],
parentEntries: parentEntries([
{
@@ -114,7 +114,7 @@ describe("Claude workflow replay", () => {
});
});
it("replays the workflow result without exposing its script or child prompt", () => {
it("replays the workflow result without exposing its script or child prompt", async () => {
const workflow = parseClaudeWorkflowRun(
JSON.stringify({
runId: RUN_ID,
@@ -128,7 +128,7 @@ describe("Claude workflow replay", () => {
expect(workflow).not.toBeNull();
if (!workflow) throw new Error("expected workflow");
const observations = observeReplayWorkflows({
const observations = await observeReplayWorkflows({
workflows: [workflow],
parentEntries: parentEntries(),
});
@@ -145,9 +145,9 @@ describe("Claude workflow replay", () => {
expect(JSON.stringify(observations)).not.toContain("SECRET CHILD PROMPT");
});
it("does not duplicate a result already present in the child transcript", () => {
it("does not duplicate a result already present in the child transcript", async () => {
const report = "What Paseo is\nA local-first coding-agent environment.";
const observations = observeReplayWorkflows({
const observations = await observeReplayWorkflows({
workflows: [
{ runId: RUN_ID, summary: "Inspect Paseo", status: "completed", result: { report } },
],
@@ -166,8 +166,8 @@ describe("Claude workflow replay", () => {
).toHaveLength(1);
});
it("fails a nonterminal persisted run because its former runtime cannot update it", () => {
const observations = observeReplayWorkflows({
it("fails a nonterminal persisted run because its former runtime cannot update it", async () => {
const observations = await observeReplayWorkflows({
workflows: [{ runId: RUN_ID, summary: "Interrupted workflow", status: "running" }],
parentEntries: parentEntries(),
});
@@ -179,9 +179,9 @@ describe("Claude workflow replay", () => {
});
});
it("drops a run that no parent Workflow tool call declared", () => {
it("drops a run that no parent Workflow tool call declared", async () => {
expect(
observeReplayWorkflows({
await observeReplayWorkflows({
workflows: [{ runId: RUN_ID, status: "completed" }],
parentEntries: [],
}),
packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts
+18 -13
View File
@@ -45,14 +45,17 @@ export function parseClaudeWorkflowRun(contents: string): ClaudeWorkflowRun | nu
* provider runtime was recreated, so replay terminalizes it as failed instead of resurrecting a
* row that can never receive another update.
*/
export function observeReplayWorkflows(input: {
export async function observeReplayWorkflows(input: {
workflows: readonly ClaudeWorkflowRun[];
parentEntries: Iterable<ClaudeWorkflowParentEntry>;
parentEntries: Iterable<ClaudeWorkflowParentEntry> | AsyncIterable<ClaudeWorkflowParentEntry>;
/** Repeatable entries in chronological order, supplied by the transcript reader. */
entriesByRunId?: ReadonlyMap<string, Iterable<ClaudeReplayEntry>>;
entriesByRunId?: ReadonlyMap<
string,
Iterable<ClaudeReplayEntry> | AsyncIterable<ClaudeReplayEntry>
>;
convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[];
}): SubagentObservation[] {
const toolCallIdByRunId = readWorkflowLinks(input.parentEntries);
}): Promise<SubagentObservation[]> {
const toolCallIdByRunId = await readWorkflowLinks(input.parentEntries);
const observations: SubagentObservation[] = [];
for (const workflow of input.workflows) {
@@ -79,7 +82,7 @@ export function observeReplayWorkflows(input: {
});
}
for (const observation of observeWorkflowTimeline({
for (const observation of await observeWorkflowTimeline({
id,
result: workflow.result,
entries: input.entriesByRunId?.get(workflow.runId) ?? [],
@@ -115,17 +118,17 @@ export function observeReplayWorkflows(input: {
return observations;
}
function observeWorkflowTimeline(input: {
async function observeWorkflowTimeline(input: {
id: string;
result: unknown;
entries: Iterable<ClaudeReplayEntry>;
entries: Iterable<ClaudeReplayEntry> | AsyncIterable<ClaudeReplayEntry>;
convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[];
finishedAt?: string;
}): SubagentObservation[] {
}): Promise<SubagentObservation[]> {
const observations: SubagentObservation[] = [];
const replayedAssistantText = new Set<string>();
for (const entry of input.entries) {
for await (const entry of input.entries) {
const timestamp = normalizeProviderReplayTimestamp(entry.timestamp);
const items = input.convertEntry?.(entry) ?? [];
for (const item of items) {
@@ -150,11 +153,13 @@ function observeWorkflowTimeline(input: {
return observations;
}
function readWorkflowLinks(entries: Iterable<ClaudeWorkflowParentEntry>): Map<string, string> {
async function readWorkflowLinks(
entries: Iterable<ClaudeWorkflowParentEntry> | AsyncIterable<ClaudeWorkflowParentEntry>,
): Promise<Map<string, string>> {
const workflowToolCallIds = new Set<string>();
const toolCallIdByRunId = new Map<string, string>();
for (const entry of entries) {
for await (const entry of entries) {
const content = entry.message?.content;
if (!Array.isArray(content)) continue;
for (const value of content) {
@@ -165,7 +170,7 @@ function readWorkflowLinks(entries: Iterable<ClaudeWorkflowParentEntry>): Map<st
}
}
for (const entry of entries) {
for await (const entry of entries) {
const content = entry.message?.content;
if (!Array.isArray(content)) continue;
for (const value of content) {
packages/server/src/server/agent/providers/claude/task-notification-tool-call.ts
+4 -2
View File
@@ -221,9 +221,11 @@ function parseTaskNotificationFromSystemRecord(record: unknown): TaskNotificatio
/** Normalize notification identity for one transcript traversal. Queue bookkeeping and delivery
* are separate records of the same occurrence; each source counts repeated identical bodies
* independently so the second event is not merged with the first. */
export function* assignTaskNotificationReplayIds<T>(entries: Iterable<T>): Generator<T> {
export async function* assignTaskNotificationReplayIds<T>(
entries: AsyncIterable<T>,
): AsyncGenerator<T> {
const occurrences = new Map<string, Map<string, number>>();
for (const entry of entries) {
for await (const entry of entries) {
const record = TaskNotificationHistoryRecordSchema.safeParse(entry);
if (!record.success) {
yield entry;
packages/server/src/server/agent/providers/claude/transcript-history.ts
+65 -42
View File
@@ -1,5 +1,6 @@
import { assignTaskNotificationReplayIds } from "./task-notification-tool-call.js";
import fs from "node:fs";
import fs, { type FileHandle } from "node:fs/promises";
import { setImmediate } from "node:timers/promises";
import path from "node:path";
import { normalizeProviderReplayTimestamp } from "../../provider-history-timestamps.js";
import {
@@ -24,11 +25,11 @@ interface RecordLocation {
}
/** Chunk in bytes so UTF-8 and multi-chunk records are decoded once, at a line boundary. */
function* locatedRecords(
async function* locatedRecords(
file: string,
fd: number,
): Generator<{ entry: TranscriptEntry; location: RecordLocation }> {
const stat = fs.fstatSync(fd);
fd: FileHandle,
): AsyncGenerator<{ entry: TranscriptEntry; location: RecordLocation }> {
const stat = await fd.stat();
// Opening a directory succeeds on some platforms; it must never look like empty history.
if (stat.isDirectory()) {
throw Object.assign(new Error(`EISDIR: illegal operation on a directory, read '${file}'`), {
@@ -43,7 +44,7 @@ function* locatedRecords(
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);
const { bytesRead: count } = await fd.read(buffer, 0, buffer.length, null);
if (count === 0) break;
remaining -= count;
let start = 0;
@@ -64,6 +65,8 @@ function* locatedRecords(
if (entry) yield { entry, location };
start = end + 1;
}
// Bound parsing work as well as IO; microtasks alone do not service timers or sockets.
await setImmediate();
if (start < count) {
pieces.push(buffer.subarray(start, count));
length += count - start;
@@ -85,20 +88,25 @@ function* locatedRecords(
/** Indexed sources parse only their own records using descriptors owned by this replay. */
function indexedRecords(
locations: readonly RecordLocation[],
descriptors: ReadonlyMap<string, number>,
): Iterable<TranscriptEntry> {
descriptors: ReadonlyMap<string, FileHandle>,
): AsyncIterable<TranscriptEntry> {
return {
*[Symbol.iterator]() {
async *[Symbol.asyncIterator]() {
yield* assignTaskNotificationReplayIds(readEntries());
},
};
function* readEntries(): Generator<TranscriptEntry> {
async function* readEntries(): AsyncGenerator<TranscriptEntry> {
for (const location of locations) {
const fd = descriptors.get(location.file)!;
const buffer = Buffer.allocUnsafe(location.length);
let read = 0;
while (read < buffer.length) {
const count = fs.readSync(fd, buffer, read, buffer.length - read, location.offset + read);
const { bytesRead: count } = await fd.read(
buffer,
read,
buffer.length - read,
location.offset + read,
);
if (count === 0)
throw new Error(`Claude transcript was truncated during replay: ${location.file}`);
read += count;
@@ -127,9 +135,9 @@ function parseRecord(line: string): TranscriptEntry | null {
}
}
function readOptionalFile(file: string): string | null {
async function readOptionalFile(file: string): Promise<string | null> {
try {
return fs.readFileSync(file, "utf8");
return await fs.readFile(file, "utf8");
} catch {
return null;
}
@@ -138,31 +146,40 @@ function readOptionalFile(file: string): string | null {
/** Discover sources without retaining transcript contents. Unrelated sidechains stay excluded
* by the replay ownership resolver; workflow children retain their separate run ownership. */
export interface ClaudeReplayHistory {
parentEntries: Iterable<TranscriptEntry>;
parentEntries: AsyncIterable<TranscriptEntry>;
subagents: {
agentId: string;
meta: ClaudeSubagentMeta | null;
entries: Iterable<TranscriptEntry>;
entries: AsyncIterable<TranscriptEntry>;
}[];
workflows: ClaudeWorkflowRun[];
workflowEntriesByRunId: Map<string, Iterable<TranscriptEntry>>;
workflowEntriesByRunId: Map<string, AsyncIterable<TranscriptEntry>>;
}
/** Own descriptors through every pass so unlink or rename cannot invalidate discovered history.
* Sources must be consumed before the callback returns. No transcript copies are needed. */
export function withClaudeReplayHistory<T>(
export async function withClaudeReplayHistory<T>(
historyPath: string,
consume: (history: ClaudeReplayHistory) => T,
): T {
const descriptors = new Map<string, number>();
consume: (history: ClaudeReplayHistory) => Promise<T>,
): Promise<T> {
const descriptors = new Map<string, FileHandle>();
try {
return consume(indexHistory(historyPath, descriptors));
return await consume(await indexHistory(historyPath, descriptors));
} finally {
for (const fd of descriptors.values()) fs.closeSync(fd);
await closeReplaySources(descriptors.values());
}
}
function indexHistory(historyPath: string, descriptors: Map<string, number>): ClaudeReplayHistory {
async function closeReplaySources(descriptors: Iterable<FileHandle>): Promise<void> {
const results = await Promise.allSettled([...descriptors].map((fd) => fd.close()));
const failure = results.find((result) => result.status === "rejected");
if (failure) throw failure.reason;
}
async function indexHistory(
historyPath: string,
descriptors: Map<string, FileHandle>,
): Promise<ClaudeReplayHistory> {
const sessionDirectory = path.join(
path.dirname(historyPath),
path.basename(historyPath, ".jsonl"),
@@ -174,8 +191,7 @@ function indexHistory(historyPath: string, descriptors: Map<string, number>): Cl
const directories = [sidechainDirectory];
while (directories.length) {
const directory = directories.pop()!;
if (!fs.existsSync(directory)) continue;
for (const entry of fs.readdirSync(directory, { withFileTypes: true })) {
for (const entry of await optionalDirectory(directory)) {
const file = path.join(directory, entry.name);
if (entry.isDirectory()) {
directories.push(file);
@@ -194,23 +210,23 @@ function indexHistory(historyPath: string, descriptors: Map<string, number>): Cl
}
const agentId = /^agent-(.+)\.meta\.json$/.exec(entry.name)?.[1];
if (!agentId) continue;
const contents = readOptionalFile(file);
const contents = await readOptionalFile(file);
const meta = contents === null ? null : parseClaudeSubagentMeta(contents);
if (meta) metaByAgentId.set(agentId, meta);
}
}
const parentLocations: RecordLocation[] = [];
const locationsByAgentId = new Map<string, RecordLocation[]>();
function scan(
async function scan(
file: string,
visit: (entry: TranscriptEntry, location: RecordLocation) => void,
): void {
const fd = fs.openSync(file, "r");
): Promise<void> {
const fd = await fs.open(file, "r");
descriptors.set(file, fd);
for (const { entry, location } of locatedRecords(file, fd)) visit(entry, location);
for await (const { entry, location } of locatedRecords(file, fd)) visit(entry, location);
}
for (const file of sidechainFiles) {
scan(file, (entry, location) => {
await scan(file, (entry, location) => {
if (file === historyPath && entry.isSidechain !== true) parentLocations.push(location);
if (entry.isSidechain !== true || typeof entry.agentId !== "string") return;
const locations = locationsByAgentId.get(entry.agentId) ?? [];
@@ -218,11 +234,11 @@ function indexHistory(historyPath: string, descriptors: Map<string, number>): Cl
locationsByAgentId.set(entry.agentId, locations);
});
}
const workflowEntriesByRunId = new Map<string, Iterable<TranscriptEntry>>();
const workflowEntriesByRunId = new Map<string, AsyncIterable<TranscriptEntry>>();
for (const [runId, files] of workflowFiles) {
const locations: RecordLocation[] = [];
for (const file of files)
scan(file, (entry, location) => {
await scan(file, (entry, location) => {
if (entry.type !== "user" || isToolResult(entry)) locations.push(location);
});
locations.sort((a, b) => {
@@ -240,21 +256,28 @@ function indexHistory(historyPath: string, descriptors: Map<string, number>): Cl
meta: metaByAgentId.get(agentId) ?? null,
entries: indexedRecords(locations, descriptors),
})),
workflows: readWorkflows(sessionDirectory),
workflows: await readWorkflows(sessionDirectory),
workflowEntriesByRunId,
};
}
function readWorkflows(sessionDirectory: string): ClaudeWorkflowRun[] {
async function readWorkflows(sessionDirectory: string): Promise<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);
}
for (const entry of await optionalDirectory(workflowDirectory)) {
if (!entry.isFile() || !entry.name.endsWith(".json")) continue;
const contents = await readOptionalFile(path.join(workflowDirectory, entry.name));
const workflow = contents === null ? null : parseClaudeWorkflowRun(contents);
if (workflow) workflows.push(workflow);
}
return workflows;
}
async function optionalDirectory(directory: string) {
try {
return await fs.readdir(directory, { withFileTypes: true });
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") return [];
throw error;
}
}