From ae4f9e04100d16cf8184cf196dbadf4520a0d39e Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 25 Mar 2026 05:09:57 +0700 Subject: [PATCH] Phase 1c: isolate provider history compatibility seam --- .../server/src/server/persistence-hooks.ts | 27 ++ ...der-history-compatibility-boundary.test.ts | 15 + ...ider-history-compatibility-service.test.ts | 395 ++++++++++++++++++ .../provider-history-compatibility-service.ts | 153 +++++++ ...er-history-compatibility-ownership.test.ts | 334 +++++++++++++++ packages/server/src/server/session.ts | 135 ++---- 6 files changed, 948 insertions(+), 111 deletions(-) create mode 100644 packages/server/src/server/provider-history-compatibility-boundary.test.ts create mode 100644 packages/server/src/server/provider-history-compatibility-service.test.ts create mode 100644 packages/server/src/server/provider-history-compatibility-service.ts create mode 100644 packages/server/src/server/session.provider-history-compatibility-ownership.test.ts diff --git a/packages/server/src/server/persistence-hooks.ts b/packages/server/src/server/persistence-hooks.ts index 2e958181a5..a35b7e7dc8 100644 --- a/packages/server/src/server/persistence-hooks.ts +++ b/packages/server/src/server/persistence-hooks.ts @@ -1,5 +1,8 @@ +import type pino from "pino"; + import type { AgentProvider, AgentSessionConfig } from "./agent/agent-sdk-types.js"; import type { StoredAgentRecord } from "./agent/agent-storage.js"; +import { isValidAgentProvider } from "./agent/provider-manifest.js"; function isKnownProvider(provider: string): provider is AgentProvider { return provider === "claude" || provider === "codex" || provider === "opencode"; @@ -36,6 +39,30 @@ export function buildSessionConfig(record: StoredAgentRecord): AgentSessionConfi }; } +export function toAgentPersistenceHandle( + logger: pino.Logger, + handle: StoredAgentRecord["persistence"], +) { + if (!handle) { + return null; + } + const provider = handle.provider; + if (!isValidAgentProvider(provider)) { + logger.warn({ provider }, `Ignoring persistence handle with unknown provider '${provider}'`); + return null; + } + if (!handle.sessionId) { + logger.warn("Ignoring persistence handle missing sessionId"); + return null; + } + return { + provider, + sessionId: handle.sessionId, + nativeHandle: handle.nativeHandle, + metadata: handle.metadata, + }; +} + export function extractTimestamps(record: StoredAgentRecord): { createdAt: Date; updatedAt: Date; diff --git a/packages/server/src/server/provider-history-compatibility-boundary.test.ts b/packages/server/src/server/provider-history-compatibility-boundary.test.ts new file mode 100644 index 0000000000..b0d49103d5 --- /dev/null +++ b/packages/server/src/server/provider-history-compatibility-boundary.test.ts @@ -0,0 +1,15 @@ +import { readFileSync } from "node:fs"; +import { describe, expect, test } from "vitest"; + +describe("provider history compatibility boundary", () => { + test("session runtime code does not directly hydrate provider history", () => { + const sessionSource = readFileSync(new URL("./session.ts", import.meta.url), "utf8"); + const compatibilitySource = readFileSync( + new URL("./provider-history-compatibility-service.ts", import.meta.url), + "utf8", + ); + + expect(sessionSource).not.toMatch(/hydrateTimelineFromProvider\s*\(/); + expect(compatibilitySource.match(/hydrateTimelineFromProvider\s*\(/g)).toHaveLength(1); + }); +}); diff --git a/packages/server/src/server/provider-history-compatibility-service.test.ts b/packages/server/src/server/provider-history-compatibility-service.test.ts new file mode 100644 index 0000000000..95a1a74131 --- /dev/null +++ b/packages/server/src/server/provider-history-compatibility-service.test.ts @@ -0,0 +1,395 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import pino from "pino"; +import { describe, expect, test, vi } from "vitest"; + +import { AgentManager } from "./agent/agent-manager.js"; +import { AgentStorage, type StoredAgentRecord } from "./agent/agent-storage.js"; +import { ProviderHistoryCompatibilityService } from "./provider-history-compatibility-service.js"; +import { createTestAgentClients } from "./test-utils/fake-agent-client.js"; + +function createStoredAgentRecord(overrides?: Partial): StoredAgentRecord { + const now = "2026-03-25T00:00:00.000Z"; + return { + id: "agent-compat-1", + provider: "codex", + cwd: "/tmp/project", + createdAt: now, + updatedAt: now, + title: null, + labels: {}, + lastStatus: "idle", + config: { + model: "gpt-5.1-codex-mini", + }, + persistence: { + provider: "codex", + sessionId: "provider-session-1", + }, + ...overrides, + }; +} + +function createDeferred() { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +function createCompatibilitySnapshot(overrides?: Partial>) { + return { + id: "agent-compat-1", + provider: "codex", + cwd: "/tmp/project", + persistence: { + provider: "codex", + sessionId: "provider-session-1", + }, + ...overrides, + }; +} + +describe("ProviderHistoryCompatibilityService", () => { + test("ensureAgentLoaded replays provider history for an unloaded persisted agent", async () => { + const workspaceRoot = mkdtempSync(path.join(os.tmpdir(), "provider-history-compat-load-")); + const logger = pino({ level: "silent" }); + + try { + const storage = new AgentStorage(path.join(workspaceRoot, "agents"), logger); + const manager = new AgentManager({ + clients: createTestAgentClients(), + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000301", + }); + const service = new ProviderHistoryCompatibilityService({ + agentManager: manager as any, + agentStorage: storage as any, + logger, + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workspaceRoot, + model: "gpt-5.1-codex-mini", + }); + await manager.runAgent(snapshot.id, "say 'timeline test'"); + await manager.flush(); + await storage.flush(); + await manager.closeAgent(snapshot.id); + + const loaded = await service.ensureAgentLoaded({ agentId: snapshot.id }); + + expect(loaded.id).toBe(snapshot.id); + expect(manager.getTimeline(snapshot.id)).toEqual([{ type: "assistant_message", text: "timeline test" }]); + } finally { + rmSync(workspaceRoot, { recursive: true, force: true }); + } + }); + + test("ensureAgentLoaded succeeds when provider history is absent", async () => { + const workspaceRoot = mkdtempSync(path.join(os.tmpdir(), "provider-history-compat-empty-")); + const logger = pino({ level: "silent" }); + + try { + const storage = new AgentStorage(path.join(workspaceRoot, "agents"), logger); + const manager = new AgentManager({ + clients: createTestAgentClients(), + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000302", + }); + const service = new ProviderHistoryCompatibilityService({ + agentManager: manager as any, + agentStorage: storage as any, + logger, + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workspaceRoot, + model: "gpt-5.1-codex-mini", + }); + await manager.flush(); + await storage.flush(); + await manager.closeAgent(snapshot.id); + + const loaded = await service.ensureAgentLoaded({ agentId: snapshot.id }); + + expect(loaded.id).toBe(snapshot.id); + expect(manager.getTimeline(snapshot.id)).toEqual([]); + } finally { + rmSync(workspaceRoot, { recursive: true, force: true }); + } + }); + + test("ensureAgentLoaded dedupes concurrent cold-load bootstrap", async () => { + const deferred = createDeferred(); + let currentAgent: any = null; + const snapshot = createCompatibilitySnapshot({ id: "agent-compat-dedupe" }); + const agentStorage = { + get: vi.fn(async () => + createStoredAgentRecord({ + id: "agent-compat-dedupe", + cwd: "/tmp/dedupe", + persistence: { + provider: "codex", + sessionId: "provider-session-dedupe", + }, + }), + ), + }; + const agentManager = { + getAgent: vi.fn(() => currentAgent), + resumeAgentFromPersistence: vi.fn(async () => deferred.promise), + createAgent: vi.fn(), + reloadAgentSession: vi.fn(), + hydrateTimelineFromProvider: vi.fn(async () => { + currentAgent = snapshot; + }), + }; + const logger = { + child: () => logger, + info: vi.fn(), + warn: vi.fn(), + }; + const service = new ProviderHistoryCompatibilityService({ + agentManager: agentManager as any, + agentStorage: agentStorage as any, + logger: logger as any, + }); + + const firstLoad = service.ensureAgentLoaded({ agentId: "agent-compat-dedupe" }); + const secondLoad = service.ensureAgentLoaded({ agentId: "agent-compat-dedupe" }); + deferred.resolve(snapshot); + + const [firstResult, secondResult] = await Promise.all([firstLoad, secondLoad]); + + expect(firstResult).toEqual(snapshot); + expect(secondResult).toEqual(snapshot); + expect(agentStorage.get).toHaveBeenCalledTimes(1); + expect(agentManager.resumeAgentFromPersistence).toHaveBeenCalledTimes(1); + expect(agentManager.hydrateTimelineFromProvider).toHaveBeenCalledTimes(1); + }); + + test("resumeAgent delegates to manager resume then compatibility hydrate", async () => { + const snapshot = createCompatibilitySnapshot({ id: "agent-compat-resume" }); + let currentAgent: any = null; + const agentManager = { + getAgent: vi.fn(() => currentAgent), + resumeAgentFromPersistence: vi.fn(async () => snapshot), + createAgent: vi.fn(), + reloadAgentSession: vi.fn(), + hydrateTimelineFromProvider: vi.fn(async () => { + currentAgent = snapshot; + }), + }; + const logger = { + child: () => logger, + info: vi.fn(), + warn: vi.fn(), + }; + const service = new ProviderHistoryCompatibilityService({ + agentManager: agentManager as any, + agentStorage: { + get: async () => null, + } as any, + logger: logger as any, + }); + + const result = await service.resumeAgent({ + handle: { + provider: "codex", + sessionId: "provider-session-resume", + }, + overrides: { + model: "gpt-5.4", + }, + }); + + expect(agentManager.resumeAgentFromPersistence).toHaveBeenCalledWith( + { + provider: "codex", + sessionId: "provider-session-resume", + }, + { + model: "gpt-5.4", + }, + ); + expect(agentManager.hydrateTimelineFromProvider).toHaveBeenCalledWith("agent-compat-resume"); + expect(result).toEqual(snapshot); + }); + + test("refreshAgent reloads loaded persisted agents through the compatibility hydrate path", async () => { + const existing = createCompatibilitySnapshot({ id: "agent-compat-refresh-loaded" }); + const reloaded = createCompatibilitySnapshot({ id: "agent-compat-refresh-loaded" }); + let currentAgent: any = existing; + const agentManager = { + getAgent: vi.fn(() => currentAgent), + resumeAgentFromPersistence: vi.fn(), + createAgent: vi.fn(), + reloadAgentSession: vi.fn(async () => { + currentAgent = reloaded; + return reloaded; + }), + hydrateTimelineFromProvider: vi.fn(async () => undefined), + }; + const logger = { + child: () => logger, + info: vi.fn(), + warn: vi.fn(), + }; + const service = new ProviderHistoryCompatibilityService({ + agentManager: agentManager as any, + agentStorage: { + get: async () => null, + } as any, + logger: logger as any, + }); + + const result = await service.refreshAgent({ agentId: "agent-compat-refresh-loaded" }); + + expect(agentManager.reloadAgentSession).toHaveBeenCalledWith("agent-compat-refresh-loaded"); + expect(agentManager.hydrateTimelineFromProvider).toHaveBeenCalledWith( + "agent-compat-refresh-loaded", + ); + expect(result).toEqual(reloaded); + }); + + test("refreshAgent rehydrates loaded non-persisted agents without reloading", async () => { + const existing = createCompatibilitySnapshot({ + id: "agent-compat-refresh-live", + persistence: null, + }); + const agentManager = { + getAgent: vi.fn(() => existing), + resumeAgentFromPersistence: vi.fn(), + createAgent: vi.fn(), + reloadAgentSession: vi.fn(), + hydrateTimelineFromProvider: vi.fn(async () => undefined), + }; + const logger = { + child: () => logger, + info: vi.fn(), + warn: vi.fn(), + }; + const service = new ProviderHistoryCompatibilityService({ + agentManager: agentManager as any, + agentStorage: { + get: async () => null, + } as any, + logger: logger as any, + }); + + const result = await service.refreshAgent({ agentId: "agent-compat-refresh-live" }); + + expect(agentManager.reloadAgentSession).not.toHaveBeenCalled(); + expect(agentManager.hydrateTimelineFromProvider).toHaveBeenCalledWith( + "agent-compat-refresh-live", + ); + expect(result).toEqual(existing); + }); + + test("refreshAgent resumes unloaded persisted agents through the compatibility hydrate path", async () => { + const snapshot = createCompatibilitySnapshot({ id: "agent-compat-refresh-cold" }); + let currentAgent: any = null; + const record = createStoredAgentRecord({ + id: "agent-compat-refresh-cold", + cwd: "/tmp/refresh-cold", + persistence: { + provider: "codex", + sessionId: "provider-session-refresh-cold", + }, + }); + const agentManager = { + getAgent: vi.fn(() => currentAgent), + resumeAgentFromPersistence: vi.fn(async () => snapshot), + createAgent: vi.fn(), + reloadAgentSession: vi.fn(), + hydrateTimelineFromProvider: vi.fn(async () => { + currentAgent = snapshot; + }), + }; + const logger = { + child: () => logger, + info: vi.fn(), + warn: vi.fn(), + }; + const service = new ProviderHistoryCompatibilityService({ + agentManager: agentManager as any, + agentStorage: { + get: vi.fn(async () => record), + } as any, + logger: logger as any, + }); + + const result = await service.refreshAgent({ agentId: "agent-compat-refresh-cold" }); + + expect(agentManager.resumeAgentFromPersistence).toHaveBeenCalledWith( + { + provider: "codex", + sessionId: "provider-session-refresh-cold", + nativeHandle: undefined, + metadata: undefined, + }, + { + cwd: "/tmp/refresh-cold", + modeId: undefined, + model: "gpt-5.1-codex-mini", + thinkingOptionId: undefined, + title: undefined, + extra: undefined, + systemPrompt: undefined, + mcpServers: undefined, + }, + "agent-compat-refresh-cold", + { + createdAt: new Date("2026-03-25T00:00:00.000Z"), + updatedAt: new Date("2026-03-25T00:00:00.000Z"), + lastUserMessageAt: null, + labels: {}, + }, + ); + expect(agentManager.hydrateTimelineFromProvider).toHaveBeenCalledWith( + "agent-compat-refresh-cold", + ); + expect(result).toEqual(snapshot); + }); + + test("refreshAgent preserves the unloaded no-persistence error", async () => { + const service = new ProviderHistoryCompatibilityService({ + agentManager: { + getAgent: vi.fn(() => null), + resumeAgentFromPersistence: vi.fn(), + createAgent: vi.fn(), + reloadAgentSession: vi.fn(), + hydrateTimelineFromProvider: vi.fn(), + } as any, + agentStorage: { + get: async () => + createStoredAgentRecord({ + id: "agent-compat-no-persistence", + persistence: null, + }), + } as any, + logger: { + child: () => ({ + child: () => null, + info: vi.fn(), + warn: vi.fn(), + }), + info: vi.fn(), + warn: vi.fn(), + } as any, + }); + + await expect( + service.refreshAgent({ agentId: "agent-compat-no-persistence" }), + ).rejects.toThrow("Agent agent-compat-no-persistence cannot be refreshed because it lacks persistence"); + }); +}); diff --git a/packages/server/src/server/provider-history-compatibility-service.ts b/packages/server/src/server/provider-history-compatibility-service.ts new file mode 100644 index 0000000000..203b2a5b9b --- /dev/null +++ b/packages/server/src/server/provider-history-compatibility-service.ts @@ -0,0 +1,153 @@ +import type pino from "pino"; + +import type { ManagedAgent } from "./agent/agent-manager.js"; +import type { AgentManager } from "./agent/agent-manager.js"; +import type { AgentPersistenceHandle, AgentSessionConfig } from "./agent/agent-sdk-types.js"; +import type { AgentStorage } from "./agent/agent-storage.js"; +import { + buildConfigOverrides, + buildSessionConfig, + extractTimestamps, + toAgentPersistenceHandle, +} from "./persistence-hooks.js"; + +const pendingAgentBootstrapLoads = new Map>(); + +export type ProviderHistoryCompatibilityServiceOptions = { + agentManager: Pick< + AgentManager, + | "createAgent" + | "getAgent" + | "hydrateTimelineFromProvider" + | "reloadAgentSession" + | "resumeAgentFromPersistence" + >; + agentStorage: Pick; + logger: pino.Logger; +}; + +// Compatibility-only bridge for runtime paths that still need provider history +// replay during cold load, explicit resume, or refresh. +export class ProviderHistoryCompatibilityService { + private readonly agentManager: ProviderHistoryCompatibilityServiceOptions["agentManager"]; + private readonly agentStorage: ProviderHistoryCompatibilityServiceOptions["agentStorage"]; + private readonly logger: pino.Logger; + + constructor(options: ProviderHistoryCompatibilityServiceOptions) { + this.agentManager = options.agentManager; + this.agentStorage = options.agentStorage; + this.logger = options.logger.child({ component: "provider-history-compatibility" }); + } + + async ensureAgentLoaded(options: { agentId: string }): Promise { + const existing = this.agentManager.getAgent(options.agentId); + if (existing) { + return existing; + } + + const inflight = pendingAgentBootstrapLoads.get(options.agentId); + if (inflight) { + return inflight; + } + + const initPromise = this.loadStoredAgent(options); + pendingAgentBootstrapLoads.set(options.agentId, initPromise); + + try { + return await initPromise; + } finally { + const current = pendingAgentBootstrapLoads.get(options.agentId); + if (current === initPromise) { + pendingAgentBootstrapLoads.delete(options.agentId); + } + } + } + + async resumeAgent(options: { + handle: AgentPersistenceHandle; + overrides?: Partial; + }): Promise { + const snapshot = await this.agentManager.resumeAgentFromPersistence( + options.handle, + options.overrides, + ); + return this.hydrateCompatibilityTimeline({ agentId: snapshot.id, fallbackSnapshot: snapshot }); + } + + async refreshAgent(options: { agentId: string }): Promise { + const existing = this.agentManager.getAgent(options.agentId); + if (existing) { + const snapshot = existing.persistence + ? await this.agentManager.reloadAgentSession(options.agentId) + : existing; + return this.hydrateCompatibilityTimeline({ + agentId: options.agentId, + fallbackSnapshot: snapshot, + }); + } + + const record = await this.agentStorage.get(options.agentId); + if (!record) { + throw new Error(`Agent not found: ${options.agentId}`); + } + + const handle = toAgentPersistenceHandle(this.logger, record.persistence); + if (!handle) { + throw new Error(`Agent ${options.agentId} cannot be refreshed because it lacks persistence`); + } + + const snapshot = await this.agentManager.resumeAgentFromPersistence( + handle, + buildConfigOverrides(record), + options.agentId, + extractTimestamps(record), + ); + return this.hydrateCompatibilityTimeline({ + agentId: options.agentId, + fallbackSnapshot: snapshot, + }); + } + + private async loadStoredAgent(options: { agentId: string }): Promise { + const record = await this.agentStorage.get(options.agentId); + if (!record) { + throw new Error(`Agent not found: ${options.agentId}`); + } + + const handle = toAgentPersistenceHandle(this.logger, record.persistence); + let snapshot: ManagedAgent; + if (handle) { + snapshot = await this.agentManager.resumeAgentFromPersistence( + handle, + buildConfigOverrides(record), + options.agentId, + extractTimestamps(record), + ); + this.logger.info( + { agentId: options.agentId, provider: record.provider }, + "Agent resumed from persistence", + ); + } else { + snapshot = await this.agentManager.createAgent(buildSessionConfig(record), options.agentId, { + labels: record.labels, + }); + this.logger.info( + { agentId: options.agentId, provider: record.provider }, + "Agent created from stored config", + ); + } + + return this.hydrateCompatibilityTimeline({ + agentId: options.agentId, + fallbackSnapshot: snapshot, + }); + } + + private async hydrateCompatibilityTimeline(options: { + agentId: string; + fallbackSnapshot: ManagedAgent; + }): Promise { + await this.agentManager.hydrateTimelineFromProvider(options.agentId); + return this.agentManager.getAgent(options.agentId) ?? options.fallbackSnapshot; + } +} diff --git a/packages/server/src/server/session.provider-history-compatibility-ownership.test.ts b/packages/server/src/server/session.provider-history-compatibility-ownership.test.ts new file mode 100644 index 0000000000..f67d94e81e --- /dev/null +++ b/packages/server/src/server/session.provider-history-compatibility-ownership.test.ts @@ -0,0 +1,334 @@ +import { describe, expect, test, vi } from "vitest"; + +import { Session } from "./session.js"; + +function createStoredAgentRecord(overrides?: Partial>) { + return { + id: "agent-1", + provider: "codex", + cwd: "/tmp/project", + createdAt: "2026-03-24T00:00:00.000Z", + updatedAt: "2026-03-24T00:00:00.000Z", + title: null, + labels: {}, + lastStatus: "idle", + config: null, + persistence: { + provider: "codex", + sessionId: "provider-session-1", + }, + archivedAt: null, + requiresAttention: false, + attentionReason: null, + attentionTimestamp: null, + ...overrides, + }; +} + +function createCompatibilitySnapshot(overrides?: Partial>) { + return { + id: "agent-1", + provider: "codex", + cwd: "/tmp/project", + persistence: { + provider: "codex", + sessionId: "provider-session-1", + }, + ...overrides, + }; +} + +function createSessionForOwnershipTests(options?: { + providerHistoryCompatibilityService?: { + ensureAgentLoaded?: (options: { agentId: string }) => Promise; + resumeAgent?: (options: { + handle: { provider: string; sessionId: string }; + overrides?: Record; + }) => Promise; + refreshAgent?: (options: { agentId: string }) => Promise; + }; + storedRecord?: Record | null; + loadedAgent?: Record | null; + timelineRows?: Array<{ seq: number; item: Record; timestamp: Date }>; +}) { + const emitted: Array<{ type: string; payload: unknown }> = []; + const logger = { + child: () => logger, + trace: vi.fn(), + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; + + const agentManager = { + subscribe: () => () => {}, + listAgents: () => [], + getAgent: vi.fn(() => options?.loadedAgent ?? null), + createAgent: vi.fn(async () => { + throw new Error( + "Session should delegate unloaded bootstrap to ProviderHistoryCompatibilityService", + ); + }), + resumeAgentFromPersistence: vi.fn(async () => { + throw new Error( + "Session should delegate persistence resume to ProviderHistoryCompatibilityService", + ); + }), + reloadAgentSession: vi.fn(async () => { + throw new Error("Session should delegate refresh reload to ProviderHistoryCompatibilityService"); + }), + hydrateTimelineFromProvider: vi.fn(async () => { + throw new Error("Session should not call hydrateTimelineFromProvider directly"); + }), + fetchTimeline: vi.fn(() => ({ + rows: options?.timelineRows ?? [], + hasOlder: false, + hasNewer: false, + })), + recordUserMessage: vi.fn(), + waitForAgentRunStart: vi.fn(async () => undefined), + getTimeline: vi.fn(() => []), + }; + + const session = new Session({ + clientId: "test-client", + onMessage: (message) => emitted.push(message as any), + logger: logger as any, + downloadTokenStore: {} as any, + pushTokenStore: {} as any, + paseoHome: "/tmp/paseo-test", + agentManager: agentManager as any, + agentStorage: { + list: async () => (options?.storedRecord ? [options.storedRecord as any] : []), + get: async () => (options?.storedRecord as any) ?? null, + } as any, + projectRegistry: { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => [], + get: async () => null, + upsert: async () => {}, + archive: async () => {}, + remove: async () => {}, + } as any, + workspaceRegistry: { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => [], + get: async () => null, + upsert: async () => {}, + archive: async () => {}, + remove: async () => {}, + } as any, + createAgentMcpTransport: async () => { + throw new Error("not used"); + }, + stt: null, + tts: null, + terminalManager: null, + providerHistoryCompatibilityService: options?.providerHistoryCompatibilityService, + } as any) as any; + + return { session, emitted, agentManager }; +} + +describe("provider history compatibility ownership", () => { + test("fetch_agent_timeline_request delegates unloaded bootstrap through the compatibility seam", async () => { + const ensureAgentLoaded = vi.fn(async () => createCompatibilitySnapshot()); + const { session, emitted } = createSessionForOwnershipTests({ + storedRecord: createStoredAgentRecord(), + timelineRows: [ + { + seq: 1, + item: { type: "assistant_message", text: "rehydrated from provider history" }, + timestamp: new Date("2026-03-24T00:00:01.000Z"), + }, + ], + providerHistoryCompatibilityService: { + ensureAgentLoaded, + }, + }); + + session.buildAgentPayload = vi.fn(async () => ({ id: "agent-1" })); + + await session.handleMessage({ + type: "fetch_agent_timeline_request", + requestId: "req-fetch", + agentId: "agent-1", + }); + + expect(ensureAgentLoaded).toHaveBeenCalledWith({ agentId: "agent-1" }); + expect(emitted).toContainEqual({ + type: "fetch_agent_timeline_response", + payload: expect.objectContaining({ + requestId: "req-fetch", + agentId: "agent-1", + error: null, + entries: [ + expect.objectContaining({ + seq: 1, + }), + ], + }), + }); + }); + + test("send_agent_message_request delegates unloaded bootstrap before recording and streaming", async () => { + const ensureAgentLoaded = vi.fn(async () => createCompatibilitySnapshot()); + const { session, agentManager, emitted } = createSessionForOwnershipTests({ + storedRecord: createStoredAgentRecord(), + providerHistoryCompatibilityService: { + ensureAgentLoaded, + }, + }); + + session.resolveAgentIdentifier = vi.fn(async () => ({ ok: true, agentId: "agent-1" })); + session.unarchiveAgentState = vi.fn(async () => true); + session.buildAgentPrompt = vi.fn((text: string) => text); + session.startAgentStream = vi.fn(() => ({ ok: true })); + + await session.handleMessage({ + type: "send_agent_message_request", + requestId: "req-send", + agentId: "agent-1", + text: "hello", + images: [], + messageId: "msg-1", + }); + + expect(ensureAgentLoaded).toHaveBeenCalledWith({ agentId: "agent-1" }); + expect(ensureAgentLoaded.mock.invocationCallOrder[0]).toBeLessThan( + agentManager.recordUserMessage.mock.invocationCallOrder[0], + ); + expect(agentManager.recordUserMessage).toHaveBeenCalledWith("agent-1", "hello", { + messageId: "msg-1", + emitState: false, + }); + expect(session.startAgentStream).toHaveBeenCalledWith("agent-1", "hello"); + expect(emitted).toContainEqual({ + type: "send_agent_message_response", + payload: { + requestId: "req-send", + agentId: "agent-1", + accepted: true, + error: null, + }, + }); + }); + + test("resume_agent_request delegates persistence bootstrap through the compatibility seam", async () => { + const resumeAgent = vi.fn(async () => createCompatibilitySnapshot()); + const { session, emitted } = createSessionForOwnershipTests({ + providerHistoryCompatibilityService: { + resumeAgent, + }, + }); + + session.unarchiveAgentByHandle = vi.fn(async () => undefined); + session.unarchiveAgentState = vi.fn(async () => true); + session.forwardAgentUpdate = vi.fn(async () => undefined); + session.getAgentPayloadById = vi.fn(async () => ({ id: "agent-1" })); + + await session.handleMessage({ + type: "resume_agent_request", + requestId: "req-resume", + handle: { + provider: "codex", + sessionId: "provider-session-1", + }, + overrides: { + model: "gpt-5.4", + }, + }); + + expect(resumeAgent).toHaveBeenCalledWith({ + handle: { + provider: "codex", + sessionId: "provider-session-1", + }, + overrides: { + model: "gpt-5.4", + }, + }); + expect(emitted).toContainEqual({ + type: "status", + payload: expect.objectContaining({ + status: "agent_resumed", + requestId: "req-resume", + agentId: "agent-1", + }), + }); + }); + + test("refresh_agent_request delegates loaded persisted refresh through the compatibility seam", async () => { + const refreshAgent = vi.fn(async () => + createCompatibilitySnapshot({ + persistence: { + provider: "codex", + sessionId: "provider-session-1", + }, + }), + ); + const { session, emitted } = createSessionForOwnershipTests({ + loadedAgent: createCompatibilitySnapshot(), + providerHistoryCompatibilityService: { + refreshAgent, + }, + }); + + session.unarchiveAgentState = vi.fn(async () => true); + session.interruptAgentIfRunning = vi.fn(async () => undefined); + session.forwardAgentUpdate = vi.fn(async () => undefined); + + await session.handleMessage({ + type: "refresh_agent_request", + requestId: "req-refresh-loaded", + agentId: "agent-1", + }); + + expect(session.interruptAgentIfRunning).toHaveBeenCalledWith("agent-1"); + expect(refreshAgent).toHaveBeenCalledWith({ agentId: "agent-1" }); + expect(emitted).toContainEqual({ + type: "status", + payload: { + status: "agent_refreshed", + requestId: "req-refresh-loaded", + agentId: "agent-1", + timelineSize: 0, + }, + }); + }); + + test("refresh_agent_request delegates unloaded persisted refresh through the compatibility seam", async () => { + const refreshAgent = vi.fn(async () => createCompatibilitySnapshot()); + const { session, emitted } = createSessionForOwnershipTests({ + storedRecord: createStoredAgentRecord(), + providerHistoryCompatibilityService: { + refreshAgent, + }, + }); + + session.unarchiveAgentState = vi.fn(async () => true); + session.interruptAgentIfRunning = vi.fn(async () => undefined); + session.forwardAgentUpdate = vi.fn(async () => undefined); + + await session.handleMessage({ + type: "refresh_agent_request", + requestId: "req-refresh-unloaded", + agentId: "agent-1", + }); + + expect(session.interruptAgentIfRunning).not.toHaveBeenCalled(); + expect(refreshAgent).toHaveBeenCalledWith({ agentId: "agent-1" }); + expect(emitted).toContainEqual({ + type: "status", + payload: { + status: "agent_refreshed", + requestId: "req-refresh-unloaded", + agentId: "agent-1", + timelineSize: 0, + }, + }); + }); +}); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index ef5beeb5e4..c3198233bc 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -54,9 +54,7 @@ import { type VoiceTurnController, } from "./voice/voice-turn-controller.js"; import { - buildConfigOverrides, - buildSessionConfig, - extractTimestamps, + toAgentPersistenceHandle, } from "./persistence-hooks.js"; import { experimental_createMCPClient } from "ai"; import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; @@ -109,6 +107,7 @@ import type { WorkspaceRegistry, } from "./workspace-registry.js"; import { WorkspaceReconciliationService } from "./workspace-reconciliation-service.js"; +import { ProviderHistoryCompatibilityService } from "./provider-history-compatibility-service.js"; import { buildVoiceAgentMcpServerConfig, buildVoiceModeSystemPrompt, @@ -172,7 +171,6 @@ const READ_ONLY_GIT_ENV: NodeJS.ProcessEnv = { ...process.env, GIT_OPTIONAL_LOCKS: "0", }; -const pendingAgentInitializations = new Map>(); const DEFAULT_AGENT_PROVIDER = AGENT_PROVIDER_IDS[0]; const CHECKOUT_DIFF_WATCH_DEBOUNCE_MS = 150; const CHECKOUT_DIFF_FALLBACK_REFRESH_MS = 5_000; @@ -394,6 +392,7 @@ export type SessionOptions = { projectRegistry: ProjectRegistry; workspaceRegistry: WorkspaceRegistry; workspaceReconciliationService?: WorkspaceReconciliationService; + providerHistoryCompatibilityService?: ProviderHistoryCompatibilityService; createAgentMcpTransport: AgentMcpTransportFactory; stt: Resolvable; tts: Resolvable; @@ -502,30 +501,6 @@ function coerceAgentProvider(logger: pino.Logger, value: string, agentId?: strin return DEFAULT_AGENT_PROVIDER; } -function toAgentPersistenceHandle( - logger: pino.Logger, - handle: StoredAgentRecord["persistence"], -): AgentPersistenceHandle | null { - if (!handle) { - return null; - } - const provider = handle.provider; - if (!isValidAgentProvider(provider)) { - logger.warn({ provider }, `Ignoring persistence handle with unknown provider '${provider}'`); - return null; - } - if (!handle.sessionId) { - logger.warn("Ignoring persistence handle missing sessionId"); - return null; - } - return { - provider, - sessionId: handle.sessionId, - nativeHandle: handle.nativeHandle, - metadata: handle.metadata, - } satisfies AgentPersistenceHandle; -} - /** * Session represents a single connected client session. * It owns all state management, orchestration logic, and message processing. @@ -578,6 +553,7 @@ export class Session { private readonly projectRegistry: ProjectRegistry; private readonly workspaceRegistry: WorkspaceRegistry; private readonly workspaceReconciliationService: WorkspaceReconciliationService; + private readonly providerHistoryCompatibilityService: ProviderHistoryCompatibilityService; private readonly createAgentMcpTransport: AgentMcpTransportFactory; private readonly downloadTokenStore: DownloadTokenStore; private readonly pushTokenStore: PushTokenStore; @@ -639,6 +615,7 @@ export class Session { projectRegistry, workspaceRegistry, workspaceReconciliationService, + providerHistoryCompatibilityService, createAgentMcpTransport, stt, tts, @@ -657,6 +634,11 @@ export class Session { this.downloadTokenStore = downloadTokenStore; this.pushTokenStore = pushTokenStore; this.paseoHome = paseoHome; + this.sessionLogger = logger.child({ + module: "session", + clientId: this.clientId, + sessionId: this.sessionId, + }); this.agentManager = agentManager; this.agentStorage = agentStorage; this.projectRegistry = projectRegistry; @@ -672,6 +654,13 @@ export class Session { this.syncWorkspaceGitWatchTarget(cwd, options), removeWorkspaceGitWatchTarget: async (cwd) => this.removeWorkspaceGitWatchTarget(cwd), }); + this.providerHistoryCompatibilityService = + providerHistoryCompatibilityService ?? + new ProviderHistoryCompatibilityService({ + agentManager: this.agentManager, + agentStorage: this.agentStorage, + logger: this.sessionLogger, + }); this.createAgentMcpTransport = createAgentMcpTransport; this.terminalManager = terminalManager; if (this.terminalManager) { @@ -699,11 +688,6 @@ export class Session { this.getSpeechReadiness = dictation?.getSpeechReadiness; this.agentProviderRuntimeSettings = agentProviderRuntimeSettings; this.abortController = new AbortController(); - this.sessionLogger = logger.child({ - module: "session", - clientId: this.clientId, - sessionId: this.sessionId, - }); this.providerRegistry = buildProviderRegistry(this.sessionLogger, { runtimeSettings: this.agentProviderRuntimeSettings, }); @@ -1097,58 +1081,7 @@ export class Session { } private async ensureAgentLoaded(agentId: string): Promise { - const existing = this.agentManager.getAgent(agentId); - if (existing) { - return existing; - } - - const inflight = pendingAgentInitializations.get(agentId); - if (inflight) { - return inflight; - } - - const initPromise = (async () => { - const record = await this.agentStorage.get(agentId); - if (!record) { - throw new Error(`Agent not found: ${agentId}`); - } - - const handle = toAgentPersistenceHandle(this.sessionLogger, record.persistence); - let snapshot: ManagedAgent; - if (handle) { - snapshot = await this.agentManager.resumeAgentFromPersistence( - handle, - buildConfigOverrides(record), - agentId, - extractTimestamps(record), - ); - this.sessionLogger.info( - { agentId, provider: record.provider }, - "Agent resumed from persistence", - ); - } else { - const config = buildSessionConfig(record); - snapshot = await this.agentManager.createAgent(config, agentId, { labels: record.labels }); - this.sessionLogger.info( - { agentId, provider: record.provider }, - "Agent created from stored config", - ); - } - - await this.agentManager.hydrateTimelineFromProvider(agentId); - return this.agentManager.getAgent(agentId) ?? snapshot; - })(); - - pendingAgentInitializations.set(agentId, initPromise); - - try { - return await initPromise; - } finally { - const current = pendingAgentInitializations.get(agentId); - if (current === initPromise) { - pendingAgentInitializations.delete(agentId); - } - } + return this.providerHistoryCompatibilityService.ensureAgentLoaded({ agentId }); } private matchesAgentFilter(options: { @@ -2547,9 +2480,11 @@ export class Session { ); try { await this.unarchiveAgentByHandle(handle); - const snapshot = await this.agentManager.resumeAgentFromPersistence(handle, overrides); + const snapshot = await this.providerHistoryCompatibilityService.resumeAgent({ + handle, + overrides, + }); await this.unarchiveAgentState(snapshot.id); - await this.agentManager.hydrateTimelineFromProvider(snapshot.id); await this.forwardAgentUpdate(snapshot); const timelineSize = this.agentManager.getTimeline(snapshot.id).length; if (requestId) { @@ -2590,32 +2525,10 @@ export class Session { try { await this.unarchiveAgentState(agentId); - let snapshot: ManagedAgent; - const existing = this.agentManager.getAgent(agentId); - if (existing) { + if (this.agentManager.getAgent(agentId)) { await this.interruptAgentIfRunning(agentId); - if (existing.persistence) { - snapshot = await this.agentManager.reloadAgentSession(agentId); - } else { - snapshot = existing; - } - } else { - const record = await this.agentStorage.get(agentId); - if (!record) { - throw new Error(`Agent not found: ${agentId}`); - } - const handle = toAgentPersistenceHandle(this.sessionLogger, record.persistence); - if (!handle) { - throw new Error(`Agent ${agentId} cannot be refreshed because it lacks persistence`); - } - snapshot = await this.agentManager.resumeAgentFromPersistence( - handle, - buildConfigOverrides(record), - agentId, - extractTimestamps(record), - ); } - await this.agentManager.hydrateTimelineFromProvider(agentId); + const snapshot = await this.providerHistoryCompatibilityService.refreshAgent({ agentId }); await this.forwardAgentUpdate(snapshot); const timelineSize = this.agentManager.getTimeline(agentId).length; if (requestId) {