diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index df98ffac04..23646351d4 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -1,10 +1,12 @@ import { describe, expect, test, vi } from "vitest"; -import { mkdtempSync } from "node:fs"; +import { mkdtempSync, rmSync } from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; import { randomUUID } from "node:crypto"; import { createTestLogger } from "../../test-utils/test-logger.js"; +import { DbAgentTimelineStore } from "../db/db-agent-timeline-store.js"; +import { openPaseoDatabase } from "../db/pglite-database.js"; import { AgentManager } from "./agent-manager.js"; import { AgentStorage } from "./agent-storage.js"; import type { @@ -1060,7 +1062,7 @@ describe("AgentManager", () => { }); } - const baseline = manager.fetchTimeline(snapshot.id, { + const baseline = await manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0, }); @@ -1075,7 +1077,7 @@ describe("AgentManager", () => { text: "finalized reply", }); - const result = manager.fetchTimeline(snapshot.id, { + const result = await manager.fetchTimeline(snapshot.id, { direction: "after", cursor: { seq: 120, @@ -1091,6 +1093,114 @@ describe("AgentManager", () => { }); }); + test("fetchTimeline and getTimelineRows prefer the durable store while live helpers stay in-memory", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-durable-read-authority-")); + const storagePath = join(workdir, "agents"); + const dataDir = join(workdir, "db"); + const storage = new AgentStorage(storagePath, logger); + const database = await openPaseoDatabase(dataDir); + + try { + const durableTimelineStore = new DbAgentTimelineStore(database.db); + const manager = new AgentManager({ + clients: { + codex: new TestAgentClient(), + }, + registry: storage, + durableTimelineStore, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000139", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + }); + + const durableOnlyItem: AgentTimelineItem = { + type: "assistant_message", + text: "durable only", + }; + const durableOnlyRow = { + seq: 1, + timestamp: "2026-03-24T00:00:01.000Z", + item: durableOnlyItem, + }; + + await durableTimelineStore.bulkInsert(snapshot.id, [durableOnlyRow]); + + expect(manager.getTimeline(snapshot.id)).toEqual([]); + expect(manager.getLastAssistantMessage(snapshot.id)).toBeNull(); + await expect(manager.getTimelineRows(snapshot.id)).resolves.toEqual([durableOnlyRow]); + await expect( + manager.fetchTimeline(snapshot.id, { + direction: "tail", + limit: 0, + }), + ).resolves.toEqual({ + direction: "tail", + window: { + minSeq: 1, + maxSeq: 1, + nextSeq: 2, + }, + hasOlder: false, + hasNewer: false, + rows: [durableOnlyRow], + }); + } finally { + await database.close(); + rmSync(workdir, { recursive: true, force: true }); + } + }); + + test("getTimelineRows falls back to the in-memory timeline when no durable store is configured", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-rows-fallback-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const manager = new AgentManager({ + clients: { + codex: new TestAgentClient(), + }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000140", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + }); + + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: "row one", + }); + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: "row two", + }); + + await expect(manager.getTimelineRows(snapshot.id)).resolves.toEqual([ + { + seq: 1, + timestamp: expect.any(String), + item: { + type: "assistant_message", + text: "row one", + }, + }, + { + seq: 2, + timestamp: expect.any(String), + item: { + type: "assistant_message", + text: "row two", + }, + }, + ]); + }); + test("getAgent does not expose committed history internals once manager owns the seam", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-boundary-")); const storagePath = join(workdir, "agents"); @@ -1136,7 +1246,7 @@ describe("AgentManager", () => { }, ]); - const fetched = manager.fetchTimeline(snapshot.id, { + const fetched = await manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0, }); @@ -1216,7 +1326,7 @@ describe("AgentManager", () => { text: "final reply", }, ]); - const fetched = manager.fetchTimeline(snapshot.id, { + const fetched = await manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0, }); @@ -1266,7 +1376,7 @@ describe("AgentManager", () => { text: "fifth", }); - const result = manager.fetchTimeline(snapshot.id, { + const result = await manager.fetchTimeline(snapshot.id, { direction: "before", cursor: { seq: 5, @@ -1312,7 +1422,7 @@ describe("AgentManager", () => { text: "third", }); - const fetched = manager.fetchTimeline(snapshot.id, { + const fetched = await manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0, }); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 961417585a..2c2ec83e8d 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -484,13 +484,22 @@ export class AgentManager { return this.timelineStore.getItems(id); } - getTimelineRows(id: string): AgentTimelineRow[] { + async getTimelineRows(id: string): Promise { this.requireAgent(id); + if (this.durableTimelineStore) { + return await this.durableTimelineStore.getCommittedRows(id); + } return this.timelineStore.getRows(id); } - fetchTimeline(id: string, options?: AgentTimelineFetchOptions): AgentTimelineFetchResult { + async fetchTimeline( + id: string, + options?: AgentTimelineFetchOptions, + ): Promise { this.requireAgent(id); + if (this.durableTimelineStore) { + return await this.durableTimelineStore.fetchCommitted(id, options); + } return this.timelineStore.fetch(id, options); } diff --git a/packages/server/src/server/session.provider-history-compatibility-ownership.test.ts b/packages/server/src/server/session.provider-history-compatibility-ownership.test.ts index f67d94e81e..bdfdb29b60 100644 --- a/packages/server/src/server/session.provider-history-compatibility-ownership.test.ts +++ b/packages/server/src/server/session.provider-history-compatibility-ownership.test.ts @@ -81,7 +81,7 @@ function createSessionForOwnershipTests(options?: { hydrateTimelineFromProvider: vi.fn(async () => { throw new Error("Session should not call hydrateTimelineFromProvider directly"); }), - fetchTimeline: vi.fn(() => ({ + fetchTimeline: vi.fn(async () => ({ rows: options?.timelineRows ?? [], hasOlder: false, hasNewer: false, diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 42ec5ef08d..83a3757143 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -6073,7 +6073,7 @@ export class Session { try { const snapshot = await this.ensureAgentLoaded(msg.agentId); const agentPayload = await this.buildAgentPayload(snapshot); - const timeline = this.agentManager.fetchTimeline(msg.agentId, { + const timeline = await this.agentManager.fetchTimeline(msg.agentId, { direction, cursor, limit,