Phase 1c: isolate provider history compatibility seam

This commit is contained in:
Mohamed Boudra
2026-03-25 20:39:07 +07:00
parent b3cdfb203b
commit ae4f9e0410
6 changed files with 948 additions and 111 deletions
@@ -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;
@@ -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);
});
});
@@ -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>): 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<T>() {
let resolve!: (value: T) => void;
let reject!: (error: unknown) => void;
const promise = new Promise<T>((res, rej) => {
resolve = res;
reject = rej;
});
return { promise, resolve, reject };
}
function createCompatibilitySnapshot(overrides?: Partial<Record<string, unknown>>) {
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<any>();
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");
});
});
@@ -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<string, Promise<ManagedAgent>>();
export type ProviderHistoryCompatibilityServiceOptions = {
agentManager: Pick<
AgentManager,
| "createAgent"
| "getAgent"
| "hydrateTimelineFromProvider"
| "reloadAgentSession"
| "resumeAgentFromPersistence"
>;
agentStorage: Pick<AgentStorage, "get">;
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<ManagedAgent> {
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<AgentSessionConfig>;
}): Promise<ManagedAgent> {
const snapshot = await this.agentManager.resumeAgentFromPersistence(
options.handle,
options.overrides,
);
return this.hydrateCompatibilityTimeline({ agentId: snapshot.id, fallbackSnapshot: snapshot });
}
async refreshAgent(options: { agentId: string }): Promise<ManagedAgent> {
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<ManagedAgent> {
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<ManagedAgent> {
await this.agentManager.hydrateTimelineFromProvider(options.agentId);
return this.agentManager.getAgent(options.agentId) ?? options.fallbackSnapshot;
}
}
@@ -0,0 +1,334 @@
import { describe, expect, test, vi } from "vitest";
import { Session } from "./session.js";
function createStoredAgentRecord(overrides?: Partial<Record<string, unknown>>) {
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<Record<string, unknown>>) {
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<any>;
resumeAgent?: (options: {
handle: { provider: string; sessionId: string };
overrides?: Record<string, unknown>;
}) => Promise<any>;
refreshAgent?: (options: { agentId: string }) => Promise<any>;
};
storedRecord?: Record<string, unknown> | null;
loadedAgent?: Record<string, unknown> | null;
timelineRows?: Array<{ seq: number; item: Record<string, unknown>; 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,
},
});
});
});
+24 -111
View File
@@ -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<string, Promise<ManagedAgent>>();
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<SpeechToTextProvider | null>;
tts: Resolvable<TextToSpeechProvider | null>;
@@ -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<ManagedAgent> {
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) {