diff --git a/docs/architecture.md b/docs/architecture.md index d630adbcf..a871dc348 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -327,7 +327,7 @@ initializing → idle ⇄ running client-side dedup; the default fetch page is 200 items. - Timeline row `timestamp` values are canonical daemon-owned timestamps. Providers may supply original replay timestamps, but clients must not guess timestamp trust or hide time UI based on local clock heuristics. - Events stream to connected clients in real time; correctness is backed by authoritative timeline fetches and paged-to-completion catch-up. -- Agent state persists to `$PASEO_HOME/agents/{cwd-with-dashes}/{agent-id}.json` (timeline rows live alongside the record). That storage path is derived from `cwd`, not from workspace id. +- Agent state persists to `$PASEO_HOME/agents/{cwd-with-dashes}/{agent-id}.json`. Timeline rows are runtime memory; provider history is the durable transcript authority and resumed agents rebuild from it. That storage path is derived from `cwd`, not from workspace id. ## Right-sidebar boundary: directory-backed vs workspace-owned @@ -402,7 +402,7 @@ Providers that can accept native tool definitions should set `supportsNativePase ``` $PASEO_HOME/ -├── agents/{cwd-with-dashes}/{agent-id}.json # Agent record + persisted timeline rows +├── agents/{cwd-with-dashes}/{agent-id}.json # Agent record ├── projects/projects.json # Project registry ├── projects/workspaces.json # Workspace registry ├── projects/icons/ # Custom project icon images diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index b8780c3c6..01efe84c1 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -5,6 +5,9 @@ Agent chat delivery has two paths: 1. **Live stream** — `agent_stream` WebSocket messages for immediacy. These may be delta-shaped lifecycle updates. 2. **Authoritative history** — `fetch_agent_timeline_request` for correctness. This always returns full projected timeline items, never lifecycle deltas. +The daemon keeps canonical rows only for its runtime. Provider history is the durable transcript +authority and repopulates those rows when an agent resumes. + The invariants are: > A continuously subscribed client applies every committed row in order. Opening or resuming an @@ -12,7 +15,7 @@ The invariants are: > through backward pagination. Tool output is bounded before it enters either delivery path. Canonical shell tool output is sliced -to 64 KiB, and the same bounded item is used for durable timeline rows and live stream events. +to 64 KiB, and the same bounded item is used for runtime timeline rows and live stream events. Provider history hydration applies the same rule so reopening an agent cannot restore an oversized tool payload. diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 7331688f6..99088c084 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -14,7 +14,6 @@ import { type ManagedAgent, } from "./agent-manager.js"; import { AgentStorage } from "./agent-storage.js"; -import { FileAgentTimelineStore } from "./file-agent-timeline-store.js"; import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js"; import { toAgentPayload } from "./agent-projections.js"; import { getOpenAgentTabLabel, PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels"; @@ -52,627 +51,6 @@ import type { ProviderDefinition } from "./provider-registry.js"; const DESKTOP_OPEN_AGENT_TAB_LABEL = getOpenAgentTabLabel("desktop-client"); const MOBILE_OPEN_AGENT_TAB_LABEL = getOpenAgentTabLabel("mobile-client"); -test("marks a fresh agent canonical timeline complete", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-fresh-complete-")); - const timelineDirectory = join(workdir, "agent-timelines"); - const manager = new AgentManager({ - clients: { codex: new TestAgentClient() }, - durableTimelineStore: new FileAgentTimelineStore(timelineDirectory), - logger, - }); - let agentId: string | null = null; - try { - const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - agentId = agent.id; - expect( - await new FileAgentTimelineStore(timelineDirectory).getCommittedSnapshot(agent.id), - ).toEqual({ rows: [], historyComplete: true }); - } finally { - if (agentId) await manager.closeAgent(agentId).catch(() => undefined); - rmSync(workdir, { recursive: true, force: true }); - } -}); - -test("restores durable canonical turn membership when an agent is reopened", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-durable-turn-")); - const storage = new AgentStorage(join(workdir, "agents"), logger); - const durablePath = join(workdir, "agent-timelines"); - const first = new AgentManager({ - clients: { codex: new TestAgentClient() }, - registry: storage, - durableTimelineStore: new FileAgentTimelineStore(durablePath), - logger, - }); - try { - const created = await first.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - const durable = new FileAgentTimelineStore(durablePath); - await durable.appendCommitted( - created.id, - { type: "user_message", text: "initial", clientMessageId: "initial-client" }, - { turnId: "turn-1" }, - ); - await durable.appendCommitted( - created.id, - { - type: "user_message", - text: "hello", - clientMessageId: "hello-client", - messageId: "hello-client", - }, - { turnId: "turn-1" }, - ); - await first.closeAgent(created.id); - await storage.flush(); - const second = new AgentManager({ - clients: { codex: new TestAgentClient() }, - registry: storage, - durableTimelineStore: new FileAgentTimelineStore(durablePath), - logger, - }); - await ensureAgentLoaded(created.id, { agentManager: second, agentStorage: storage, logger }); - const hello = second - .fetchTimeline(created.id, { limit: 0 }) - .rows.filter( - (row) => row.item.type === "user_message" && row.item.clientMessageId === "hello-client", - ); - expect(hello).toHaveLength(1); - expect(hello[0]).toMatchObject({ - turnId: "turn-1", - item: { messageId: "hello-client", clientMessageId: "hello-client" }, - }); - await second.closeAgent(created.id); - } finally { - await storage.flush().catch(() => undefined); - rmSync(workdir, { recursive: true, force: true }); - } -}); - -test("reconciles provider history whenever a persisted agent starts a new runtime", async () => { - for (const scenario of [{ historyComplete: true }, { historyComplete: false }]) { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-completeness-")); - const storage = new AgentStorage(join(workdir, "agents"), logger); - const timelineDirectory = join(workdir, "agent-timelines"); - const first = new AgentManager({ - clients: { codex: new TestAgentClient() }, - registry: storage, - durableTimelineStore: new FileAgentTimelineStore(timelineDirectory), - logger, - }); - let agentId: string | null = null; - let second: AgentManager | null = null; - try { - const created = await first.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - agentId = created.id; - await first.closeAgent(created.id); - await storage.flush(); - - const canonicalRow: AgentTimelineRow = { - seq: 1, - timestamp: "2026-01-01T00:00:00.000Z", - item: { type: "user_message", text: "initial", clientMessageId: "initial-client" }, - turnId: "turn-1", - providerMessageId: "provider-initial", - }; - await new FileAgentTimelineStore(timelineDirectory).replaceCommittedSnapshot(created.id, { - rows: [canonicalRow], - historyComplete: scenario.historyComplete, - }); - - let historyCalls = 0; - class HistorySession extends TestAgentSession { - override async *streamHistory(): AsyncGenerator { - historyCalls += 1; - yield { - type: "timeline", - provider: "codex", - item: { type: "user_message", text: "initial" }, - }; - yield { - type: "timeline", - provider: "codex", - item: { type: "assistant_message", text: "tail" }, - }; - } - } - class HistoryClient extends TestAgentClient { - override async resumeSession( - _handle: AgentPersistenceHandle, - config?: Partial, - ): Promise { - return new HistorySession({ provider: "codex", cwd: config?.cwd ?? workdir }); - } - } - second = new AgentManager({ - clients: { codex: new HistoryClient() }, - registry: storage, - durableTimelineStore: new FileAgentTimelineStore(timelineDirectory), - logger, - }); - await ensureAgentLoaded(created.id, { agentManager: second, agentStorage: storage, logger }); - - const rows = second.fetchTimeline(created.id, { limit: 0 }).rows; - expect(historyCalls).toBe(1); - const textRows = rows.flatMap((row) => { - if (row.item.type === "user_message" || row.item.type === "assistant_message") { - return [row.item.text]; - } - return []; - }); - expect(textRows).toEqual(["initial", "tail"]); - expect( - rows.filter((row) => row.item.type === "user_message" && row.item.text === "initial"), - ).toEqual([canonicalRow]); - const reopened = await new FileAgentTimelineStore(timelineDirectory).getCommittedSnapshot( - created.id, - ); - expect(reopened.historyComplete).toBe(true); - expect( - reopened.rows.filter( - (row) => row.item.type === "assistant_message" && row.item.text === "tail", - ), - ).toHaveLength(1); - } finally { - if (second && agentId) await second.closeAgent(agentId).catch(() => undefined); - await storage.flush().catch(() => undefined); - rmSync(workdir, { recursive: true, force: true }); - } - } -}); - -test("retains a canonical steer when incomplete provider history lags on reopen", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-lagging-history-")); - const storage = new AgentStorage(join(workdir, "agents"), logger); - const timelineDirectory = join(workdir, "agent-timelines"); - const first = new AgentManager({ - clients: { codex: new TestAgentClient() }, - registry: storage, - durableTimelineStore: new FileAgentTimelineStore(timelineDirectory), - logger, - }); - let second: AgentManager | null = null; - let agentId: string | null = null; - try { - const created = await first.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - agentId = created.id; - await first.closeAgent(created.id); - await storage.flush(); - await new FileAgentTimelineStore(timelineDirectory).replaceCommittedSnapshot(created.id, { - historyComplete: false, - rows: [ - { - seq: 1, - timestamp: "2026-01-01T00:00:00.000Z", - item: { type: "user_message", text: "initial", clientMessageId: "initial" }, - turnId: "turn-1", - }, - { - seq: 2, - timestamp: "2026-01-02T00:00:00.000Z", - item: { type: "user_message", text: "hello", clientMessageId: "hello" }, - turnId: "turn-1", - }, - ], - }); - - class LaggingHistorySession extends TestAgentSession { - override async *streamHistory(): AsyncGenerator { - yield { - type: "timeline", - provider: "codex", - item: { type: "assistant_message", text: "provider prefix" }, - }; - yield { - type: "timeline", - provider: "codex", - item: { type: "user_message", text: "initial" }, - }; - } - } - class LaggingHistoryClient extends TestAgentClient { - override async resumeSession( - _handle: AgentPersistenceHandle, - config?: Partial, - ): Promise { - return new LaggingHistorySession({ provider: "codex", cwd: config?.cwd ?? workdir }); - } - } - second = new AgentManager({ - clients: { codex: new LaggingHistoryClient() }, - registry: storage, - durableTimelineStore: new FileAgentTimelineStore(timelineDirectory), - logger, - }); - const events: AgentManagerEvent[] = []; - second.subscribe((event) => events.push(event), { - agentId: created.id, - replayState: false, - }); - await ensureAgentLoaded(created.id, { - agentManager: second, - agentStorage: storage, - logger, - broadcastTimeline: true, - }); - - expect(second.fetchTimeline(created.id, { limit: 0 }).rows).toMatchObject([ - { item: { type: "assistant_message", text: "provider prefix" } }, - { item: { text: "initial", clientMessageId: "initial" }, turnId: "turn-1" }, - { item: { text: "hello", clientMessageId: "hello" }, turnId: "turn-1" }, - ]); - expect( - events.flatMap((event) => { - if (event.type !== "agent_stream" || event.event.type !== "timeline") return []; - return [ - { - seq: event.seq, - item: event.event.item, - turnId: event.event.turnId, - }, - ]; - }), - ).toMatchObject([ - { seq: 1, item: { type: "assistant_message", text: "provider prefix" } }, - { - seq: 2, - item: { type: "user_message", text: "initial", clientMessageId: "initial" }, - turnId: "turn-1", - }, - { - seq: 3, - item: { type: "user_message", text: "hello", clientMessageId: "hello" }, - turnId: "turn-1", - }, - ]); - expect( - await new FileAgentTimelineStore(timelineDirectory).getCommittedSnapshot(created.id), - ).toMatchObject({ historyComplete: true }); - } finally { - if (second && agentId) await second.closeAgent(agentId).catch(() => undefined); - await storage.flush().catch(() => undefined); - rmSync(workdir, { recursive: true, force: true }); - } -}); - -test("retries an interrupted initial history hydration after reopen without a durable prefix", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-retry-")); - const storage = new AgentStorage(join(workdir, "agents"), logger); - const timelineDirectory = join(workdir, "agent-timelines"); - let agentId: string | null = null; - let first: AgentManager | null = null; - let second: AgentManager | null = null; - try { - class InterruptedHistorySession extends TestAgentSession { - override async *streamHistory(): AsyncGenerator { - yield { - type: "timeline", - provider: "codex", - item: { type: "user_message", text: "prefix" }, - }; - throw new Error("history interrupted"); - } - } - class InterruptedHistoryClient extends TestAgentClient { - override async resumeSession( - _handle: AgentPersistenceHandle, - config?: Partial, - ): Promise { - return new InterruptedHistorySession({ - provider: "codex", - cwd: config?.cwd ?? workdir, - }); - } - } - first = new AgentManager({ - clients: { codex: new InterruptedHistoryClient() }, - registry: storage, - durableTimelineStore: new FileAgentTimelineStore(timelineDirectory), - logger, - }); - const created = await first.resumeAgentFromPersistence( - { - provider: "codex", - sessionId: "interrupted-history", - metadata: { cwd: workdir }, - }, - { cwd: workdir }, - ); - agentId = created.id; - await expect(first.hydrateTimelineFromProvider(created.id)).rejects.toThrow( - "history interrupted", - ); - expect( - await new FileAgentTimelineStore(timelineDirectory).getCommittedSnapshot(created.id), - ).toEqual({ - rows: [], - historyComplete: false, - }); - await first.closeAgent(created.id); - first = null; - await storage.flush(); - - let historyCalls = 0; - class CompleteHistorySession extends TestAgentSession { - override async *streamHistory(): AsyncGenerator { - historyCalls += 1; - yield { - type: "timeline", - provider: "codex", - item: { type: "user_message", text: "prefix" }, - }; - yield { - type: "timeline", - provider: "codex", - item: { type: "assistant_message", text: "tail" }, - }; - } - } - class CompleteHistoryClient extends TestAgentClient { - override async resumeSession( - _handle: AgentPersistenceHandle, - config?: Partial, - ): Promise { - return new CompleteHistorySession({ provider: "codex", cwd: config?.cwd ?? workdir }); - } - } - second = new AgentManager({ - clients: { codex: new CompleteHistoryClient() }, - registry: storage, - durableTimelineStore: new FileAgentTimelineStore(timelineDirectory), - logger, - }); - await ensureAgentLoaded(created.id, { agentManager: second, agentStorage: storage, logger }); - - expect(historyCalls).toBe(1); - const rows = second.fetchTimeline(created.id, { limit: 0 }).rows; - expect(rows.map((row) => row.item)).toEqual([ - { type: "user_message", text: "prefix" }, - { type: "assistant_message", text: "tail" }, - ]); - expect(rows.every((row) => row.turnId === undefined)).toBe(true); - expect( - await new FileAgentTimelineStore(timelineDirectory).getCommittedSnapshot(created.id), - ).toEqual({ - rows, - historyComplete: true, - }); - } finally { - if (first && agentId) await first.closeAgent(agentId).catch(() => undefined); - if (second && agentId) await second.closeAgent(agentId).catch(() => undefined); - await storage.flush().catch(() => undefined); - rmSync(workdir, { recursive: true, force: true }); - } -}); - -test("force hydration preserves a completed steered turn's canonical membership across durable reopen", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-force-steer-history-")); - const storage = new AgentStorage(join(workdir, "agents"), logger); - const timelineDirectory = join(workdir, "timelines"); - const first = new AgentManager({ - clients: { codex: new TestAgentClient() }, - registry: storage, - logger, - }); - let agentId: string | null = null; - let second: AgentManager | null = null; - try { - const created = await first.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - agentId = created.id; - await first.closeAgent(created.id); - await storage.flush(); - const rows: AgentTimelineRow[] = [ - { - seq: 1, - timestamp: "2026-01-01T00:00:00.000Z", - item: { type: "user_message", text: "initial", clientMessageId: "initial" }, - turnId: "turn-1", - }, - { - seq: 2, - timestamp: "2026-01-01T00:00:01.000Z", - item: { type: "assistant_message", text: "working" }, - turnId: "turn-1", - }, - { - seq: 3, - timestamp: "2026-01-01T00:00:02.000Z", - item: { type: "user_message", text: "hello", clientMessageId: "hello", messageId: "hello" }, - turnId: "turn-1", - providerMessageId: "provider-hello", - }, - { - seq: 4, - timestamp: "2026-01-01T00:00:03.000Z", - item: { type: "assistant_message", text: "done" }, - turnId: "turn-1", - }, - ]; - await new FileAgentTimelineStore(timelineDirectory).replaceCommittedSnapshot(created.id, { - rows, - historyComplete: true, - }); - class HistorySession extends TestAgentSession { - override async *streamHistory(): AsyncGenerator { - yield { - type: "timeline", - provider: "codex", - item: { type: "user_message", text: "initial" }, - }; - yield { - type: "timeline", - provider: "codex", - item: { type: "assistant_message", text: "working" }, - }; - yield { - type: "timeline", - provider: "codex", - item: { type: "user_message", text: "hello" }, - }; - yield { - type: "timeline", - provider: "codex", - item: { type: "assistant_message", text: "done" }, - }; - } - } - class HistoryClient extends TestAgentClient { - override async resumeSession( - _handle: AgentPersistenceHandle, - config?: Partial, - ): Promise { - return new HistorySession({ provider: "codex", cwd: config?.cwd ?? workdir }); - } - } - second = new AgentManager({ - clients: { codex: new HistoryClient() }, - registry: storage, - durableTimelineStore: new FileAgentTimelineStore(timelineDirectory), - logger, - }); - await ensureAgentLoaded(created.id, { agentManager: second, agentStorage: storage, logger }); - await second.hydrateTimelineFromProvider(created.id, { force: true }); - await second.flush(); - const reopened = await new FileAgentTimelineStore(timelineDirectory).getCommittedSnapshot( - created.id, - ); - const hello = reopened.rows.filter( - (row) => row.item.type === "user_message" && row.item.text === "hello", - ); - expect(reopened.historyComplete).toBe(true); - const textRows = reopened.rows.flatMap((row) => { - if (row.item.type === "user_message" || row.item.type === "assistant_message") { - return [row.item.text]; - } - return []; - }); - expect(textRows).toEqual(["initial", "working", "hello", "done"]); - expect(hello).toEqual([ - expect.objectContaining({ - turnId: "turn-1", - providerMessageId: "provider-hello", - item: expect.objectContaining({ clientMessageId: "hello", messageId: "hello" }), - }), - ]); - } finally { - if (second && agentId) await second.closeAgent(agentId).catch(() => undefined); - await storage.flush().catch(() => undefined); - rmSync(workdir, { recursive: true, force: true }); - } -}); - -test("forced rewind history removes omitted rows while retaining canonical steer identity", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-force-rewind-history-")); - const storage = new AgentStorage(join(workdir, "agents"), logger); - const timelines = join(workdir, "timelines"); - const first = new AgentManager({ - clients: { codex: new TestAgentClient() }, - registry: storage, - logger, - }); - let agentId: string | null = null; - let manager: AgentManager | null = null; - try { - const created = await first.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - agentId = created.id; - await first.closeAgent(created.id); - await storage.flush(); - await new FileAgentTimelineStore(timelines).replaceCommittedSnapshot(created.id, { - historyComplete: true, - rows: [ - { - seq: 1, - timestamp: "2026-01-01T00:00:00.000Z", - item: { type: "user_message", text: "initial", clientMessageId: "initial" }, - turnId: "turn-1", - }, - { - seq: 2, - timestamp: "2026-01-01T00:00:01.000Z", - item: { - type: "user_message", - text: "hello", - clientMessageId: "hello", - messageId: "hello", - }, - turnId: "turn-1", - providerMessageId: "provider-hello", - }, - { - seq: 3, - timestamp: "2026-01-01T00:00:02.000Z", - item: { type: "assistant_message", text: "removed later" }, - turnId: "turn-2", - }, - ], - }); - class RewoundSession extends TestAgentSession { - override async *streamHistory(): AsyncGenerator { - yield { - type: "timeline", - provider: "codex", - item: { type: "user_message", text: "initial" }, - }; - yield { - type: "timeline", - provider: "codex", - item: { type: "user_message", text: "hello" }, - }; - } - } - class RewoundClient extends TestAgentClient { - override async resumeSession( - _handle: AgentPersistenceHandle, - config?: Partial, - ): Promise { - return new RewoundSession({ provider: "codex", cwd: config?.cwd ?? workdir }); - } - } - manager = new AgentManager({ - clients: { codex: new RewoundClient() }, - registry: storage, - durableTimelineStore: new FileAgentTimelineStore(timelines), - logger, - }); - await ensureAgentLoaded(created.id, { agentManager: manager, agentStorage: storage, logger }); - await manager.hydrateTimelineFromProvider(created.id, { force: true }); - await manager.flush(); - const reopened = await new FileAgentTimelineStore(timelines).getCommittedSnapshot(created.id); - expect(reopened.historyComplete).toBe(true); - const userTexts = reopened.rows.flatMap((row) => { - if (row.item.type === "user_message") return [row.item.text]; - return []; - }); - expect(userTexts).toEqual(["initial", "hello"]); - expect(reopened.rows).not.toContainEqual( - expect.objectContaining({ item: { text: "removed later" } }), - ); - expect( - reopened.rows.filter((row) => row.item.type === "user_message" && row.item.text === "hello"), - ).toEqual([ - expect.objectContaining({ - turnId: "turn-1", - providerMessageId: "provider-hello", - item: expect.objectContaining({ clientMessageId: "hello", messageId: "hello" }), - }), - ]); - } finally { - if (manager && agentId) await manager.closeAgent(agentId).catch(() => undefined); - await storage.flush().catch(() => undefined); - rmSync(workdir, { recursive: true, force: true }); - } -}); - interface Deferred { promise: Promise; resolve: (value: T) => void; @@ -721,6 +99,73 @@ 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 { + this.ensure(agentId); + return this.memory.append(agentId, item, options); + } + + async fetchCommitted( + agentId: string, + options?: AgentTimelineFetchOptions, + ): Promise { + this.ensure(agentId); + return this.memory.fetch(agentId, options); + } + + async getLatestCommittedSeq(agentId: string): Promise { + return this.memory.has(agentId) ? (this.memory.getRows(agentId).at(-1)?.seq ?? 0) : 0; + } + + async getCommittedRows(agentId: string): Promise { + return this.memory.has(agentId) ? this.memory.getRows(agentId) : []; + } + + async getLastItem(agentId: string): Promise { + return this.memory.has(agentId) ? this.memory.getLastItem(agentId) : null; + } + + async getLastAssistantMessage(agentId: string): Promise { + return this.memory.has(agentId) ? this.memory.getLastAssistantMessage(agentId) : null; + } + + async deleteAgent(agentId: string): Promise { + this.memory.delete(agentId); + } + + async bulkInsert(agentId: string, rows: readonly AgentTimelineRow[]): Promise { + 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 { + 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", @@ -1183,6 +628,86 @@ async function startAndSteerThroughManager( return { manager, agentId: agent.id, workdir }; } +test("uses an injected timeline store without making it a production requirement", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-store-")); + const store = new RecordingTimelineStore(); + const manager = new AgentManager({ + clients: { codex: new TestAgentClient() }, + durableTimelineStore: store, + logger, + }); + let agentId: string | null = null; + try { + const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + agentId = agent.id; + await manager.appendTimelineItem(agent.id, { type: "assistant_message", text: "stored" }); + await manager.flush(); + + expect(store.writes.flat()).toContainEqual( + expect.objectContaining({ item: { type: "assistant_message", text: "stored" } }), + ); + await expect(manager.getTimelineRows(agent.id)).resolves.toContainEqual( + expect.objectContaining({ item: { type: "assistant_message", text: "stored" } }), + ); + } finally { + if (agentId) await manager.closeAgent(agentId).catch(() => undefined); + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("retries provider history hydration after a stream failure", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-retry-")); + let attempts = 0; + class RetryingHistorySession extends TestAgentSession { + override async *streamHistory(): AsyncGenerator { + attempts += 1; + if (attempts === 1) throw new Error("history interrupted"); + yield { + type: "timeline", + provider: "codex", + item: { type: "assistant_message", text: "provider history" }, + }; + } + } + const manager = new AgentManager({ + clients: { + codex: new (class extends TestAgentClient { + override async resumeSession( + _handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + return new RetryingHistorySession({ provider: "codex", cwd: config?.cwd ?? workdir }); + } + })(), + }, + logger, + }); + let agentId: string | null = null; + try { + const agent = await manager.resumeAgentFromPersistence( + { provider: "codex", sessionId: "retry-history" }, + { cwd: workdir }, + ); + agentId = agent.id; + await expect(manager.hydrateTimelineFromProvider(agent.id)).rejects.toThrow( + "history interrupted", + ); + expect(manager.getAgent(agent.id)?.historyPrimed).toBe(false); + + await manager.hydrateTimelineFromProvider(agent.id); + expect(attempts).toBe(2); + expect(manager.getTimeline(agent.id)).toContainEqual({ + type: "assistant_message", + text: "provider history", + }); + } finally { + if (agentId) await manager.closeAgent(agentId).catch(() => undefined); + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("unavailable steer interrupts once and starts one replacement turn", async () => { const session = new SteeringTestSession({ provider: "codex", cwd: process.cwd() }); session.steerResult = "unavailable"; @@ -1199,306 +724,6 @@ test("unavailable steer interrupts once and starts one replacement turn", async } }); -test("surfaces an ambiguous persistence error after one accepted steer without fallback", async () => { - const session = new SteeringTestSession({ provider: "codex", cwd: process.cwd() }); - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-steer-persist-failure-")); - let durableWriteCount = 0; - const manager = new AgentManager({ - clients: { - codex: new (class extends TestAgentClient { - override async createSession(): Promise { - return session; - } - })(), - }, - durableTimelineStore: new FileAgentTimelineStore(join(workdir, "timelines"), { - writeJson: async () => { - durableWriteCount += 1; - if (durableWriteCount > 1) throw new Error("disk full"); - }, - }), - logger, - }); - let agentId: string | null = null; - try { - const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - agentId = agent.id; - const run = manager.streamAgent(agent.id, "initial"); - void (async () => { - for await (const _event of run) { - } - })(); - await manager.waitForAgentRunStart(agent.id); - - await expect( - manager.steerAgentRun(agent.id, "hello", { clientMessageId: "hello-client" }), - ).rejects.toThrow("Steering was accepted but its canonical message could not be persisted"); - expect(session.steerCount).toBe(1); - expect(session.interruptCount).toBe(0); - expect(session.startCount).toBe(1); - expect( - manager - .getTimeline(agent.id) - .filter((item) => item.type === "user_message" && item.clientMessageId === "hello-client"), - ).toHaveLength(1); - } finally { - if (agentId) await manager.closeAgent(agentId).catch(() => undefined); - rmSync(workdir, { recursive: true, force: true }); - } -}); - -test("persists an immediate provider echo behind its queued canonical row", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-durable-echo-")); - const timelineDirectory = join(workdir, "timelines"); - const session = new SteeringTestSession({ provider: "codex", cwd: workdir }); - const manager = new AgentManager({ - clients: { - codex: new (class extends TestAgentClient { - override async createSession(): Promise { - return session; - } - })(), - }, - durableTimelineStore: new FileAgentTimelineStore(timelineDirectory), - durableTimelineCoalesceWindowMs: 60_000, - durableTimelineTimers: { - setTimeout: () => 0 as unknown as ReturnType, - clearTimeout: () => undefined, - }, - logger, - }); - let agentId: string | null = null; - try { - const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - agentId = agent.id; - const run = manager.streamAgent(agent.id, "initial"); - void (async () => { - for await (const _event of run) { - } - })(); - await manager.waitForAgentRunStart(agent.id); - await manager.steerAgentRun(agent.id, "hello", { clientMessageId: "hello-client" }); - await manager.flush(); - - const rows = await new FileAgentTimelineStore(timelineDirectory).getCommittedRows(agent.id); - const hello = rows.filter( - (row) => row.item.type === "user_message" && row.item.clientMessageId === "hello-client", - ); - expect(hello).toEqual([ - expect.objectContaining({ - turnId: "active-turn-1", - providerMessageId: "provider-hello-client", - item: expect.objectContaining({ clientMessageId: "hello-client" }), - }), - ]); - } finally { - if (agentId) await manager.closeAgent(agentId).catch(() => undefined); - rmSync(workdir, { recursive: true, force: true }); - } -}); - -test("does not resurrect a deleted pending timeline and permits fresh agent-id reuse", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-durable-delete-")); - const timelineDirectory = join(workdir, "timelines"); - const manager = new AgentManager({ - clients: { codex: new TestAgentClient() }, - durableTimelineStore: new FileAgentTimelineStore(timelineDirectory), - durableTimelineCoalesceWindowMs: 60_000, - durableTimelineTimers: { - setTimeout: () => 0 as unknown as ReturnType, - clearTimeout: () => undefined, - }, - logger, - }); - const agentId = "00000000-0000-4000-8000-000000000201"; - try { - await manager.createAgent({ provider: "codex", cwd: workdir }, agentId, { - workspaceId: undefined, - }); - await manager.appendTimelineItem(agentId, { type: "assistant_message", text: "deleted" }); - await manager.deleteCommittedTimeline(agentId); - await manager.flush(); - expect(await new FileAgentTimelineStore(timelineDirectory).getCommittedRows(agentId)).toEqual( - [], - ); - - await manager.closeAgent(agentId); - await manager.createAgent({ provider: "codex", cwd: workdir }, agentId, { - workspaceId: undefined, - }); - await manager.appendTimelineItem(agentId, { type: "assistant_message", text: "fresh" }); - await manager.flush(); - expect(await new FileAgentTimelineStore(timelineDirectory).getCommittedRows(agentId)).toEqual([ - expect.objectContaining({ item: { type: "assistant_message", text: "fresh" } }), - ]); - } finally { - await manager.closeAgent(agentId).catch(() => undefined); - rmSync(workdir, { recursive: true, force: true }); - } -}); - -test("flush drains a pending durable timeline batch before its scheduler fires", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-durable-flush-")); - const timelineDirectory = join(workdir, "timelines"); - const manager = new AgentManager({ - clients: { codex: new TestAgentClient() }, - durableTimelineStore: new FileAgentTimelineStore(timelineDirectory), - durableTimelineCoalesceWindowMs: 60_000, - durableTimelineTimers: { - setTimeout: () => 0 as unknown as ReturnType, - clearTimeout: () => undefined, - }, - logger, - }); - let agentId: string | null = null; - try { - const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - agentId = agent.id; - await manager.appendTimelineItem(agent.id, { type: "assistant_message", text: "pending" }); - await manager.flush(); - - expect(await new FileAgentTimelineStore(timelineDirectory).getCommittedRows(agent.id)).toEqual([ - expect.objectContaining({ item: { type: "assistant_message", text: "pending" } }), - ]); - } finally { - if (agentId) await manager.closeAgent(agentId).catch(() => undefined); - rmSync(workdir, { recursive: true, force: true }); - } -}); - -test("waits for durable canonical steer commit before resolving accepted", async () => { - const entered = deferred(); - const release = deferred(); - class HeldStore implements AgentTimelineStore { - 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 { - this.ensure(agentId); - return this.memory.append(agentId, item, options); - } - - async fetchCommitted( - agentId: string, - options?: AgentTimelineFetchOptions, - ): Promise { - this.ensure(agentId); - return this.memory.fetch(agentId, options); - } - - async getLatestCommittedSeq(agentId: string): Promise { - return this.memory.has(agentId) ? (this.memory.getRows(agentId).at(-1)?.seq ?? 0) : 0; - } - - async getCommittedRows(agentId: string): Promise { - return this.memory.has(agentId) ? this.memory.getRows(agentId) : []; - } - - async getCommittedSnapshot(agentId: string) { - return { rows: await this.getCommittedRows(agentId), historyComplete: false }; - } - - async getLastItem(agentId: string): Promise { - return this.memory.has(agentId) ? this.memory.getLastItem(agentId) : null; - } - - async getLastAssistantMessage(agentId: string): Promise { - return this.memory.has(agentId) ? this.memory.getLastAssistantMessage(agentId) : null; - } - - async deleteAgent(agentId: string): Promise { - this.memory.delete(agentId); - } - - override async bulkInsert(agentId: string, rows: readonly AgentTimelineRow[]): Promise { - entered.resolve(); - await release.promise; - 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 { - this.ensure(agentId); - const rows = this.memory.getRows(agentId); - const index = rows.findIndex((candidate) => candidate.seq === row.seq); - if (index < 0) { - this.memory.append(agentId, row.item, { timestamp: row.timestamp, turnId: row.turnId }); - } - } - async replaceCommittedSnapshot( - agentId: string, - snapshot: { rows: AgentTimelineRow[] }, - ): Promise { - this.ensure(agentId); - this.memory.initialize(agentId, { rows: snapshot.rows }); - } - } - const store = new HeldStore(); - const session = new SteeringTestSession({ provider: "codex", cwd: process.cwd() }); - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-held-durable-")); - const client = new (class extends TestAgentClient { - override async createSession() { - return session; - } - })(); - const manager = new AgentManager({ - clients: { codex: client }, - durableTimelineStore: store, - logger, - }); - let agentId: string | null = null; - try { - const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - agentId = agent.id; - const run = manager.streamAgent(agent.id, "initial"); - void (async () => { - for await (const _ of run) { - } - })(); - await manager.waitForAgentRunStart(agent.id); - const steer = manager.steerAgentRun(agent.id, "hello", { clientMessageId: "hello-client" }); - await entered.promise; - let settled = false; - void steer.finally(() => { - settled = true; - }); - await Promise.resolve(); - expect(settled).toBe(false); - release.resolve(); - await expect(steer).resolves.toEqual({ status: "accepted" }); - expect( - manager - .getTimeline(agent.id) - .filter((item) => item.type === "user_message" && item.clientMessageId === "hello-client"), - ).toHaveLength(1); - } finally { - release.resolve(); - if (agentId) { - await manager.closeAgent(agentId).catch(() => undefined); - } - rmSync(workdir, { recursive: true, force: true }); - } -}); - test("orders an accepted steer before output emitted while acknowledgement is pending", async () => { const entered = deferred(); const release = deferred(); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index c8e209643..263432450 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -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(); private readonly backgroundTasks = new Set>(); private readonly agentRegistrationTasks = new Set>(); @@ -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 { - if (!this.durableTimelineStore) { - return; - } - await this.durableTimelineBuffer?.discard(agentId); - await this.durableTimelineStore.deleteAgent(agentId); - } - async deleteAgentState(agentId: string): Promise { this.discardRetainedAgentState(agentId); await this.deleteCommittedTimeline(agentId); } + async deleteCommittedTimeline(agentId: string): Promise { + await this.durableTimelineStore?.deleteAgent(agentId); + } + async getLastAssistantMessage(agentId: string): Promise { 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 { 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 { + ): Promise { 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 { const deferredBroadcast = typeof broadcast === "function"; - let timelineRows: AgentTimelineRow[] = []; + const timelineEvents: Array<{ + event: Extract; + row: AgentTimelineRow; + }> = []; const providerSubagentEvents: AgentManagerEvent[] = []; - const historyEvents: Extract[] = []; 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 { - 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 { this.agentStreamCoalescer.flushAll(); - await this.durableTimelineBuffer?.flushAll(); // Drain tasks, including tasks spawned while awaiting. while ( this.backgroundTasks.size > 0 || diff --git a/packages/server/src/server/agent/agent-timeline-store-types.ts b/packages/server/src/server/agent/agent-timeline-store-types.ts index 13fe840f7..d73f52299 100644 --- a/packages/server/src/server/agent/agent-timeline-store-types.ts +++ b/packages/server/src/server/agent/agent-timeline-store-types.ts @@ -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; getLatestCommittedSeq(agentId: string): Promise; getCommittedRows(agentId: string): Promise; - getCommittedSnapshot(agentId: string): Promise; getLastItem(agentId: string): Promise; getLastAssistantMessage(agentId: string): Promise; deleteAgent(agentId: string): Promise; bulkInsert(agentId: string, rows: readonly AgentTimelineRow[]): Promise; - replaceCommittedSnapshot(agentId: string, snapshot: AgentTimelineSnapshot): Promise; updateCommittedRow(agentId: string, row: AgentTimelineRow): Promise; } diff --git a/packages/server/src/server/agent/durable-timeline-buffer.test.ts b/packages/server/src/server/agent/durable-timeline-buffer.test.ts deleted file mode 100644 index c6fc513a1..000000000 --- a/packages/server/src/server/agent/durable-timeline-buffer.test.ts +++ /dev/null @@ -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 { - throw new Error("not used"); - } - async fetchCommitted(): Promise { - throw new Error("not used"); - } - async getLatestCommittedSeq(): Promise { - return 0; - } - async getCommittedRows(): Promise { - return []; - } - async getCommittedSnapshot(): Promise { - return { rows: [], historyComplete: false }; - } - async getLastItem(): Promise { - return null; - } - async getLastAssistantMessage(): Promise { - return null; - } - async deleteAgent(): Promise {} - async bulkInsert(_agentId: string, rows: readonly AgentTimelineRow[]): Promise { - this.batches.push([...rows]); - if (this.failuresRemaining > 0) { - this.failuresRemaining -= 1; - throw new Error("disk full"); - } - } - async replaceCommittedSnapshot(): Promise {} - async updateCommittedRow(): Promise {} -} - -class ControlledTimers { - readonly delays: number[] = []; - private readonly callbacks = new Map void>(); - private nextId = 1; - - setTimeout = (callback: () => void, delayMs: number): ReturnType => { - const id = this.nextId++; - this.delays.push(delayMs); - this.callbacks.set(id, callback); - return id as unknown as ReturnType; - }; - - clearTimeout = (timer: ReturnType): void => { - this.callbacks.delete(timer as unknown as number); - }; - - async runNext(): Promise { - 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, - 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); -} diff --git a/packages/server/src/server/agent/durable-timeline-buffer.ts b/packages/server/src/server/agent/durable-timeline-buffer.ts deleted file mode 100644 index 31642ec22..000000000 --- a/packages/server/src/server/agent/durable-timeline-buffer.ts +++ /dev/null @@ -1,118 +0,0 @@ -import type { AgentTimelineRow, AgentTimelineStore } from "./agent-timeline-store-types.js"; - -export interface DurableTimelineTimers { - setTimeout(callback: () => void, delayMs: number): ReturnType; - clearTimeout(timer: ReturnType): void; -} - -export interface DurableTimelineRetryOptions { - baseDelayMs?: number; - maxDelayMs?: number; -} - -const DEFAULT_RETRY_MAX_DELAY_MS = 5_000; - -interface PendingAgentRows { - rows: AgentTimelineRow[]; - timer: ReturnType | null; - flushing: Promise | null; - retryAttempt: number; -} - -/** Keeps each agent's durable transcript contiguous while coalescing live writes. */ -export class DurableTimelineBuffer { - private readonly pending = new Map(); - - 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 { - 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 { - await Promise.all([...this.pending.keys()].map((agentId) => this.flush(agentId))); - } - - async discard(agentId: string): Promise { - 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): void { - const candidate = timer as unknown as { unref?: () => void }; - candidate.unref?.(); - } -} diff --git a/packages/server/src/server/agent/file-agent-timeline-store.test.ts b/packages/server/src/server/agent/file-agent-timeline-store.test.ts deleted file mode 100644 index 4001e8d45..000000000 --- a/packages/server/src/server/agent/file-agent-timeline-store.test.ts +++ /dev/null @@ -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 { - 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((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([]); - }); -}); diff --git a/packages/server/src/server/agent/file-agent-timeline-store.ts b/packages/server/src/server/agent/file-agent-timeline-store.ts deleted file mode 100644 index 2abafc4de..000000000 --- a/packages/server/src/server/agent/file-agent-timeline-store.ts +++ /dev/null @@ -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 = 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; - -export interface FileAgentTimelineStoreOptions { - writeJson?: (filePath: string, value: unknown) => Promise; -} - -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(); - private readonly loading = new Map>(); - private readonly mutationTails = new Map>(); - private readonly writeJson: (filePath: string, value: unknown) => Promise; - - constructor( - private readonly directory: string, - options?: FileAgentTimelineStoreOptions, - ) { - this.writeJson = options?.writeJson ?? writeJsonFileAtomic; - } - - async appendCommitted( - agentId: string, - item: AgentTimelineItem, - options?: { timestamp?: string; turnId?: string }, - ): Promise { - 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 { - return selectRows(await this.getDocument(agentId), options); - } - - async getLatestCommittedSeq(agentId: string): Promise { - return (await this.getDocument(agentId)).rows.at(-1)?.seq ?? 0; - } - - async getCommittedRows(agentId: string): Promise { - return (await this.getDocument(agentId)).rows.map(cloneRow); - } - - async getCommittedSnapshot(agentId: string): Promise { - const document = await this.getDocument(agentId); - return { rows: document.rows.map(cloneRow), historyComplete: document.historyComplete }; - } - - async getLastItem(agentId: string): Promise { - const item = (await this.getDocument(agentId)).rows.at(-1)?.item; - return item ? structuredClone(item) : null; - } - - async getLastAssistantMessage(agentId: string): Promise { - 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 { - 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 { - 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 { - 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 { - 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( - agentId: string, - mutation: (document: TimelineDocument) => T, - ): Promise { - 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 { - 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 { - 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(agentId: string, operation: () => Promise): Promise { - 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)); - } -} diff --git a/packages/server/src/server/agent/history-reconciliation.test.ts b/packages/server/src/server/agent/history-reconciliation.test.ts deleted file mode 100644 index 8b14315c2..000000000 --- a/packages/server/src/server/agent/history-reconciliation.test.ts +++ /dev/null @@ -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([]); - }); -}); diff --git a/packages/server/src/server/agent/history-reconciliation.ts b/packages/server/src/server/agent/history-reconciliation.ts deleted file mode 100644 index 327bc487f..000000000 --- a/packages/server/src/server/agent/history-reconciliation.ts +++ /dev/null @@ -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(); - - 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, -): { 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 { - const counts = new Map(); - 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 } : {}), - }; -} diff --git a/packages/server/src/server/bootstrap.smoke.test.ts b/packages/server/src/server/bootstrap.smoke.test.ts index 68bafda19..47e3823a0 100644 --- a/packages/server/src/server/bootstrap.smoke.test.ts +++ b/packages/server/src/server/bootstrap.smoke.test.ts @@ -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"); diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index d2e093a07..50d32b9c1 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -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 { 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,