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

refactor(server): give agent timeline history a single awaitable owner

Browse Source 
Agent timeline history had several writers that appended and published
independently, so a persistent backend could not be swapped in without
reordering or losing items. AgentTimeline now owns each agent's history:
one FIFO queue per agent (child agents share the parent's queue) commits
each item and then publishes it, so subscribers never see an item before it
is recorded. Backends may be async, which prepares the SQLite swap.

Terminal turn events wait for queued writes so a finished turn never races
its own last chunks. A subscriber that throws after commit leaves the item
in history. A failed commit surfaces as an agent error ("Conversation
history could not be saved") via the existing lastError/attention path; it
survives the turn's terminal transition and clears on the next run.
This commit is contained in:
Mohamed Boudra
2026-11-10 08:11:00 +01:00
parent 254166e7c1
commit 4178497d3d
38 changed files with 2153 additions and 1394 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/agent-loading.test.ts
+3 -3
View File
@@ -185,9 +185,9 @@ test("loads an archived agent's history after its working directory is removed",
expect(loaded.id).toBe(agentId);
// The transcript is replayed from the provider's persisted history, so the reply the
// agent gave before the worktree went away is still readable.
const replies = manager
.getTimeline(agentId)
.filter((item) => item.type === "assistant_message");
const replies = (await manager.getTimeline(agentId)).filter(
(item) => item.type === "assistant_message",
);
expect(replies.length).toBeGreaterThan(0);
expect(replies.every((item) => item.type === "assistant_message" && item.text.length > 0)).toBe(
true,
packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts
+86 -2
View File
@@ -3,12 +3,16 @@ import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, describe, expect, test, vi } from "vitest";
import pino, { type Logger } from "pino";
import { createTestLogger } from "../../test-utils/test-logger.js";
import { AgentManager, type AgentManagerEvent } from "./agent-manager.js";
import { AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS } from "./agent-stream-coalescer.js";
import { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
import type { AgentTimelineRow } from "./agent-timeline-store-types.js";
import { projectTimelineRows } from "./timeline-projection.js";
import { AgentTimeline } from "./timeline/index.js";
import { DeferredTimelineBackend } from "./timeline/testing.js";
import type {
AgentCapabilityFlags,
AgentClient,
@@ -243,13 +247,18 @@ interface Harness {
cleanup: () => void;
}
function createHarness(options?: { provider?: AgentProvider }): Harness {
function createHarness(options?: {
provider?: AgentProvider;
timeline?: AgentTimeline;
logger?: Logger;
}): Harness {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-stream-coalescing-"));
const client = new TestAgentClient(options?.provider ?? "codex");
const manager = new AgentManager({
clients: { [client.provider]: client },
idFactory: createIdFactory(),
logger: createTestLogger(),
logger: options?.logger ?? createTestLogger(),
timeline: options?.timeline,
});
const events: AgentManagerEvent[] = [];
manager.subscribe((event) => events.push(event), { replayState: false });
@@ -1037,6 +1046,81 @@ describe("target coalesced behavior", () => {
}
});
test.each(["turn_completed", "turn_failed", "turn_canceled"] as const)(
"%s waits for the final chunk an asynchronous timeline is still committing",
async (terminalType) => {
const harness = createHarness({ timeline: new AgentTimeline(new DeferredTimelineBackend()) });
try {
const { agentId, session } = await createManagedSession(harness);
session.pushEvent(assistant("final chunk", "codex", "turn"));
session.pushEvent(terminalEvent(terminalType, "turn"));
const streamTypes = () =>
getStreamEvents(harness.events, agentId).map((event) =>
event.type === "agent_stream" ? event.event.type : event.type,
);
await expect.poll(streamTypes).toContain(terminalType);
const streamEvents = getStreamEvents(harness.events, agentId);
expect(streamEvents[0]).toMatchObject({
event: { type: "timeline", item: { type: "assistant_message", text: "final chunk" } },
seq: 1,
});
expect(streamTypes().indexOf(terminalType)).toBeGreaterThan(0);
} finally {
await harness.manager.flush();
harness.cleanup();
}
},
);
test("a final chunk the timeline fails to commit surfaces a history error after the turn", async () => {
const backend = new DeferredTimelineBackend();
const logs: Array<Record<string, unknown>> = [];
const harness = createHarness({
timeline: new AgentTimeline(backend),
logger: pino({ level: "error" }, { write: (line: string) => logs.push(JSON.parse(line)) }),
});
try {
const { agentId, session } = await createManagedSession(harness);
session.pushEvent(assistant("earlier chunk", "codex", "earlier"));
session.pushEvent(terminalEvent("turn_completed", "earlier"));
await expect.poll(() => getTimelineStreamEvents(harness.events, agentId)).toHaveLength(1);
await harness.manager.flush();
session.pushEvent({ type: "turn_started", provider: "codex", turnId: "turn" });
await expect.poll(() => harness.manager.getAgent(agentId)?.lifecycle).toBe("running");
backend.failNextAppend = true;
session.pushEvent(assistant("final chunk", "codex", "turn"));
session.pushEvent(terminalEvent("turn_completed", "turn"));
const completions = () =>
getStreamEvents(harness.events, agentId).filter(
(event) => event.type === "agent_stream" && event.event.type === "turn_completed",
);
await expect.poll(completions).toHaveLength(2);
await harness.manager.flush();
expect(harness.manager.getAgent(agentId)).toMatchObject({
lifecycle: "error",
lastError: "Conversation history could not be saved: disk full",
attention: { requiresAttention: true, attentionReason: "error" },
});
const earlier = { type: "assistant_message", text: "earlier chunk" };
expect(getTimelineStreamEvents(harness.events, agentId)).toEqual([
expect.objectContaining({ event: expect.objectContaining({ item: earlier }) }),
]);
expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([earlier]);
expect(logs).toContainEqual(
expect.objectContaining({
msg: "Failed to record timeline item",
agentId,
err: expect.objectContaining({ message: "disk full" }),
}),
);
} finally {
harness.cleanup();
}
});
test("lifecycle paths are timer-safe", async () => {
vi.useFakeTimers();
packages/server/src/server/agent/agent-manager.test.ts
+230 -194
View File
@@ -11,6 +11,7 @@ import { join } from "node:path";
import { tmpdir } from "node:os";
import { randomUUID } from "node:crypto";
import pino from "pino";
import { createTestLogger } from "../../test-utils/test-logger.js";
import {
AgentManager,
@@ -20,7 +21,8 @@ import {
type ManagedAgent,
} from "./agent-manager.js";
import { AgentStorage } from "./agent-storage.js";
import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js";
import { AgentTimeline } from "./timeline/index.js";
import { DeferredTimelineBackend } from "./timeline/testing.js";
import { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
import { toAgentPayload } from "./agent-projections.js";
import { projectTimelineRows } from "./timeline-projection.js";
@@ -29,12 +31,6 @@ import { startAgentRun } from "./agent-prompt.js";
import { StaleProviderSessionError } from "./stale-provider-session-error.js";
import { ensureAgentLoaded, ensureUnarchivedAgentLoaded } from "./agent-loading.js";
import type { StoredAgentRecord } from "./agent-storage.js";
import type {
AgentTimelineFetchOptions,
AgentTimelineFetchResult,
AgentTimelineRow,
AgentTimelineStore,
} from "./agent-timeline-store-types.js";
import type {
AgentClient,
AgentCreateSessionOptions,
@@ -115,73 +111,6 @@ const TEST_CAPABILITIES = {
supportsToolInvocations: false,
} as const;
class RecordingTimelineStore implements AgentTimelineStore {
readonly writes: AgentTimelineRow[][] = [];
private readonly memory = new InMemoryAgentTimelineStore();
private ensure(agentId: string): void {
if (!this.memory.has(agentId)) this.memory.initialize(agentId);
}
async appendCommitted(
agentId: string,
item: AgentTimelineItem,
options?: { timestamp?: string; turnId?: string },
): Promise<AgentTimelineRow> {
this.ensure(agentId);
return this.memory.append(agentId, item, options);
}
async fetchCommitted(
agentId: string,
options?: AgentTimelineFetchOptions,
): Promise<AgentTimelineFetchResult> {
this.ensure(agentId);
return this.memory.fetch(agentId, options);
}
async getLatestCommittedSeq(agentId: string): Promise<number> {
return this.memory.has(agentId) ? (this.memory.getRows(agentId).at(-1)?.seq ?? 0) : 0;
}
async getCommittedRows(agentId: string): Promise<AgentTimelineRow[]> {
return this.memory.has(agentId) ? this.memory.getRows(agentId) : [];
}
async getLastItem(agentId: string): Promise<AgentTimelineItem | null> {
return this.memory.has(agentId) ? this.memory.getLastItem(agentId) : null;
}
async getLastAssistantMessage(agentId: string): Promise<string | null> {
return this.memory.has(agentId) ? this.memory.getLastAssistantMessage(agentId) : null;
}
async deleteAgent(agentId: string): Promise<void> {
this.memory.delete(agentId);
}
async bulkInsert(agentId: string, rows: readonly AgentTimelineRow[]): Promise<void> {
this.writes.push(rows.map((row) => ({ ...row })));
this.ensure(agentId);
for (const row of rows) {
this.memory.append(agentId, row.item, {
timestamp: row.timestamp,
turnId: row.turnId,
});
}
}
async updateCommittedRow(agentId: string, row: AgentTimelineRow): Promise<void> {
this.ensure(agentId);
const rows = this.memory.getRows(agentId);
const index = rows.findIndex((candidate) => candidate.seq === row.seq);
if (index >= 0) {
rows[index] = row;
this.memory.initialize(agentId, { rows });
}
}
}
function createFeature(args: { id: string; label: string; value: boolean }): AgentFeature {
return {
type: "toggle",
@@ -670,12 +599,12 @@ async function startAndSteerThroughManager(
return { manager, agentId: agent.id, workdir };
}
test("uses an injected timeline store without making it a production requirement", async () => {
test("commits timeline items through an injected asynchronous backend", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-store-"));
const store = new RecordingTimelineStore();
const backend = new DeferredTimelineBackend();
const manager = new AgentManager({
clients: { codex: new TestAgentClient() },
durableTimelineStore: store,
timeline: new AgentTimeline(backend),
logger,
});
let agentId: string | null = null;
@@ -687,7 +616,7 @@ test("uses an injected timeline store without making it a production requirement
await manager.appendTimelineItem(agent.id, { type: "assistant_message", text: "stored" });
await manager.flush();
expect(store.writes.flat()).toContainEqual(
expect(backend.appends.flat()).toContainEqual(
expect.objectContaining({ item: { type: "assistant_message", text: "stored" } }),
);
await expect(manager.getTimelineRows(agent.id)).resolves.toContainEqual(
@@ -699,9 +628,113 @@ test("uses an injected timeline store without making it a production requirement
}
});
test("refreshing an agent replaces the injected timeline store instead of appending a second copy", async () => {
test("a subscriber that throws on publication leaves the committed item in history", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-publish-failure-"));
const logs: Array<Record<string, unknown>> = [];
const manager = new AgentManager({
clients: { codex: new TestAgentClient() },
timeline: new AgentTimeline(new DeferredTimelineBackend()),
logger: pino({ level: "error" }, { write: (line: string) => logs.push(JSON.parse(line)) }),
});
let throwOnTimeline = true;
manager.subscribe(
(event) => {
if (throwOnTimeline && event.type === "agent_stream" && event.event.type === "timeline") {
throwOnTimeline = false;
throw new Error("subscriber failed");
}
},
{ replayState: false },
);
let agentId: string | null = null;
try {
const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
workspaceId: undefined,
});
agentId = agent.id;
await expect(
manager.appendTimelineItem(agent.id, { type: "assistant_message", text: "kept" }),
).resolves.toMatchObject({ seq: 1 });
await expect(
manager.appendTimelineItem(agent.id, { type: "assistant_message", text: " and next" }),
).resolves.toMatchObject({ seq: 2 });
expect(await manager.getLastAssistantMessage(agent.id)).toBe("kept and next");
expect(logs).toContainEqual(
expect.objectContaining({
msg: "Failed to publish committed timeline item",
agentId: agent.id,
err: expect.objectContaining({ message: "subscriber failed" }),
}),
);
} finally {
if (agentId) await manager.closeAgent(agentId).catch(() => undefined);
rmSync(workdir, { recursive: true, force: true });
}
});
test("an accepted prompt the timeline fails to commit surfaces a history error", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-failure-"));
const backend = new DeferredTimelineBackend();
const logs: Array<Record<string, unknown>> = [];
const manager = new AgentManager({
clients: { codex: new TestAgentClient() },
timeline: new AgentTimeline(backend),
logger: pino({ level: "error" }, { write: (line: string) => logs.push(JSON.parse(line)) }),
});
const published: AgentTimelineItem[] = [];
manager.subscribe(
(event) => {
if (event.type === "agent_stream" && event.event.type === "timeline") {
published.push(event.event.item);
}
},
{ replayState: false },
);
let agentId: string | null = null;
try {
const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
workspaceId: undefined,
});
agentId = agent.id;
const earlier: AgentTimelineItem = { type: "assistant_message", text: "Earlier reply" };
await manager.appendTimelineItem(agent.id, earlier);
await manager.flush();
backend.failNextAppend = true;
await manager.runAgent(agent.id, "hello", { clientMessageId: "hello-client" });
await manager.flush();
expect(manager.getAgent(agent.id)).toMatchObject({
lifecycle: "error",
lastError: "Conversation history could not be saved: disk full",
attention: { requiresAttention: true, attentionReason: "error" },
});
expect(await manager.getTimeline(agent.id)).toEqual([earlier]);
expect(published).toEqual([earlier]);
expect(logs).toContainEqual(
expect.objectContaining({
msg: "Failed to record timeline item",
agentId: agent.id,
err: expect.objectContaining({ message: "disk full" }),
}),
);
// The next run clears the error and records normally.
await manager.runAgent(agent.id, "again", { clientMessageId: "again-client" });
await manager.flush();
expect(manager.getAgent(agent.id)).toMatchObject({ lifecycle: "idle" });
expect(manager.getAgent(agent.id)?.lastError).toBeUndefined();
expect(await manager.getTimeline(agent.id)).toEqual([
earlier,
expect.objectContaining({ type: "user_message", clientMessageId: "again-client" }),
]);
} finally {
if (agentId) await manager.closeAgent(agentId).catch(() => undefined);
rmSync(workdir, { recursive: true, force: true });
}
});
test("refreshing an agent replaces its timeline instead of appending a second copy", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-rehydrate-"));
const store = new RecordingTimelineStore();
class HistorySession extends TestAgentSession {
override async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
yield {
@@ -730,7 +763,7 @@ test("refreshing an agent replaces the injected timeline store instead of append
}
const manager = new AgentManager({
clients: { codex: new HistoryClient() },
durableTimelineStore: store,
timeline: new AgentTimeline(new DeferredTimelineBackend()),
logger,
});
let agentId: string | null = null;
@@ -758,9 +791,8 @@ test("refreshing an agent replaces the injected timeline store instead of append
}
});
test("a failed history replay leaves the committed timeline intact", async () => {
test("a failed history replay commits none of the replay", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-replay-fail-"));
const store = new RecordingTimelineStore();
let failReplay = false;
class FlakyHistorySession extends TestAgentSession {
override async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
@@ -793,7 +825,7 @@ test("a failed history replay leaves the committed timeline intact", async () =>
}
const manager = new AgentManager({
clients: { codex: new FlakyHistoryClient() },
durableTimelineStore: store,
timeline: new AgentTimeline(new DeferredTimelineBackend()),
logger,
});
let agentId: string | null = null;
@@ -814,7 +846,13 @@ test("a failed history replay leaves the committed timeline intact", async () =>
).rejects.toThrow("history stream closed");
await manager.flush();
expect(await manager.getTimelineRows(agent.id)).toEqual(committed);
// Reload with rehydration discards the active history before the replay (phase 1 keeps
// that); the failed replay must not leave the rows it streamed before failing.
expect(committed.map((row) => row.item)).toEqual([
{ type: "user_message", text: "continue" },
{ type: "assistant_message", text: "done" },
]);
expect(await manager.getTimelineRows(agent.id)).toEqual([]);
// A retry once the provider recovers replaces that history rather than stacking on it.
failReplay = false;
@@ -871,7 +909,7 @@ test("retries provider history hydration after a stream failure", async () => {
await manager.hydrateTimelineFromProvider(agent.id);
expect(attempts).toBe(2);
expect(manager.getTimeline(agent.id)).toContainEqual({
expect(await manager.getTimeline(agent.id)).toContainEqual({
type: "assistant_message",
text: "provider history",
});
@@ -924,7 +962,7 @@ test("notification batches share the run they start or join", async () => {
session.pushEvent({ type: "turn_completed", provider: "codex", turnId: "active-turn-2" });
await second;
expect(
manager.fetchTimeline(agent.id, { limit: 0 }).rows.map((row) => ({
(await manager.fetchTimeline(agent.id, { limit: 0 })).rows.map((row) => ({
turnId: row.turnId,
id: row.item.type === "tool_call" ? row.item.callId : row.item.type,
})),
@@ -948,7 +986,7 @@ test("unavailable steer interrupts once and starts one replacement turn", async
try {
expect(session.interruptCount).toBe(1);
expect(session.startCount).toBe(2);
expect(manager.getTimeline(agentId)).toContainEqual(
expect(await manager.getTimeline(agentId)).toContainEqual(
expect.objectContaining({ type: "user_message", clientMessageId: "replacement-client" }),
);
} finally {
@@ -1006,9 +1044,8 @@ test("steers a send that arrives while the previous turn is still starting", asy
expect(session.startPrompts).toEqual(["FIRST"]);
expect(session.interruptCount).toBe(0);
expect(
manager
.fetchTimeline(agent.id, { limit: 0 })
.rows.filter((row) => row.item.type === "user_message")
(await manager.fetchTimeline(agent.id, { limit: 0 })).rows
.filter((row) => row.item.type === "user_message")
.map((row) => ({
text: row.item.type === "user_message" ? row.item.text : null,
turnId: row.turnId,
@@ -1137,14 +1174,14 @@ test("orders an accepted steer before output emitted while acknowledgement is pe
provider: "codex",
turnId: "active-turn-1",
});
expect(manager.getTimeline(agent.id)).not.toContainEqual(
expect(await manager.getTimeline(agent.id)).not.toContainEqual(
expect.objectContaining({ type: "assistant_message", text: "terminal output" }),
);
release.resolve();
await expect(steer).resolves.toEqual({ status: "accepted" });
await consume;
const rows = manager.fetchTimeline(agent.id, { limit: 0 }).rows.filter((row) => {
const rows = (await manager.fetchTimeline(agent.id, { limit: 0 })).rows.filter((row) => {
return (
(row.item.type === "user_message" && row.item.text === "hello") ||
(row.item.type === "assistant_message" && row.item.text === "terminal output")
@@ -1201,7 +1238,7 @@ test("orders buffered pre-steer output before an immediately accepted steer", as
manager.steerAgentRun(agent.id, "hello", { clientMessageId: "hello-client" }),
).resolves.toEqual({ status: "accepted" });
const rows = manager.fetchTimeline(agent.id, { limit: 0 }).rows.filter((row) => {
const rows = (await manager.fetchTimeline(agent.id, { limit: 0 })).rows.filter((row) => {
return (
(row.item.type === "assistant_message" && row.item.text === "pre-steer output") ||
(row.item.type === "user_message" && row.item.text === "hello")
@@ -1281,9 +1318,8 @@ test("orders a concurrent replacement after a pending accepted steer", async ()
expect(session.interruptCount).toBe(1);
expect(
manager
.fetchTimeline(agent.id, { limit: 0 })
.rows.filter((row) => row.item.type === "user_message")
(await manager.fetchTimeline(agent.id, { limit: 0 })).rows
.filter((row) => row.item.type === "user_message")
.map((row) => row.item),
).toMatchObject([
{ text: "hello", clientMessageId: "hello-client" },
@@ -1351,9 +1387,9 @@ test("does not replace a newer foreground turn after unavailable steer fallback
expect(session.interruptCount).toBe(0);
expect(session.startPrompts).not.toContain("hello");
expect(
manager
.getTimeline(agent.id)
.some((item) => item.type === "user_message" && item.clientMessageId === "hello-client"),
(await manager.getTimeline(agent.id)).some(
(item) => item.type === "user_message" && item.clientMessageId === "hello-client",
),
).toBe(false);
} finally {
release.resolve();
@@ -1395,7 +1431,7 @@ test("steers a tracked autonomous turn without creating a replacement run", asyn
expect(session.interruptCount).toBe(0);
expect(manager.getAgent(agent.id)?.activeTurnId).toBe("active-turn-0");
expect(manager.getAgent(agent.id)?.activeForegroundTurnId).toBeNull();
expect(manager.getTimeline(agent.id)).toContainEqual(
expect(await manager.getTimeline(agent.id)).toContainEqual(
expect.objectContaining({
type: "user_message",
text: "autonomous follow-up",
@@ -1421,7 +1457,7 @@ test("isolated rewind falls back from steering to the normal replacement path",
await manager.waitForAgentRunStart(agentId);
expect(session.interruptCount).toBe(2);
expect(session.startPrompts).toContain("/rewind submitted-message-id");
expect(manager.getTimeline(agentId)).toContainEqual(
expect(await manager.getTimeline(agentId)).toContainEqual(
expect.objectContaining({ type: "user_message", clientMessageId: "rewind-client" }),
);
} finally {
@@ -1436,7 +1472,7 @@ test("missing steer operation interrupts once and starts one replacement turn",
try {
expect(session.interruptCount).toBe(1);
expect(session.startCount).toBe(2);
expect(manager.getTimeline(agentId)).toContainEqual(
expect(await manager.getTimeline(agentId)).toContainEqual(
expect.objectContaining({ type: "user_message", clientMessageId: "replacement-client" }),
);
} finally {
@@ -1476,7 +1512,7 @@ test("ambiguous steer failure leaves the active turn untouched", async () => {
).rejects.toThrow("connection lost after delivery");
expect(session.interruptCount).toBe(0);
expect(session.startCount).toBe(1);
expect(manager.getTimeline(agent.id)).not.toContainEqual(
expect(await manager.getTimeline(agent.id)).not.toContainEqual(
expect.objectContaining({ clientMessageId: "ambiguous-client" }),
);
} finally {
@@ -1510,7 +1546,9 @@ test("steering records concurrent early echoes as canonical submitted prompts",
manager.steerAgentRun(agent.id, "one", { clientMessageId: "client-one" }),
manager.steerAgentRun(agent.id, "two", { clientMessageId: "client-two" }),
]);
const rows = manager.getTimeline(agent.id).filter((item) => item.type === "user_message");
const rows = (await manager.getTimeline(agent.id)).filter(
(item) => item.type === "user_message",
);
expect(rows).toEqual(
expect.arrayContaining([
expect.objectContaining({ messageId: "client-one", clientMessageId: "client-one" }),
@@ -4235,7 +4273,7 @@ test("a prompt after provider replacement reopens the stale session", async () =
const result = await manager.waitForAgentEvent(created.id);
expect(result.status).toBe("idle");
expect(replacement.resumeOverrides).toHaveLength(1);
const timeline = manager.getTimeline(created.id);
const timeline = await manager.getTimeline(created.id);
expect(timeline.filter((item) => item.type === "user_message")).toMatchObject([
{ text: "continue after reload", clientMessageId: "stale-session-retry" },
]);
@@ -4590,7 +4628,7 @@ test("importProviderSession imports the selected session without listing and pub
});
expect(imported.lifecycle).toBe("idle");
expect(imported.historyPrimed).toBe(true);
expect(manager.getTimeline(imported.id)).toEqual([
expect(await manager.getTimeline(imported.id)).toEqual([
{ type: "user_message", text: "Trace provider imports" },
{ type: "assistant_message", text: "Done" },
limitAgentTimelineItemContent({
@@ -4610,7 +4648,7 @@ test("importProviderSession imports the selected session without listing and pub
expect(manager.listProviderSubagents(imported.id)).toEqual([
expect.objectContaining({ id: "thread-child", title: "Imported child", status: "completed" }),
]);
expect(manager.fetchProviderSubagentTimeline(imported.id, "thread-child").rows).toEqual([
expect((await manager.fetchProviderSubagentTimeline(imported.id, "thread-child")).rows).toEqual([
expect.objectContaining({ item: { type: "assistant_message", text: "Child result" } }),
]);
expect(events).toHaveLength(3);
@@ -4780,18 +4818,18 @@ test("reloadAgentSession preserves timeline and does not force history replay",
text: "keep this timeline in memory",
});
await manager.hydrateTimelineFromProvider(snapshot.id);
const beforeReload = manager.getTimeline(snapshot.id);
const beforeReload = await manager.getTimeline(snapshot.id);
expect(beforeReload).toHaveLength(1);
await manager.reloadAgentSession(snapshot.id, {
systemPrompt: "reloaded prompt",
});
const afterReload = manager.getTimeline(snapshot.id);
const afterReload = await manager.getTimeline(snapshot.id);
expect(afterReload).toEqual(beforeReload);
// If reload resets historyPrimed, this would replay provider history and append another item.
await manager.hydrateTimelineFromProvider(snapshot.id);
const afterHydrate = manager.getTimeline(snapshot.id);
const afterHydrate = await manager.getTimeline(snapshot.id);
expect(afterHydrate).toEqual(beforeReload);
});
@@ -5961,13 +5999,13 @@ test("fetchTimeline returns a bounded reset window when cursor epoch is stale",
text: "three",
});
const baseline = manager.fetchTimeline(snapshot.id, {
const baseline = await manager.fetchTimeline(snapshot.id, {
direction: "tail",
limit: 2,
});
expect(baseline.rows).toHaveLength(2);
const result = manager.fetchTimeline(snapshot.id, {
const result = await manager.fetchTimeline(snapshot.id, {
direction: "after",
cursor: {
epoch: "stale-epoch",
@@ -5984,7 +6022,7 @@ test("fetchTimeline returns a bounded reset window when cursor epoch is stale",
expect(result.rows[result.rows.length - 1]?.seq).toBe(3);
expect(result.hasOlder).toBe(true);
const older = manager.fetchTimeline(snapshot.id, {
const older = await manager.fetchTimeline(snapshot.id, {
direction: "before",
cursor: {
epoch: result.epoch,
@@ -6081,7 +6119,7 @@ test("getAgent does not expose committed history internals once manager owns the
expect("timelineRows" in live).toBe(false);
expect("timelineNextSeq" in live).toBe(false);
expect(manager.getTimeline(snapshot.id)).toEqual([
expect(await manager.getTimeline(snapshot.id)).toEqual([
{
type: "user_message",
text: "hello boundary",
@@ -6178,7 +6216,7 @@ test("streams coalesced assistant chunks and retains the projected message", asy
epoch: expect.any(String),
});
expect(manager.getTimeline(snapshot.id)).toEqual([
expect(await manager.getTimeline(snapshot.id)).toEqual([
{
type: "assistant_message",
text: "final reply",
@@ -6252,7 +6290,7 @@ test("fetchTimeline supports older-history pagination with before seq", async ()
const result = await manager.fetchTimeline(snapshot.id, {
direction: "before",
cursor: {
epoch: manager.fetchTimeline(snapshot.id).epoch,
epoch: (await manager.fetchTimeline(snapshot.id)).epoch,
seq: 5,
},
limit: 2,
@@ -6396,7 +6434,7 @@ test("hydrateTimeline preserves assistant chunk, reasoning, and tool timeline hi
await manager.hydrateTimelineFromProvider(snapshot.id, { force: true });
expect(manager.getTimeline(snapshot.id)).toEqual([
expect(await manager.getTimeline(snapshot.id)).toEqual([
{ type: "assistant_message", text: "chunk one chunk two" },
{ type: "reasoning", text: "internal" },
{
@@ -6478,7 +6516,7 @@ test("hydrateTimeline preserves reasoning between assistant chunks", async () =>
await manager.hydrateTimelineFromProvider(snapshot.id, { force: true });
expect(manager.getTimeline(snapshot.id)).toEqual([
expect(await manager.getTimeline(snapshot.id)).toEqual([
{
type: "assistant_message",
text: "before reasoning ",
@@ -7084,13 +7122,11 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the
capturedSession!.pushEvent({ type: "turn_completed", provider: "codex", turnId: "turn-1" });
await new Promise((resolve) => setTimeout(resolve, 0));
expect(
manager
.fetchTimeline(snapshot.id, { direction: "tail", limit: 0 })
.rows.find(
(row) =>
row.item.type === "user_message" &&
row.item.clientMessageId === "replacement-client-message",
)?.item,
(await manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0 })).rows.find(
(row) =>
row.item.type === "user_message" &&
row.item.clientMessageId === "replacement-client-message",
)?.item,
).toMatchObject({ messageId: "replacement-client-message" });
allowSecondTurnToComplete.resolve();
await firstRunDrain;
@@ -7196,7 +7232,7 @@ test("applies live autonomous events and preserves usage omitted from completion
contextWindowMaxTokens: 200_000,
contextWindowUsedTokens: 175,
});
expect(manager.getTimeline(snapshot.id)).toContainEqual({
expect(await manager.getTimeline(snapshot.id)).toContainEqual({
type: "assistant_message",
text: "AUTONOMOUS_PUMP_MESSAGE",
});
@@ -7258,7 +7294,7 @@ test("ignores stale autonomous terminals without lowering the active turn lifecy
name: "Current turn permission",
kind: "tool",
});
const timelineBeforeStaleTerminals = manager.getTimeline(snapshot.id);
const timelineBeforeStaleTerminals = await manager.getTimeline(snapshot.id);
const staleTerminals: AgentStreamEvent[] = [
{
@@ -7294,7 +7330,7 @@ test("ignores stale autonomous terminals without lowering the active turn lifecy
expect(stillRunning?.lastError).toBe("turn-b marker");
expect(stillRunning?.lastUsage).toEqual({ inputTokens: 7 });
expect(stillRunning?.pendingPermissions.has("turn-b-permission")).toBe(true);
expect(manager.getTimeline(snapshot.id)).toEqual(timelineBeforeStaleTerminals);
expect(await manager.getTimeline(snapshot.id)).toEqual(timelineBeforeStaleTerminals);
}
capturedSession!.pushEvent({ type: "turn_completed", provider: "codex", turnId: "turn-b" });
@@ -7339,7 +7375,7 @@ test("preserves terminal fallback when no active turn identity was observed", as
expect(failed?.lifecycle).toBe("error");
expect(failed?.lastError).toBe("untracked failure");
});
expect(manager.getTimeline(snapshot.id)).toContainEqual(
expect(await manager.getTimeline(snapshot.id)).toContainEqual(
expect.objectContaining({
type: "assistant_message",
text: expect.stringContaining("untracked failure"),
@@ -7643,8 +7679,8 @@ test("autonomous events arriving during foreground run are processed via subscri
turnId: autonomousTurnId,
});
await vi.waitFor(() =>
expect(manager.getTimeline(snapshot.id)).toContainEqual({
await vi.waitFor(async () =>
expect(await manager.getTimeline(snapshot.id)).toContainEqual({
type: "assistant_message",
text: "AUTONOMOUS_DURING_FOREGROUND",
}),
@@ -7670,7 +7706,7 @@ test("autonomous events arriving during foreground run are processed via subscri
).toBe(false);
// Autonomous timeline item should still be recorded in the agent timeline
expect(manager.getTimeline(snapshot.id)).toContainEqual({
expect(await manager.getTimeline(snapshot.id)).toContainEqual({
type: "assistant_message",
text: "AUTONOMOUS_DURING_FOREGROUND",
});
@@ -7751,7 +7787,7 @@ test("subscribe error isolation: throwing subscriber does not break event flow",
expect(receivedEvents).toContain("turn_started");
expect(receivedEvents).toContain("timeline");
expect(receivedEvents).toContain("turn_completed");
expect(manager.getTimeline(snapshot.id)).toContainEqual({
expect(await manager.getTimeline(snapshot.id)).toContainEqual({
type: "assistant_message",
text: "EVENT_AFTER_ERROR",
});
@@ -8371,9 +8407,9 @@ test("subscribe hides provider subagents of internal parents from global subscri
expect(() => manager.getProviderSubagent(internalAgentId, "hidden-child")).toThrow(
`Unknown agent '${internalAgentId}'`,
);
expect(() => manager.fetchProviderSubagentTimeline(internalAgentId, "hidden-child")).toThrow(
`Unknown agent '${internalAgentId}'`,
);
await expect(
manager.fetchProviderSubagentTimeline(internalAgentId, "hidden-child"),
).rejects.toThrow(`Unknown agent '${internalAgentId}'`);
expect(manager.listProviderSubagentActivity()).toEqual([]);
});
@@ -8677,7 +8713,7 @@ test("streamAgent clears pending run when startTurn fails before a turn id exist
clientMessageId: "rejected-prompt",
}),
).rejects.toThrow("Invalid request: missing field `text`");
expect(manager.getTimeline(agent.id)).toEqual([
expect(await manager.getTimeline(agent.id)).toEqual([
{
type: "user_message",
text: "fail before turn id",
@@ -9577,12 +9613,10 @@ test("turn_failed emits a system error assistant timeline message and keeps erro
expect(snapshot?.lifecycle).toBe("error");
expect(snapshot?.lastError).toBe("invalid model id");
const systemErrors = manager
.getTimeline(agent.id)
.filter(
(item): item is Extract<AgentTimelineItem, { type: "assistant_message" }> =>
item.type === "assistant_message" && item.text.includes("[System Error]"),
);
const systemErrors = (await manager.getTimeline(agent.id)).filter(
(item): item is Extract<AgentTimelineItem, { type: "assistant_message" }> =>
item.type === "assistant_message" && item.text.includes("[System Error]"),
);
expect(systemErrors).toHaveLength(1);
expect(systemErrors[0]?.text).toContain("invalid model id");
});
@@ -9653,12 +9687,10 @@ test("turn_failed surfaces provider code and diagnostic in system error message"
expect(manager.getAgent(agent.id)?.lastError).toBe("Provider execution failed");
const systemError = manager
.getTimeline(agent.id)
.find(
(item): item is Extract<AgentTimelineItem, { type: "assistant_message" }> =>
item.type === "assistant_message" && item.text.includes("[System Error]"),
);
const systemError = (await manager.getTimeline(agent.id)).find(
(item): item is Extract<AgentTimelineItem, { type: "assistant_message" }> =>
item.type === "assistant_message" && item.text.includes("[System Error]"),
);
expect(systemError?.text).toContain("Provider execution failed");
expect(systemError?.text).toContain("code: 126");
expect(systemError?.text).toContain("No preset version installed for command claude");
@@ -10913,7 +10945,7 @@ test("hydrateTimeline keeps provider user_message items when no canonical user h
await manager.hydrateTimelineFromProvider(snapshot.id, { force: true });
const timeline = manager.getTimeline(snapshot.id);
const timeline = await manager.getTimeline(snapshot.id);
const userMessages = timeline.filter((item) => item.type === "user_message");
const assistantMessages = timeline.filter((item) => item.type === "assistant_message");
expect(userMessages).toHaveLength(2);
@@ -10977,7 +11009,7 @@ test("hydrateTimeline preserves provider replay timestamps and marks missing one
);
await manager.hydrateTimelineFromProvider(snapshot.id, { force: true });
const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0 }).rows;
const timeline = (await manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0 })).rows;
expect(timeline).toHaveLength(2);
expect(timeline[0]).toMatchObject({
@@ -11042,7 +11074,7 @@ test("provider user_message is recorded from the live stream", async () => {
await manager.runAgent(snapshot.id, { text: "do something" });
const timeline = manager.getTimeline(snapshot.id);
const timeline = await manager.getTimeline(snapshot.id);
const userMessages = timeline.filter((item) => item.type === "user_message");
// Provider's user_message should be recorded (no canonical to dedup against)
@@ -11178,7 +11210,8 @@ test("canonical submitted prompt keeps wire identity while rewind resolves provi
{ type: "turn_completed" },
]);
const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }).rows;
const timeline = (await manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }))
.rows;
expect(timeline).toMatchObject([
{
seq: 1,
@@ -11267,6 +11300,8 @@ test("authoritative timeline records a daemon-handled submitted prompt before it
runOptions: { clientMessageId: "msg-client-daemon-handled" },
});
await commandCompleted.promise;
// The command's output publishes once the timeline commits it.
await manager.flush();
expect(
events.flatMap((event) =>
@@ -11281,7 +11316,8 @@ test("authoritative timeline records a daemon-handled submitted prompt before it
{ type: "assistant_message", text: "Handled by the daemon" },
]);
const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }).rows;
const timeline = (await manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }))
.rows;
expect(timeline.map((row) => row.item)).toEqual([
{
type: "user_message",
@@ -11695,7 +11731,7 @@ test("system-injected prompts appear as trigger tools in the live timeline", asy
await manager.runAgent(snapshot.id, { text: "do something" });
const timeline = manager.getTimeline(snapshot.id);
const timeline = await manager.getTimeline(snapshot.id);
expect(timeline).toContainEqual(
expect.objectContaining({ type: "tool_call", name: "heartbeat", status: "completed" }),
);
@@ -11739,7 +11775,7 @@ test("system-injected prompts appear as trigger tools during history replay", as
await manager.hydrateTimelineFromProvider(snapshot.id, { force: true });
const timeline = manager.getTimeline(snapshot.id);
const timeline = await manager.getTimeline(snapshot.id);
expect(timeline).toContainEqual(
expect.objectContaining({ type: "tool_call", name: "heartbeat", status: "completed" }),
);
@@ -11985,10 +12021,10 @@ test("commits startup notices once on create and after restored history", async
return new NotifyingSession({ provider: "codex", cwd: workdir });
}
})();
const store = new RecordingTimelineStore();
const backend = new DeferredTimelineBackend();
const manager = new AgentManager({
clients: { codex: client },
durableTimelineStore: store,
timeline: new AgentTimeline(backend),
logger,
});
const ids: string[] = [];
@@ -12015,7 +12051,7 @@ test("commits startup notices once on create and after restored history", async
notice,
]);
await manager.flush();
expect(store.writes.flat()).toContainEqual(expect.objectContaining({ item: notice }));
expect(backend.appends.flat()).toContainEqual(expect.objectContaining({ item: notice }));
} finally {
for (const id of ids) await manager.closeAgent(id);
rmSync(workdir, { recursive: true, force: true });
@@ -12113,7 +12149,7 @@ test.each(["live", "history"])(
});
if (path === "live") await manager.runAgent(agent.id, { text: "start" });
else await manager.hydrateTimelineFromProvider(agent.id, { force: true });
expect(manager.getTimeline(agent.id)).toContainEqual(
expect(await manager.getTimeline(agent.id)).toContainEqual(
expect.objectContaining({
type: "tool_call",
callId: "paseo-agent-message:delivery-1",
@@ -12131,9 +12167,9 @@ test.each(["live", "history"])(
},
}),
);
expect(manager.getTimeline(agent.id).filter((entry) => entry.type === "user_message")).toEqual(
[],
);
expect(
(await manager.getTimeline(agent.id)).filter((entry) => entry.type === "user_message"),
).toEqual([]);
expect(manager.getAgent(agent.id)?.lastUserMessageAt).toBeNull();
},
);
@@ -12168,15 +12204,17 @@ test.each([false, true])(
clientMessageId: delivery.messageId,
})) {
}
expect(manager.getTimeline(agent.id).filter((item) => item.type === "tool_call")).toEqual([
expect(
(await manager.getTimeline(agent.id)).filter((item) => item.type === "tool_call"),
).toEqual([
expect.objectContaining({
callId: "paseo-agent-message:accepted-delivery",
status: "completed",
}),
]);
expect(manager.getTimeline(agent.id).filter((item) => item.type === "user_message")).toEqual(
[],
);
expect(
(await manager.getTimeline(agent.id)).filter((item) => item.type === "user_message"),
).toEqual([]);
expect(manager.getAgent(agent.id)?.lastUserMessageAt).toBeNull();
} finally {
await manager.closeAgent(agent.id);
@@ -12211,7 +12249,9 @@ test("a rejected agent prompt never becomes a delivered notification", async ()
clientMessageId: delivery.messageId,
});
await expect(run.next()).rejects.toThrow("provider rejected delivery");
expect(manager.getTimeline(agent.id).filter((item) => item.type === "tool_call")).toEqual([]);
expect(
(await manager.getTimeline(agent.id)).filter((item) => item.type === "tool_call"),
).toEqual([]);
expect(manager.getAgent(agent.id)?.lastUserMessageAt).toBeNull();
} finally {
await manager.closeAgent(agent.id);
@@ -12248,7 +12288,7 @@ test.each([false, true])(
clientMessageId: delivery.messageId,
})) {
}
expect(manager.getTimeline(agent.id)).toEqual([
expect(await manager.getTimeline(agent.id)).toEqual([
{
type: "user_message",
text: example,
@@ -12295,22 +12335,15 @@ test.each([
])(
"lifecycle hooks receive no conversation for $event.type and stream dispatch never reads timeline items",
async ({ event: terminalEvent, outcome }) => {
class CountingTimelineStore extends InMemoryAgentTimelineStore {
itemReads = 0;
override getItems(agentId: string): AgentTimelineItem[] {
this.itemReads += 1;
return super.getItems(agentId);
}
}
const workdir = mkdtempSync(join(tmpdir(), "agent-hook-payload-"));
const client = new SessionRecordingAgentClient();
const timelineStore = new CountingTimelineStore();
const timelineStore = new DeferredTimelineBackend();
const events: Array<{ name: string; event: unknown }> = [];
const manager = new AgentManager({
clients: { codex: client },
registry: new AgentStorage(join(workdir, "agents"), logger),
logger,
timelineStore,
timeline: new AgentTimeline(timelineStore),
pluginLifecycle: {
before: async (_name, request) => request,
emit: (name, event) => {
@@ -12330,14 +12363,14 @@ test.each([
type: "assistant_message",
text: "A long conversation".repeat(10_000),
});
expect(manager.getTimeline(agent.id)).toHaveLength(1);
const readsBeforeDispatch = timelineStore.itemReads;
expect(await manager.getTimeline(agent.id)).toHaveLength(1);
const readsBeforeDispatch = timelineStore.readRowsCalls;
const session = client.sessions[0]!;
session.pushEvent({ type: "turn_started", provider: "codex", turnId: "hook-turn" });
await expect
.poll(() => events.filter(({ name }) => name === "agent.turn.before").length)
.toBe(1);
expect(timelineStore.itemReads).toBe(readsBeforeDispatch);
expect(timelineStore.readRowsCalls).toBe(readsBeforeDispatch);
session.pushEvent(terminalEvent);
await expect
.poll(() => events.filter(({ name }) => name === "agent.turn.after"))
@@ -12359,7 +12392,7 @@ test.each([
},
},
]);
expect(timelineStore.itemReads).toBe(readsBeforeDispatch);
expect(timelineStore.readRowsCalls).toBe(readsBeforeDispatch);
},
);
@@ -12408,8 +12441,7 @@ test.each(["live", "history", "accepted"] as const)(
}
} else await manager.runAgent(agent.id, { text: "start" });
expect(
manager
.getTimeline(agent.id)
(await manager.getTimeline(agent.id))
.filter((row) => row.type === "tool_call")
.map((row) => ({ callId: row.callId, agentMessage: row.agentMessage })),
).toEqual([
@@ -12430,9 +12462,9 @@ test.each(["live", "history", "accepted"] as const)(
},
},
]);
expect(manager.getTimeline(agent.id).filter((row) => row.type === "user_message")).toEqual(
[],
);
expect(
(await manager.getTimeline(agent.id)).filter((row) => row.type === "user_message"),
).toEqual([]);
expect(manager.getAgent(agent.id)?.lastUserMessageAt).toBeNull();
} finally {
await manager.closeAgent(agent.id);
@@ -12442,17 +12474,21 @@ test.each(["live", "history", "accepted"] as const)(
);
test("reads the last durable assistant response without loading a stored-only child", async () => {
const store = new RecordingTimelineStore();
await store.appendCommitted("stored-child", {
type: "assistant_message",
text: "Partial work saved before restart",
});
const manager = new AgentManager({ clients: {}, durableTimelineStore: store, logger });
const backend = new DeferredTimelineBackend();
// History a previous daemon process committed.
await backend.create(
{ agentId: "stored-child" },
{
entries: [{ item: { type: "assistant_message", text: "Partial work saved before restart" } }],
},
);
const manager = new AgentManager({ clients: {}, timeline: new AgentTimeline(backend), logger });
expect(manager.getAgent("stored-child")).toBeNull();
expect(await manager.getLastAssistantMessage("stored-child")).toBe(
"Partial work saved before restart",
);
expect(manager.getAgent("stored-child")).toBeNull();
expect(backend.readRowsCalls).toBe(0);
});
function runtimeCleanup(manager: AgentManager) {
packages/server/src/server/agent/agent-manager.ts
+339 -376
View File
File diff suppressed because it is too large Load Diff
packages/server/src/server/agent/agent-timeline-store-types.ts
-19
View File
@@ -47,22 +47,3 @@ export interface AgentTimelineFetchResult {
endSeq: number | null;
rows: ProjectedTimelineRow[];
}
export interface AgentTimelineStore {
appendCommitted(
agentId: string,
item: AgentTimelineItem,
options?: { timestamp?: string; turnId?: string },
): Promise<AgentTimelineRow>;
fetchCommitted(
agentId: string,
options?: AgentTimelineFetchOptions,
): Promise<AgentTimelineFetchResult>;
getLatestCommittedSeq(agentId: string): Promise<number>;
getCommittedRows(agentId: string): Promise<AgentTimelineRow[]>;
getLastItem(agentId: string): Promise<AgentTimelineItem | null>;
getLastAssistantMessage(agentId: string): Promise<string | null>;
deleteAgent(agentId: string): Promise<void>;
bulkInsert(agentId: string, rows: readonly AgentTimelineRow[]): Promise<void>;
updateCommittedRow(agentId: string, row: AgentTimelineRow): Promise<void>;
}
packages/server/src/server/agent/agent-timeline-store.test.ts
-443
View File
@@ -1,443 +0,0 @@
import { execFile } from "node:child_process";
import { promisify } from "node:util";
import { fileURLToPath } from "node:url";
import { describe, expect, it } from "vitest";
import { selectReferenceProjectedTimelinePage } from "./test-utils/timeline-page-reference.js";
import type { AgentTimelineItem } from "./agent-sdk-types.js";
import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js";
describe("InMemoryAgentTimelineStore", () => {
it("resets stale forward cursors to complete available turns", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("agent", { epoch: "current" });
for (let index = 0; index < 60; index++)
store.append(
"agent",
{
type: "tool_call",
callId: `call-${index}`,
name: "shell",
status: "completed",
error: null,
detail: { type: "unknown", input: null, output: null },
},
{ turnId: "run" },
);
const page = store.fetch("agent", {
direction: "after",
cursor: { epoch: "old", seq: 10 },
limit: 40,
wholeTurns: true,
});
expect(page.rows).toHaveLength(60);
expect(page).toMatchObject({
reset: true,
staleCursor: true,
startSeq: 1,
endSeq: 60,
hasOlder: false,
hasNewer: false,
});
});
it("clamps an overshooting before cursor into the bounded tail window", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("agent-1", {
epoch: "epoch-1",
nextSeq: 8,
rows: [
{
seq: 5,
timestamp: "2026-01-01T00:00:00.000Z",
item: { type: "assistant_message", text: "five", messageId: "five" },
},
{
seq: 6,
timestamp: "2026-01-01T00:00:01.000Z",
item: { type: "assistant_message", text: "six", messageId: "six" },
},
{
seq: 7,
timestamp: "2026-01-01T00:00:02.000Z",
item: { type: "assistant_message", text: "seven", messageId: "seven" },
},
],
});
const result = store.fetch("agent-1", {
direction: "before",
cursor: { epoch: "epoch-1", seq: 100 },
limit: 2,
});
expect(result).toMatchObject({
epoch: "epoch-1",
direction: "before",
reset: false,
staleCursor: false,
gap: false,
window: { minSeq: 5, maxSeq: 7, nextSeq: 8 },
hasOlder: true,
hasNewer: false,
rows: [
{
seq: 6,
timestamp: "2026-01-01T00:00:01.000Z",
item: { type: "assistant_message", text: "six", messageId: "six" },
},
{
seq: 7,
timestamp: "2026-01-01T00:00:02.000Z",
item: { type: "assistant_message", text: "seven", messageId: "seven" },
},
],
});
});
it("returns a bounded reset window when an after cursor is behind retained history", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("agent-1", {
epoch: "epoch-1",
nextSeq: 8,
rows: [
{
seq: 5,
timestamp: "2026-01-01T00:00:00.000Z",
item: { type: "assistant_message", text: "five", messageId: "five" },
},
{
seq: 6,
timestamp: "2026-01-01T00:00:01.000Z",
item: { type: "assistant_message", text: "six", messageId: "six" },
},
{
seq: 7,
timestamp: "2026-01-01T00:00:02.000Z",
item: { type: "assistant_message", text: "seven", messageId: "seven" },
},
],
});
const result = store.fetch("agent-1", {
direction: "after",
cursor: { epoch: "epoch-1", seq: 1 },
limit: 1,
});
expect(result).toMatchObject({
epoch: "epoch-1",
direction: "after",
reset: true,
staleCursor: false,
gap: true,
window: { minSeq: 5, maxSeq: 7, nextSeq: 8 },
hasOlder: true,
hasNewer: false,
rows: [
{
seq: 7,
timestamp: "2026-01-01T00:00:02.000Z",
item: { type: "assistant_message", text: "seven", messageId: "seven" },
},
],
});
});
});
describe("projected timeline retention", () => {
it("retains one full tool state while streaming every source update", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("agent");
for (let seq = 1; seq <= 2000; seq++) {
const item = {
type: "tool_call" as const,
callId: "child",
name: "task",
status: "running" as const,
error: null,
detail: { type: "plain_text" as const, label: "Child", text: "x".repeat(seq * 128) },
};
expect(store.append("agent", item)).toMatchObject({ seq, item });
}
const result = store.fetch("agent", { limit: 0 });
expect(result.rows).toHaveLength(1);
expect(JSON.stringify(result.rows).length).toBeLessThan(260_000);
expect(result.window.maxSeq).toBe(2000);
});
it("catches up a mid-message cursor with the complete projected message", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("agent", { epoch: "e" });
store.append("agent", { type: "assistant_message", messageId: "m", text: "A" });
store.append("agent", { type: "assistant_message", messageId: "m", text: "B" });
expect(
store.fetch("agent", { direction: "after", cursor: { epoch: "e", seq: 1 } }).rows,
).toMatchObject([
{
item: { text: "AB" },
seqStart: 1,
seqEnd: 2,
sourceSeqRanges: [{ startSeq: 1, endSeq: 2 }],
},
]);
});
it("serves a merged assistant message within the item budget", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("agent", { epoch: "e" });
const chunk = "a".repeat(200 * 1024);
for (let index = 0; index < 10; index++) {
store.append("agent", { type: "assistant_message", messageId: "m", text: chunk });
}
const [row] = store.fetch("agent", { limit: 1 }).rows;
expect(row).toMatchObject({ seqStart: 1, seqEnd: 10 });
expect(row.item).toMatchObject({ type: "assistant_message", messageId: "m" });
expect(Buffer.byteLength(JSON.stringify(row.item))).toBeLessThanOrEqual(256 * 1024);
expect(store.getLastAssistantMessage("agent")).toHaveLength(10 * chunk.length);
});
});
describe("projected sequence ownership", () => {
it("preserves fetched coverage when another source chunk arrives", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("a");
store.append("a", { type: "assistant_message", text: "A" });
const before = store.fetch("a");
store.append("a", { type: "assistant_message", text: "B" });
expect(before.rows[0]).toMatchObject({
item: { text: "A" },
seqEnd: 1,
sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }],
});
expect(store.fetch("a").rows[0]).toMatchObject({
item: { text: "AB" },
seqEnd: 2,
sourceSeqRanges: [{ startSeq: 1, endSeq: 2 }],
});
});
it("preserves source positions when seeding a projected history whose last update is an earlier tool", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("a");
const tool = {
type: "tool_call" as const,
callId: "t",
name: "shell",
error: null,
detail: { type: "plain_text" as const, label: "work" },
};
store.append("a", { ...tool, status: "running" });
store.append("a", { type: "assistant_message", text: "Answer" });
store.append("a", { ...tool, status: "completed" });
store.initialize("b", { rows: store.getRows("a") });
expect(store.getLastItem("b")).toMatchObject({ type: "tool_call", status: "completed" });
expect(store.getLastAssistantMessage("b")).toBe("Answer");
expect(store.append("b", { type: "user_message", text: "next" }).seq).toBe(4);
expect(store.getLastItem("b")).toEqual({ type: "user_message", text: "next" });
expect(store.fetch("b").rows.map((row) => row.seqStart)).toEqual([1, 2, 4]);
});
it("returns the last projected assistant message without joining different messages", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("a");
store.append("a", { type: "assistant_message", messageId: "first", text: "First" });
store.append("a", { type: "assistant_message", messageId: "second", text: "Sec" });
store.append("a", { type: "assistant_message", messageId: "second", text: "ond" });
expect(store.getLastAssistantMessage("a")).toBe("Second");
});
});
it("includes transitive tool updates in a contiguous projected tail", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("a");
const tool = (callId: string, status: "running" | "completed") => ({
type: "tool_call" as const,
callId,
name: "shell",
status,
error: null,
detail: { type: "plain_text" as const, label: "work" },
});
store.append("a", tool("first", "running"));
store.append("a", tool("second", "running"));
store.append("a", { type: "user_message", text: "Continue" });
store.append("a", tool("first", "completed"));
store.append("a", { type: "assistant_message", text: "Answer" });
store.append("a", tool("second", "completed"));
const tail = store.fetch("a", { limit: 1 });
expect(tail.rows.map((row) => row.seqStart)).toEqual([1, 2, 3, 5]);
expect(tail).toMatchObject({ startSeq: 1, endSeq: 6, hasOlder: false });
});
async function measureTimelineTail(wholeTurns: boolean) {
const fixture = fileURLToPath(
new URL("./test-utils/timeline-responsiveness.ts", import.meta.url),
);
const { stdout, stderr } = await promisify(execFile)(
process.execPath,
["--expose-gc", "--import", "tsx", fixture, String(wholeTurns)],
{ timeout: 15_000 },
);
process.stderr.write(stderr);
return JSON.parse(stdout) as {
result: ReturnType<InMemoryAgentTimelineStore["fetch"]>;
blockingMs: number;
};
}
it("excludes unrelated test-runner CPU from the timeline fetch budget", async () => {
let unrelatedWork: ReturnType<typeof setImmediate>;
const burnCpu = () => {
const threadCpuUsage: (previous?: NodeJS.CpuUsage) => NodeJS.CpuUsage = Reflect.get(
process,
"threadCpuUsage",
);
const start = threadCpuUsage();
let elapsed: NodeJS.CpuUsage;
do {
elapsed = threadCpuUsage(start);
} while (elapsed.user + elapsed.system < 80_000);
unrelatedWork = setImmediate(burnCpu);
};
unrelatedWork = setImmediate(burnCpu);
try {
const { result: tail, blockingMs } = await measureTimelineTail(false);
expect(tail.rows).toHaveLength(200);
expect(tail.startSeq).toBe(99_801);
expect(tail.endSeq).toBe(100_000);
expect(blockingMs).toBeLessThan(50);
} finally {
clearImmediate(unrelatedWork);
}
});
it.each([false, true])(
"serves tail 200 from 100,000 retained rows without stalling the loop (wholeTurns=%s)",
async (wholeTurns) => {
const { result: tail, blockingMs } = await measureTimelineTail(wholeTurns);
expect(tail.rows).toHaveLength(200);
expect(tail.startSeq).toBe(99_801);
expect(tail.endSeq).toBe(100_000);
expect(blockingMs).toBeLessThan(50);
},
);
it("matches the previous reprojecting selector on seeded randomized timelines", () => {
let seed = 0x76aabbcc;
const random = (bound: number) => {
seed = (Math.imul(seed, 1664525) + 1013904223) >>> 0;
return Math.floor((seed / 0x1_0000_0000) * bound);
};
for (let scenario = 0; scenario < 100; scenario++) {
const store = new InMemoryAgentTimelineStore();
store.initialize("a", { epoch: "e" });
for (let seq = 1; seq <= 100; seq++) {
const identity = String(random(8));
const text = String(seq);
const items: AgentTimelineItem[] = [
{ type: "user_message", text },
{ type: "assistant_message", text },
{ type: "assistant_message", text, messageId: identity },
{ type: "reasoning", text },
{
type: "tool_call",
callId: identity,
name: "shell",
status: "running",
error: null,
detail: { type: "plain_text", label: text },
},
{
type: "plugin",
id: identity,
pluginId: "test",
kind: "state",
version: 1,
data: { text },
},
];
store.append("a", items[random(items.length)], { turnId: String(random(3)) });
}
// Include rehydration of already-projected rows, with non-monotonic seqEnd.
store.initialize("b", { rows: store.getRows("a"), epoch: "e" });
for (const agentId of ["a", "b"]) {
assertSelectionMatchesReference(store, agentId);
store.append(agentId, { type: "assistant_message", text: "later" });
assertSelectionMatchesReference(store, agentId);
}
}
});
function assertSelectionMatchesReference(store: InMemoryAgentTimelineStore, agentId: string): void {
for (const direction of ["tail", "before", "after"] as const) {
for (const limit of [0, 1, 7, 200]) {
for (const cursorSeq of [0, 1, 25, 50, 99, 100, 150]) {
const actual = store.fetch(agentId, {
direction,
limit,
cursor: { epoch: "e", seq: cursorSeq },
});
const expected = selectReferenceProjectedTimelinePage({
rows: store.getRows(agentId),
bounds: actual.window,
direction,
limit,
cursorSeq,
});
expect({
...expected,
entries: undefined,
rows: expected.entries.map((entry) => Object.assign({ seq: entry.seqEnd }, entry)),
}).toEqual({
entries: undefined,
rows: actual.rows,
startSeq: actual.startSeq,
endSeq: actual.endSeq,
hasOlder: actual.hasOlder,
hasNewer: actual.hasNewer,
});
}
}
}
}
it("continues a scroll across appends and lifecycle collapse without losing source coverage", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("a", { epoch: "e" });
const tool: AgentTimelineItem = {
type: "tool_call",
callId: "t",
name: "shell",
status: "running",
error: null,
detail: { type: "plain_text", label: "work" },
};
store.append("a", tool);
for (let seq = 2; seq <= 5; seq++) store.append("a", { type: "user_message", text: String(seq) });
const first = store.fetch("a", { direction: "after", cursor: { epoch: "e", seq: 0 }, limit: 2 });
expect(first.endSeq).toBe(2);
store.append("a", { ...tool, status: "completed" });
store.append("a", { type: "assistant_message", text: "Done" });
expect(first.rows[0].item).toMatchObject({ status: "running" });
const second = store.fetch("a", {
direction: "after",
cursor: { epoch: "e", seq: first.endSeq! },
limit: 2,
});
expect(second.rows.map((row) => row.seqStart)).toEqual([3, 4]);
expect(second.endSeq).toBe(4);
const third = store.fetch("a", {
direction: "after",
cursor: { epoch: "e", seq: second.endSeq! },
limit: 2,
});
expect(third.rows.map((row) => row.seqStart)).toEqual([1, 5]);
expect(third.rows[0].item).toMatchObject({ status: "completed" });
expect(third.endSeq).toBe(6);
const last = store.fetch("a", {
direction: "after",
cursor: { epoch: "e", seq: third.endSeq! },
limit: 2,
});
expect(last.rows.map((row) => row.item)).toEqual([{ type: "assistant_message", text: "Done" }]);
expect(last.endSeq).toBe(7);
expect(last.hasNewer).toBe(false);
});
packages/server/src/server/agent/agent-timeline-store.ts
-185
View File
@@ -1,185 +0,0 @@
import { randomUUID } from "node:crypto";
import { TimelineProjection, type ProjectedTimelineRow } from "./timeline-projection.js";
import { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
import type { AgentTimelineItem } from "./agent-sdk-types.js";
import type {
AgentTimelineFetchOptions,
AgentTimelineFetchResult,
AgentTimelineRow,
} from "./agent-timeline-store-types.js";
export interface SeedAgentTimelineOptions {
items?: readonly AgentTimelineItem[];
rows?: readonly AgentTimelineRow[];
epoch?: string;
nextSeq?: number;
timestamp?: string;
}
interface AgentTimelineState {
epoch: string;
projection: TimelineProjection;
minSeq: number;
nextSeq: number;
}
const DEFAULT_TIMELINE_FETCH_LIMIT = 200;
function cloneRow<T extends AgentTimelineRow>(row: T): T {
return { ...row };
}
export class InMemoryAgentTimelineStore {
private readonly states = new Map<string, AgentTimelineState>();
has(agentId: string): boolean {
return this.states.has(agentId);
}
initialize(agentId: string, options?: SeedAgentTimelineOptions): void {
const timestamp = options?.timestamp ?? new Date().toISOString();
const rows = options?.rows?.length
? options.rows.map(cloneRow)
: this.buildRowsFromItems(options?.items ?? [], options?.nextSeq ?? 1, timestamp);
const nextSeq = rows.reduce((next, row) => Math.max(next, row.seq + 1), options?.nextSeq ?? 1);
const projection = new TimelineProjection(rows);
this.states.set(agentId, {
epoch: options?.epoch ?? randomUUID(),
projection,
minSeq: projection.getRows()[0]?.seqStart ?? 0,
nextSeq,
});
}
delete(agentId: string): void {
this.states.delete(agentId);
}
getItems(agentId: string): AgentTimelineItem[] {
return this.requireState(agentId)
.projection.getRows()
.map((row) => row.item);
}
getRows(agentId: string): ProjectedTimelineRow[] {
return this.requireState(agentId).projection.getRows().map(cloneRow);
}
getSubmittedUserMessage(agentId: string, clientMessageId: string): AgentTimelineRow | null {
const row = this.requireState(agentId)
.projection.getRows()
.find(
(candidate) =>
candidate.item.type === "user_message" &&
candidate.item.clientMessageId === clientMessageId,
);
return row ? cloneRow(row) : null;
}
enrichSubmittedUserMessage(
agentId: string,
clientMessageId: string,
providerMessageId: string,
): AgentTimelineRow | null {
return this.requireState(agentId).projection.enrichSubmittedUserMessage(
clientMessageId,
providerMessageId,
);
}
getEpoch(agentId: string): string {
return this.requireState(agentId).epoch;
}
fetch(agentId: string, options?: AgentTimelineFetchOptions): AgentTimelineFetchResult {
const state = this.requireState(agentId);
const direction = options?.direction ?? "tail";
const cursor = options?.cursor;
const hasRows = state.minSeq !== 0;
const window = { minSeq: state.minSeq, maxSeq: state.nextSeq - 1, nextSeq: state.nextSeq };
const staleCursor = cursor !== undefined && cursor.epoch !== state.epoch;
const gap =
!staleCursor &&
direction === "after" &&
cursor !== undefined &&
hasRows &&
cursor.seq < state.minSeq - 1;
const reset = staleCursor || gap;
const page = state.projection.selectPage({
bounds: window,
direction: reset ? "tail" : direction,
cursorSeq: cursor?.seq,
limit: options?.limit ?? DEFAULT_TIMELINE_FETCH_LIMIT,
wholeTurns: options?.wholeTurns,
});
return {
epoch: state.epoch,
direction,
reset,
staleCursor,
gap,
window,
hasOlder: page.hasOlder,
hasNewer: page.hasNewer,
startSeq: page.startSeq,
endSeq: page.endSeq,
// Stored rows are already within budget; this bounds assistant and reasoning merges.
rows: page.entries.map((entry) =>
Object.assign({ seq: entry.seqEnd }, entry, {
item: limitAgentTimelineItemContent(entry.item),
}),
),
};
}
append(
agentId: string,
item: AgentTimelineItem,
options?: { timestamp?: string; providerMessageId?: string; turnId?: string },
): AgentTimelineRow {
const state = this.requireState(agentId);
const row: AgentTimelineRow = {
seq: state.nextSeq,
timestamp: options?.timestamp ?? new Date().toISOString(),
item,
...(options?.turnId ? { turnId: options.turnId } : {}),
...(options?.providerMessageId ? { providerMessageId: options.providerMessageId } : {}),
};
state.nextSeq += 1;
if (state.minSeq === 0) state.minSeq = row.seq;
state.projection.append(row);
return cloneRow(row);
}
getLastItem(agentId: string): AgentTimelineItem | null {
const state = this.requireState(agentId);
return state.projection.getLastItem(state.nextSeq - 1);
}
getLastAssistantMessage(agentId: string): string | null {
return this.requireState(agentId).projection.getLastAssistantMessage();
}
private requireState(agentId: string): AgentTimelineState {
const state = this.states.get(agentId);
if (!state) {
throw new Error(`Unknown agent '${agentId}'`);
}
return state;
}
private buildRowsFromItems(
items: readonly AgentTimelineItem[],
startSeq: number,
timestamp: string,
): AgentTimelineRow[] {
let nextSeq = startSeq;
return items.map((item) => {
const row: AgentTimelineRow = {
seq: nextSeq,
timestamp,
item,
};
nextSeq += 1;
return row;
});
}
}
packages/server/src/server/agent/import-sessions.test.ts
+1 -1
View File
@@ -679,7 +679,7 @@ class ProviderImportHarness {
return this.snapshot;
},
hydrateTimelineFromProvider: async () => {},
getTimeline: () => this.timeline,
getTimeline: async () => this.timeline,
closeAgent: async (agentId: string) => {
this.closedAgentIds.push(agentId);
this.activeAgent = null;
packages/server/src/server/agent/import-sessions.ts
+2 -2
View File
@@ -239,7 +239,7 @@ async function importProviderSessionNow(
});
return {
snapshot,
timelineSize: input.agentManager.getTimeline(snapshot.id).length,
timelineSize: (await input.agentManager.getTimeline(snapshot.id)).length,
};
} catch (error) {
await rollbackArchivedImport(input, archivedRecord, archivedRecord.archivedAt);
@@ -258,7 +258,7 @@ async function importProviderSessionNow(
return {
snapshot,
timelineSize: input.agentManager.getTimeline(snapshot.id).length,
timelineSize: (await input.agentManager.getTimeline(snapshot.id)).length,
};
}
packages/server/src/server/agent/mcp-parity.e2e.test.ts
+6 -6
View File
@@ -565,12 +565,12 @@ describe("Suite A: Core Fixes", () => {
const notifications = await waitFor({
timeoutMs: 10000,
label: `${outcome} notification`,
check: () => {
const items = daemonHandle.daemon.agentManager
.getTimeline(parentAgentId)
.filter(
(item) => item.type === "tool_call" && item.agentMessage?.sender.id === childId,
);
check: async () => {
const items = (
await daemonHandle.daemon.agentManager.getTimeline(parentAgentId)
).filter(
(item) => item.type === "tool_call" && item.agentMessage?.sender.id === childId,
);
return items.length > 0 ? items : null;
},
});
packages/server/src/server/agent/mcp-shared.ts
+1 -1
View File
@@ -142,7 +142,7 @@ export async function waitForAgentWithTimeout(
} catch (error) {
if (error instanceof Error && error.message === "wait timeout") {
const snapshot = agentManager.getAgent(agentId);
const timeline = agentManager.getTimeline(agentId);
const timeline = await agentManager.getTimeline(agentId);
const recent = selectItemsByProjectedLimit({
items: timeline,
direction: "tail",
packages/server/src/server/agent/projected-timeline-contract.test.ts
+24 -15
View File
@@ -64,9 +64,13 @@ test("projects Codex child history and confines old-client degradation to the ch
.toBe(1);
await expect
.poll(
() =>
ctx.daemon.daemon.agentManager.fetchProviderSubagentTimeline(agent.id, "child-thread")
.window.maxSeq,
async () =>
(
await ctx.daemon.daemon.agentManager.fetchProviderSubagentTimeline(
agent.id,
"child-thread",
)
).window.maxSeq,
)
.toBe(3);
const child = await ctx.client.fetchProviderSubagentTimeline(agent.id, "child-thread", {
@@ -170,9 +174,10 @@ test("resuming Codex restores child descriptors and complete messages without hi
test("retains only the latest cumulative Pi tool progress payload", async () => {
const { parseToolArgs, parseToolResult, mapToolDetail } =
await import("./providers/pi/tool-call-mapper.js");
const { InMemoryAgentTimelineStore } = await import("./agent-timeline-store.js");
const store = new InMemoryAgentTimelineStore();
store.initialize("agent");
const { AgentTimeline } = await import("./timeline/index.js");
const timeline = new AgentTimeline();
const scope = { agentId: "agent" };
await timeline.create(scope);
const tool = parseToolArgs("custom_progress", {});
const messages: string[] = [];
for (let index = 0; index < 20; index++) {
@@ -181,16 +186,20 @@ test("retains only the latest cumulative Pi tool progress payload", async () =>
content: [{ type: "text", text: "running" }],
details: { results: [{ messages: [...messages] }] },
});
store.append("agent", {
type: "tool_call",
callId: "same-call",
name: "custom_progress",
status: "running",
error: null,
detail: mapToolDetail(tool, result),
});
await timeline.append(scope, [
{
item: {
type: "tool_call",
callId: "same-call",
name: "custom_progress",
status: "running",
error: null,
detail: mapToolDetail(tool, result),
},
},
]);
}
const history = store.fetch("agent", { limit: 0 });
const history = await timeline.fetch(scope, { limit: 0 });
expect(history.rows).toHaveLength(1);
expect(history.rows[0]).toMatchObject({
seqStart: 1,
packages/server/src/server/agent/provider-subagents/store.test.ts
+17 -16
View File
@@ -1,11 +1,12 @@
import { describe, expect, test } from "vitest";
import { AgentTimeline } from "../timeline/index.js";
import { ProviderSubagentStore } from "./store.js";
describe("ProviderSubagentStore", () => {
test("keeps provider children and their timelines scoped to the parent agent", () => {
const subagents = new ProviderSubagentStore();
test("keeps provider children and their timelines scoped to the parent agent", async () => {
const subagents = new ProviderSubagentStore(new AgentTimeline());
subagents.apply("parent-a", "codex", {
await subagents.apply("parent-a", "codex", {
type: "upsert",
id: "child-1",
title: "Explore",
@@ -13,19 +14,19 @@ describe("ProviderSubagentStore", () => {
status: "running",
timestamp: "2026-07-12T10:00:00.000Z",
});
subagents.apply("parent-a", "codex", {
await subagents.apply("parent-a", "codex", {
type: "timeline",
id: "child-1",
item: { type: "assistant_message", text: "Found it." },
timestamp: "2026-07-12T10:00:01.000Z",
});
subagents.apply("parent-a", "codex", {
await subagents.apply("parent-a", "codex", {
type: "upsert",
id: "child-1",
status: "completed",
timestamp: "2026-07-12T10:00:02.000Z",
});
subagents.apply("parent-b", "claude", {
await subagents.apply("parent-b", "claude", {
type: "upsert",
id: "child-1",
title: "Review",
@@ -45,7 +46,7 @@ describe("ProviderSubagentStore", () => {
updatedAt: "2026-07-12T10:00:02.000Z",
}),
]);
expect(subagents.fetchTimeline("parent-a", "child-1").rows).toMatchObject([
expect((await subagents.fetchTimeline("parent-a", "child-1")).rows).toMatchObject([
{
seq: 1,
timestamp: "2026-07-12T10:00:01.000Z",
@@ -53,17 +54,17 @@ describe("ProviderSubagentStore", () => {
},
]);
expect(subagents.list("parent-b")[0]).toMatchObject({ provider: "claude", title: "Review" });
expect(subagents.deleteParent("parent-a")).toEqual([
expect(await subagents.deleteParent("parent-a")).toEqual([
{ type: "remove", parentAgentId: "parent-a", subagentId: "child-1" },
]);
expect(subagents.list("parent-a")).toEqual([]);
expect(subagents.list("parent-b")).toHaveLength(1);
});
test("limits oversized provider child tool output before storage", () => {
const subagents = new ProviderSubagentStore();
test("limits oversized provider child tool output before storage", async () => {
const subagents = new ProviderSubagentStore(new AgentTimeline());
const output = "x".repeat(70 * 1024);
const update = subagents.apply("parent-a", "opencode", {
const update = await subagents.apply("parent-a", "opencode", {
type: "timeline",
id: "child-1",
item: {
@@ -77,24 +78,24 @@ describe("ProviderSubagentStore", () => {
});
expect(update.type).toBe("timeline");
const [row] = subagents.fetchTimeline("parent-a", "child-1").rows;
const [row] = (await subagents.fetchTimeline("parent-a", "child-1")).rows;
expect(row?.item).toMatchObject({
type: "tool_call",
detail: { type: "shell", output: expect.stringMatching(/^x+…\[truncated \d+\.\d KB\]$/) },
});
});
test("pages provider history on projected item boundaries", () => {
const subagents = new ProviderSubagentStore();
test("pages provider history on projected item boundaries", async () => {
const subagents = new ProviderSubagentStore(new AgentTimeline());
for (let index = 0; index < 101; index += 1) {
subagents.apply("parent-a", "opencode", {
await subagents.apply("parent-a", "opencode", {
type: "timeline",
id: "child-1",
item: { type: "assistant_message", text: String(index) },
});
}
const page = subagents.fetchTimeline("parent-a", "child-1", {
const page = await subagents.fetchTimeline("parent-a", "child-1", {
direction: "tail",
limit: 1,
});
packages/server/src/server/agent/provider-subagents/store.ts
+27 -26
View File
@@ -1,11 +1,10 @@
import type { AgentProvider, AgentTimelineItem } from "../agent-sdk-types.js";
import { limitAgentTimelineItemContent } from "../agent-timeline-content.js";
import { InMemoryAgentTimelineStore } from "../agent-timeline-store.js";
import type {
AgentTimelineFetchOptions,
AgentTimelineFetchResult,
AgentTimelineRow,
} from "../agent-timeline-store-types.js";
import type { AgentTimeline, TimelineScope } from "../timeline/index.js";
export type ProviderSubagentStatus = "running" | "completed" | "failed" | "canceled";
@@ -66,6 +65,10 @@ function storeKey(parentAgentId: string, subagentId: string): string {
return `${parentAgentId}\0${subagentId}`;
}
function childScope(parentAgentId: string, subagentId: string): TimelineScope {
return { agentId: parentAgentId, subagentId };
}
/**
* Sticky upsert semantics for a descriptor field: an omitted value preserves what is stored, an
* explicit `null` clears it. Providers observe these fields incrementally, so a partial upsert
@@ -77,41 +80,38 @@ function stickyField<T>(next: T | undefined, previous: T | null | undefined): T
export class ProviderSubagentStore {
private readonly descriptors = new Map<string, ProviderSubagentDescriptor>();
private readonly timelines = new InMemoryAgentTimelineStore();
apply(
constructor(private readonly timeline: AgentTimeline) {}
async apply(
parentAgentId: string,
provider: AgentProvider,
event: ProviderSubagentInputEvent,
): ProviderSubagentStoreEvent {
): Promise<ProviderSubagentStoreEvent> {
const key = storeKey(parentAgentId, event.id);
const scope = childScope(parentAgentId, event.id);
if (event.type === "remove") {
this.descriptors.delete(key);
this.timelines.delete(key);
await this.timeline.delete(scope);
return { type: "remove", parentAgentId, subagentId: event.id };
}
if (event.type === "timeline") {
if (!this.timelines.has(key)) {
this.timelines.initialize(key);
}
const row = this.timelines.append(key, limitAgentTimelineItemContent(event.item), {
timestamp: event.timestamp,
});
await this.timeline.open(scope);
const {
commits: [{ row, epoch }],
} = await this.timeline.append(scope, [{ item: event.item, timestamp: event.timestamp }]);
return {
type: "timeline",
parentAgentId,
subagentId: event.id,
provider,
row,
epoch: this.timelines.getEpoch(key),
epoch,
};
}
const previous = this.descriptors.get(key);
if (!this.timelines.has(key)) {
this.timelines.initialize(key);
}
const timestamp = event.timestamp ?? new Date().toISOString();
const subagent: ProviderSubagentDescriptor = {
id: event.id,
@@ -128,6 +128,7 @@ export class ProviderSubagentStore {
parentSubagentId: stickyField(event.parentSubagentId, previous?.parentSubagentId),
};
this.descriptors.set(key, subagent);
await this.timeline.open(scope);
return { type: "upsert", subagent };
}
@@ -149,18 +150,18 @@ export class ProviderSubagentStore {
parentAgentId: string,
subagentId: string,
options?: AgentTimelineFetchOptions,
): AgentTimelineFetchResult {
return this.timelines.fetch(storeKey(parentAgentId, subagentId), options);
): Promise<AgentTimelineFetchResult> {
return this.timeline.fetch(childScope(parentAgentId, subagentId), options);
}
deleteParent(parentAgentId: string): ProviderSubagentStoreEvent[] {
const events: ProviderSubagentStoreEvent[] = [];
for (const subagent of this.list(parentAgentId)) {
const key = storeKey(parentAgentId, subagent.id);
this.descriptors.delete(key);
this.timelines.delete(key);
events.push({ type: "remove", parentAgentId, subagentId: subagent.id });
async deleteParent(parentAgentId: string): Promise<ProviderSubagentStoreEvent[]> {
const removed = this.list(parentAgentId).map((subagent) => subagent.id);
for (const subagentId of removed) {
this.descriptors.delete(storeKey(parentAgentId, subagentId));
}
return events;
await Promise.all(
removed.map((subagentId) => this.timeline.delete(childScope(parentAgentId, subagentId))),
);
return removed.map((subagentId) => ({ type: "remove", parentAgentId, subagentId }));
}
}
packages/server/src/server/agent/providers/claude/subagents/observation.test.ts
+10 -9
View File
@@ -1,6 +1,7 @@
import { describe, expect, it } from "vitest";
import { ProviderSubagentStore } from "../../../provider-subagents/store.js";
import { AgentTimeline } from "../../../timeline/index.js";
import { foldSubagentObservations, type SubagentObservation } from "./observation.js";
describe("foldSubagentObservations", () => {
@@ -66,9 +67,9 @@ describe("foldSubagentObservations", () => {
);
});
it("accumulates through the store so a later status keeps the declared identity", () => {
it("accumulates through the store so a later status keeps the declared identity", async () => {
// The property the whole design rests on: sources emit sparse facts, the store merges them.
const store = new ProviderSubagentStore();
const store = new ProviderSubagentStore(new AgentTimeline());
const events = foldSubagentObservations([
{ kind: "declared", id: "toolu_1", title: "Explore", description: "Find the code" },
{ kind: "status", id: "toolu_1", status: "completed" },
@@ -76,7 +77,7 @@ describe("foldSubagentObservations", () => {
let descriptor = null;
for (const event of events) {
const applied = store.apply("parent", "claude", event);
const applied = await store.apply("parent", "claude", event);
if (applied.type === "upsert") descriptor = applied.subagent;
}
@@ -102,15 +103,15 @@ describe("subtitle observations", () => {
expect(event).not.toHaveProperty("status");
});
it("does not revert a finished subagent when presentation changes", () => {
const store = new ProviderSubagentStore();
it("does not revert a finished subagent when presentation changes", async () => {
const store = new ProviderSubagentStore(new AgentTimeline());
let descriptor = null;
for (const event of foldSubagentObservations([
{ kind: "declared", id: "toolu_1", title: "Explore" },
{ kind: "status", id: "toolu_1", status: "completed" },
{ kind: "subtitle", id: "toolu_1", subtitle: "Explore · Opus 5 · 32.3k tokens" },
])) {
const applied = store.apply("parent", "claude", event);
const applied = await store.apply("parent", "claude", event);
if (applied.type === "upsert") descriptor = applied.subagent;
}
expect(descriptor).toMatchObject({
@@ -119,14 +120,14 @@ describe("subtitle observations", () => {
});
});
it("keeps identity when a later subtitle arrives", () => {
const store = new ProviderSubagentStore();
it("keeps identity when a later subtitle arrives", async () => {
const store = new ProviderSubagentStore(new AgentTimeline());
let descriptor = null;
for (const event of foldSubagentObservations([
{ kind: "declared", id: "toolu_1", title: "Explore", description: "Find the code" },
{ kind: "subtitle", id: "toolu_1", subtitle: "Explore · Opus 5" },
])) {
const applied = store.apply("parent", "claude", event);
const applied = await store.apply("parent", "claude", event);
if (applied.type === "upsert") descriptor = applied.subagent;
}
expect(descriptor).toMatchObject({
packages/server/src/server/agent/providers/claude/subagents/replay-source.test.ts
+14 -11
View File
@@ -11,6 +11,7 @@ import {
ProviderSubagentStore,
type ProviderSubagentDescriptor,
} from "../../../provider-subagents/store.js";
import { AgentTimeline } from "../../../timeline/index.js";
import { ClaudeTaskProtocolSource } from "./live-source.js";
import { foldSubagentObservations, type SubagentObservation } from "./observation.js";
import {
@@ -40,11 +41,13 @@ function parentWithTaskCall(failed = false): ClaudeReplayParentFacts {
};
}
function applyToStore(observations: SubagentObservation[]): ProviderSubagentDescriptor | null {
const store = new ProviderSubagentStore();
async function applyToStore(
observations: SubagentObservation[],
): Promise<ProviderSubagentDescriptor | null> {
const store = new ProviderSubagentStore(new AgentTimeline());
let descriptor: ProviderSubagentDescriptor | null = null;
for (const event of foldSubagentObservations(observations)) {
const applied = store.apply("parent", "claude", event);
const applied = await store.apply("parent", "claude", event);
if (applied.type === "upsert") descriptor = applied.subagent;
}
return descriptor;
@@ -203,7 +206,7 @@ describe("observeReplaySubagents", () => {
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(
const descriptor = await applyToStore(
(
await observeReplaySubagents({
subagents: [
@@ -231,7 +234,7 @@ describe("observeReplaySubagents", () => {
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(
const descriptor = await applyToStore(
(
await observeReplaySubagents({
subagents: [
@@ -257,7 +260,7 @@ describe("observeReplaySubagents", () => {
});
it("reports a scraped link's failure as failed", async () => {
const descriptor = applyToStore(
const descriptor = await applyToStore(
(
await observeReplaySubagents({
subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
@@ -299,7 +302,7 @@ describe("observeReplaySubagents", () => {
});
it("finishes a terminal Task child when the parent has not recorded its outcome", async () => {
const descriptor = applyToStore(
const descriptor = await applyToStore(
(
await observeReplaySubagents({
subagents: [
@@ -575,7 +578,7 @@ describe("replay usage", () => {
});
it("lands the complete provider-owned subtitle on the descriptor", async () => {
const store = new ProviderSubagentStore();
const store = new ProviderSubagentStore(new AgentTimeline());
const observations = (
await observeReplaySubagents({
subagents: [
@@ -596,7 +599,7 @@ describe("replay usage", () => {
})
).observations;
for (const event of foldSubagentObservations(observations)) {
store.apply("parent", "claude", event);
await store.apply("parent", "claude", event);
}
expect(store.get("parent", TOOL_USE_ID)?.subtitle).toBe("general-purpose · 7 tokens");
@@ -645,8 +648,8 @@ describe("live and replay agree", () => {
})
).observations;
const fromWire = applyToStore(liveObservations);
const fromDisk = applyToStore(replayObservations);
const fromWire = await applyToStore(liveObservations);
const fromDisk = await applyToStore(replayObservations);
const identity = (descriptor: ProviderSubagentDescriptor | null) => ({
id: descriptor?.id,
packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.test.ts
+5 -4
View File
@@ -7,6 +7,7 @@ import {
import { describe, expect, it } from "vitest";
import { ProviderSubagentStore } from "../../../provider-subagents/store.js";
import { AgentTimeline } from "../../../timeline/index.js";
import { foldSubagentObservations } from "./observation.js";
import {
observeReplayWorkflows,
@@ -82,9 +83,9 @@ describe("Claude workflow replay", () => {
],
parentEntries: parentEntries(),
});
const store = new ProviderSubagentStore();
const store = new ProviderSubagentStore(new AgentTimeline());
for (const event of foldSubagentObservations(observations)) {
store.apply("parent", "claude", event);
await store.apply("parent", "claude", event);
}
expect(store.get("parent", TOOL_CALL_ID)).toMatchObject({
@@ -107,9 +108,9 @@ describe("Claude workflow replay", () => {
},
]),
});
const store = new ProviderSubagentStore();
const store = new ProviderSubagentStore(new AgentTimeline());
for (const event of foldSubagentObservations(observations)) {
store.apply("parent", "claude", event);
await store.apply("parent", "claude", event);
}
expect(store.get("parent", TOOL_CALL_ID)).toMatchObject({
packages/server/src/server/agent/providers/codex-app-server-agent.test.ts
+9 -9
View File
@@ -625,7 +625,7 @@ function createReopenableCodexAgent(threads: Record<string, unknown[]>) {
// Cold-loads the agent, closes its runtime, and reopens it while Paseo keeps its timeline.
async loadCloseAndReopen(): Promise<AgentTimelineItem[]> {
await open();
const projected = manager.getTimeline(agentId);
const projected = await manager.getTimeline(agentId);
await manager.closeAgent(agentId);
await open();
return projected;
@@ -2072,7 +2072,7 @@ describe("Codex app-server provider", () => {
"thread/read",
]);
expect(codex.threadMethods(1)).toEqual(["thread/loaded/list", "thread/resume"]);
expect(codex.manager.getTimeline(codex.agentId)).toEqual(projected);
expect(await codex.manager.getTimeline(codex.agentId)).toEqual(projected);
} finally {
await codex.close();
}
@@ -2098,7 +2098,7 @@ describe("Codex app-server provider", () => {
expect(codex.latestAppServer().recordedRollbacks).toEqual([
{ threadId: "forked-thread", numTurns: 1 },
]);
expect(codex.manager.getTimeline(codex.agentId)).toEqual([
expect(await codex.manager.getTimeline(codex.agentId)).toEqual([
expect.objectContaining({ type: "user_message", messageId: "user-1", text: "First" }),
expect.objectContaining({ type: "assistant_message", text: "First answer." }),
]);
@@ -2151,9 +2151,9 @@ describe("Codex app-server provider", () => {
appServer.completeTurn({ threadId: "saved-thread" });
await run;
const toolCalls = codex.manager
.getTimeline(codex.agentId)
.filter((item) => item.type === "tool_call");
const toolCalls = (await codex.manager.getTimeline(codex.agentId)).filter(
(item) => item.type === "tool_call",
);
expect(new Set(toolCalls.map((item) => item.callId))).toEqual(new Set(["spawn-child"]));
expect(toolCalls.at(-1)).toMatchObject({ callId: "spawn-child", status: "completed" });
expect(codex.manager.getProviderSubagent(codex.agentId, "child-thread")).toMatchObject({
@@ -2161,9 +2161,9 @@ describe("Codex app-server provider", () => {
status: "completed",
});
expect(
codex.manager
.fetchProviderSubagentTimeline(codex.agentId, "child-thread")
.rows.map((row) => row.item),
(await codex.manager.fetchProviderSubagentTimeline(codex.agentId, "child-thread")).rows.map(
(row) => row.item,
),
).toEqual([
expect.objectContaining({ type: "assistant_message", text: "Child started." }),
expect.objectContaining({ type: "assistant_message", text: "Child replied." }),
packages/server/src/server/agent/providers/mock-load-test-agent.test.ts
+1 -1
View File
@@ -535,7 +535,7 @@ describe("MockLoadTestAgentClient", () => {
await vi.advanceTimersByTimeAsync(10_000);
await resultPromise;
const timeline = manager.getTimeline(agent.id);
const timeline = await manager.getTimeline(agent.id);
const assistantMessages = timeline.filter((item) => item.type === "assistant_message");
const toolCalls = timeline.filter((item) => item.type === "tool_call");
packages/server/src/server/agent/providers/pi/extensions/edxeth-pi-subagents/index.test.ts
+4 -3
View File
@@ -8,6 +8,7 @@ import { join } from "node:path";
import { edxethPiSubagents } from "./index.js";
import { streamPiHistory } from "../../history-mapper.js";
import { ProviderSubagentStore } from "../../../../provider-subagents/store.js";
import { AgentTimeline } from "../../../../timeline/index.js";
const args = {
agent: "scout",
@@ -274,13 +275,13 @@ describe("edxeth/pi-subagents adapter", () => {
from: `/tmp/paseo-edxeth-fixture/${mode}/child-session.jsonl`,
to: fileURLToPath(new URL(`./fixtures/${mode}-child-session.jsonl`, import.meta.url)),
});
const store = new ProviderSubagentStore();
const store = new ProviderSubagentStore(new AgentTimeline());
for await (const event of streamPiHistory("pi", fixture.messages)) {
if (event.type === "provider_subagent") store.apply("parent", "pi", event.event);
if (event.type === "provider_subagent") await store.apply("parent", "pi", event.event);
}
const children = store.list("parent");
expect(children).toEqual([expect.objectContaining({ status: "completed", title: "marker" })]);
const transcript = store.fetchTimeline("parent", children[0]!.id);
const transcript = await store.fetchTimeline("parent", children[0]!.id);
expect(transcript.rows).toContainEqual(
expect.objectContaining({
item: expect.objectContaining({
packages/server/src/server/agent/providers/pi/extensions/gotgenes-pi-subagents/index.test.ts
+8 -3
View File
@@ -1,6 +1,7 @@
import { describe, expect, test } from "vitest";
import { fileURLToPath } from "node:url";
import { ProviderSubagentStore } from "../../../../provider-subagents/store.js";
import { AgentTimeline } from "../../../../timeline/index.js";
import { createPiExtensionHost } from "../index.js";
import { readSubagentFixture, verifySubagentFixture } from "../subagent-fixture-test.js";
import { streamPiHistory } from "../../history-mapper.js";
@@ -43,9 +44,13 @@ describe("@gotgenes/pi-subagents adapter", () => {
if (block.type === "toolCall") block.arguments = start.args;
}
const events = await verifySubagentFixture(fixture);
const store = new ProviderSubagentStore();
store.apply("parent", "pi", { type: "upsert", id: "previous-child", status: "completed" });
for (const event of events) store.apply("parent", "pi", event.event);
const store = new ProviderSubagentStore(new AgentTimeline());
await store.apply("parent", "pi", {
type: "upsert",
id: "previous-child",
status: "completed",
});
for (const event of events) await store.apply("parent", "pi", event.event);
expect(events.map(({ event }) => event.type)).toEqual(["upsert", "remove"]);
expect(store.list("parent")).toMatchObject([{ id: "previous-child", status: "completed" }]);
},
packages/server/src/server/agent/rewind/rewind.test.ts
+5 -5
View File
@@ -75,9 +75,9 @@ describe("AgentManager rewind", () => {
expect(session.recordedRewinds).toEqual([{ mode: "conversation", messageId: "message-1" }]);
expect(session.historyReadCount).toBe(1);
expect(manager.fetchTimeline(agentId, { limit: 0 }).rows.map((row) => row.item)).toEqual([
{ type: "user_message", text: "before", messageId: "message-1" },
]);
expect(
(await manager.fetchTimeline(agentId, { limit: 0 })).rows.map((row) => row.item),
).toEqual([{ type: "user_message", text: "before", messageId: "message-1" }]);
});
test("replaces the canonical epoch without replaying reconstructed parent rows", async () => {
@@ -87,7 +87,7 @@ describe("AgentManager rewind", () => {
text: `rewound ${index}`,
}));
session.history = history;
const epochBefore = manager.fetchTimeline(agentId, { limit: 0 }).epoch;
const epochBefore = (await manager.fetchTimeline(agentId, { limit: 0 })).epoch;
const events: string[] = [];
const unsubscribe = manager.subscribe((event) => events.push(event.type), {
replayState: false,
@@ -96,7 +96,7 @@ describe("AgentManager rewind", () => {
await manager.rewind(agentId, "message-1", "conversation");
unsubscribe();
const replacement = manager.fetchTimeline(agentId, { limit: 0 });
const replacement = await manager.fetchTimeline(agentId, { limit: 0 });
expect(replacement.epoch).not.toBe(epochBefore);
expect(replacement.rows).toEqual([
expect.objectContaining({
packages/server/src/server/agent/test-utils/plugin-stream-memory.ts
+5 -5
View File
@@ -6,7 +6,7 @@ import { setTimeout as delay } from "node:timers/promises";
import type { ProviderEvent, ProviderRegistration } from "@getpaseo/plugin/server/provider";
import pino from "pino";
import { PluginAgentClientRegistry } from "../plugin-provider.js";
import { InMemoryAgentTimelineStore } from "../agent-timeline-store.js";
import { AgentTimeline } from "../timeline/index.js";
import { AgentManager } from "../agent-manager.js";
import { AgentStorage } from "../agent-storage.js";
import { buildProviderRegistry } from "../provider-registry.js";
@@ -67,12 +67,12 @@ const registrations: ProviderRegistration[] = Array.from({ length: 3 }, (_, inde
},
}));
plugins.replace(registrations);
const store = new InMemoryAgentTimelineStore();
const timeline = new AgentTimeline();
const manager = new AgentManager({
clients: plugins.clients(),
providerDefinitions: buildProviderRegistry(logger, { pluginProviders: plugins.definitions() }),
registry: new AgentStorage(root, logger),
timelineStore: store,
timeline,
logger,
});
const agents = await Promise.all(
@@ -143,7 +143,7 @@ global.gc();
const retained = process.memoryUsage().heapUsed - before;
assert(retained < 24 * 1024 * 1024, `Retained heap exceeded 24 MiB: ${retained} bytes`);
for (let index = 0; index < sessions.length; index++) {
const items = store.getItems(agents[index]!.id);
const items = (await timeline.readRows({ agentId: agents[index]!.id })).map((row) => row.item);
assert.equal(items.length, steps * 2);
for (let step = 0; step < steps; step++) {
const item = items[step * 2]!;
@@ -160,7 +160,7 @@ for (let index = 0; index < sessions.length; index++) {
assert.equal(rootItems, steps * 2);
assert.equal(childItems, steps * 2);
await manager.archiveAgent(agents[index]!.id);
assert.equal(store.has(agents[index]!.id), false);
assert.equal(await timeline.describe({ agentId: agents[index]!.id }), null);
assert.equal(
manager.listAgents().some((agent) => agent.id === agents[index]!.id),
false,
packages/server/src/server/agent/test-utils/timeline-responsiveness.ts
+13 -9
View File
@@ -1,23 +1,27 @@
import { setImmediate as nextTurn } from "node:timers/promises";
import { measureMainThreadWork } from "../../../test-utils/main-thread-work.js";
import { InMemoryAgentTimelineStore } from "../agent-timeline-store.js";
import { AgentTimeline } from "../timeline/index.js";
// Measure a cold fetch on its own thread, without Vitest reporting or callbacks.
const wholeTurns = process.argv[2] === "true";
const store = new InMemoryAgentTimelineStore();
store.initialize("large");
for (let index = 0; index < 100_000; index++) {
store.append("large", { type: "user_message", text: String(index) });
}
const timeline = new AgentTimeline();
const scope = { agentId: "large" };
await timeline.create(scope);
await timeline.append(
scope,
Array.from({ length: 100_000 }, (_, index) => ({
item: { type: "user_message" as const, text: String(index) },
})),
);
// Finish fixture allocation and collect its temporary garbage before measuring.
globalThis.gc!();
await nextTurn();
const measured = await measureMainThreadWork(
`timeline tail main-thread work (wholeTurns=${wholeTurns})`,
() => {
let page = store.fetch("large", { limit: 200, wholeTurns });
async () => {
let page = await timeline.fetch(scope, { limit: 200, wholeTurns });
for (let repeat = 1; repeat < 8; repeat++)
page = store.fetch("large", { limit: 200, wholeTurns });
page = await timeline.fetch(scope, { limit: 200, wholeTurns });
return page;
},
);
packages/server/src/server/agent/timeline-projection.ts
+17 -4
View File
@@ -308,7 +308,6 @@ interface ProjectedTimelinePageOptions {
export class TimelineProjection {
private readonly rows: ProjectedTimelineRow[] = [];
private readonly identities = new Map<string, number>();
private lastAssistantRowIndex: number | null = null;
// Disjoint source-coverage spans in display order. A tail starting inside a
// span must include its first anchor to cover interleaved lifecycle updates.
private readonly spans: { start: number; endSeq: number }[] = [];
@@ -356,7 +355,6 @@ export class TimelineProjection {
return this.rows.length - 1;
}
if (identity !== null) this.identities.set(identity, this.rows.length);
if (entry.item.type === "assistant_message") this.lastAssistantRowIndex = this.rows.length;
this.rows.push({ ...entry, seq: entry.seqEnd });
const last = this.spans.at(-1);
if (last && last.endSeq >= entry.seqStart) {
@@ -394,6 +392,10 @@ export class TimelineProjection {
return this.rows.map((row) => ({ ...row }));
}
get size(): number {
return this.rows.length;
}
selectPage(input: ProjectedTimelinePageOptions): ProjectedTimelinePageSelection {
if (this.rows.length === 0) {
const cursor =
@@ -522,9 +524,20 @@ export class TimelineProjection {
return last?.endSeq === seq ? this.rows[last.rowIndex].item : null;
}
/**
* The text of the trailing run of assistant rows, joined in order. Rows after the run are
* skipped; the first non-assistant row before it ends the run.
*/
getLastAssistantMessage(): string | null {
const row = this.lastAssistantRowIndex === null ? null : this.rows[this.lastAssistantRowIndex];
return row?.item.type === "assistant_message" ? row.item.text : null;
let end = this.rows.length;
while (end > 0 && this.rows[end - 1].item.type !== "assistant_message") end -= 1;
let start = end;
while (start > 0 && this.rows[start - 1].item.type === "assistant_message") start -= 1;
if (start === end) return null;
return this.rows
.slice(start, end)
.map((row) => (row.item.type === "assistant_message" ? row.item.text : ""))
.join("");
}
enrichSubmittedUserMessage(
packages/server/src/server/agent/timeline/agent-timeline.test.ts
+659
View File
@@ -0,0 +1,659 @@
import { execFile } from "node:child_process";
import { promisify } from "node:util";
import { fileURLToPath } from "node:url";
import { describe, expect, it } from "vitest";
import type { AgentTimelineItem } from "../agent-sdk-types.js";
import { limitAgentTimelineItemContent } from "../agent-timeline-content.js";
import type { AgentTimelineFetchResult, AgentTimelineRow } from "../agent-timeline-store-types.js";
import { DeferredTimelineBackend } from "./testing.js";
import { selectReferenceProjectedTimelinePage } from "../test-utils/timeline-page-reference.js";
import { AgentTimeline, type TimelineCommit, type TimelineScope } from "./index.js";
import { UnknownTimelineError } from "./internal/backend.js";
const root = (agentId: string): TimelineScope => ({ agentId });
async function append(
timeline: AgentTimeline,
agentId: string,
item: AgentTimelineItem,
options?: { turnId?: string },
): Promise<AgentTimelineRow> {
const { commits } = await timeline.append(root(agentId), [{ item, ...options }]);
const [commit] = commits;
return commit.row;
}
describe.each([
["memory", () => new AgentTimeline()],
["deferred", () => new AgentTimeline(new DeferredTimelineBackend())],
])("AgentTimeline over a %s backend", (_name, createTimeline) => {
it("lets a read observe writes called before it without awaiting them", async () => {
const timeline = createTimeline();
void timeline.create(root("a"), { epoch: "e" });
const published: number[] = [];
for (const text of ["one", "two", "three"]) {
void timeline.ingest(
root("a"),
{ item: { type: "user_message", text } },
{ publish: ({ row }) => published.push(row.seq) },
);
}
const rows = await timeline.readRows(root("a"));
expect(rows.map((row) => row.item)).toEqual([
{ type: "user_message", text: "one" },
{ type: "user_message", text: "two" },
{ type: "user_message", text: "three" },
]);
expect(published).toEqual([1, 2, 3]);
});
it("publishes each commit before the next operation starts", async () => {
const timeline = createTimeline();
await timeline.create(root("a"), { epoch: "e" });
const order: string[] = [];
const first = timeline.ingest(
root("a"),
{ item: { type: "assistant_message", text: "A" } },
{ publish: () => order.push("publish A") },
);
const settled = timeline.settled(root("a")).then(() => order.push("settled"));
const second = timeline.ingest(
root("a"),
{ item: { type: "assistant_message", text: "B" } },
{ publish: () => order.push("publish B") },
);
await Promise.all([first, settled, second]);
expect(order).toEqual(["publish A", "settled", "publish B"]);
});
it("rejects a failed write without publishing it or consuming a sequence number", async () => {
const backend = new DeferredTimelineBackend();
const timeline = new AgentTimeline(backend);
await timeline.create(root("a"), { epoch: "e" });
const published: TimelineCommit[] = [];
backend.failNextAppend = true;
const failed = timeline.ingest(
root("a"),
{ item: { type: "user_message", text: "lost" } },
{ publish: (commit) => published.push(commit) },
);
const next = timeline.ingest(
root("a"),
{ item: { type: "user_message", text: "kept" } },
{ publish: (commit) => published.push(commit) },
);
await expect(failed).rejects.toThrow("disk full");
await expect(next).resolves.toMatchObject({ kind: "recorded", row: { seq: 1 } });
expect(published.map(({ row }) => row.item)).toEqual([{ type: "user_message", text: "kept" }]);
expect((await timeline.fetch(root("a"))).rows.map((row) => row.seq)).toEqual([1]);
});
it("records a submitted prompt once per client message id", async () => {
const timeline = createTimeline();
await timeline.create(root("a"), { epoch: "e" });
const prompt: AgentTimelineItem = { type: "user_message", text: "Hi", clientMessageId: "c1" };
const published: number[] = [];
const publish = ({ row }: TimelineCommit) => published.push(row.seq);
const first = await timeline.ingest(root("a"), { item: prompt, turnId: "t1" }, { publish });
const echo = await timeline.ingest(root("a"), { item: prompt }, { publish });
expect(first).toMatchObject({ kind: "recorded", epoch: "e", row: { seq: 1, turnId: "t1" } });
expect(echo).toMatchObject({ kind: "duplicate", epoch: "e", row: { seq: 1 } });
expect(published).toEqual([1]);
await timeline.attachProviderMessageId(root("a"), "c1", "provider-1");
expect(await timeline.findSubmittedPrompt(root("a"), "c1")).toMatchObject({
seq: 1,
providerMessageId: "provider-1",
});
});
it("keeps a commit whose publication throws and lets later writes proceed", async () => {
const timeline = createTimeline();
await timeline.create(root("a"), { epoch: "e" });
const prompt: AgentTimelineItem = { type: "user_message", text: "Hi", clientMessageId: "c1" };
const failure = new Error("subscriber failed");
const throwing = () => {
throw failure;
};
const published: number[] = [];
const unpublished = timeline.ingest(root("a"), { item: prompt }, { publish: throwing });
const next = timeline.ingest(
root("a"),
{ item: { type: "assistant_message", text: "reply" } },
{ publish: ({ row }) => published.push(row.seq) },
);
const batch = timeline.append(root("a"), [{ item: { type: "user_message", text: "batch" } }], {
publish: throwing,
});
await expect(unpublished).resolves.toMatchObject({
kind: "unpublished",
epoch: "e",
row: { seq: 1, item: prompt },
error: failure,
});
await expect(next).resolves.toMatchObject({ kind: "recorded", row: { seq: 2 } });
await expect(batch).resolves.toMatchObject({
kind: "unpublished",
commits: [{ row: { seq: 3 } }],
error: failure,
});
expect(published).toEqual([2]);
// The committed prompt is not recorded twice when ingested again.
await expect(timeline.ingest(root("a"), { item: prompt })).resolves.toMatchObject({
kind: "duplicate",
row: { seq: 1 },
});
expect((await timeline.readRows(root("a"))).map((row) => row.seq)).toEqual([1, 2, 3]);
});
it("reads the trailing run of assistant messages as the last assistant message", async () => {
const tool = (callId: string): AgentTimelineItem => ({
type: "tool_call",
callId,
name: "shell",
status: "completed",
error: null,
detail: { type: "unknown", input: null, output: null },
});
const timeline = createTimeline();
await timeline.create(root("a"));
expect(await timeline.lastAssistantMessage(root("a"))).toBeNull();
await timeline.append(root("a"), [{ item: tool("only") }]);
expect(await timeline.lastAssistantMessage(root("a"))).toBeNull();
await timeline.append(root("a"), [
{ item: { type: "assistant_message", text: "Earlier reply" } },
{ item: { type: "user_message", text: "Next" } },
{ item: { type: "assistant_message", text: "Before the tool. " } },
{ item: tool("call-1") },
{ item: { type: "assistant_message", text: "Hel", messageId: "m1" } },
{ item: { type: "assistant_message", text: "lo", messageId: "m1" } },
{ item: { type: "assistant_message", text: " world", messageId: "m2" } },
{ item: tool("call-2") },
]);
expect(await timeline.lastAssistantMessage(root("a"))).toBe("Hello world");
await timeline.append(root("a"), [
{ item: { type: "user_message", text: "Again" } },
{ item: { type: "assistant_message", text: "Bye" } },
]);
expect(await timeline.lastAssistantMessage(root("a"))).toBe("Bye");
});
it("keeps provider child history apart from its parent's", async () => {
const timeline = createTimeline();
const child = { agentId: "a", subagentId: "child" };
await timeline.create(root("a"), { epoch: "parent" });
await timeline.open(child);
await append(timeline, "a", { type: "assistant_message", text: "parent" });
await timeline.append(child, [{ item: { type: "assistant_message", text: "child" } }]);
expect((await timeline.readRows(child)).map((row) => [row.seq, row.item])).toEqual([
[1, { type: "assistant_message", text: "child" }],
]);
await timeline.delete(child);
expect(await timeline.describe(child)).toBeNull();
expect(await timeline.describe(root("a"))).toEqual({ epoch: "parent", size: 1 });
await expect(timeline.readRows(child)).rejects.toBeInstanceOf(UnknownTimelineError);
});
it("gives a recreated history a new identity", async () => {
const timeline = createTimeline();
const before = await timeline.create(root("a"));
await append(timeline, "a", { type: "user_message", text: "old" });
const after = await timeline.create(root("a"));
expect(after.epoch).not.toBe(before.epoch);
expect(after.size).toBe(0);
expect(timeline.residentSize(root("a"))).toBe(0);
expect(await timeline.open(root("a"))).toEqual(after);
});
it("bounds ingested content", async () => {
const timeline = createTimeline();
await timeline.create(root("a"));
const text = "a".repeat(400 * 1024);
const { row } = await timeline.ingest(root("a"), {
item: { type: "assistant_message", text },
});
expect(Buffer.byteLength(JSON.stringify(row.item))).toBeLessThanOrEqual(256 * 1024);
});
});
describe("history paging", () => {
it("resets stale forward cursors to complete available turns", async () => {
const timeline = new AgentTimeline();
await timeline.create(root("agent"), { epoch: "current" });
for (let index = 0; index < 60; index++)
await append(
timeline,
"agent",
{
type: "tool_call",
callId: `call-${index}`,
name: "shell",
status: "completed",
error: null,
detail: { type: "unknown", input: null, output: null },
},
{ turnId: "run" },
);
const page = await timeline.fetch(root("agent"), {
direction: "after",
cursor: { epoch: "old", seq: 10 },
limit: 40,
wholeTurns: true,
});
expect(page.rows).toHaveLength(60);
expect(page).toMatchObject({
reset: true,
staleCursor: true,
startSeq: 1,
endSeq: 60,
hasOlder: false,
hasNewer: false,
});
});
const seededRows: AgentTimelineRow[] = [
{
seq: 5,
timestamp: "2026-01-01T00:00:00.000Z",
item: { type: "assistant_message", text: "five", messageId: "five" },
},
{
seq: 6,
timestamp: "2026-01-01T00:00:01.000Z",
item: { type: "assistant_message", text: "six", messageId: "six" },
},
{
seq: 7,
timestamp: "2026-01-01T00:00:02.000Z",
item: { type: "assistant_message", text: "seven", messageId: "seven" },
},
];
it("clamps an overshooting before cursor into the bounded tail window", async () => {
const timeline = new AgentTimeline();
await timeline.create(root("agent-1"), { epoch: "epoch-1", nextSeq: 8, rows: seededRows });
const result = await timeline.fetch(root("agent-1"), {
direction: "before",
cursor: { epoch: "epoch-1", seq: 100 },
limit: 2,
});
expect(result).toMatchObject({
epoch: "epoch-1",
direction: "before",
reset: false,
staleCursor: false,
gap: false,
window: { minSeq: 5, maxSeq: 7, nextSeq: 8 },
hasOlder: true,
hasNewer: false,
rows: [
{
seq: 6,
timestamp: "2026-01-01T00:00:01.000Z",
item: { type: "assistant_message", text: "six", messageId: "six" },
},
{
seq: 7,
timestamp: "2026-01-01T00:00:02.000Z",
item: { type: "assistant_message", text: "seven", messageId: "seven" },
},
],
});
});
it("returns a bounded reset window when an after cursor is behind retained history", async () => {
const timeline = new AgentTimeline();
await timeline.create(root("agent-1"), { epoch: "epoch-1", nextSeq: 8, rows: seededRows });
const result = await timeline.fetch(root("agent-1"), {
direction: "after",
cursor: { epoch: "epoch-1", seq: 1 },
limit: 1,
});
expect(result).toMatchObject({
epoch: "epoch-1",
direction: "after",
reset: true,
staleCursor: false,
gap: true,
window: { minSeq: 5, maxSeq: 7, nextSeq: 8 },
hasOlder: true,
hasNewer: false,
rows: [
{
seq: 7,
timestamp: "2026-01-01T00:00:02.000Z",
item: { type: "assistant_message", text: "seven", messageId: "seven" },
},
],
});
});
});
describe("projected timeline retention", () => {
it("retains one full tool state while streaming every source update", async () => {
const timeline = new AgentTimeline();
await timeline.create(root("agent"));
for (let seq = 1; seq <= 2000; seq++) {
const item = {
type: "tool_call" as const,
callId: "child",
name: "task",
status: "running" as const,
error: null,
detail: { type: "plain_text" as const, label: "Child", text: "x".repeat(seq * 128) },
};
expect(await append(timeline, "agent", item)).toMatchObject({
seq,
item: limitAgentTimelineItemContent(item),
});
}
const result = await timeline.fetch(root("agent"), { limit: 0 });
expect(result.rows).toHaveLength(1);
expect(JSON.stringify(result.rows).length).toBeLessThan(260_000);
expect(result.window.maxSeq).toBe(2000);
});
it("catches up a mid-message cursor with the complete projected message", async () => {
const timeline = new AgentTimeline();
await timeline.create(root("agent"), { epoch: "e" });
await append(timeline, "agent", { type: "assistant_message", messageId: "m", text: "A" });
await append(timeline, "agent", { type: "assistant_message", messageId: "m", text: "B" });
expect(
(await timeline.fetch(root("agent"), { direction: "after", cursor: { epoch: "e", seq: 1 } }))
.rows,
).toMatchObject([
{
item: { text: "AB" },
seqStart: 1,
seqEnd: 2,
sourceSeqRanges: [{ startSeq: 1, endSeq: 2 }],
},
]);
});
it("serves a merged assistant message within the item budget", async () => {
const timeline = new AgentTimeline();
await timeline.create(root("agent"), { epoch: "e" });
const chunk = "a".repeat(200 * 1024);
for (let index = 0; index < 10; index++) {
await append(timeline, "agent", { type: "assistant_message", messageId: "m", text: chunk });
}
const [row] = (await timeline.fetch(root("agent"), { limit: 1 })).rows;
expect(row).toMatchObject({ seqStart: 1, seqEnd: 10 });
expect(row.item).toMatchObject({ type: "assistant_message", messageId: "m" });
expect(Buffer.byteLength(JSON.stringify(row.item))).toBeLessThanOrEqual(256 * 1024);
// The resident projection keeps the whole message for completion notifications.
const [resident] = await timeline.readRows(root("agent"));
expect(resident.item.type === "assistant_message" && resident.item.text).toHaveLength(
10 * chunk.length,
);
});
});
describe("projected sequence ownership", () => {
it("preserves fetched coverage when another source chunk arrives", async () => {
const timeline = new AgentTimeline();
await timeline.create(root("a"));
await append(timeline, "a", { type: "assistant_message", text: "A" });
const before = await timeline.fetch(root("a"));
await append(timeline, "a", { type: "assistant_message", text: "B" });
expect(before.rows[0]).toMatchObject({
item: { text: "A" },
seqEnd: 1,
sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }],
});
expect((await timeline.fetch(root("a"))).rows[0]).toMatchObject({
item: { text: "AB" },
seqEnd: 2,
sourceSeqRanges: [{ startSeq: 1, endSeq: 2 }],
});
});
it("preserves source positions when seeding a projected history whose last update is an earlier tool", async () => {
const timeline = new AgentTimeline();
await timeline.create(root("a"));
const tool = {
type: "tool_call" as const,
callId: "t",
name: "shell",
error: null,
detail: { type: "plain_text" as const, label: "work" },
};
await append(timeline, "a", { ...tool, status: "running" });
await append(timeline, "a", { type: "assistant_message", text: "Answer" });
await append(timeline, "a", { ...tool, status: "completed" });
await timeline.create(root("b"), { rows: await timeline.readRows(root("a")) });
expect(await timeline.lastItem(root("b"))).toMatchObject({
type: "tool_call",
status: "completed",
});
expect((await timeline.readRows(root("b")))[1]?.item).toEqual({
type: "assistant_message",
text: "Answer",
});
expect((await append(timeline, "b", { type: "user_message", text: "next" })).seq).toBe(4);
expect(await timeline.lastItem(root("b"))).toEqual({ type: "user_message", text: "next" });
expect((await timeline.fetch(root("b"))).rows.map((row) => row.seqStart)).toEqual([1, 2, 4]);
});
it("keeps differently identified assistant messages as separate rows", async () => {
const timeline = new AgentTimeline();
await timeline.create(root("a"));
await append(timeline, "a", { type: "assistant_message", messageId: "first", text: "First" });
await append(timeline, "a", { type: "assistant_message", messageId: "second", text: "Sec" });
await append(timeline, "a", { type: "assistant_message", messageId: "second", text: "ond" });
expect((await timeline.readRows(root("a"))).map((row) => row.item)).toEqual([
{ type: "assistant_message", messageId: "first", text: "First" },
{ type: "assistant_message", messageId: "second", text: "Second" },
]);
});
});
it("includes transitive tool updates in a contiguous projected tail", async () => {
const timeline = new AgentTimeline();
await timeline.create(root("a"));
const tool = (callId: string, status: "running" | "completed") => ({
type: "tool_call" as const,
callId,
name: "shell",
status,
error: null,
detail: { type: "plain_text" as const, label: "work" },
});
await append(timeline, "a", tool("first", "running"));
await append(timeline, "a", tool("second", "running"));
await append(timeline, "a", { type: "user_message", text: "Continue" });
await append(timeline, "a", tool("first", "completed"));
await append(timeline, "a", { type: "assistant_message", text: "Answer" });
await append(timeline, "a", tool("second", "completed"));
const tail = await timeline.fetch(root("a"), { limit: 1 });
expect(tail.rows.map((row) => row.seqStart)).toEqual([1, 2, 3, 5]);
expect(tail).toMatchObject({ startSeq: 1, endSeq: 6, hasOlder: false });
});
async function measureTimelineTail(wholeTurns: boolean) {
const fixture = fileURLToPath(
new URL("../test-utils/timeline-responsiveness.ts", import.meta.url),
);
const { stdout, stderr } = await promisify(execFile)(
process.execPath,
["--expose-gc", "--import", "tsx", fixture, String(wholeTurns)],
{ timeout: 15_000 },
);
process.stderr.write(stderr);
return JSON.parse(stdout) as {
result: AgentTimelineFetchResult;
blockingMs: number;
};
}
it("excludes unrelated test-runner CPU from the timeline fetch budget", async () => {
let unrelatedWork: ReturnType<typeof setImmediate>;
const burnCpu = () => {
const threadCpuUsage: (previous?: NodeJS.CpuUsage) => NodeJS.CpuUsage = Reflect.get(
process,
"threadCpuUsage",
);
const start = threadCpuUsage();
let elapsed: NodeJS.CpuUsage;
do {
elapsed = threadCpuUsage(start);
} while (elapsed.user + elapsed.system < 80_000);
unrelatedWork = setImmediate(burnCpu);
};
unrelatedWork = setImmediate(burnCpu);
try {
const { result: tail, blockingMs } = await measureTimelineTail(false);
expect(tail.rows).toHaveLength(200);
expect(tail.startSeq).toBe(99_801);
expect(tail.endSeq).toBe(100_000);
expect(blockingMs).toBeLessThan(50);
} finally {
clearImmediate(unrelatedWork);
}
});
it.each([false, true])(
"serves tail 200 from 100,000 retained rows without stalling the loop (wholeTurns=%s)",
async (wholeTurns) => {
const { result: tail, blockingMs } = await measureTimelineTail(wholeTurns);
expect(tail.rows).toHaveLength(200);
expect(tail.startSeq).toBe(99_801);
expect(tail.endSeq).toBe(100_000);
expect(blockingMs).toBeLessThan(50);
},
);
it("matches the previous reprojecting selector on seeded randomized timelines", async () => {
let seed = 0x76aabbcc;
const random = (bound: number) => {
seed = (Math.imul(seed, 1664525) + 1013904223) >>> 0;
return Math.floor((seed / 0x1_0000_0000) * bound);
};
for (let scenario = 0; scenario < 100; scenario++) {
const timeline = new AgentTimeline();
await timeline.create(root("a"), { epoch: "e" });
for (let seq = 1; seq <= 100; seq++) {
const identity = String(random(8));
const text = String(seq);
const items: AgentTimelineItem[] = [
{ type: "user_message", text },
{ type: "assistant_message", text },
{ type: "assistant_message", text, messageId: identity },
{ type: "reasoning", text },
{
type: "tool_call",
callId: identity,
name: "shell",
status: "running",
error: null,
detail: { type: "plain_text", label: text },
},
{
type: "plugin",
id: identity,
pluginId: "test",
kind: "state",
version: 1,
data: { text },
},
];
await append(timeline, "a", items[random(items.length)], { turnId: String(random(3)) });
}
// Include rehydration of already-projected rows, with non-monotonic seqEnd.
await timeline.create(root("b"), { rows: await timeline.readRows(root("a")), epoch: "e" });
for (const agentId of ["a", "b"]) {
await assertSelectionMatchesReference(timeline, agentId);
await append(timeline, agentId, { type: "assistant_message", text: "later" });
await assertSelectionMatchesReference(timeline, agentId);
}
}
});
async function assertSelectionMatchesReference(
timeline: AgentTimeline,
agentId: string,
): Promise<void> {
const rows = await timeline.readRows(root(agentId));
for (const direction of ["tail", "before", "after"] as const) {
for (const limit of [0, 1, 7, 200]) {
for (const cursorSeq of [0, 1, 25, 50, 99, 100, 150]) {
const actual = await timeline.fetch(root(agentId), {
direction,
limit,
cursor: { epoch: "e", seq: cursorSeq },
});
const expected = selectReferenceProjectedTimelinePage({
rows,
bounds: actual.window,
direction,
limit,
cursorSeq,
});
expect({
...expected,
entries: undefined,
rows: expected.entries.map((entry) => Object.assign({ seq: entry.seqEnd }, entry)),
}).toEqual({
entries: undefined,
rows: actual.rows,
startSeq: actual.startSeq,
endSeq: actual.endSeq,
hasOlder: actual.hasOlder,
hasNewer: actual.hasNewer,
});
}
}
}
}
it("continues a scroll across appends and lifecycle collapse without losing source coverage", async () => {
const timeline = new AgentTimeline();
await timeline.create(root("a"), { epoch: "e" });
const tool: AgentTimelineItem = {
type: "tool_call",
callId: "t",
name: "shell",
status: "running",
error: null,
detail: { type: "plain_text", label: "work" },
};
await append(timeline, "a", tool);
for (let seq = 2; seq <= 5; seq++)
await append(timeline, "a", { type: "user_message", text: String(seq) });
const first = await timeline.fetch(root("a"), {
direction: "after",
cursor: { epoch: "e", seq: 0 },
limit: 2,
});
expect(first.endSeq).toBe(2);
await append(timeline, "a", { ...tool, status: "completed" });
await append(timeline, "a", { type: "assistant_message", text: "Done" });
expect(first.rows[0].item).toMatchObject({ status: "running" });
const second = await timeline.fetch(root("a"), {
direction: "after",
cursor: { epoch: "e", seq: first.endSeq! },
limit: 2,
});
expect(second.rows.map((row) => row.seqStart)).toEqual([3, 4]);
expect(second.endSeq).toBe(4);
const third = await timeline.fetch(root("a"), {
direction: "after",
cursor: { epoch: "e", seq: second.endSeq! },
limit: 2,
});
expect(third.rows.map((row) => row.seqStart)).toEqual([1, 5]);
expect(third.rows[0].item).toMatchObject({ status: "completed" });
expect(third.endSeq).toBe(6);
const last = await timeline.fetch(root("a"), {
direction: "after",
cursor: { epoch: "e", seq: third.endSeq! },
limit: 2,
});
expect(last.rows.map((row) => row.item)).toEqual([{ type: "assistant_message", text: "Done" }]);
expect(last.endSeq).toBe(7);
expect(last.hasNewer).toBe(false);
});
packages/server/src/server/agent/timeline/index.ts
+10
View File
@@ -0,0 +1,10 @@
export {
AgentTimeline,
type TimelineAppendResult,
type TimelineCommit,
type TimelineEntry,
type TimelineHead,
type TimelineIngestResult,
type TimelineScope,
type TimelineSeed,
} from "./internal/agent-timeline.js";
packages/server/src/server/agent/timeline/internal/agent-timeline.ts
+242
View File
@@ -0,0 +1,242 @@
import type { AgentTimelineItem } from "../../agent-sdk-types.js";
import { limitAgentTimelineItemContent } from "../../agent-timeline-content.js";
import type {
AgentTimelineFetchOptions,
AgentTimelineFetchResult,
AgentTimelineRow,
} from "../../agent-timeline-store-types.js";
import type { ProjectedTimelineRow } from "../../timeline-projection.js";
import type {
AgentTimelineBackend,
TimelineEntry,
TimelineHead,
TimelineScope,
TimelineSeed,
} from "./backend.js";
import { UnknownTimelineError } from "./backend.js";
import { MemoryTimelineBackend } from "./memory-backend.js";
export type { TimelineEntry, TimelineHead, TimelineScope, TimelineSeed } from "./backend.js";
export interface TimelineCommit {
row: AgentTimelineRow;
epoch: string;
}
export type TimelineIngestResult =
| { kind: "recorded"; row: AgentTimelineRow; epoch: string }
/**
* The row is committed but `publish` threw. The commit stands and nothing retries the
* publication; ingesting the same submitted prompt again returns `duplicate`.
*/
| { kind: "unpublished"; row: AgentTimelineRow; epoch: string; error: unknown }
/** A submitted prompt with this `clientMessageId` is already recorded; nothing was published. */
| { kind: "duplicate"; row: AgentTimelineRow; epoch: string };
export type TimelineAppendResult =
| { kind: "recorded"; commits: TimelineCommit[] }
/** The rows are committed but `publish` threw; see `TimelineIngestResult`. */
| { kind: "unpublished"; commits: TimelineCommit[]; error: unknown };
export interface TimelineIngestOptions {
/**
* Runs right after the commit, before any later operation on the agent's timeline starts, so
* publications leave in commit order. Not called when the commit fails or is a duplicate.
* A throw does not undo the commit: the write resolves `unpublished` with the error.
*/
publish?: (commit: TimelineCommit) => void;
}
export interface TimelineAppendOptions {
/** See `TimelineIngestOptions.publish`. */
publish?: (commits: TimelineCommit[]) => void;
}
/**
* Owner of agent timeline history: item identity, sequence numbers, projection, and reads.
*
* Operations on one agent (its root history and its provider children) run in call order. A read
* observes every write called before it, whether or not the caller awaited that write. A write
* that fails rejects, consumes no sequence number, and publishes nothing.
*/
export class AgentTimeline {
private readonly tails = new Map<string, Promise<void>>();
private readonly sizes = new Map<string, number>();
constructor(private readonly backend: AgentTimelineBackend = new MemoryTimelineBackend()) {}
describe(scope: TimelineScope): Promise<TimelineHead | null> {
return this.run(scope, () => this.backend.head(scope));
}
/** Creates the scope's history with a new identity, replacing any existing history. */
create(scope: TimelineScope, seed: TimelineSeed = {}): Promise<TimelineHead> {
return this.run(scope, async () => this.track(scope, await this.backend.create(scope, seed)));
}
/** Returns the existing history, or creates an empty one. */
open(scope: TimelineScope): Promise<TimelineHead> {
return this.run(scope, async () => {
const existing = await this.backend.head(scope);
return existing ?? this.track(scope, await this.backend.create(scope, {}));
});
}
delete(scope: TimelineScope): Promise<void> {
return this.run(scope, async () => {
await this.backend.delete(scope);
this.sizes.delete(sizeKey(scope));
});
}
ingest(
scope: TimelineScope,
entry: TimelineEntry,
options?: TimelineIngestOptions,
): Promise<TimelineIngestResult> {
return this.run(scope, async () => {
const clientMessageId = submittedPromptId(entry.item);
if (clientMessageId) {
const existing = await this.backend.findSubmittedPrompt(scope, clientMessageId);
if (existing) return { kind: "duplicate", row: existing, epoch: await this.epoch(scope) };
}
const [commit] = await this.commit(scope, [entry]);
const failure = publishCommitted(options?.publish, commit);
return failure
? { kind: "unpublished", ...commit, error: failure.error }
: { kind: "recorded", ...commit };
});
}
/** Appends provider history entries in one commit, without submitted-prompt deduplication. */
append(
scope: TimelineScope,
entries: readonly TimelineEntry[],
options?: TimelineAppendOptions,
): Promise<TimelineAppendResult> {
return this.run(scope, async () => {
const commits = entries.length === 0 ? [] : await this.commit(scope, entries);
const failure = publishCommitted(options?.publish, commits);
return failure
? { kind: "unpublished", commits, error: failure.error }
: { kind: "recorded", commits };
});
}
attachProviderMessageId(
scope: TimelineScope,
clientMessageId: string,
providerMessageId: string,
): Promise<AgentTimelineRow | null> {
return this.run(scope, () =>
this.backend.attachProviderMessageId(scope, clientMessageId, providerMessageId),
);
}
findSubmittedPrompt(
scope: TimelineScope,
clientMessageId: string,
): Promise<AgentTimelineRow | null> {
return this.run(scope, () => this.backend.findSubmittedPrompt(scope, clientMessageId));
}
fetch(
scope: TimelineScope,
options: AgentTimelineFetchOptions = {},
): Promise<AgentTimelineFetchResult> {
return this.run(scope, () => this.backend.fetch(scope, options));
}
readRows(scope: TimelineScope): Promise<ProjectedTimelineRow[]> {
return this.run(scope, () => this.backend.readRows(scope));
}
lastItem(scope: TimelineScope): Promise<AgentTimelineItem | null> {
return this.run(scope, () => this.backend.lastItem(scope));
}
/**
* The text of the trailing run of assistant messages, joined in order (providers stream one
* reply as several messages). Non-assistant items after the run are skipped; the first one
* before it ends the run. Null when the history has no assistant message.
*/
lastAssistantMessage(scope: TimelineScope): Promise<string | null> {
return this.run(scope, () => this.backend.lastAssistantMessage(scope));
}
/**
* Resolves once every operation called before it has finished, including its publication.
* Callers use it so an event published outside the timeline cannot overtake queued writes.
*/
settled(scope: TimelineScope): Promise<void> {
return this.run(scope, settledOperation);
}
/** Projected row count as of the last committed write, for diagnostics. */
residentSize(scope: TimelineScope): number | null {
return this.sizes.get(sizeKey(scope)) ?? null;
}
private async commit(
scope: TimelineScope,
entries: readonly TimelineEntry[],
): Promise<TimelineCommit[]> {
const bounded = entries.map((entry) => ({
...entry,
item: limitAgentTimelineItemContent(entry.item),
}));
const { rows, head } = await this.backend.append(scope, bounded);
this.track(scope, head);
return rows.map((row) => ({ row, epoch: head.epoch }));
}
private async epoch(scope: TimelineScope): Promise<string> {
const head = await this.backend.head(scope);
if (!head) throw new UnknownTimelineError(scope);
return head.epoch;
}
private track(scope: TimelineScope, head: TimelineHead): TimelineHead {
this.sizes.set(sizeKey(scope), head.size);
return head;
}
private run<T>(scope: TimelineScope, operation: () => T | Promise<T>): Promise<T> {
const key = scope.agentId;
const previous = this.tails.get(key) ?? Promise.resolve();
const result = previous.then(operation);
const tail = result.then(settledOperation, settledOperation);
this.tails.set(key, tail);
void tail.finally(() => {
if (this.tails.get(key) === tail) this.tails.delete(key);
});
return result;
}
}
function settledOperation(): void {}
interface PublishFailure {
error: unknown;
}
/** Returns what `publish` threw, or null. A committed write stays committed either way. */
function publishCommitted<T>(
publish: ((committed: T) => void) | undefined,
committed: T,
): PublishFailure | null {
try {
publish?.(committed);
return null;
} catch (error) {
return { error };
}
}
function sizeKey(scope: TimelineScope): string {
return scope.subagentId === undefined ? scope.agentId : `${scope.agentId}\0${scope.subagentId}`;
}
function submittedPromptId(item: AgentTimelineItem): string | null {
return item.type === "user_message" && item.clientMessageId ? item.clientMessageId : null;
}
packages/server/src/server/agent/timeline/internal/backend.ts
+88
View File
@@ -0,0 +1,88 @@
import type { AgentTimelineItem } from "../../agent-sdk-types.js";
import type {
AgentTimelineFetchOptions,
AgentTimelineFetchResult,
AgentTimelineRow,
} from "../../agent-timeline-store-types.js";
import type { ProjectedTimelineRow } from "../../timeline-projection.js";
/** Root history uses `agentId` alone; provider-child history adds the child's id. */
export interface TimelineScope {
agentId: string;
subagentId?: string;
}
export interface TimelineEntry {
item: AgentTimelineItem;
timestamp?: string;
turnId?: string;
providerMessageId?: string;
}
export interface TimelineSeed {
entries?: readonly TimelineEntry[];
/** Already-sequenced rows, e.g. an imported session. Takes precedence over `entries`. */
rows?: readonly AgentTimelineRow[];
nextSeq?: number;
epoch?: string;
timestamp?: string;
}
export interface TimelineHead {
epoch: string;
/** Projected (displayed) row count. */
size: number;
}
export interface TimelineBackendAppendResult {
rows: AgentTimelineRow[];
head: TimelineHead;
}
type MaybePromise<T> = T | Promise<T>;
/**
* Storage for the timeline owner. Each mutating call is one atomic commit: it either applies
* fully and returns, or throws having changed nothing. The owner serializes calls per agent, so
* a backend never sees concurrent calls for one agent. Callers outside the timeline module never
* see a backend.
*/
export interface AgentTimelineBackend {
create(scope: TimelineScope, seed: TimelineSeed): MaybePromise<TimelineHead>;
delete(scope: TimelineScope): MaybePromise<void>;
head(scope: TimelineScope): MaybePromise<TimelineHead | null>;
/** Allocates the next sequence numbers and folds the entries into the projection. */
append(
scope: TimelineScope,
entries: readonly TimelineEntry[],
): MaybePromise<TimelineBackendAppendResult>;
attachProviderMessageId(
scope: TimelineScope,
clientMessageId: string,
providerMessageId: string,
): MaybePromise<AgentTimelineRow | null>;
findSubmittedPrompt(
scope: TimelineScope,
clientMessageId: string,
): MaybePromise<AgentTimelineRow | null>;
fetch(
scope: TimelineScope,
options: AgentTimelineFetchOptions,
): MaybePromise<AgentTimelineFetchResult>;
readRows(scope: TimelineScope): MaybePromise<ProjectedTimelineRow[]>;
lastItem(scope: TimelineScope): MaybePromise<AgentTimelineItem | null>;
/** The trailing run of assistant messages joined, skipping later non-assistant rows. */
lastAssistantMessage(scope: TimelineScope): MaybePromise<string | null>;
}
/** The scope has no history: it was never created, or it was deleted. */
export class UnknownTimelineError extends Error {
constructor(readonly scope: TimelineScope) {
super(
scope.subagentId === undefined
? `Unknown agent '${scope.agentId}'`
: `Unknown timeline '${scope.agentId}/${scope.subagentId}'`,
);
this.name = "UnknownTimelineError";
}
}
packages/server/src/server/agent/timeline/internal/memory-backend.ts
+177
View File
@@ -0,0 +1,177 @@
import { randomUUID } from "node:crypto";
import { limitAgentTimelineItemContent } from "../../agent-timeline-content.js";
import type { AgentTimelineItem } from "../../agent-sdk-types.js";
import type {
AgentTimelineFetchOptions,
AgentTimelineFetchResult,
AgentTimelineRow,
} from "../../agent-timeline-store-types.js";
import { TimelineProjection, type ProjectedTimelineRow } from "../../timeline-projection.js";
import type {
AgentTimelineBackend,
TimelineBackendAppendResult,
TimelineEntry,
TimelineHead,
TimelineScope,
TimelineSeed,
} from "./backend.js";
import { UnknownTimelineError } from "./backend.js";
interface MemoryTimelineState {
epoch: string;
projection: TimelineProjection;
minSeq: number;
nextSeq: number;
}
const DEFAULT_TIMELINE_FETCH_LIMIT = 200;
function scopeKey(scope: TimelineScope): string {
return scope.subagentId === undefined ? scope.agentId : `${scope.agentId}\0${scope.subagentId}`;
}
function cloneRow<T extends AgentTimelineRow>(row: T): T {
return { ...row };
}
function buildRow(entry: TimelineEntry, seq: number, timestamp: string): AgentTimelineRow {
return {
seq,
timestamp: entry.timestamp ?? timestamp,
item: entry.item,
...(entry.turnId ? { turnId: entry.turnId } : {}),
...(entry.providerMessageId ? { providerMessageId: entry.providerMessageId } : {}),
};
}
/** Volatile backend: history lives in process memory and is lost when the daemon exits. */
export class MemoryTimelineBackend implements AgentTimelineBackend {
private readonly states = new Map<string, MemoryTimelineState>();
create(scope: TimelineScope, seed: TimelineSeed): TimelineHead {
const timestamp = seed.timestamp ?? new Date().toISOString();
const startSeq = seed.nextSeq ?? 1;
const rows = seed.rows?.length
? seed.rows.map(cloneRow)
: (seed.entries ?? []).map((entry, index) => buildRow(entry, startSeq + index, timestamp));
const nextSeq = rows.reduce((next, row) => Math.max(next, row.seq + 1), startSeq);
const projection = new TimelineProjection(rows);
const state: MemoryTimelineState = {
epoch: seed.epoch ?? randomUUID(),
projection,
minSeq: projection.getRows()[0]?.seqStart ?? 0,
nextSeq,
};
this.states.set(scopeKey(scope), state);
return this.describe(state);
}
delete(scope: TimelineScope): void {
this.states.delete(scopeKey(scope));
}
head(scope: TimelineScope): TimelineHead | null {
const state = this.states.get(scopeKey(scope));
return state ? this.describe(state) : null;
}
append(scope: TimelineScope, entries: readonly TimelineEntry[]): TimelineBackendAppendResult {
const state = this.requireState(scope);
const timestamp = new Date().toISOString();
const rows = entries.map((entry, index) => buildRow(entry, state.nextSeq + index, timestamp));
for (const row of rows) {
state.projection.append(row);
state.nextSeq = row.seq + 1;
if (state.minSeq === 0) state.minSeq = row.seq;
}
return { rows: rows.map(cloneRow), head: this.describe(state) };
}
attachProviderMessageId(
scope: TimelineScope,
clientMessageId: string,
providerMessageId: string,
): AgentTimelineRow | null {
return this.requireState(scope).projection.enrichSubmittedUserMessage(
clientMessageId,
providerMessageId,
);
}
findSubmittedPrompt(scope: TimelineScope, clientMessageId: string): AgentTimelineRow | null {
const row = this.requireState(scope)
.projection.getRows()
.find(
(candidate) =>
candidate.item.type === "user_message" &&
candidate.item.clientMessageId === clientMessageId,
);
return row ? cloneRow(row) : null;
}
fetch(scope: TimelineScope, options: AgentTimelineFetchOptions): AgentTimelineFetchResult {
const state = this.requireState(scope);
const direction = options.direction ?? "tail";
const cursor = options.cursor;
const hasRows = state.minSeq !== 0;
const window = { minSeq: state.minSeq, maxSeq: state.nextSeq - 1, nextSeq: state.nextSeq };
const staleCursor = cursor !== undefined && cursor.epoch !== state.epoch;
const gap =
!staleCursor &&
direction === "after" &&
cursor !== undefined &&
hasRows &&
cursor.seq < state.minSeq - 1;
const reset = staleCursor || gap;
const page = state.projection.selectPage({
bounds: window,
direction: reset ? "tail" : direction,
cursorSeq: cursor?.seq,
limit: options.limit ?? DEFAULT_TIMELINE_FETCH_LIMIT,
wholeTurns: options.wholeTurns,
});
return {
epoch: state.epoch,
direction,
reset,
staleCursor,
gap,
window,
hasOlder: page.hasOlder,
hasNewer: page.hasNewer,
startSeq: page.startSeq,
endSeq: page.endSeq,
// Stored rows are already within budget; this bounds assistant and reasoning merges.
rows: page.entries.map((entry) =>
Object.assign({ seq: entry.seqEnd }, entry, {
item: limitAgentTimelineItemContent(entry.item),
}),
),
};
}
readRows(scope: TimelineScope): ProjectedTimelineRow[] {
return this.requireState(scope).projection.getRows().map(cloneRow);
}
lastItem(scope: TimelineScope): AgentTimelineItem | null {
const state = this.requireState(scope);
return state.projection.getLastItem(state.nextSeq - 1);
}
lastAssistantMessage(scope: TimelineScope): string | null {
return this.requireState(scope).projection.getLastAssistantMessage();
}
private describe(state: MemoryTimelineState): TimelineHead {
return { epoch: state.epoch, size: state.projection.size };
}
private requireState(scope: TimelineScope): MemoryTimelineState {
const state = this.states.get(scopeKey(scope));
if (!state) {
throw new UnknownTimelineError(scope);
}
return state;
}
}
packages/server/src/server/agent/timeline/testing.ts
+90
View File
@@ -0,0 +1,90 @@
import { setImmediate as nextTurn } from "node:timers/promises";
import type { AgentTimelineItem } from "../agent-sdk-types.js";
import type {
AgentTimelineFetchOptions,
AgentTimelineFetchResult,
AgentTimelineRow,
} from "../agent-timeline-store-types.js";
import type { ProjectedTimelineRow } from "../timeline-projection.js";
import type {
AgentTimelineBackend,
TimelineBackendAppendResult,
TimelineEntry,
TimelineHead,
TimelineScope,
TimelineSeed,
} from "./internal/backend.js";
import { MemoryTimelineBackend } from "./internal/memory-backend.js";
/**
* Testing entry point of the timeline module. Production code imports `./index.js`.
*
* Defers every call by a turn of the event loop, the shape of an I/O-backed store, and records
* the committed appends and whole-history reads.
*/
export class DeferredTimelineBackend implements AgentTimelineBackend {
private readonly memory = new MemoryTimelineBackend();
readonly appends: TimelineEntry[][] = [];
readRowsCalls = 0;
failNextAppend = false;
private async defer<T>(call: () => T): Promise<T> {
await nextTurn();
return call();
}
create(scope: TimelineScope, seed: TimelineSeed): Promise<TimelineHead> {
return this.defer(() => this.memory.create(scope, seed));
}
delete(scope: TimelineScope): Promise<void> {
return this.defer(() => this.memory.delete(scope));
}
head(scope: TimelineScope): Promise<TimelineHead | null> {
return this.defer(() => this.memory.head(scope));
}
append(
scope: TimelineScope,
entries: readonly TimelineEntry[],
): Promise<TimelineBackendAppendResult> {
return this.defer(() => {
if (this.failNextAppend) {
this.failNextAppend = false;
throw new Error("disk full");
}
const result = this.memory.append(scope, entries);
this.appends.push([...entries]);
return result;
});
}
attachProviderMessageId(
scope: TimelineScope,
clientMessageId: string,
providerMessageId: string,
): Promise<AgentTimelineRow | null> {
return this.defer(() =>
this.memory.attachProviderMessageId(scope, clientMessageId, providerMessageId),
);
}
findSubmittedPrompt(
scope: TimelineScope,
clientMessageId: string,
): Promise<AgentTimelineRow | null> {
return this.defer(() => this.memory.findSubmittedPrompt(scope, clientMessageId));
}
fetch(
scope: TimelineScope,
options: AgentTimelineFetchOptions,
): Promise<AgentTimelineFetchResult> {
return this.defer(() => this.memory.fetch(scope, options));
}
readRows(scope: TimelineScope): Promise<ProjectedTimelineRow[]> {
this.readRowsCalls += 1;
return this.defer(() => this.memory.readRows(scope));
}
lastItem(scope: TimelineScope): Promise<AgentTimelineItem | null> {
return this.defer(() => this.memory.lastItem(scope));
}
lastAssistantMessage(scope: TimelineScope): Promise<string | null> {
return this.defer(() => this.memory.lastAssistantMessage(scope));
}
}
packages/server/src/server/agent/tools/paseo-tools.ts
+1 -1
View File
@@ -3173,7 +3173,7 @@ function createToolCatalog(
agentStorage,
logger: childLogger,
});
const timeline = agentManager.getTimeline(agentId);
const timeline = await agentManager.getTimeline(agentId);
const snapshot = agentManager.getAgent(agentId);
return {
packages/server/src/server/daemon-e2e/checkout-ship.e2e.test.ts
+7 -4
View File
@@ -183,13 +183,15 @@ describe("daemon checkout ship loop", () => {
expect(diffUncommitted.error).toBeNull();
expect(diffUncommitted.files.length).toBeGreaterThan(0);
const timelineBeforeCommit = ctx.daemon.daemon.agentManager.getTimeline(agent.id).length;
const timelineBeforeCommit = (await ctx.daemon.daemon.agentManager.getTimeline(agent.id))
.length;
const commitResult = await ctx.client.checkoutCommit(worktree.worktreePath, {
addAll: true,
});
expect(commitResult.error).toBeNull();
expect(commitResult.success).toBe(true);
const timelineAfterCommit = ctx.daemon.daemon.agentManager.getTimeline(agent.id).length;
const timelineAfterCommit = (await ctx.daemon.daemon.agentManager.getTimeline(agent.id))
.length;
expect(timelineAfterCommit).toBe(timelineBeforeCommit);
const diffAfterCommit = await ctx.client.getCheckoutDiff(worktree.worktreePath, {
@@ -203,13 +205,14 @@ describe("daemon checkout ship loop", () => {
});
expect(baseDiff.files.length).toBeGreaterThan(0);
const timelineBeforePr = ctx.daemon.daemon.agentManager.getTimeline(agent.id).length;
const timelineBeforePr = (await ctx.daemon.daemon.agentManager.getTimeline(agent.id))
.length;
const prCreate = await ctx.client.checkoutPrCreate(worktree.worktreePath, {
baseRef: "main",
});
expect(prCreate.error).toBeNull();
expect(prCreate.url).toContain(repoName);
const timelineAfterPr = ctx.daemon.daemon.agentManager.getTimeline(agent.id).length;
const timelineAfterPr = (await ctx.daemon.daemon.agentManager.getTimeline(agent.id)).length;
expect(timelineAfterPr).toBe(timelineBeforePr);
const prStatus = await ctx.client.checkoutPrStatus(worktree.worktreePath);
packages/server/src/server/daemon-e2e/timeline-window.e2e.test.ts
+18 -12
View File
@@ -333,10 +333,12 @@ describe("daemon E2E - timeline window", () => {
text: `row ${seq}`,
});
}
const epoch = ctx.daemon.daemon.agentManager.fetchTimeline(agent.id, {
direction: "tail",
limit: 1,
}).epoch;
const epoch = (
await ctx.daemon.daemon.agentManager.fetchTimeline(agent.id, {
direction: "tail",
limit: 1,
})
).epoch;
const timeline = await ctx.client.fetchAgentTimeline(agent.id, {
direction: "after",
@@ -374,10 +376,12 @@ describe("daemon E2E - timeline window", () => {
text: `message ${seq - 200}`,
});
}
const epoch = ctx.daemon.daemon.agentManager.fetchTimeline(agent.id, {
direction: "tail",
limit: 1,
}).epoch;
const epoch = (
await ctx.daemon.daemon.agentManager.fetchTimeline(agent.id, {
direction: "tail",
limit: 1,
})
).epoch;
const timeline = await ctx.client.fetchAgentTimeline(agent.id, {
direction: "after",
@@ -476,10 +480,12 @@ describe("daemon E2E - timeline window", () => {
text: `row ${seq}`,
});
}
const epoch = ctx.daemon.daemon.agentManager.fetchTimeline(agent.id, {
direction: "tail",
limit: 1,
}).epoch;
const epoch = (
await ctx.daemon.daemon.agentManager.fetchTimeline(agent.id, {
direction: "tail",
limit: 1,
})
).epoch;
const timeline = await ctx.client.fetchAgentTimeline(agent.id, {
direction: "before",
packages/server/src/server/hub/provider-policy.real.e2e.test.ts
+4 -1
View File
@@ -149,7 +149,10 @@ describe.skipIf(!RUN_REAL_HUB_POLICY)("Hub provider policy (real providers)", ()
provider,
scenario: "replying_worker",
actions: sink.calls,
agent: redactDiagnostic(hub.ownedAgentDiagnostic(created.agentId), hub.repoRoot()),
agent: redactDiagnostic(
await hub.ownedAgentDiagnostic(created.agentId),
hub.repoRoot(),
),
permissionsRequested: permissionRequests(events),
toolOutcomes: toolOutcomes(events),
})}`,
packages/server/src/server/hub/test-utils/relationship-harness.ts
+3 -3
View File
@@ -1036,15 +1036,15 @@ export class HubRelationshipHarness {
);
}
ownedAgentDiagnostic(agentId: string): {
async ownedAgentDiagnostic(agentId: string): Promise<{
lifecycle: string;
lastError?: string;
assistantText?: string;
timelineTypes: string[];
} {
}> {
const agent = this.daemon!.agentManager.getAgent(agentId);
if (!agent) throw new Error(`Owned agent ${agentId} does not exist`);
const timeline = this.daemon!.agentManager.getTimeline(agentId);
const timeline = await this.daemon!.agentManager.getTimeline(agentId);
const assistantText = timeline
.flatMap((item) => (item.type === "assistant_message" ? [item.text] : []))
.join("");
packages/server/src/server/session.ts
+15 -13
View File
@@ -1989,7 +1989,7 @@ export class Session {
this.unsubscribeAgentEvents = this.agentManager.subscribe(
(event) => {
if (event.type === "timeline_replacement") {
this.deliverTimelineReplacement(event.agentId, this.rewindInitiators.get(event.agentId));
this.deliverTimelineReplacement(event, this.rewindInitiators.get(event.agentId));
return;
}
@@ -4642,7 +4642,7 @@ export class Session {
await unarchiveAgentState(this.agentStorage, this.agentManager, snapshot.id);
await this.agentManager.hydrateTimelineFromProvider(snapshot.id);
await this.agentUpdates.forwardLiveAgent(snapshot);
const timelineSize = this.agentManager.getTimeline(snapshot.id).length;
const timelineSize = (await this.agentManager.getTimeline(snapshot.id)).length;
if (requestId) {
const agentPayload = await this.buildAgentPayload(snapshot);
this.emit({
@@ -4791,7 +4791,7 @@ export class Session {
}
await this.agentManager.hydrateTimelineFromProvider(agentId, { broadcast: true });
await this.agentUpdates.forwardLiveAgent(snapshot);
const timelineSize = this.agentManager.getTimeline(agentId).length;
const timelineSize = (await this.agentManager.getTimeline(agentId)).length;
if (requestId) {
this.emit({
type: "status",
@@ -4910,11 +4910,13 @@ export class Session {
}
}
private deliverTimelineReplacement(agentId: string, initiatingSource?: object): void {
private deliverTimelineReplacement(
timeline: Extract<AgentManagerEvent, { type: "timeline_replacement" }>,
initiatingSource?: object,
): void {
const { agentId, epoch } = timeline;
const agent = this.agentManager.getAgent(agentId);
if (!agent) return;
const timeline = this.agentManager.fetchTimeline(agentId, { limit: 0 });
const epoch = timeline.epoch;
if (this.clientSources.size === 0 || !this.onMessageToSource) {
if (!this.supports(CLIENT_CAPS.timelineReplacementInvalidation)) {
@@ -7843,7 +7845,7 @@ export class Session {
});
const agentPayload = await this.buildAgentPayload(snapshot);
const fetchedControlTimeline = this.agentManager.fetchTimeline(msg.agentId, {
const fetchedControlTimeline = await this.agentManager.fetchTimeline(msg.agentId, {
direction,
cursor,
limit: msg.limit,
@@ -7973,14 +7975,14 @@ export class Session {
logger: this.sessionLogger,
});
const rows = await this.agentManager.getTimelineRows(msg.agentId);
const { epoch } = this.agentManager.fetchTimeline(msg.agentId, {
const { epoch } = await this.agentManager.fetchTimeline(msg.agentId, {
direction: "tail",
limit: 1,
});
const result = await searchTimeline({ rows, query: msg.query, cursor: msg.cursor });
if (
this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", limit: 1 }).epoch !==
epoch
(await this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", limit: 1 }))
.epoch !== epoch
) {
throw new Error("History changed; search again");
}
@@ -8026,7 +8028,7 @@ export class Session {
logger: this.sessionLogger,
});
const rows = await this.agentManager.getTimelineRows(msg.agentId);
const timeline = this.agentManager.fetchTimeline(msg.agentId, {
const timeline = await this.agentManager.fetchTimeline(msg.agentId, {
direction: "tail",
limit: 1,
});
@@ -8114,7 +8116,7 @@ export class Session {
const supportsProjection = source
? this.supportsForSource(CLIENT_CAPS.projectedSubagentTimeline, source)
: this.supports(CLIENT_CAPS.projectedSubagentTimeline);
const timeline = this.agentManager.fetchProviderSubagentTimeline(
const timeline = await this.agentManager.fetchProviderSubagentTimeline(
msg.parentAgentId,
msg.subagentId,
{
@@ -8205,7 +8207,7 @@ export class Session {
logger: this.sessionLogger,
});
const agentPayload = await this.buildAgentPayload(snapshot);
const timeline = this.agentManager.fetchTimeline(msg.agentId, {
const timeline = await this.agentManager.fetchTimeline(msg.agentId, {
direction: "tail",
limit: 0,
});
packages/server/src/server/wire-compat.test.ts
+12 -8
View File
@@ -22,7 +22,7 @@ import { OWNER_PERMISSIONS } from "./authorization/index.js";
import { DirectorySyncService } from "./directory-sync/index.js";
import { createProviderSnapshotManagerStub } from "./test-utils/session-stubs.js";
import type { AgentTimelineRow } from "./agent/agent-manager.js";
import { InMemoryAgentTimelineStore } from "./agent/agent-timeline-store.js";
import { AgentTimeline } from "./agent/timeline/index.js";
import type { AgentTimelineFetchOptions } from "./agent/agent-timeline-store-types.js";
import { handleCreatePaseoWorktreeRequest } from "./worktree-session.js";
import { createPersistedProjectRecord } from "./workspace-registry.js";
@@ -60,14 +60,18 @@ interface SessionInternals {
}
class InMemoryAgentManager {
private readonly timeline = new InMemoryAgentTimelineStore();
private readonly timeline = new AgentTimeline();
constructor(rows: AgentTimelineRow[]) {
this.timeline.initialize("agent-1", {
epoch: "epoch-1",
rows,
nextSeq: (rows.at(-1)?.seq ?? 0) + 1,
});
// The timeline runs operations in call order, so every fetch observes this seed.
void this.timeline.create(
{ agentId: "agent-1" },
{
epoch: "epoch-1",
rows,
nextSeq: (rows.at(-1)?.seq ?? 0) + 1,
},
);
}
getAgent() {
@@ -115,7 +119,7 @@ class InMemoryAgentManager {
}
fetchTimeline(_agentId: string, options?: AgentTimelineFetchOptions) {
return this.timeline.fetch("agent-1", options);
return this.timeline.fetch({ agentId: "agent-1" }, options);
}
listAgents() {