fix(server): keep agent timelines in memory (#3647)
The filesystem timeline store rewrote each agent's full transcript on every buffered update, causing write amplification that grew with conversation length. Return production timelines to runtime memory and provider-history hydration while preserving the dormant storage interface for a future intentional design.
This commit is contained in:
Mohamed Boudra
File diff suppressed because it is too large
Load Diff
@@ -61,8 +61,6 @@ import type {
|
||||
AgentTimelineRow,
|
||||
AgentTimelineStore,
|
||||
} from "./agent-timeline-store-types.js";
|
||||
import { DurableTimelineBuffer, type DurableTimelineTimers } from "./durable-timeline-buffer.js";
|
||||
import { reconcileProviderHistory } from "./history-reconciliation.js";
|
||||
import {
|
||||
AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS,
|
||||
AgentStreamCoalescer,
|
||||
@@ -281,8 +279,6 @@ export interface AgentManagerOptions {
|
||||
paseoToolCatalogFactory?: PaseoToolCatalogFactory;
|
||||
appendSystemPrompt?: string;
|
||||
agentStreamCoalesceWindowMs?: number;
|
||||
durableTimelineCoalesceWindowMs?: number;
|
||||
durableTimelineTimers?: DurableTimelineTimers;
|
||||
rescueTimeouts?: AgentManagerRescueTimeouts;
|
||||
beforeSteerUnavailableFallback?: (input: {
|
||||
agentId: string;
|
||||
@@ -474,7 +470,7 @@ function attachManagedTurnIdentity(
|
||||
}
|
||||
case "timeline": {
|
||||
// Live provider items belong to the foreground turn that owns their dispatch.
|
||||
// History deliberately keeps absent IDs so old persisted rows retain legacy grouping.
|
||||
// Provider history deliberately keeps absent IDs because it has no daemon turn identity.
|
||||
const turnId = agent.activeForegroundTurnId ?? agent.activeTurnId ?? undefined;
|
||||
return turnId ? { event: { ...event, turnId }, turnId } : { event, turnId };
|
||||
}
|
||||
@@ -682,7 +678,6 @@ export class AgentManager {
|
||||
private readonly idFactory: () => string;
|
||||
private readonly registry?: AgentStorage;
|
||||
private readonly durableTimelineStore?: AgentTimelineStore;
|
||||
private readonly durableTimelineBuffer?: DurableTimelineBuffer;
|
||||
private readonly previousStatuses = new Map<string, AgentLifecycleStatus>();
|
||||
private readonly backgroundTasks = new Set<Promise<void>>();
|
||||
private readonly agentRegistrationTasks = new Set<Promise<void>>();
|
||||
@@ -713,7 +708,6 @@ export class AgentManager {
|
||||
this.configurePaseoTools(options);
|
||||
this.appendSystemPrompt = options.appendSystemPrompt ?? "";
|
||||
this.logger = options.logger.child({ module: "agent", component: "agent-manager" });
|
||||
this.durableTimelineBuffer = this.createDurableTimelineBuffer(options);
|
||||
this.rescueTimeouts = {
|
||||
reloadSessionCloseMs:
|
||||
options.rescueTimeouts?.reloadSessionCloseMs ?? RELOAD_SESSION_CLOSE_TIMEOUT_MS,
|
||||
@@ -735,20 +729,6 @@ export class AgentManager {
|
||||
});
|
||||
}
|
||||
|
||||
private createDurableTimelineBuffer(
|
||||
options: AgentManagerOptions,
|
||||
): DurableTimelineBuffer | undefined {
|
||||
if (!this.durableTimelineStore) return undefined;
|
||||
return new DurableTimelineBuffer(
|
||||
this.durableTimelineStore,
|
||||
options.durableTimelineCoalesceWindowMs ?? 25,
|
||||
options.durableTimelineTimers ?? { setTimeout, clearTimeout },
|
||||
(agentId, error) => {
|
||||
this.logger.error({ err: error, agentId }, "Failed to persist queued timeline rows");
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
private configurePaseoTools(options: AgentManagerOptions): void {
|
||||
this.paseoToolsEnabled = options.paseoToolsEnabled ?? true;
|
||||
this.paseoToolCatalogFactory = options.paseoToolCatalogFactory ?? null;
|
||||
@@ -1404,10 +1384,8 @@ export class AgentManager {
|
||||
}
|
||||
|
||||
if (rehydrateFromDisk) {
|
||||
// Wipe both durable and in-memory timeline so registerSession mints a
|
||||
// new epoch and hydrateTimelineFromProvider re-streams the freshly read
|
||||
// provider history into an empty timeline.
|
||||
await this.deleteCommittedTimeline(agentId);
|
||||
// Wipe the in-memory timeline so registerSession mints a new epoch and
|
||||
// hydrateTimelineFromProvider re-streams the freshly read provider history.
|
||||
this.timelineStore.delete(agentId);
|
||||
for (const event of this.providerSubagents.deleteParent(agentId)) {
|
||||
this.dispatch({ type: "provider_subagent", event });
|
||||
@@ -2558,15 +2536,6 @@ export class AgentManager {
|
||||
messageId: clientMessageId,
|
||||
turnId: expectedTurnId,
|
||||
});
|
||||
if (this.durableTimelineBuffer) {
|
||||
try {
|
||||
await this.durableTimelineBuffer.flush(agent.id);
|
||||
} catch (error) {
|
||||
throw new Error("Steering was accepted but its canonical message could not be persisted", {
|
||||
cause: error,
|
||||
});
|
||||
}
|
||||
}
|
||||
this.emitState(agent);
|
||||
}
|
||||
|
||||
@@ -2816,9 +2785,7 @@ export class AgentManager {
|
||||
}
|
||||
|
||||
/**
|
||||
* Hydrates the timeline from provider history if the agent's durable
|
||||
* timeline is empty (e.g., imported agents that have provider history
|
||||
* on disk but no persisted timeline rows). No-ops if already hydrated.
|
||||
* Hydrates the runtime timeline from provider history. No-ops if already hydrated.
|
||||
*/
|
||||
async hydrateTimelineFromProvider(
|
||||
agentId: string,
|
||||
@@ -2874,19 +2841,15 @@ export class AgentManager {
|
||||
}
|
||||
}
|
||||
|
||||
async deleteCommittedTimeline(agentId: string): Promise<void> {
|
||||
if (!this.durableTimelineStore) {
|
||||
return;
|
||||
}
|
||||
await this.durableTimelineBuffer?.discard(agentId);
|
||||
await this.durableTimelineStore.deleteAgent(agentId);
|
||||
}
|
||||
|
||||
async deleteAgentState(agentId: string): Promise<void> {
|
||||
this.discardRetainedAgentState(agentId);
|
||||
await this.deleteCommittedTimeline(agentId);
|
||||
}
|
||||
|
||||
async deleteCommittedTimeline(agentId: string): Promise<void> {
|
||||
await this.durableTimelineStore?.deleteAgent(agentId);
|
||||
}
|
||||
|
||||
async getLastAssistantMessage(agentId: string): Promise<string | null> {
|
||||
const agent = this.agents.get(agentId);
|
||||
if (!agent) {
|
||||
@@ -2936,33 +2899,23 @@ export class AgentManager {
|
||||
if (!this.durableTimelineStore) {
|
||||
return liveSegment?.text ?? null;
|
||||
}
|
||||
|
||||
if (!liveSegment) {
|
||||
return await this.durableTimelineStore.getLastAssistantMessage(agentId);
|
||||
}
|
||||
|
||||
if (!liveSegment.startsAtBeginning) {
|
||||
return liveSegment.text;
|
||||
}
|
||||
|
||||
const lastDurableItem = await this.durableTimelineStore.getLastItem(agentId);
|
||||
if (lastDurableItem?.type !== "assistant_message") {
|
||||
return liveSegment.text;
|
||||
}
|
||||
|
||||
const durableMessage = await this.durableTimelineStore.getLastAssistantMessage(agentId);
|
||||
return durableMessage ? `${durableMessage}${liveSegment.text}` : liveSegment.text;
|
||||
}
|
||||
|
||||
private async getLastItemFromStores(agentId: string): Promise<AgentTimelineItem | null> {
|
||||
const lastLiveItem = this.timelineStore.getLastItem(agentId);
|
||||
if (lastLiveItem) {
|
||||
return lastLiveItem;
|
||||
}
|
||||
if (!this.durableTimelineStore) {
|
||||
return null;
|
||||
}
|
||||
return await this.durableTimelineStore.getLastItem(agentId);
|
||||
return lastLiveItem ?? (await this.durableTimelineStore?.getLastItem(agentId)) ?? null;
|
||||
}
|
||||
|
||||
async waitForAgentEvent(
|
||||
@@ -3163,7 +3116,7 @@ export class AgentManager {
|
||||
);
|
||||
|
||||
const now = new Date();
|
||||
const { durableHistoryComplete } = await this.initializeAgentTimelineForRegister({
|
||||
const { durableTimelineHasRows } = await this.initializeAgentTimelineForRegister({
|
||||
agentId: resolvedAgentId,
|
||||
now,
|
||||
options,
|
||||
@@ -3174,12 +3127,10 @@ export class AgentManager {
|
||||
session,
|
||||
config,
|
||||
now,
|
||||
durableTimelineHasRows,
|
||||
options,
|
||||
});
|
||||
|
||||
if (managed.historyPrimed && !durableHistoryComplete) {
|
||||
await this.commitCompleteHistorySnapshot(resolvedAgentId);
|
||||
}
|
||||
this.assertAcceptingAgentRegistrations();
|
||||
this.agents.set(resolvedAgentId, managed);
|
||||
registered = true;
|
||||
@@ -3259,19 +3210,18 @@ export class AgentManager {
|
||||
updatedAt?: Date;
|
||||
}
|
||||
| undefined;
|
||||
}): Promise<{ durableHistoryComplete: boolean }> {
|
||||
}): Promise<{ durableTimelineHasRows: boolean }> {
|
||||
const { agentId, now, options } = params;
|
||||
const timelineAlreadyPrimed = this.timelineStore.has(agentId);
|
||||
const explicitTimelineSeed = buildExplicitTimelineSeedForRegister(now, options);
|
||||
const shouldSeedFromDurable =
|
||||
!explicitTimelineSeed &&
|
||||
!this.timelineStore.has(agentId) &&
|
||||
this.durableTimelineStore !== undefined;
|
||||
!explicitTimelineSeed && !this.timelineStore.has(agentId) && this.durableTimelineStore;
|
||||
const durableTimelineSeed = shouldSeedFromDurable
|
||||
? await this.loadCommittedTimelineSeed(agentId, now)
|
||||
: null;
|
||||
const durableHistoryComplete =
|
||||
timelineAlreadyPrimed || durableTimelineSeed?.historyComplete === true;
|
||||
const durableTimelineHasRows =
|
||||
timelineAlreadyPrimed ||
|
||||
(durableTimelineSeed != null && (durableTimelineSeed.nextSeq ?? 1) > 1);
|
||||
const timelineSeed = explicitTimelineSeed ?? durableTimelineSeed;
|
||||
if (timelineSeed || !this.timelineStore.has(agentId)) {
|
||||
this.timelineStore.initialize(agentId, timelineSeed ?? { timestamp: now.toISOString() });
|
||||
@@ -3279,7 +3229,7 @@ export class AgentManager {
|
||||
if (options?.timelineRows?.length) {
|
||||
this.enqueueDurableTimelineBulkInsert(agentId, options.timelineRows);
|
||||
}
|
||||
return { durableHistoryComplete };
|
||||
return { durableTimelineHasRows };
|
||||
}
|
||||
|
||||
private buildManagedAgentForRegister(params: {
|
||||
@@ -3287,6 +3237,7 @@ export class AgentManager {
|
||||
session: AgentSession;
|
||||
config: AgentSessionConfig;
|
||||
now: Date;
|
||||
durableTimelineHasRows: boolean;
|
||||
options:
|
||||
| {
|
||||
createdAt?: Date;
|
||||
@@ -3303,7 +3254,7 @@ export class AgentManager {
|
||||
}
|
||||
| undefined;
|
||||
}): ActiveManagedAgent {
|
||||
const { resolvedAgentId, session, config, now, options } = params;
|
||||
const { resolvedAgentId, session, config, now, durableTimelineHasRows, options } = params;
|
||||
return {
|
||||
id: resolvedAgentId,
|
||||
provider: config.provider,
|
||||
@@ -3333,7 +3284,7 @@ export class AgentManager {
|
||||
options?.persistence ?? session.describePersistence(),
|
||||
config.cwd,
|
||||
),
|
||||
historyPrimed: options?.historyPrimed ?? false,
|
||||
historyPrimed: options?.historyPrimed ?? durableTimelineHasRows,
|
||||
lastUserMessageAt: options?.lastUserMessageAt ?? null,
|
||||
lastUsage: options?.lastUsage,
|
||||
lastError: options?.lastError,
|
||||
@@ -3346,18 +3297,13 @@ export class AgentManager {
|
||||
private async loadCommittedTimelineSeed(
|
||||
agentId: string,
|
||||
now: Date,
|
||||
): Promise<SeedAgentTimelineOptions & { historyComplete?: boolean }> {
|
||||
): Promise<SeedAgentTimelineOptions> {
|
||||
if (!this.durableTimelineStore) {
|
||||
return { timestamp: now.toISOString() };
|
||||
}
|
||||
|
||||
const snapshot = await this.durableTimelineStore.getCommittedSnapshot(agentId);
|
||||
const rows = snapshot.rows;
|
||||
return {
|
||||
rows,
|
||||
nextSeq: rows.length > 0 ? rows[rows.length - 1]!.seq + 1 : 1,
|
||||
nextSeq: (await this.durableTimelineStore.getLatestCommittedSeq(agentId)) + 1,
|
||||
timestamp: now.toISOString(),
|
||||
historyComplete: snapshot.historyComplete,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -3673,17 +3619,11 @@ export class AgentManager {
|
||||
}
|
||||
}
|
||||
|
||||
const reconciledRows = reconcileProviderHistory(
|
||||
this.timelineStore.getRows(agent.id),
|
||||
historyEvents.map((event) => ({ item: event.item, timestamp: event.timestamp })),
|
||||
{ mode: "force" },
|
||||
);
|
||||
this.agentStreamCoalescer.flushAndDiscard(agent.id);
|
||||
await this.durableTimelineBuffer?.flush(agent.id);
|
||||
await this.deleteCommittedTimeline(agent.id);
|
||||
this.timelineStore.delete(agent.id);
|
||||
this.timelineStore.initialize(agent.id, { rows: reconciledRows });
|
||||
agent.historyPrimed = false;
|
||||
this.timelineStore.initialize(agent.id, { timestamp: new Date().toISOString() });
|
||||
agent.historyPrimed = true;
|
||||
|
||||
for (const event of this.providerSubagents.deleteParent(agent.id)) {
|
||||
if (broadcast) {
|
||||
@@ -3696,11 +3636,20 @@ export class AgentManager {
|
||||
this.dispatch({ type: "provider_subagent", event: update });
|
||||
}
|
||||
}
|
||||
if (broadcast) {
|
||||
this.dispatchReconciledTimelineRows(agent, reconciledRows);
|
||||
for (const event of historyEvents) {
|
||||
const row = this.recordTimeline(
|
||||
agent.id,
|
||||
event.item,
|
||||
event.timestamp ? { timestamp: event.timestamp } : undefined,
|
||||
);
|
||||
if (broadcast) {
|
||||
this.dispatchStream(agent.id, event, {
|
||||
seq: row.seq,
|
||||
epoch: this.timelineStore.getEpoch(agent.id),
|
||||
timestamp: row.timestamp,
|
||||
});
|
||||
}
|
||||
}
|
||||
await this.commitCompleteHistorySnapshot(agent.id);
|
||||
agent.historyPrimed = true;
|
||||
this.touchUpdatedAt(agent);
|
||||
this.emitState(agent);
|
||||
}
|
||||
@@ -3710,9 +3659,11 @@ export class AgentManager {
|
||||
broadcast: boolean | (() => boolean),
|
||||
): Promise<void> {
|
||||
const deferredBroadcast = typeof broadcast === "function";
|
||||
let timelineRows: AgentTimelineRow[] = [];
|
||||
const timelineEvents: Array<{
|
||||
event: Extract<AgentStreamEvent, { type: "timeline" }>;
|
||||
row: AgentTimelineRow;
|
||||
}> = [];
|
||||
const providerSubagentEvents: AgentManagerEvent[] = [];
|
||||
const historyEvents: Extract<AgentStreamEvent, { type: "timeline" }>[] = [];
|
||||
agent.historyPrimed = false;
|
||||
try {
|
||||
for await (const rawEvent of agent.session.streamHistory()) {
|
||||
@@ -3733,28 +3684,26 @@ export class AgentManager {
|
||||
if (event.item.type === "user_message" && isSystemInjectedEnvelope(event.item.text)) {
|
||||
continue;
|
||||
}
|
||||
historyEvents.push(event);
|
||||
const row = this.recordTimeline(
|
||||
agent.id,
|
||||
event.item,
|
||||
event.timestamp ? { timestamp: event.timestamp } : undefined,
|
||||
);
|
||||
if (deferredBroadcast) {
|
||||
timelineEvents.push({ event, row });
|
||||
} else if (broadcast) {
|
||||
this.dispatchStream(agent.id, event, {
|
||||
seq: row.seq,
|
||||
epoch: this.timelineStore.getEpoch(agent.id),
|
||||
timestamp: row.timestamp,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const reconciledRows = reconcileProviderHistory(
|
||||
this.timelineStore.getRows(agent.id),
|
||||
historyEvents.map((event) => ({ item: event.item, timestamp: event.timestamp })),
|
||||
);
|
||||
this.timelineStore.initialize(agent.id, { rows: reconciledRows });
|
||||
if (deferredBroadcast) {
|
||||
timelineRows = reconciledRows;
|
||||
} else if (broadcast) {
|
||||
this.dispatchReconciledTimelineRows(agent, reconciledRows);
|
||||
}
|
||||
await this.commitCompleteHistorySnapshot(agent.id);
|
||||
agent.historyPrimed = true;
|
||||
} catch (error) {
|
||||
this.logger.warn(
|
||||
{ err: error, agentId: agent.id },
|
||||
"Failed to commit provider history snapshot",
|
||||
);
|
||||
this.logger.warn({ err: error, agentId: agent.id }, "Failed to hydrate provider history");
|
||||
throw error;
|
||||
}
|
||||
agent.historyPrimed = true;
|
||||
|
||||
if (typeof broadcast !== "function" || !broadcast()) {
|
||||
return;
|
||||
@@ -3762,42 +3711,15 @@ export class AgentManager {
|
||||
for (const event of providerSubagentEvents) {
|
||||
this.dispatch(event);
|
||||
}
|
||||
this.dispatchReconciledTimelineRows(agent, timelineRows);
|
||||
}
|
||||
|
||||
private dispatchReconciledTimelineRows(
|
||||
agent: ActiveManagedAgent,
|
||||
rows: readonly AgentTimelineRow[],
|
||||
): void {
|
||||
const epoch = this.timelineStore.getEpoch(agent.id);
|
||||
for (const row of rows) {
|
||||
this.dispatchStream(
|
||||
agent.id,
|
||||
{
|
||||
type: "timeline",
|
||||
provider: agent.provider,
|
||||
item: row.item,
|
||||
...(row.turnId ? { turnId: row.turnId } : {}),
|
||||
timestamp: row.timestamp,
|
||||
},
|
||||
{
|
||||
seq: row.seq,
|
||||
epoch,
|
||||
timestamp: row.timestamp,
|
||||
},
|
||||
);
|
||||
for (const { event, row } of timelineEvents) {
|
||||
this.dispatchStream(agent.id, event, {
|
||||
seq: row.seq,
|
||||
epoch: this.timelineStore.getEpoch(agent.id),
|
||||
timestamp: row.timestamp,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async commitCompleteHistorySnapshot(agentId: string): Promise<void> {
|
||||
if (!this.durableTimelineStore) return;
|
||||
await this.durableTimelineBuffer?.flush(agentId);
|
||||
await this.durableTimelineStore.replaceCommittedSnapshot(agentId, {
|
||||
rows: this.timelineStore.getRows(agentId),
|
||||
historyComplete: true,
|
||||
});
|
||||
}
|
||||
|
||||
private notifyForegroundTurnWaiters(agentId: string, event: AgentStreamEvent): void {
|
||||
const turnId = getAgentStreamEventTurnId(event);
|
||||
if (turnId == null) {
|
||||
@@ -4447,14 +4369,11 @@ export class AgentManager {
|
||||
timestamp?: string;
|
||||
providerMessageId?: string;
|
||||
turnId?: string;
|
||||
persist?: boolean;
|
||||
},
|
||||
): AgentTimelineRow {
|
||||
item = limitAgentTimelineItemContent(item);
|
||||
const row = this.timelineStore.append(agentId, item, options);
|
||||
if (options?.persist !== false) {
|
||||
this.enqueueDurableTimelineAppend(agentId, row);
|
||||
}
|
||||
this.enqueueDurableTimelineAppend(agentId, row);
|
||||
return row;
|
||||
}
|
||||
|
||||
@@ -4541,10 +4460,16 @@ export class AgentManager {
|
||||
}
|
||||
|
||||
private enqueueDurableTimelineAppend(agentId: string, row: AgentTimelineRow): void {
|
||||
if (!this.durableTimelineBuffer) {
|
||||
if (!this.durableTimelineStore) {
|
||||
return;
|
||||
}
|
||||
this.durableTimelineBuffer.enqueue(agentId, row);
|
||||
const task = this.durableTimelineStore.bulkInsert(agentId, [row]).catch((err) => {
|
||||
this.logger.error(
|
||||
{ err, agentId, seq: row.seq, itemType: row.item.type },
|
||||
"Failed to append timeline row to durable store",
|
||||
);
|
||||
});
|
||||
this.trackBackgroundTask(task);
|
||||
}
|
||||
|
||||
private enqueueDurableTimelineBulkInsert(
|
||||
@@ -4554,15 +4479,18 @@ export class AgentManager {
|
||||
if (!this.durableTimelineStore || rows.length === 0) {
|
||||
return;
|
||||
}
|
||||
for (const row of rows) this.durableTimelineBuffer?.enqueue(agentId, row);
|
||||
const task = this.durableTimelineStore.bulkInsert(agentId, rows).catch((err) => {
|
||||
this.logger.error(
|
||||
{ err, agentId, rowCount: rows.length },
|
||||
"Failed to seed durable timeline store",
|
||||
);
|
||||
});
|
||||
this.trackBackgroundTask(task);
|
||||
}
|
||||
|
||||
private enqueueDurableTimelineUpdate(agentId: string, row: AgentTimelineRow): void {
|
||||
if (!this.durableTimelineStore) return;
|
||||
const task = (async () => {
|
||||
await this.durableTimelineBuffer?.flush(agentId);
|
||||
await this.durableTimelineStore?.updateCommittedRow(agentId, row);
|
||||
})().catch((err) => {
|
||||
const task = this.durableTimelineStore.updateCommittedRow(agentId, row).catch((err) => {
|
||||
this.logger.error(
|
||||
{ err, agentId, seq: row.seq, itemType: row.item.type },
|
||||
"Failed to enrich durable timeline row",
|
||||
@@ -4609,7 +4537,6 @@ export class AgentManager {
|
||||
|
||||
private async flushTasks(options: { includeAgentRegistrations: boolean }): Promise<void> {
|
||||
this.agentStreamCoalescer.flushAll();
|
||||
await this.durableTimelineBuffer?.flushAll();
|
||||
// Drain tasks, including tasks spawned while awaiting.
|
||||
while (
|
||||
this.backgroundTasks.size > 0 ||
|
||||
|
||||
@@ -32,12 +32,6 @@ export interface AgentTimelineWindow {
|
||||
nextSeq: number;
|
||||
}
|
||||
|
||||
/** A committed canonical transcript and whether provider history is fully represented. */
|
||||
export interface AgentTimelineSnapshot {
|
||||
rows: AgentTimelineRow[];
|
||||
historyComplete: boolean;
|
||||
}
|
||||
|
||||
export interface AgentTimelineFetchResult {
|
||||
epoch: string;
|
||||
direction: AgentTimelineFetchDirection;
|
||||
@@ -62,11 +56,9 @@ export interface AgentTimelineStore {
|
||||
): Promise<AgentTimelineFetchResult>;
|
||||
getLatestCommittedSeq(agentId: string): Promise<number>;
|
||||
getCommittedRows(agentId: string): Promise<AgentTimelineRow[]>;
|
||||
getCommittedSnapshot(agentId: string): Promise<AgentTimelineSnapshot>;
|
||||
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>;
|
||||
replaceCommittedSnapshot(agentId: string, snapshot: AgentTimelineSnapshot): Promise<void>;
|
||||
updateCommittedRow(agentId: string, row: AgentTimelineRow): Promise<void>;
|
||||
}
|
||||
|
||||
@@ -1,226 +0,0 @@
|
||||
import { describe, expect, test } from "vitest";
|
||||
|
||||
import { DurableTimelineBuffer } from "./durable-timeline-buffer.js";
|
||||
import type {
|
||||
AgentTimelineFetchResult,
|
||||
AgentTimelineRow,
|
||||
AgentTimelineSnapshot,
|
||||
AgentTimelineStore,
|
||||
} from "./agent-timeline-store-types.js";
|
||||
import type { AgentTimelineItem } from "./agent-sdk-types.js";
|
||||
|
||||
const row = (seq: number): AgentTimelineRow => ({
|
||||
seq,
|
||||
timestamp: "2026-01-01T00:00:00.000Z",
|
||||
item: { type: "assistant_message", text: String(seq) },
|
||||
});
|
||||
|
||||
class RecordingStore implements AgentTimelineStore {
|
||||
readonly batches: AgentTimelineRow[][] = [];
|
||||
failuresRemaining = 0;
|
||||
|
||||
async appendCommitted(): Promise<AgentTimelineRow> {
|
||||
throw new Error("not used");
|
||||
}
|
||||
async fetchCommitted(): Promise<AgentTimelineFetchResult> {
|
||||
throw new Error("not used");
|
||||
}
|
||||
async getLatestCommittedSeq(): Promise<number> {
|
||||
return 0;
|
||||
}
|
||||
async getCommittedRows(): Promise<AgentTimelineRow[]> {
|
||||
return [];
|
||||
}
|
||||
async getCommittedSnapshot(): Promise<AgentTimelineSnapshot> {
|
||||
return { rows: [], historyComplete: false };
|
||||
}
|
||||
async getLastItem(): Promise<AgentTimelineItem | null> {
|
||||
return null;
|
||||
}
|
||||
async getLastAssistantMessage(): Promise<string | null> {
|
||||
return null;
|
||||
}
|
||||
async deleteAgent(): Promise<void> {}
|
||||
async bulkInsert(_agentId: string, rows: readonly AgentTimelineRow[]): Promise<void> {
|
||||
this.batches.push([...rows]);
|
||||
if (this.failuresRemaining > 0) {
|
||||
this.failuresRemaining -= 1;
|
||||
throw new Error("disk full");
|
||||
}
|
||||
}
|
||||
async replaceCommittedSnapshot(): Promise<void> {}
|
||||
async updateCommittedRow(): Promise<void> {}
|
||||
}
|
||||
|
||||
class ControlledTimers {
|
||||
readonly delays: number[] = [];
|
||||
private readonly callbacks = new Map<number, () => void>();
|
||||
private nextId = 1;
|
||||
|
||||
setTimeout = (callback: () => void, delayMs: number): ReturnType<typeof setTimeout> => {
|
||||
const id = this.nextId++;
|
||||
this.delays.push(delayMs);
|
||||
this.callbacks.set(id, callback);
|
||||
return id as unknown as ReturnType<typeof setTimeout>;
|
||||
};
|
||||
|
||||
clearTimeout = (timer: ReturnType<typeof setTimeout>): void => {
|
||||
this.callbacks.delete(timer as unknown as number);
|
||||
};
|
||||
|
||||
async runNext(): Promise<void> {
|
||||
const next = this.callbacks.entries().next().value as [number, () => void] | undefined;
|
||||
if (!next) throw new Error("Expected a scheduled timer");
|
||||
this.callbacks.delete(next[0]);
|
||||
next[1]();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
}
|
||||
|
||||
get size(): number {
|
||||
return this.callbacks.size;
|
||||
}
|
||||
}
|
||||
|
||||
const timers = {
|
||||
setTimeout: () => 0 as unknown as ReturnType<typeof setTimeout>,
|
||||
clearTimeout: () => undefined,
|
||||
};
|
||||
|
||||
describe("DurableTimelineBuffer", () => {
|
||||
test("retries a failed prefix with later rows without a sequence hole", async () => {
|
||||
const store = new RecordingStore();
|
||||
store.failuresRemaining = 1;
|
||||
const buffer = new DurableTimelineBuffer(store, 25, timers, () => undefined);
|
||||
|
||||
buffer.enqueue("agent", row(1));
|
||||
await expect(buffer.flush("agent")).rejects.toThrow("disk full");
|
||||
buffer.enqueue("agent", row(2));
|
||||
await buffer.flush("agent");
|
||||
|
||||
const batchSequences = store.batches.map(batchSequencesFor);
|
||||
expect(batchSequences).toEqual([[1], [1, 2]]);
|
||||
});
|
||||
|
||||
test("coalesces a burst into one bounded atomic write and flush drains it", async () => {
|
||||
const store = new RecordingStore();
|
||||
const buffer = new DurableTimelineBuffer(store, 25, timers, () => undefined);
|
||||
|
||||
for (let seq = 1; seq <= 50; seq += 1) buffer.enqueue("agent", row(seq));
|
||||
await buffer.flushAll();
|
||||
|
||||
expect(store.batches).toHaveLength(1);
|
||||
expect(store.batches[0]?.map((entry) => entry.seq)).toEqual(
|
||||
Array.from({ length: 50 }, (_, index) => index + 1),
|
||||
);
|
||||
});
|
||||
|
||||
test("discard drops a pending batch so a later global flush cannot resurrect it", async () => {
|
||||
const store = new RecordingStore();
|
||||
const buffer = new DurableTimelineBuffer(store, 25, timers, () => undefined);
|
||||
|
||||
buffer.enqueue("agent", row(1));
|
||||
await buffer.discard("agent");
|
||||
await buffer.flushAll();
|
||||
buffer.enqueue("agent", row(1));
|
||||
await buffer.flushAll();
|
||||
|
||||
expect(store.batches).toEqual([[row(1)]]);
|
||||
});
|
||||
|
||||
test("retries a final failed batch without another enqueue", async () => {
|
||||
const store = new RecordingStore();
|
||||
store.failuresRemaining = 1;
|
||||
const controlledTimers = new ControlledTimers();
|
||||
const errors: unknown[] = [];
|
||||
const buffer = new DurableTimelineBuffer(
|
||||
store,
|
||||
25,
|
||||
controlledTimers,
|
||||
(_agentId, error) => errors.push(error),
|
||||
{ baseDelayMs: 10, maxDelayMs: 40 },
|
||||
);
|
||||
|
||||
buffer.enqueue("agent", row(1));
|
||||
await controlledTimers.runNext();
|
||||
expect(store.batches.map(batchSequencesFor)).toEqual([[1]]);
|
||||
expect(errors).toHaveLength(1);
|
||||
expect(controlledTimers.delays).toEqual([25, 10]);
|
||||
|
||||
await controlledTimers.runNext();
|
||||
expect(store.batches.map(batchSequencesFor)).toEqual([[1], [1]]);
|
||||
expect(controlledTimers.size).toBe(0);
|
||||
});
|
||||
|
||||
test("grows retry delays to a deterministic cap", async () => {
|
||||
const store = new RecordingStore();
|
||||
store.failuresRemaining = 4;
|
||||
const controlledTimers = new ControlledTimers();
|
||||
const buffer = new DurableTimelineBuffer(store, 25, controlledTimers, () => undefined, {
|
||||
baseDelayMs: 10,
|
||||
maxDelayMs: 40,
|
||||
});
|
||||
|
||||
buffer.enqueue("agent", row(1));
|
||||
await controlledTimers.runNext();
|
||||
await controlledTimers.runNext();
|
||||
await controlledTimers.runNext();
|
||||
await controlledTimers.runNext();
|
||||
await controlledTimers.runNext();
|
||||
|
||||
expect(controlledTimers.delays).toEqual([25, 10, 20, 40, 40]);
|
||||
expect(store.batches).toHaveLength(5);
|
||||
});
|
||||
|
||||
test("joins a later enqueue behind the retained failed prefix", async () => {
|
||||
const store = new RecordingStore();
|
||||
store.failuresRemaining = 1;
|
||||
const controlledTimers = new ControlledTimers();
|
||||
const buffer = new DurableTimelineBuffer(store, 25, controlledTimers, () => undefined, {
|
||||
baseDelayMs: 10,
|
||||
});
|
||||
|
||||
buffer.enqueue("agent", row(1));
|
||||
await controlledTimers.runNext();
|
||||
buffer.enqueue("agent", row(2));
|
||||
await controlledTimers.runNext();
|
||||
|
||||
expect(store.batches.map(batchSequencesFor)).toEqual([[1], [1, 2]]);
|
||||
});
|
||||
|
||||
test("discard cancels a retry after failure", async () => {
|
||||
const store = new RecordingStore();
|
||||
store.failuresRemaining = 1;
|
||||
const controlledTimers = new ControlledTimers();
|
||||
const buffer = new DurableTimelineBuffer(store, 25, controlledTimers, () => undefined, {
|
||||
baseDelayMs: 10,
|
||||
});
|
||||
|
||||
buffer.enqueue("agent", row(1));
|
||||
await controlledTimers.runNext();
|
||||
await buffer.discard("agent");
|
||||
|
||||
expect(controlledTimers.size).toBe(0);
|
||||
expect(store.batches.map(batchSequencesFor)).toEqual([[1]]);
|
||||
});
|
||||
|
||||
test("manual flush rejects while retaining an autonomous retry", async () => {
|
||||
const store = new RecordingStore();
|
||||
store.failuresRemaining = 1;
|
||||
const controlledTimers = new ControlledTimers();
|
||||
const buffer = new DurableTimelineBuffer(store, 25, controlledTimers, () => undefined, {
|
||||
baseDelayMs: 10,
|
||||
});
|
||||
|
||||
buffer.enqueue("agent", row(1));
|
||||
await expect(buffer.flush("agent")).rejects.toThrow("disk full");
|
||||
expect(controlledTimers.delays).toEqual([25, 10]);
|
||||
|
||||
await controlledTimers.runNext();
|
||||
expect(store.batches.map(batchSequencesFor)).toEqual([[1], [1]]);
|
||||
});
|
||||
});
|
||||
|
||||
function batchSequencesFor(batch: AgentTimelineRow[]): number[] {
|
||||
return batch.map((entry) => entry.seq);
|
||||
}
|
||||
@@ -1,118 +0,0 @@
|
||||
import type { AgentTimelineRow, AgentTimelineStore } from "./agent-timeline-store-types.js";
|
||||
|
||||
export interface DurableTimelineTimers {
|
||||
setTimeout(callback: () => void, delayMs: number): ReturnType<typeof setTimeout>;
|
||||
clearTimeout(timer: ReturnType<typeof setTimeout>): void;
|
||||
}
|
||||
|
||||
export interface DurableTimelineRetryOptions {
|
||||
baseDelayMs?: number;
|
||||
maxDelayMs?: number;
|
||||
}
|
||||
|
||||
const DEFAULT_RETRY_MAX_DELAY_MS = 5_000;
|
||||
|
||||
interface PendingAgentRows {
|
||||
rows: AgentTimelineRow[];
|
||||
timer: ReturnType<typeof setTimeout> | null;
|
||||
flushing: Promise<void> | null;
|
||||
retryAttempt: number;
|
||||
}
|
||||
|
||||
/** Keeps each agent's durable transcript contiguous while coalescing live writes. */
|
||||
export class DurableTimelineBuffer {
|
||||
private readonly pending = new Map<string, PendingAgentRows>();
|
||||
|
||||
constructor(
|
||||
private readonly store: AgentTimelineStore,
|
||||
private readonly delayMs: number,
|
||||
private readonly timers: DurableTimelineTimers,
|
||||
private readonly onError: (agentId: string, error: unknown) => void,
|
||||
private readonly retryOptions: DurableTimelineRetryOptions = {},
|
||||
) {}
|
||||
|
||||
enqueue(agentId: string, row: AgentTimelineRow): void {
|
||||
const state = this.stateFor(agentId);
|
||||
state.rows.push(row);
|
||||
this.schedule(agentId, state, this.delayMs);
|
||||
}
|
||||
|
||||
async flush(agentId: string): Promise<void> {
|
||||
const state = this.pending.get(agentId);
|
||||
if (!state) return;
|
||||
if (state.timer !== null) {
|
||||
this.timers.clearTimeout(state.timer);
|
||||
state.timer = null;
|
||||
}
|
||||
if (state.flushing) {
|
||||
await state.flushing;
|
||||
return this.flush(agentId);
|
||||
}
|
||||
if (state.rows.length === 0) {
|
||||
this.pending.delete(agentId);
|
||||
return;
|
||||
}
|
||||
const batch = [...state.rows];
|
||||
state.flushing = this.store.bulkInsert(agentId, batch);
|
||||
try {
|
||||
await state.flushing;
|
||||
state.rows.splice(0, batch.length);
|
||||
state.retryAttempt = 0;
|
||||
} catch (error) {
|
||||
this.scheduleRetry(agentId, state);
|
||||
throw error;
|
||||
} finally {
|
||||
state.flushing = null;
|
||||
if (state.rows.length === 0 && state.timer === null) {
|
||||
this.pending.delete(agentId);
|
||||
} else if (state.rows.length > 0 && state.timer === null) {
|
||||
this.schedule(agentId, state, this.delayMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async flushAll(): Promise<void> {
|
||||
await Promise.all([...this.pending.keys()].map((agentId) => this.flush(agentId)));
|
||||
}
|
||||
|
||||
async discard(agentId: string): Promise<void> {
|
||||
const state = this.pending.get(agentId);
|
||||
if (!state) return;
|
||||
if (state.timer !== null) this.timers.clearTimeout(state.timer);
|
||||
state.timer = null;
|
||||
state.rows.length = 0;
|
||||
await state.flushing?.catch(() => undefined);
|
||||
this.pending.delete(agentId);
|
||||
}
|
||||
|
||||
private stateFor(agentId: string): PendingAgentRows {
|
||||
let state = this.pending.get(agentId);
|
||||
if (!state) {
|
||||
state = { rows: [], timer: null, flushing: null, retryAttempt: 0 };
|
||||
this.pending.set(agentId, state);
|
||||
}
|
||||
return state;
|
||||
}
|
||||
|
||||
private schedule(agentId: string, state: PendingAgentRows, delayMs: number): void {
|
||||
if (state.timer !== null || state.rows.length === 0) return;
|
||||
state.timer = this.timers.setTimeout(() => {
|
||||
state.timer = null;
|
||||
void this.flush(agentId).catch((error) => this.onError(agentId, error));
|
||||
}, delayMs);
|
||||
this.unref(state.timer);
|
||||
}
|
||||
|
||||
private scheduleRetry(agentId: string, state: PendingAgentRows): void {
|
||||
const baseDelayMs = this.retryOptions.baseDelayMs ?? this.delayMs;
|
||||
const maxDelayMs = this.retryOptions.maxDelayMs ?? DEFAULT_RETRY_MAX_DELAY_MS;
|
||||
const delayMs = Math.min(baseDelayMs * 2 ** state.retryAttempt, maxDelayMs);
|
||||
state.retryAttempt += 1;
|
||||
this.schedule(agentId, state, delayMs);
|
||||
}
|
||||
|
||||
private unref(timer: ReturnType<typeof setTimeout>): void {
|
||||
const candidate = timer as unknown as { unref?: () => void };
|
||||
candidate.unref?.();
|
||||
}
|
||||
}
|
||||
@@ -1,226 +0,0 @@
|
||||
import { mkdtemp, readdir, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import { writeJsonFileAtomic } from "../atomic-file.js";
|
||||
import { createTestLogger } from "../../test-utils/test-logger.js";
|
||||
import { AgentStorage } from "./agent-storage.js";
|
||||
import { FileAgentTimelineStore } from "./file-agent-timeline-store.js";
|
||||
import type { AgentTimelineRow } from "./agent-timeline-store-types.js";
|
||||
|
||||
const user = (text: string) => ({ type: "user_message" as const, text });
|
||||
|
||||
async function storeDirectory(): Promise<string> {
|
||||
return await mkdtemp(path.join(tmpdir(), "paseo-timeline-"));
|
||||
}
|
||||
|
||||
function filePath(directory: string, agentId: string): string {
|
||||
return path.join(directory, `agent-${Buffer.from(agentId, "utf8").toString("base64url")}.json`);
|
||||
}
|
||||
|
||||
function document(rows: AgentTimelineRow[], nextSeq: number) {
|
||||
return { version: 1, epoch: "epoch-1", nextSeq, rows };
|
||||
}
|
||||
|
||||
describe("FileAgentTimelineStore", () => {
|
||||
it("loads a missing agent file as an empty timeline", async () => {
|
||||
const store = new FileAgentTimelineStore(await storeDirectory());
|
||||
expect(await store.getCommittedRows("agent")).toEqual([]);
|
||||
});
|
||||
|
||||
it("loads legacy rows without turnId", async () => {
|
||||
const directory = await storeDirectory();
|
||||
await writeJsonFileAtomic(
|
||||
filePath(directory, "agent"),
|
||||
document([{ seq: 1, timestamp: "2026-01-01T00:00:00.000Z", item: user("old") }], 2),
|
||||
);
|
||||
|
||||
expect(
|
||||
(await new FileAgentTimelineStore(directory).getCommittedRows("agent"))[0]?.turnId,
|
||||
).toBeUndefined();
|
||||
});
|
||||
|
||||
it("treats existing rows as incomplete until an atomic history snapshot commits", async () => {
|
||||
const directory = await storeDirectory();
|
||||
const store = new FileAgentTimelineStore(directory);
|
||||
const row = { seq: 1, timestamp: "2026-01-01T00:00:00.000Z", item: user("old") };
|
||||
|
||||
await store.bulkInsert("agent", [row]);
|
||||
expect((await store.getCommittedSnapshot("agent")).historyComplete).toBe(false);
|
||||
|
||||
await store.replaceCommittedSnapshot("agent", { rows: [row], historyComplete: true });
|
||||
expect(await new FileAgentTimelineStore(directory).getCommittedSnapshot("agent")).toEqual({
|
||||
rows: [row],
|
||||
historyComplete: true,
|
||||
});
|
||||
});
|
||||
|
||||
it("round-trips supplied canonical rows without changing sequence or identity", async () => {
|
||||
const directory = await storeDirectory();
|
||||
const row: AgentTimelineRow = {
|
||||
seq: 7,
|
||||
timestamp: "2026-01-01T00:00:00.000Z",
|
||||
item: { type: "user_message", text: "hello", clientMessageId: "hello-client" },
|
||||
turnId: "turn-1",
|
||||
providerMessageId: "provider-hello",
|
||||
};
|
||||
|
||||
await new FileAgentTimelineStore(directory).bulkInsert("agent", [row]);
|
||||
const reopened = new FileAgentTimelineStore(directory);
|
||||
expect(await reopened.getCommittedRows("agent")).toEqual([row]);
|
||||
expect((await reopened.fetchCommitted("agent", { limit: 0 })).window.nextSeq).toBe(8);
|
||||
});
|
||||
|
||||
it("treats duplicate supplied rows as idempotent and rejects conflicting sequence reuse", async () => {
|
||||
const directory = await storeDirectory();
|
||||
let writeCount = 0;
|
||||
const store = new FileAgentTimelineStore(directory, {
|
||||
writeJson: async (file, value) => {
|
||||
writeCount += 1;
|
||||
await writeJsonFileAtomic(file, value);
|
||||
},
|
||||
});
|
||||
const row: AgentTimelineRow = {
|
||||
seq: 3,
|
||||
timestamp: "2026-01-01T00:00:00.000Z",
|
||||
item: user("hello"),
|
||||
providerMessageId: "provider-hello",
|
||||
turnId: "turn-1",
|
||||
};
|
||||
|
||||
await store.bulkInsert("agent", [row]);
|
||||
await store.bulkInsert("agent", [row]);
|
||||
expect(writeCount).toBe(1);
|
||||
await expect(store.bulkInsert("agent", [{ ...row, item: user("different") }])).rejects.toThrow(
|
||||
"Conflicting timeline row sequence 3",
|
||||
);
|
||||
expect(await store.getCommittedRows("agent")).toEqual([row]);
|
||||
});
|
||||
|
||||
it("serializes concurrent mutations for one agent without losing rows", async () => {
|
||||
const store = new FileAgentTimelineStore(await storeDirectory());
|
||||
await Promise.all([
|
||||
store.appendCommitted("agent", user("one")),
|
||||
store.appendCommitted("agent", user("two")),
|
||||
]);
|
||||
expect((await store.getCommittedRows("agent")).map((row) => row.seq)).toEqual([1, 2]);
|
||||
});
|
||||
|
||||
it("updates an existing row exactly and rejects a missing row sequence", async () => {
|
||||
const store = new FileAgentTimelineStore(await storeDirectory());
|
||||
await store.bulkInsert("agent", [
|
||||
{ seq: 1, timestamp: "2026-01-01T00:00:00.000Z", item: user("before") },
|
||||
]);
|
||||
const replacement: AgentTimelineRow = {
|
||||
seq: 1,
|
||||
timestamp: "2026-01-01T00:00:01.000Z",
|
||||
item: user("after"),
|
||||
turnId: "turn-1",
|
||||
providerMessageId: "provider-1",
|
||||
};
|
||||
|
||||
await store.updateCommittedRow("agent", replacement);
|
||||
expect(await store.getCommittedRows("agent")).toEqual([replacement]);
|
||||
await expect(store.updateCommittedRow("agent", { ...replacement, seq: 2 })).rejects.toThrow(
|
||||
"Cannot update missing timeline row sequence 2",
|
||||
);
|
||||
});
|
||||
|
||||
it("recovers an agent mutation lane after a rejected write without committing the failed row", async () => {
|
||||
const directory = await storeDirectory();
|
||||
let writeCount = 0;
|
||||
const store = new FileAgentTimelineStore(directory, {
|
||||
writeJson: async (file, value) => {
|
||||
writeCount += 1;
|
||||
if (writeCount === 1) throw new Error("disk full");
|
||||
await writeJsonFileAtomic(file, value);
|
||||
},
|
||||
});
|
||||
|
||||
await expect(store.appendCommitted("agent", user("lost"))).rejects.toThrow("disk full");
|
||||
expect(await store.getCommittedRows("agent")).toEqual([]);
|
||||
await expect(store.appendCommitted("agent", user("saved"))).resolves.toMatchObject({ seq: 1 });
|
||||
expect(await store.getCommittedRows("agent")).toEqual([
|
||||
expect.objectContaining({ seq: 1, item: user("saved") }),
|
||||
]);
|
||||
});
|
||||
|
||||
it("isolates corrupt files and failed mutation lanes to their own agent", async () => {
|
||||
const directory = await storeDirectory();
|
||||
await writeFile(filePath(directory, "broken"), "not json");
|
||||
const store = new FileAgentTimelineStore(directory, {
|
||||
writeJson: async (file, value) => {
|
||||
if (file === filePath(directory, "failing")) throw new Error("failing agent write");
|
||||
await writeJsonFileAtomic(file, value);
|
||||
},
|
||||
});
|
||||
|
||||
await expect(store.getCommittedRows("broken")).rejects.toThrow();
|
||||
await expect(store.appendCommitted("valid", user("works"))).resolves.toMatchObject({ seq: 1 });
|
||||
await expect(store.appendCommitted("failing", user("fails"))).rejects.toThrow(
|
||||
"failing agent write",
|
||||
);
|
||||
await expect(store.appendCommitted("valid", user("still works"))).resolves.toMatchObject({
|
||||
seq: 2,
|
||||
});
|
||||
});
|
||||
|
||||
it("does not serialize agent B behind a held agent A write", async () => {
|
||||
const directory = await storeDirectory();
|
||||
let releaseA!: () => void;
|
||||
const aWrite = new Promise<void>((resolve) => {
|
||||
releaseA = resolve;
|
||||
});
|
||||
const store = new FileAgentTimelineStore(directory, {
|
||||
writeJson: async (file, value) => {
|
||||
if (file === filePath(directory, "agent-a")) await aWrite;
|
||||
await writeJsonFileAtomic(file, value);
|
||||
},
|
||||
});
|
||||
|
||||
const appendA = store.appendCommitted("agent-a", user("held"));
|
||||
await expect(store.appendCommitted("agent-b", user("independent"))).resolves.toMatchObject({
|
||||
seq: 1,
|
||||
});
|
||||
releaseA();
|
||||
await expect(appendA).resolves.toMatchObject({ seq: 1 });
|
||||
});
|
||||
|
||||
it.each([
|
||||
{
|
||||
name: "invalid timeline item",
|
||||
value: document([{ seq: 1, timestamp: "now", item: { type: "not-a-timeline-item" } }], 2),
|
||||
},
|
||||
{
|
||||
name: "out-of-order rows",
|
||||
value: document(
|
||||
[
|
||||
{ seq: 2, timestamp: "now", item: user("two") },
|
||||
{ seq: 1, timestamp: "now", item: user("one") },
|
||||
],
|
||||
3,
|
||||
),
|
||||
},
|
||||
{
|
||||
name: "next sequence at the committed maximum",
|
||||
value: document([{ seq: 1, timestamp: "now", item: user("one") }], 1),
|
||||
},
|
||||
])("rejects persisted $name", async ({ value }) => {
|
||||
const directory = await storeDirectory();
|
||||
await writeJsonFileAtomic(filePath(directory, "agent"), value);
|
||||
await expect(new FileAgentTimelineStore(directory).getCommittedRows("agent")).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("writes outside AgentStorage discovery", async () => {
|
||||
const root = await storeDirectory();
|
||||
const agentStorageDirectory = path.join(root, "agents");
|
||||
const timelineDirectory = path.join(root, "timelines");
|
||||
const store = new FileAgentTimelineStore(timelineDirectory);
|
||||
|
||||
await store.appendCommitted("agent", user("hello"));
|
||||
|
||||
expect(await readdir(timelineDirectory)).toEqual(["agent-YWdlbnQ.json"]);
|
||||
expect(await new AgentStorage(agentStorageDirectory, createTestLogger()).list()).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -1,274 +0,0 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { isDeepStrictEqual } from "node:util";
|
||||
import { z } from "zod";
|
||||
|
||||
import { writeJsonFileAtomic } from "../atomic-file.js";
|
||||
import type { AgentTimelineItem } from "./agent-sdk-types.js";
|
||||
import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js";
|
||||
import { AgentTimelineItemPayloadSchema } from "@getpaseo/protocol/messages";
|
||||
import type {
|
||||
AgentTimelineFetchOptions,
|
||||
AgentTimelineFetchResult,
|
||||
AgentTimelineRow,
|
||||
AgentTimelineSnapshot,
|
||||
AgentTimelineStore,
|
||||
} from "./agent-timeline-store-types.js";
|
||||
|
||||
const TimelineRowSchema: z.ZodType<AgentTimelineRow, unknown> = z.object({
|
||||
seq: z.number().int().positive(),
|
||||
timestamp: z.string(),
|
||||
item: AgentTimelineItemPayloadSchema,
|
||||
turnId: z.string().optional(),
|
||||
providerMessageId: z.string().optional(),
|
||||
});
|
||||
|
||||
const TimelineDocumentSchema = z.object({
|
||||
version: z.literal(1),
|
||||
epoch: z.string(),
|
||||
nextSeq: z.number().int().positive(),
|
||||
rows: z.array(TimelineRowSchema),
|
||||
// Documents written before history completeness existed are intentionally incomplete.
|
||||
historyComplete: z.boolean().optional().default(false),
|
||||
});
|
||||
|
||||
type TimelineDocument = z.infer<typeof TimelineDocumentSchema>;
|
||||
|
||||
export interface FileAgentTimelineStoreOptions {
|
||||
writeJson?: (filePath: string, value: unknown) => Promise<void>;
|
||||
}
|
||||
|
||||
function cloneRow(row: AgentTimelineRow): AgentTimelineRow {
|
||||
return { ...row, item: structuredClone(row.item) };
|
||||
}
|
||||
|
||||
function cloneDocument(document: TimelineDocument): TimelineDocument {
|
||||
return { ...document, rows: document.rows.map(cloneRow) };
|
||||
}
|
||||
|
||||
function validateDocument(value: unknown): TimelineDocument {
|
||||
const document = TimelineDocumentSchema.parse(value);
|
||||
let previousSeq = 0;
|
||||
for (const row of document.rows) {
|
||||
if (row.seq <= previousSeq) {
|
||||
throw new Error("Timeline rows must have strictly increasing sequence numbers");
|
||||
}
|
||||
previousSeq = row.seq;
|
||||
}
|
||||
if (document.nextSeq <= previousSeq) {
|
||||
throw new Error("Timeline nextSeq must be greater than every row sequence number");
|
||||
}
|
||||
return document;
|
||||
}
|
||||
|
||||
function emptyDocument(): TimelineDocument {
|
||||
return { version: 1, epoch: randomUUID(), nextSeq: 1, rows: [], historyComplete: false };
|
||||
}
|
||||
|
||||
function fileNameForAgent(agentId: string): string {
|
||||
return `agent-${Buffer.from(agentId, "utf8").toString("base64url")}.json`;
|
||||
}
|
||||
|
||||
function selectRows(
|
||||
document: TimelineDocument,
|
||||
options?: AgentTimelineFetchOptions,
|
||||
): AgentTimelineFetchResult {
|
||||
const memory = new InMemoryAgentTimelineStore();
|
||||
memory.initialize("timeline", document);
|
||||
return memory.fetch("timeline", options);
|
||||
}
|
||||
|
||||
/** Durable canonical timeline rows, isolated as one atomic document per agent. */
|
||||
export class FileAgentTimelineStore implements AgentTimelineStore {
|
||||
private readonly documents = new Map<string, TimelineDocument>();
|
||||
private readonly loading = new Map<string, Promise<TimelineDocument>>();
|
||||
private readonly mutationTails = new Map<string, Promise<void>>();
|
||||
private readonly writeJson: (filePath: string, value: unknown) => Promise<void>;
|
||||
|
||||
constructor(
|
||||
private readonly directory: string,
|
||||
options?: FileAgentTimelineStoreOptions,
|
||||
) {
|
||||
this.writeJson = options?.writeJson ?? writeJsonFileAtomic;
|
||||
}
|
||||
|
||||
async appendCommitted(
|
||||
agentId: string,
|
||||
item: AgentTimelineItem,
|
||||
options?: { timestamp?: string; turnId?: string },
|
||||
): Promise<AgentTimelineRow> {
|
||||
return this.mutate(agentId, (document) => {
|
||||
const row: AgentTimelineRow = {
|
||||
seq: document.nextSeq,
|
||||
timestamp: options?.timestamp ?? new Date().toISOString(),
|
||||
item: structuredClone(item),
|
||||
...(options?.turnId ? { turnId: options.turnId } : {}),
|
||||
};
|
||||
document.rows.push(row);
|
||||
document.nextSeq += 1;
|
||||
return cloneRow(row);
|
||||
});
|
||||
}
|
||||
|
||||
async fetchCommitted(
|
||||
agentId: string,
|
||||
options?: AgentTimelineFetchOptions,
|
||||
): Promise<AgentTimelineFetchResult> {
|
||||
return selectRows(await this.getDocument(agentId), options);
|
||||
}
|
||||
|
||||
async getLatestCommittedSeq(agentId: string): Promise<number> {
|
||||
return (await this.getDocument(agentId)).rows.at(-1)?.seq ?? 0;
|
||||
}
|
||||
|
||||
async getCommittedRows(agentId: string): Promise<AgentTimelineRow[]> {
|
||||
return (await this.getDocument(agentId)).rows.map(cloneRow);
|
||||
}
|
||||
|
||||
async getCommittedSnapshot(agentId: string): Promise<AgentTimelineSnapshot> {
|
||||
const document = await this.getDocument(agentId);
|
||||
return { rows: document.rows.map(cloneRow), historyComplete: document.historyComplete };
|
||||
}
|
||||
|
||||
async getLastItem(agentId: string): Promise<AgentTimelineItem | null> {
|
||||
const item = (await this.getDocument(agentId)).rows.at(-1)?.item;
|
||||
return item ? structuredClone(item) : null;
|
||||
}
|
||||
|
||||
async getLastAssistantMessage(agentId: string): Promise<string | null> {
|
||||
const rows = (await this.getDocument(agentId)).rows;
|
||||
const chunks: string[] = [];
|
||||
for (let index = rows.length - 1; index >= 0; index -= 1) {
|
||||
const item = rows[index]!.item;
|
||||
if (item.type !== "assistant_message") {
|
||||
if (chunks.length > 0) break;
|
||||
continue;
|
||||
}
|
||||
chunks.push(item.text);
|
||||
}
|
||||
return chunks.length > 0 ? chunks.toReversed().join("") : null;
|
||||
}
|
||||
|
||||
async deleteAgent(agentId: string): Promise<void> {
|
||||
await this.runMutation(agentId, async () => {
|
||||
await fs.rm(this.filePath(agentId), { force: true });
|
||||
this.documents.delete(agentId);
|
||||
});
|
||||
}
|
||||
|
||||
async bulkInsert(agentId: string, rows: readonly AgentTimelineRow[]): Promise<void> {
|
||||
await this.runMutation(agentId, async () => {
|
||||
const current = await this.getDocument(agentId);
|
||||
const proposed = cloneDocument(current);
|
||||
let changed = false;
|
||||
for (const row of rows) {
|
||||
const parsed = TimelineRowSchema.parse(row);
|
||||
const existing = proposed.rows.find((candidate) => candidate.seq === parsed.seq);
|
||||
if (existing) {
|
||||
if (!isDeepStrictEqual(existing, parsed)) {
|
||||
throw new Error(`Conflicting timeline row sequence ${parsed.seq}`);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
proposed.rows.push(cloneRow(parsed));
|
||||
changed = true;
|
||||
}
|
||||
if (!changed) return;
|
||||
proposed.rows.sort((left, right) => left.seq - right.seq);
|
||||
const maxSeq = proposed.rows.at(-1)?.seq ?? 0;
|
||||
proposed.nextSeq = Math.max(proposed.nextSeq, maxSeq + 1);
|
||||
validateDocument(proposed);
|
||||
await this.writeJson(this.filePath(agentId), proposed);
|
||||
this.documents.set(agentId, proposed);
|
||||
});
|
||||
}
|
||||
|
||||
async replaceCommittedSnapshot(agentId: string, snapshot: AgentTimelineSnapshot): Promise<void> {
|
||||
await this.runMutation(agentId, async () => {
|
||||
const current = await this.getDocument(agentId);
|
||||
const rows = snapshot.rows.map((row) => cloneRow(TimelineRowSchema.parse(row)));
|
||||
const nextSeq = (rows.at(-1)?.seq ?? 0) + 1;
|
||||
const proposed: TimelineDocument = {
|
||||
version: 1,
|
||||
epoch: current.epoch,
|
||||
nextSeq,
|
||||
rows,
|
||||
historyComplete: snapshot.historyComplete,
|
||||
};
|
||||
validateDocument(proposed);
|
||||
await this.writeJson(this.filePath(agentId), proposed);
|
||||
this.documents.set(agentId, proposed);
|
||||
});
|
||||
}
|
||||
|
||||
async updateCommittedRow(agentId: string, row: AgentTimelineRow): Promise<void> {
|
||||
await this.mutate(agentId, (document) => {
|
||||
const parsed = TimelineRowSchema.parse(row);
|
||||
const index = document.rows.findIndex((candidate) => candidate.seq === parsed.seq);
|
||||
if (index < 0) {
|
||||
throw new Error(`Cannot update missing timeline row sequence ${parsed.seq}`);
|
||||
}
|
||||
document.rows[index] = cloneRow(parsed);
|
||||
});
|
||||
}
|
||||
|
||||
private async mutate<T>(
|
||||
agentId: string,
|
||||
mutation: (document: TimelineDocument) => T,
|
||||
): Promise<T> {
|
||||
return this.runMutation(agentId, async () => {
|
||||
const current = await this.getDocument(agentId);
|
||||
const proposed = cloneDocument(current);
|
||||
const result = mutation(proposed);
|
||||
validateDocument(proposed);
|
||||
await this.writeJson(this.filePath(agentId), proposed);
|
||||
this.documents.set(agentId, proposed);
|
||||
return result;
|
||||
});
|
||||
}
|
||||
|
||||
private async getDocument(agentId: string): Promise<TimelineDocument> {
|
||||
const cached = this.documents.get(agentId);
|
||||
if (cached) return cached;
|
||||
let load = this.loading.get(agentId);
|
||||
if (!load) {
|
||||
load = this.loadDocument(agentId);
|
||||
this.loading.set(agentId, load);
|
||||
}
|
||||
try {
|
||||
const document = await load;
|
||||
this.documents.set(agentId, document);
|
||||
return document;
|
||||
} finally {
|
||||
if (this.loading.get(agentId) === load) this.loading.delete(agentId);
|
||||
}
|
||||
}
|
||||
|
||||
private async loadDocument(agentId: string): Promise<TimelineDocument> {
|
||||
try {
|
||||
return validateDocument(JSON.parse(await fs.readFile(this.filePath(agentId), "utf8")));
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") return emptyDocument();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async runMutation<T>(agentId: string, operation: () => Promise<T>): Promise<T> {
|
||||
const previous = this.mutationTails.get(agentId) ?? Promise.resolve();
|
||||
const result = previous.catch(() => undefined).then(operation);
|
||||
const tail = result.then(
|
||||
() => undefined,
|
||||
() => undefined,
|
||||
);
|
||||
this.mutationTails.set(agentId, tail);
|
||||
void tail.finally(() => {
|
||||
if (this.mutationTails.get(agentId) === tail) this.mutationTails.delete(agentId);
|
||||
});
|
||||
return result;
|
||||
}
|
||||
|
||||
private filePath(agentId: string): string {
|
||||
return path.join(this.directory, fileNameForAgent(agentId));
|
||||
}
|
||||
}
|
||||
@@ -1,135 +0,0 @@
|
||||
import { describe, expect, test } from "vitest";
|
||||
|
||||
import { reconcileProviderHistory } from "./history-reconciliation.js";
|
||||
|
||||
const user = (text: string, id?: string) => ({
|
||||
type: "user_message" as const,
|
||||
text,
|
||||
...(id ? { clientMessageId: id } : {}),
|
||||
});
|
||||
|
||||
describe("reconcileProviderHistory", () => {
|
||||
test("puts missing provider prefix before a newer durable suffix while preserving suffix metadata", () => {
|
||||
const rows = reconcileProviderHistory(
|
||||
[
|
||||
{
|
||||
seq: 2,
|
||||
timestamp: "2026-01-02T00:00:00.000Z",
|
||||
item: user("suffix", "suffix"),
|
||||
turnId: "turn-1",
|
||||
providerMessageId: "p-suffix",
|
||||
},
|
||||
],
|
||||
[
|
||||
{ item: user("prefix"), timestamp: "2026-01-01T00:00:00.000Z" },
|
||||
{ item: user("suffix"), timestamp: "2026-01-02T00:00:00.000Z" },
|
||||
],
|
||||
);
|
||||
expect(rows).toMatchObject([
|
||||
{ seq: 1, item: { text: "prefix" } },
|
||||
{
|
||||
seq: 2,
|
||||
turnId: "turn-1",
|
||||
providerMessageId: "p-suffix",
|
||||
item: { text: "suffix", clientMessageId: "suffix" },
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
test("pairs repeated identical user text by ordered occurrence", () => {
|
||||
const rows = reconcileProviderHistory(
|
||||
[
|
||||
{
|
||||
seq: 1,
|
||||
timestamp: "2026-01-01T00:00:00.000Z",
|
||||
item: user("same", "one"),
|
||||
turnId: "turn-1",
|
||||
},
|
||||
{
|
||||
seq: 2,
|
||||
timestamp: "2026-01-02T00:00:00.000Z",
|
||||
item: user("same", "two"),
|
||||
turnId: "turn-2",
|
||||
},
|
||||
],
|
||||
[{ item: user("same") }, { item: user("same") }],
|
||||
);
|
||||
expect(
|
||||
rows.map((row) => [
|
||||
row.item.type === "user_message" ? row.item.clientMessageId : null,
|
||||
row.turnId,
|
||||
]),
|
||||
).toEqual([
|
||||
["one", "turn-1"],
|
||||
["two", "turn-2"],
|
||||
]);
|
||||
});
|
||||
|
||||
test("retains a canonical suffix when provider history is lagging", () => {
|
||||
const rows = reconcileProviderHistory(
|
||||
[
|
||||
{
|
||||
seq: 1,
|
||||
timestamp: "2026-01-01T00:00:00.000Z",
|
||||
item: user("initial", "initial"),
|
||||
turnId: "turn-1",
|
||||
},
|
||||
{
|
||||
seq: 2,
|
||||
timestamp: "2026-01-02T00:00:00.000Z",
|
||||
item: user("hello", "hello"),
|
||||
turnId: "turn-1",
|
||||
},
|
||||
],
|
||||
[{ item: user("initial") }],
|
||||
);
|
||||
|
||||
expect(rows).toMatchObject([
|
||||
{ seq: 1, item: { text: "initial", clientMessageId: "initial" }, turnId: "turn-1" },
|
||||
{ seq: 2, item: { text: "hello", clientMessageId: "hello" }, turnId: "turn-1" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("does not transfer provider identity between ambiguous repeated prompts", () => {
|
||||
const rows = reconcileProviderHistory(
|
||||
[
|
||||
{
|
||||
seq: 1,
|
||||
timestamp: "2026-01-01T00:00:00.000Z",
|
||||
item: user("same", "one"),
|
||||
turnId: "turn-1",
|
||||
},
|
||||
{
|
||||
seq: 2,
|
||||
timestamp: "2026-01-02T00:00:00.000Z",
|
||||
item: user("same", "two"),
|
||||
turnId: "turn-2",
|
||||
},
|
||||
],
|
||||
[{ item: { type: "user_message", text: "same", messageId: "provider-two" } }],
|
||||
);
|
||||
|
||||
expect(rows).toMatchObject([
|
||||
{ item: { clientMessageId: "one" }, turnId: "turn-1" },
|
||||
{ item: { clientMessageId: "two" }, turnId: "turn-2" },
|
||||
]);
|
||||
expect(rows.some((row) => row.providerMessageId === "provider-two")).toBe(false);
|
||||
});
|
||||
|
||||
test("does not invent turn membership for provider-only rows", () => {
|
||||
expect(
|
||||
reconcileProviderHistory([], [{ item: { type: "assistant_message", text: "provider" } }])[0]
|
||||
?.turnId,
|
||||
).toBeUndefined();
|
||||
});
|
||||
|
||||
test("clears unmatched rows for an authoritative forced history", () => {
|
||||
expect(
|
||||
reconcileProviderHistory(
|
||||
[{ seq: 1, timestamp: "now", item: user("old", "old"), turnId: "turn-1" }],
|
||||
[],
|
||||
{ mode: "force" },
|
||||
),
|
||||
).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -1,157 +0,0 @@
|
||||
import { isDeepStrictEqual } from "node:util";
|
||||
|
||||
import type { AgentTimelineItem } from "./agent-sdk-types.js";
|
||||
import type { AgentTimelineRow } from "./agent-timeline-store-types.js";
|
||||
|
||||
export interface ProviderHistoryTimelineEntry {
|
||||
item: AgentTimelineItem;
|
||||
timestamp?: string;
|
||||
}
|
||||
|
||||
/** Reconciles canonical metadata onto provider-ordered history without inventing membership. */
|
||||
export function reconcileProviderHistory(
|
||||
canonicalRows: readonly AgentTimelineRow[],
|
||||
providerEntries: readonly ProviderHistoryTimelineEntry[],
|
||||
options?: { mode?: "incomplete" | "force" },
|
||||
): AgentTimelineRow[] {
|
||||
if (providerEntries.length === 0) {
|
||||
return options?.mode === "force"
|
||||
? []
|
||||
: canonicalRows.map((row, index) => ({ ...row, seq: index + 1 }));
|
||||
}
|
||||
const remaining = canonicalRows.map((row, canonicalIndex) => ({
|
||||
row,
|
||||
canonicalIndex,
|
||||
used: false,
|
||||
}));
|
||||
const structuralCounts = countStructuralOccurrences(canonicalRows, providerEntries);
|
||||
const providerRows = providerEntries.map((entry) => {
|
||||
const match = takeMatch(remaining, entry.item, structuralCounts);
|
||||
return { entry, match };
|
||||
});
|
||||
const rows: AgentTimelineRow[] = [];
|
||||
const emittedCanonicalIndexes = new Set<number>();
|
||||
|
||||
for (const { entry, match } of providerRows) {
|
||||
if (match) {
|
||||
for (const candidate of remaining) {
|
||||
if (candidate.canonicalIndex >= match.canonicalIndex) break;
|
||||
if (!emittedCanonicalIndexes.has(candidate.canonicalIndex)) {
|
||||
rows.push({ ...candidate.row });
|
||||
emittedCanonicalIndexes.add(candidate.canonicalIndex);
|
||||
}
|
||||
}
|
||||
rows.push(
|
||||
match.transferProviderIdentity ? mergeMatchedRow(match.row, entry) : { ...match.row },
|
||||
);
|
||||
emittedCanonicalIndexes.add(match.canonicalIndex);
|
||||
continue;
|
||||
}
|
||||
rows.push({
|
||||
seq: 0,
|
||||
timestamp: entry.timestamp ?? new Date(0).toISOString(),
|
||||
item: entry.item,
|
||||
});
|
||||
}
|
||||
|
||||
if (options?.mode !== "force") {
|
||||
for (const candidate of remaining) {
|
||||
if (!emittedCanonicalIndexes.has(candidate.canonicalIndex)) {
|
||||
rows.push({ ...candidate.row });
|
||||
}
|
||||
}
|
||||
}
|
||||
rows.forEach((row, index) => {
|
||||
row.seq = index + 1;
|
||||
});
|
||||
return rows;
|
||||
}
|
||||
|
||||
function takeMatch(
|
||||
remaining: Array<{ row: AgentTimelineRow; canonicalIndex: number; used: boolean }>,
|
||||
provider: AgentTimelineItem,
|
||||
structuralCounts: Map<string, { canonical: number; provider: number }>,
|
||||
): { row: AgentTimelineRow; canonicalIndex: number; transferProviderIdentity: boolean } | null {
|
||||
const strong = remaining.find(
|
||||
(candidate) => !candidate.used && hasSharedIdentity(candidate.row, provider),
|
||||
);
|
||||
const structural =
|
||||
strong ??
|
||||
remaining.find(
|
||||
(candidate) => !candidate.used && structurallyMatches(candidate.row.item, provider),
|
||||
);
|
||||
if (!structural) return null;
|
||||
structural.used = true;
|
||||
const key = structuralKey(provider);
|
||||
const counts = structuralCounts.get(key);
|
||||
return {
|
||||
row: structural.row,
|
||||
canonicalIndex: structural.canonicalIndex,
|
||||
transferProviderIdentity:
|
||||
strong !== undefined || (counts?.canonical === 1 && counts.provider === 1),
|
||||
};
|
||||
}
|
||||
|
||||
function mergeMatchedRow(
|
||||
canonical: AgentTimelineRow,
|
||||
provider: ProviderHistoryTimelineEntry,
|
||||
): AgentTimelineRow {
|
||||
return {
|
||||
...canonical,
|
||||
item: mergeCanonicalIdentity(canonical.item, provider.item),
|
||||
};
|
||||
}
|
||||
|
||||
function countStructuralOccurrences(
|
||||
canonicalRows: readonly AgentTimelineRow[],
|
||||
providerEntries: readonly ProviderHistoryTimelineEntry[],
|
||||
): Map<string, { canonical: number; provider: number }> {
|
||||
const counts = new Map<string, { canonical: number; provider: number }>();
|
||||
for (const row of canonicalRows) {
|
||||
const key = structuralKey(row.item);
|
||||
const count = counts.get(key) ?? { canonical: 0, provider: 0 };
|
||||
count.canonical += 1;
|
||||
counts.set(key, count);
|
||||
}
|
||||
for (const entry of providerEntries) {
|
||||
const key = structuralKey(entry.item);
|
||||
const count = counts.get(key) ?? { canonical: 0, provider: 0 };
|
||||
count.provider += 1;
|
||||
counts.set(key, count);
|
||||
}
|
||||
return counts;
|
||||
}
|
||||
|
||||
function structuralKey(item: AgentTimelineItem): string {
|
||||
return item.type === "user_message"
|
||||
? `user:${item.text}`
|
||||
: `${item.type}:${JSON.stringify(item)}`;
|
||||
}
|
||||
|
||||
function hasSharedIdentity(row: AgentTimelineRow, provider: AgentTimelineItem): boolean {
|
||||
if (row.item.type !== "user_message" || provider.type !== "user_message") return false;
|
||||
const identities = [row.item.clientMessageId, row.item.messageId, row.providerMessageId].filter(
|
||||
Boolean,
|
||||
);
|
||||
return identities.some(
|
||||
(identity) => identity === provider.clientMessageId || identity === provider.messageId,
|
||||
);
|
||||
}
|
||||
|
||||
function structurallyMatches(left: AgentTimelineItem, right: AgentTimelineItem): boolean {
|
||||
if (left.type === "user_message" && right.type === "user_message")
|
||||
return left.text === right.text;
|
||||
return isDeepStrictEqual(left, right);
|
||||
}
|
||||
|
||||
function mergeCanonicalIdentity(
|
||||
canonical: AgentTimelineItem,
|
||||
provider: AgentTimelineItem,
|
||||
): AgentTimelineItem {
|
||||
if (canonical.type !== "user_message" || provider.type !== "user_message") return provider;
|
||||
return {
|
||||
...provider,
|
||||
...(canonical.clientMessageId ? { clientMessageId: canonical.clientMessageId } : {}),
|
||||
...(canonical.messageId ? { messageId: canonical.messageId } : {}),
|
||||
};
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
import os from "node:os";
|
||||
import http from "node:http";
|
||||
import path from "node:path";
|
||||
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { access, mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import pino from "pino";
|
||||
import { afterEach, describe, expect, test, vi } from "vitest";
|
||||
import { WebSocket } from "ws";
|
||||
@@ -79,6 +79,66 @@ describe("paseo daemon bootstrap", () => {
|
||||
}
|
||||
});
|
||||
|
||||
test("keeps timeline activity in memory and removes obsolete timeline files at startup", async () => {
|
||||
const paseoHomeRoot = await mkdtemp(path.join(os.tmpdir(), "paseo-timeline-cleanup-"));
|
||||
const paseoHome = path.join(paseoHomeRoot, ".paseo");
|
||||
const obsoleteTimelineDirectory = path.join(paseoHome, "agent-timelines");
|
||||
const agentCwd = await mkdtemp(path.join(os.tmpdir(), "paseo-timeline-agent-"));
|
||||
await mkdir(obsoleteTimelineDirectory, { recursive: true });
|
||||
await writeFile(path.join(obsoleteTimelineDirectory, "obsolete.json"), "{}\n", "utf-8");
|
||||
|
||||
const daemonHandle = await createTestPaseoDaemon({ paseoHomeRoot, cleanup: false });
|
||||
try {
|
||||
await expect(access(obsoleteTimelineDirectory)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
|
||||
const agent = await daemonHandle.daemon.agentManager.createAgent(
|
||||
{ provider: "codex", cwd: agentCwd },
|
||||
undefined,
|
||||
{ workspaceId: undefined },
|
||||
);
|
||||
await daemonHandle.daemon.agentManager.appendTimelineItem(agent.id, {
|
||||
type: "assistant_message",
|
||||
text: "timeline stays in memory",
|
||||
});
|
||||
await daemonHandle.daemon.agentManager.flush();
|
||||
|
||||
await expect(access(obsoleteTimelineDirectory)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} finally {
|
||||
await daemonHandle.close();
|
||||
await Promise.all([
|
||||
rm(paseoHomeRoot, { recursive: true, force: true }),
|
||||
rm(agentCwd, { recursive: true, force: true }),
|
||||
]);
|
||||
}
|
||||
});
|
||||
|
||||
test("does not create a timeline directory for live timeline activity", async () => {
|
||||
const paseoHomeRoot = await mkdtemp(path.join(os.tmpdir(), "paseo-timeline-memory-"));
|
||||
const agentCwd = await mkdtemp(path.join(os.tmpdir(), "paseo-timeline-agent-"));
|
||||
const daemonHandle = await createTestPaseoDaemon({ paseoHomeRoot, cleanup: false });
|
||||
const timelineDirectory = path.join(daemonHandle.paseoHome, "agent-timelines");
|
||||
try {
|
||||
const agent = await daemonHandle.daemon.agentManager.createAgent(
|
||||
{ provider: "codex", cwd: agentCwd },
|
||||
undefined,
|
||||
{ workspaceId: undefined },
|
||||
);
|
||||
await daemonHandle.daemon.agentManager.appendTimelineItem(agent.id, {
|
||||
type: "assistant_message",
|
||||
text: "timeline stays in memory",
|
||||
});
|
||||
await daemonHandle.daemon.agentManager.flush();
|
||||
|
||||
await expect(access(timelineDirectory)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} finally {
|
||||
await daemonHandle.close();
|
||||
await Promise.all([
|
||||
rm(paseoHomeRoot, { recursive: true, force: true }),
|
||||
rm(agentCwd, { recursive: true, force: true }),
|
||||
]);
|
||||
}
|
||||
});
|
||||
|
||||
test("reload applies live HTTP, MCP, Git, provider, relay, and app policies", async () => {
|
||||
const paseoHomeRoot = await mkdtemp(path.join(os.tmpdir(), "paseo-config-reload-runtime-"));
|
||||
const paseoHome = path.join(paseoHomeRoot, ".paseo");
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import express from "express";
|
||||
import { createServer as createHTTPServer, type IncomingMessage, type ServerResponse } from "http";
|
||||
import { constants, existsSync, unlinkSync } from "fs";
|
||||
import { open } from "fs/promises";
|
||||
import { open, rm } from "fs/promises";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { hostname as getHostname } from "node:os";
|
||||
import path from "node:path";
|
||||
@@ -129,7 +129,6 @@ import type { LocalSpeechProviderConfig } from "./speech/providers/local/config.
|
||||
import type { RequestedSpeechProviders } from "./speech/speech-types.js";
|
||||
import { createSpeechService } from "./speech/speech-runtime.js";
|
||||
import { AgentManager } from "./agent/agent-manager.js";
|
||||
import { FileAgentTimelineStore } from "./agent/file-agent-timeline-store.js";
|
||||
import { AgentStorage } from "./agent/agent-storage.js";
|
||||
import { attachAgentStoragePersistence } from "./persistence-hooks.js";
|
||||
import { createAgentMcpServer } from "./agent/mcp-server.js";
|
||||
@@ -576,6 +575,13 @@ export async function createPaseoDaemon(
|
||||
): Promise<PaseoDaemon> {
|
||||
configureGitProcessPolicy(config.git ?? resolveGitProcessPolicy({ env: process.env }));
|
||||
const logger = rootLogger.child({ module: "bootstrap" });
|
||||
const obsoleteTimelineDirectory = path.join(config.paseoHome, "agent-timelines");
|
||||
await rm(obsoleteTimelineDirectory, { recursive: true, force: true }).catch((error) => {
|
||||
logger.warn(
|
||||
{ err: error, path: obsoleteTimelineDirectory },
|
||||
"Failed to remove obsolete agent timeline data",
|
||||
);
|
||||
});
|
||||
const bootstrapStart = performance.now();
|
||||
const elapsed = () => `${(performance.now() - bootstrapStart).toFixed(0)}ms`;
|
||||
const daemonVersion = config.daemonVersion ?? resolveDaemonVersion(import.meta.url);
|
||||
@@ -851,7 +857,6 @@ export async function createPaseoDaemon(
|
||||
}
|
||||
|
||||
const agentStorage = new AgentStorage(config.agentStoragePath, logger);
|
||||
const timelineStore = new FileAgentTimelineStore(path.join(config.paseoHome, "agent-timelines"));
|
||||
const projectRegistry = new FileBackedProjectRegistry(
|
||||
path.join(config.paseoHome, "projects", "projects.json"),
|
||||
logger,
|
||||
@@ -904,7 +909,6 @@ export async function createPaseoDaemon(
|
||||
});
|
||||
const initialAgentManagerState = providerSnapshotManager.getAgentManagerProviderState();
|
||||
const agentManager = new AgentManager({
|
||||
durableTimelineStore: timelineStore,
|
||||
clients: initialAgentManagerState.clients,
|
||||
providerDefinitions: initialAgentManagerState.providerDefinitions,
|
||||
registry: agentStorage,
|
||||
|
||||
Reference in New Issue
Block a user