diff --git a/packages/app/e2e/browser/agent-message-rewind.spec.ts b/packages/app/e2e/browser/agent-message-rewind.spec.ts index 14aa4ca95f..da2a54a5c7 100644 --- a/packages/app/e2e/browser/agent-message-rewind.spec.ts +++ b/packages/app/e2e/browser/agent-message-rewind.spec.ts @@ -11,7 +11,13 @@ import { import { expectNearBottom, scrollChatAwayFromBottom } from "../support/helpers/agent-bottom-anchor"; import { holdRewindCompletion } from "../support/helpers/agent-timeline-gate"; import { openAgentRoute, seedMockAgentWorkspace } from "../support/helpers/mock-agent"; -import { primeAdditionalPage, resetSeededPageState } from "../support/helpers/archive-tab"; +import { userScrollsTimelineToHistoryStart } from "../support/helpers/timeline-pagination"; +import { + closeWorkspaceAgentTab, + fetchAgentArchivedAt, + primeAdditionalPage, + resetSeededPageState, +} from "../support/helpers/archive-tab"; async function completeSubmittedTurn( page: Page, @@ -64,6 +70,47 @@ async function expectPrompts(page: Page, prompts: string[]): Promise { await expect(messages).toContainText(prompts); } +async function reloadAgentFromTabMenu(page: Page, agentId: string): Promise { + const tabId = `workspace-tab-agent_${agentId}`; + await page.getByTestId(tabId).filter({ visible: true }).first().click({ button: "right" }); + await page.getByTestId(`workspace-tab-context-agent_${agentId}-reload-agent`).click(); +} + +async function archiveAgentByClosingTab( + page: Page, + agent: Awaited>, +): Promise { + await closeWorkspaceAgentTab(page, agent.agentId); + await expect.poll(() => fetchAgentArchivedAt(agent.client, agent.agentId)).not.toBeNull(); +} + +async function unarchiveAgent(page: Page): Promise { + await page.getByRole("button", { name: "Unarchive", exact: true }).click(); + await expect(page.getByText("This agent is archived", { exact: true })).toHaveCount(0, { + timeout: 30_000, + }); +} + +async function expectToast(page: Page, message: string): Promise { + await expect(page.getByTestId("app-toast-message")).toHaveText(message, { timeout: 30_000 }); +} + +// The mock provider's resumed session holds no history, so it never matches what Paseo shows. +const UNSYNCED_RELOAD = + "Reloaded agent, but the provider's history doesn't match this chat, so it wasn't synced"; + +/** The prompts of the whole conversation, after scrolling up through its older pages. */ +async function expectConversationPrompts(page: Page, prompts: string[]): Promise { + const messages = page.getByTestId("user-message"); + await expect + .poll(async () => { + await userScrollsTimelineToHistoryStart(page); + return await messages.count(); + }) + .toBe(prompts.length); + await expect(messages).toContainText(prompts); +} + // A row React remounts loses this attribute; one that stays mounted keeps it. async function markMounted(row: Locator): Promise { await row.evaluate((element) => element.setAttribute("data-e2e-mounted", "before-rewind")); @@ -143,6 +190,62 @@ test.describe("Agent message rewind", () => { } }); + test("keeps the rewound conversation through Reload agent, archive and unarchive", async ({ + page, + }) => { + test.setTimeout(150_000); + const keptPrompt = "Keep this turn through reload and archive."; + const rewoundPrompt = "Rewind this turn before reloading."; + const nextPrompt = "Continue after the rewind, then reload."; + const agent = await seedMockAgentWorkspace({ + repoPrefix: "message-rewind-reload-archive-", + title: "Message rewind reload archive", + model: "ten-second-stream", + }); + + try { + await openAgentRoute(page, agent); + await expectComposerVisible(page); + await completeSubmittedTurn(page, agent, keptPrompt); + const rewound = await completeSubmittedTurn(page, agent, rewoundPrompt); + await rewindConversation(page, rewound, rewoundPrompt); + await expectPrompts(page, [keptPrompt]); + await completeSubmittedTurn(page, agent, nextPrompt); + await expectPrompts(page, [keptPrompt, nextPrompt]); + + await test.step("Reload agent keeps the chat and says the provider's history was not synced", async () => { + await reloadAgentFromTabMenu(page, agent.agentId); + await expectToast(page, UNSYNCED_RELOAD); + await expectPrompts(page, [keptPrompt, nextPrompt]); + await expectComposerVisible(page); + }); + + await test.step("the archived chat shows the rewound conversation", async () => { + await archiveAgentByClosingTab(page, agent); + await openAgentRoute(page, agent); + await expect(page.getByText("This agent is archived", { exact: true })).toBeVisible({ + timeout: 30_000, + }); + await expectConversationPrompts(page, [keptPrompt, nextPrompt]); + }); + + await test.step("unarchiving keeps the rewound conversation", async () => { + await unarchiveAgent(page); + await expectToast(page, UNSYNCED_RELOAD); + await expectComposerVisible(page); + await expectConversationPrompts(page, [keptPrompt, nextPrompt]); + }); + + await test.step("a reconnecting client loads the same conversation", async () => { + await page.reload(); + await expectComposerVisible(page); + await expectConversationPrompts(page, [keptPrompt, nextPrompt]); + }); + } finally { + await agent.cleanup(); + } + }); + test("keeps a pre-acknowledgement turn running after rewind is rejected", async ({ page }) => { test.setTimeout(90_000); const prompt = "Delay synthetic user message by 2000ms."; diff --git a/packages/app/src/components/archived-agent-callout.tsx b/packages/app/src/components/archived-agent-callout.tsx index b5f6a3ea54..ed29423173 100644 --- a/packages/app/src/components/archived-agent-callout.tsx +++ b/packages/app/src/components/archived-agent-callout.tsx @@ -6,6 +6,7 @@ import { FOOTER_HEIGHT } from "@/constants/layout"; import { useHostRuntimeClient, useHostRuntimeIsConnected } from "@/runtime/host-runtime"; import { KeyboardTranslateView } from "@/keyboard/shift"; import { Button } from "@/components/ui/button"; +import { useToast } from "@/contexts/toast-context"; import type { Theme } from "@/styles/theme"; import { toErrorMessage } from "@/utils/error-messages"; @@ -18,6 +19,7 @@ export function ArchivedAgentCallout({ serverId, agentId }: ArchivedAgentCallout const { t } = useTranslation(); const client = useHostRuntimeClient(serverId); const isConnected = useHostRuntimeIsConnected(serverId); + const toast = useToast(); const [isUnarchiving, setIsUnarchiving] = useState(false); const [unarchiveError, setUnarchiveError] = useState(null); @@ -26,12 +28,15 @@ export function ArchivedAgentCallout({ serverId, agentId }: ArchivedAgentCallout setIsUnarchiving(true); setUnarchiveError(null); try { - await client.refreshAgent(agentId); + const refreshed = await client.refreshAgent(agentId); + if (refreshed.historyDiverged) { + toast.show(t("workspace.tabs.toasts.reloadedAgentHistoryDiverged"), { variant: "warning" }); + } } catch (error) { setUnarchiveError(toErrorMessage(error)); setIsUnarchiving(false); } - }, [client, isConnected, isUnarchiving, agentId]); + }, [client, isConnected, isUnarchiving, agentId, toast, t]); return ( diff --git a/packages/app/src/i18n/resources/ar.ts b/packages/app/src/i18n/resources/ar.ts index 8d7c2a2950..36803d4df7 100644 --- a/packages/app/src/i18n/resources/ar.ts +++ b/packages/app/src/i18n/resources/ar.ts @@ -699,6 +699,8 @@ export const ar: TranslationResources = { resumeCommandUnavailable: "أمر الاستئناف غير متوفر", reloadingAgent: "وكيل إعادة التحميل...", reloadedAgent: "وكيل إعادة تحميل", + reloadedAgentHistoryDiverged: + "تمت إعادة تحميل الوكيل، لكن سجل المزوّد لا يطابق هذه المحادثة، لذا لم تتم مزامنته", failedToReloadAgent: "فشل في إعادة تحميل الوكيل", failedToCloseAgent: "فشل في إغلاق الوكيل", }, diff --git a/packages/app/src/i18n/resources/en.ts b/packages/app/src/i18n/resources/en.ts index 1f61a83f12..d3ca03acbf 100644 --- a/packages/app/src/i18n/resources/en.ts +++ b/packages/app/src/i18n/resources/en.ts @@ -696,6 +696,8 @@ export const en = { resumeCommandUnavailable: "Resume command not available", reloadingAgent: "Reloading agent...", reloadedAgent: "Reloaded agent", + reloadedAgentHistoryDiverged: + "Reloaded agent, but the provider's history doesn't match this chat, so it wasn't synced", failedToReloadAgent: "Failed to reload agent", failedToCloseAgent: "Failed to close agent", }, diff --git a/packages/app/src/i18n/resources/es.ts b/packages/app/src/i18n/resources/es.ts index 7e5a1d41e6..fa7291b298 100644 --- a/packages/app/src/i18n/resources/es.ts +++ b/packages/app/src/i18n/resources/es.ts @@ -706,6 +706,8 @@ export const es: TranslationResources = { resumeCommandUnavailable: "Comando de reanudación no disponible", reloadingAgent: "Agente de recarga...", reloadedAgent: "Agente recargado", + reloadedAgentHistoryDiverged: + "Agente recargado, pero el historial del proveedor no coincide con este chat, así que no se sincronizó", failedToReloadAgent: "No se pudo recargar el agente", failedToCloseAgent: "No se pudo cerrar el agente", }, diff --git a/packages/app/src/i18n/resources/fr.ts b/packages/app/src/i18n/resources/fr.ts index 551f4a4158..46f3d40117 100644 --- a/packages/app/src/i18n/resources/fr.ts +++ b/packages/app/src/i18n/resources/fr.ts @@ -706,6 +706,8 @@ export const fr: TranslationResources = { resumeCommandUnavailable: "Commande de reprise indisponible", reloadingAgent: "Rechargement de l’agent…", reloadedAgent: "Agent rechargé", + reloadedAgentHistoryDiverged: + "Agent rechargé, mais l’historique du fournisseur ne correspond pas à cette conversation, il n’a donc pas été synchronisé", failedToReloadAgent: "Impossible de recharger l’agent", failedToCloseAgent: "Impossible de fermer l’agent", }, diff --git a/packages/app/src/i18n/resources/ja.ts b/packages/app/src/i18n/resources/ja.ts index 4b1b8b6f64..7d6d4fe337 100644 --- a/packages/app/src/i18n/resources/ja.ts +++ b/packages/app/src/i18n/resources/ja.ts @@ -706,6 +706,8 @@ export const ja: TranslationResources = { resumeCommandUnavailable: "再開コマンドが利用できません", reloadingAgent: "エージェントを再読み込み中...", reloadedAgent: "エージェントを再読み込みしました", + reloadedAgentHistoryDiverged: + "エージェントを再読み込みしましたが、プロバイダーの履歴がこのチャットと一致しないため同期しませんでした", failedToReloadAgent: "エージェントの再読み込みに失敗しました", failedToCloseAgent: "エージェントを閉じられませんでした", }, diff --git a/packages/app/src/i18n/resources/ko.ts b/packages/app/src/i18n/resources/ko.ts index b9e33b4eba..128484861a 100644 --- a/packages/app/src/i18n/resources/ko.ts +++ b/packages/app/src/i18n/resources/ko.ts @@ -701,6 +701,8 @@ export const ko: TranslationResources = { resumeCommandUnavailable: "재개 명령을 사용할 수 없습니다", reloadingAgent: "에이전트 다시 로드 중...", reloadedAgent: "에이전트를 다시 로드했습니다", + reloadedAgentHistoryDiverged: + "에이전트를 다시 로드했지만 제공자 기록이 이 채팅과 일치하지 않아 동기화하지 않았습니다", failedToReloadAgent: "에이전트를 다시 로드하지 못했습니다", failedToCloseAgent: "에이전트를 닫지 못했습니다", }, diff --git a/packages/app/src/i18n/resources/pt-BR.ts b/packages/app/src/i18n/resources/pt-BR.ts index 2fa6c30c8e..54bd7b19ac 100644 --- a/packages/app/src/i18n/resources/pt-BR.ts +++ b/packages/app/src/i18n/resources/pt-BR.ts @@ -704,6 +704,8 @@ export const ptBR: TranslationResources = { resumeCommandUnavailable: "Comando de retomada indisponível", reloadingAgent: "Recarregando agente...", reloadedAgent: "Agente recarregado", + reloadedAgentHistoryDiverged: + "Agente recarregado, mas o histórico do provedor não corresponde a este chat, então não foi sincronizado", failedToReloadAgent: "Falha ao recarregar agente", failedToCloseAgent: "Falha ao fechar agente", }, diff --git a/packages/app/src/i18n/resources/ru.ts b/packages/app/src/i18n/resources/ru.ts index 9a6eb5b0e2..750a74e8d0 100644 --- a/packages/app/src/i18n/resources/ru.ts +++ b/packages/app/src/i18n/resources/ru.ts @@ -706,6 +706,8 @@ export const ru: TranslationResources = { resumeCommandUnavailable: "Команда возобновления недоступна", reloadingAgent: "Перезагрузка агента...", reloadedAgent: "Агент перезагружен", + reloadedAgentHistoryDiverged: + "Агент перезагружен, но история провайдера не совпадает с этим чатом, поэтому она не синхронизирована", failedToReloadAgent: "Не удалось перезагрузить агента", failedToCloseAgent: "Не удалось закрыть агента", }, diff --git a/packages/app/src/i18n/resources/zh-CN.ts b/packages/app/src/i18n/resources/zh-CN.ts index 553d8212b4..674bdcf1a9 100644 --- a/packages/app/src/i18n/resources/zh-CN.ts +++ b/packages/app/src/i18n/resources/zh-CN.ts @@ -699,6 +699,8 @@ export const zhCN: TranslationResources = { resumeCommandUnavailable: "恢复命令不可用", reloadingAgent: "正在重新加载 Agent...", reloadedAgent: "已重新加载 Agent", + reloadedAgentHistoryDiverged: + "已重新加载 Agent，但提供方的历史记录与此对话不一致，因此未同步", failedToReloadAgent: "重新加载 Agent 失败", failedToCloseAgent: "关闭 Agent 失败", }, diff --git a/packages/app/src/screens/workspace/workspace-screen.tsx b/packages/app/src/screens/workspace/workspace-screen.tsx index dd806e1fa1..508156222b 100644 --- a/packages/app/src/screens/workspace/workspace-screen.tsx +++ b/packages/app/src/screens/workspace/workspace-screen.tsx @@ -2162,11 +2162,10 @@ function WorkspaceScreenContent({ toast.show(t("workspace.tabs.toasts.reloadingAgent"), { durationMs: null }); try { - await client.refreshAgent(agentId); - // Send the existing cursor so the server detects the new epoch and - // returns reset:true. Without a cursor, the server returns reset:false - // and the client takes the incremental path, where new-epoch rows are - // dropped against the stale cursor. + const refreshed = await client.refreshAgent(agentId); + // Send the existing cursor: Reload agent keeps the history, so the server + // returns only the rows it added. When the history was imported again + // under a new epoch, the cursor reads as stale and the server resets. const sessionState = useSessionStore.getState().sessions[normalizedServerId]; const currentCursor = sessionState?.agentTimelineCursor.get(agentId); await getHostRuntimeStore().fetchAgentTimeline(normalizedServerId, agentId, { @@ -2175,7 +2174,13 @@ function WorkspaceScreenContent({ ? { cursor: { epoch: currentCursor.epoch, seq: currentCursor.endSeq } } : {}), }); - toast.show(t("workspace.tabs.toasts.reloadedAgent"), { variant: "success" }); + if (refreshed.historyDiverged) { + toast.show(t("workspace.tabs.toasts.reloadedAgentHistoryDiverged"), { + variant: "warning", + }); + } else { + toast.show(t("workspace.tabs.toasts.reloadedAgent"), { variant: "success" }); + } } catch (error) { toast.error( error instanceof Error ? error.message : t("workspace.tabs.toasts.failedToReloadAgent"), diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 959a32e3a1..a9d1e236b6 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -3938,6 +3938,8 @@ export const AgentResumedStatusPayloadSchema = z export const AgentRefreshedStatusPayloadSchema = z .object({ status: z.literal("agent_refreshed"), + /** The provider's history differs from the kept history, which Reload agent did not change. */ + historyDiverged: z.boolean().optional(), }) .extend(AgentStatusWithTimelineSchema.shape); diff --git a/packages/server/src/server/agent/agent-loading.test.ts b/packages/server/src/server/agent/agent-loading.test.ts index dbae7adb46..9df22ba26b 100644 --- a/packages/server/src/server/agent/agent-loading.test.ts +++ b/packages/server/src/server/agent/agent-loading.test.ts @@ -5,8 +5,7 @@ import { expect, test } from "vitest"; import { createTestLogger } from "../../test-utils/test-logger.js"; import { AgentManager } from "./agent-manager.js"; -import { ensureAgentLoaded } from "./agent-loading.js"; -import { startAgentRun } from "./agent-prompt.js"; +import { AgentNotFoundError, ensureAgentLoaded, loadAgentHistory } from "./agent-loading.js"; import { AgentStorage } from "./agent-storage.js"; import type { AgentClient, @@ -160,13 +159,8 @@ test("loads an archived agent's history after its working directory is removed", const agent = await manager.createAgent({ provider: "codex", cwd: worktree }, agentId, { workspaceId: "workspace-worktree", }); - await startAgentRun(manager, agent.id, "what did you change", logger, {}); - // Dispatching a run does not finish it: the provider appends the reply to its - // history afterwards, and the turn is finalized only once that append lands. - // Archive after the turn is finalized so the transcript this test reads back is - // already on disk when the worktree goes away. - const finished = await manager.waitForAgentEvent(agent.id); - expect(finished.status).toBe("idle"); + // Archive once the turn has finished, so the reply is part of the history archive keeps. + await manager.runAgent(agent.id, "what did you change"); await manager.archiveAgent(agent.id); await manager.closeAgent(agent.id); await manager.flush(); @@ -183,8 +177,8 @@ test("loads an archived agent's history after its working directory is removed", }); expect(loaded.id).toBe(agentId); - // The transcript is replayed from the provider's persisted history, so the reply the - // agent gave before the worktree went away is still readable. + // Archive kept the transcript, so the reply the agent gave before the worktree went away + // is still readable from the resumed agent. const replies = (await manager.getTimeline(agentId)).filter( (item) => item.type === "assistant_message", ); @@ -199,3 +193,106 @@ test("loads an archived agent's history after its working directory is removed", await rm(root, { recursive: true, force: true }); } }); + +test("reads stored history without starting or requiring the provider, and imports history that is not stored", async () => { + const root = await mkdtemp(path.join(tmpdir(), "agent-loading-history-")); + const logger = createTestLogger(); + const storage = new AgentStorage(path.join(root, "agents"), logger); + const baseClient = createTestAgentClients().codex; + if (!baseClient) { + throw new Error("expected Codex test client"); + } + let resumes = 0; + const client: AgentClient = { + provider: baseClient.provider, + capabilities: baseClient.capabilities, + createSession: async (config: AgentSessionConfig, launchContext?: AgentLaunchContext) => + await baseClient.createSession(config, launchContext), + resumeSession: async ( + handle: AgentPersistenceHandle, + overrides?: Partial, + launchContext?: AgentLaunchContext, + ) => { + resumes += 1; + return await baseClient.resumeSession(handle, overrides, launchContext); + }, + fetchCatalog: async (options) => await baseClient.fetchCatalog(options), + isAvailable: async () => await baseClient.isAvailable(), + }; + const agentId = "00000000-0000-4000-8000-000000000601"; + const managers: AgentManager[] = []; + function startDaemon(): AgentManager { + const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger }); + managers.push(manager); + return manager; + } + + try { + const manager = startDaemon(); + const agent = await manager.createAgent({ provider: "codex", cwd: root }, agentId, { + workspaceId: "workspace-a", + }); + await manager.runAgent(agent.id, "hello"); + const transcript = await manager.getTimeline(agent.id); + expect(transcript.map((item) => item.type)).toContain("assistant_message"); + await manager.closeAgent(agent.id); + await manager.flush(); + await storage.flush(); + + const stored = await loadAgentHistory(agentId, { + agentManager: manager, + agentStorage: storage, + logger, + }); + expect(stored).toMatchObject({ kind: "stored", record: { id: agentId } }); + expect(await manager.getTimeline(agentId)).toEqual(transcript); + expect(manager.getAgent(agentId)).toBeNull(); + // Stored history needs no provider, so it still reads once the provider is removed. + await expect( + loadAgentHistory(agentId, { + agentManager: manager, + agentStorage: storage, + logger, + validProviders: [], + }), + ).resolves.toMatchObject({ kind: "stored", record: { id: agentId } }); + expect(resumes).toBe(0); + + // A restarted daemon has no history in memory; the provider's own history is imported, + // which needs the provider. + const restarted = startDaemon(); + await expect( + loadAgentHistory(agentId, { + agentManager: restarted, + agentStorage: storage, + logger, + validProviders: [], + }), + ).rejects.toThrow("references unavailable provider 'codex'"); + const loaded = await loadAgentHistory(agentId, { + agentManager: restarted, + agentStorage: storage, + logger, + }); + expect(loaded).toMatchObject({ kind: "live", agent: { id: agentId } }); + expect(resumes).toBe(1); + expect((await restarted.getTimeline(agentId)).map((item) => item.type)).toEqual( + transcript.map((item) => item.type), + ); + + await expect( + loadAgentHistory("00000000-0000-4000-8000-000000000602", { + agentManager: restarted, + agentStorage: storage, + logger, + }), + ).rejects.toBeInstanceOf(AgentNotFoundError); + } finally { + for (const manager of managers) { + await manager.closeAgent(agentId).catch(() => undefined); + await manager.flush().catch(() => undefined); + } + await storage.flush().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/packages/server/src/server/agent/agent-loading.ts b/packages/server/src/server/agent/agent-loading.ts index ba444a394d..245da8840e 100644 --- a/packages/server/src/server/agent/agent-loading.ts +++ b/packages/server/src/server/agent/agent-loading.ts @@ -2,7 +2,7 @@ import type { Logger } from "pino"; import type { AgentProvider } from "./agent-sdk-types.js"; import type { AgentManager, ManagedAgent } from "./agent-manager.js"; -import type { AgentStorage } from "./agent-storage.js"; +import type { AgentStorage, StoredAgentRecord } from "./agent-storage.js"; import { buildConfigOverrides, buildSessionConfig, @@ -19,20 +19,11 @@ export class AgentNotFoundError extends Error { } } -interface PendingAgentInitialization { - promise: Promise; - options: { broadcastTimeline: boolean }; -} - -const pendingAgentInitializations = new Map(); +const pendingAgentInitializations = new Map>(); export type AgentLoaderManager = Pick< AgentManager, - | "createAgent" - | "getAgent" - | "getRegisteredProviderIds" - | "hydrateTimelineFromProvider" - | "resumeAgentFromPersistence" + "createAgent" | "getAgent" | "getRegisteredProviderIds" | "resumeAgentFromPersistence" > & Partial>; @@ -40,10 +31,53 @@ export interface EnsureAgentLoadedDeps { agentManager: AgentLoaderManager; agentStorage: AgentStorage; validProviders?: Iterable; - broadcastTimeline?: boolean; logger: Logger; } +/** The agent's history to read: from its runtime, or stored, with no provider session opened. */ +export type AgentHistorySource = + | { kind: "live"; agent: ManagedAgent } + | { kind: "stored"; record: StoredAgentRecord }; + +export interface LoadAgentHistoryDeps extends EnsureAgentLoadedDeps { + agentManager: AgentLoaderManager & Pick; +} + +/** + * Resolves an agent for reading its history. A loaded agent is read live. A stored agent whose + * history the timeline holds is read without resuming its provider, so it reads even when its + * provider is no longer available. Otherwise the provider is resumed, which imports its history + * and requires the provider. + */ +export async function loadAgentHistory( + agentId: string, + deps: LoadAgentHistoryDeps, +): Promise { + const live = deps.agentManager.getAgent(agentId); + if (live) return { kind: "live", agent: live }; + const record = await deps.agentStorage.get(agentId); + if (!record) throw new AgentNotFoundError(agentId); + if (await deps.agentManager.hasStoredHistory(agentId)) { + return { kind: "stored", record }; + } + return { kind: "live", agent: await ensureAgentLoaded(agentId, deps) }; +} + +async function requireLoadableRecord( + agentId: string, + deps: EnsureAgentLoadedDeps, +): Promise { + const record = await deps.agentStorage.get(agentId); + if (!record) { + throw new AgentNotFoundError(agentId); + } + const validProviders = deps.validProviders ?? deps.agentManager.getRegisteredProviderIds(); + if (!isStoredAgentProviderAvailable(record, validProviders)) { + throw new Error(`Agent ${agentId} references unavailable provider '${record.provider}'`); + } + return record; +} + export async function ensureUnarchivedAgentLoaded( agentId: string, deps: EnsureAgentLoadedDeps & { @@ -75,8 +109,7 @@ export async function ensureAgentLoaded( const inflight = pendingAgentInitializations.get(agentId); if (inflight) { - inflight.options.broadcastTimeline ||= deps.broadcastTimeline === true; - return inflight.promise; + return inflight; } const existing = deps.agentManager.getAgent(agentId); @@ -91,24 +124,12 @@ export async function ensureAgentLoaded( const laterInflight = pendingAgentInitializations.get(agentId); if (laterInflight) { - laterInflight.options.broadcastTimeline ||= deps.broadcastTimeline === true; - return laterInflight.promise; + return laterInflight; } - const pendingOptions = { - broadcastTimeline: deps.broadcastTimeline === true, - }; const initPromise = (async () => { - const record = await deps.agentStorage.get(agentId); - if (!record) { - throw new AgentNotFoundError(agentId); - } - + const record = await requireLoadableRecord(agentId, deps); const validProviders = deps.validProviders ?? deps.agentManager.getRegisteredProviderIds(); - if (!isStoredAgentProviderAvailable(record, validProviders)) { - throw new Error(`Agent ${agentId} references unavailable provider '${record.provider}'`); - } - const handle = toAgentPersistenceHandle(validProviders, record.persistence); let snapshot: ManagedAgent; @@ -140,20 +161,16 @@ export async function ensureAgentLoaded( deps.logger.info({ agentId, provider: record.provider }, "Agent created from stored config"); } - await deps.agentManager.hydrateTimelineFromProvider(agentId, { - broadcast: () => pendingOptions.broadcastTimeline, - }); return deps.agentManager.getAgent(agentId) ?? snapshot; })(); - const pending: PendingAgentInitialization = { promise: initPromise, options: pendingOptions }; - pendingAgentInitializations.set(agentId, pending); + pendingAgentInitializations.set(agentId, initPromise); try { return await initPromise; } finally { const current = pendingAgentInitializations.get(agentId); - if (current === pending) { + if (current === initPromise) { pendingAgentInitializations.delete(agentId); } } diff --git a/packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts b/packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts index 5c57915ad5..f95b0a7b2c 100644 --- a/packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts +++ b/packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts @@ -182,6 +182,7 @@ class TestAgentSession implements AgentSession { class TestAgentClient implements AgentClient { readonly capabilities = TEST_CAPABILITIES; private sessionCounter = 0; + private history: AgentStreamEvent[] = []; readonly sessions = new Map(); constructor(readonly provider: AgentProvider = "codex") {} @@ -195,6 +196,7 @@ class TestAgentClient implements AgentClient { config, `${config.provider}-session-${++this.sessionCounter}`, ); + session.setHistory(this.history); this.sessions.set(config.cwd, session); return session; } @@ -230,6 +232,11 @@ class TestAgentClient implements AgentClient { return true; } + /** The provider history of the sessions created from now on. */ + setHistory(events: AgentStreamEvent[]): void { + this.history = [...events]; + } + getSession(cwd: string): TestAgentSession { const session = this.sessions.get(cwd); if (!session) { @@ -277,6 +284,20 @@ function createIdFactory(): () => string { return () => AGENT_IDS[index++] ?? `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`; } +/** Resumes an agent whose history Paseo does not hold, so it imports the provider's. */ +async function resumeWithProviderHistory( + harness: Harness, + history: AgentStreamEvent[], +): Promise<{ agentId: string }> { + harness.client.setHistory(history); + const agent = await harness.manager.resumeAgentFromPersistence( + { provider: harness.client.provider, sessionId: "provider-history" }, + { cwd: harness.workdir }, + AGENT_IDS[0], + ); + return { agentId: agent.id }; +} + async function createManagedSession( harness: Harness, options?: { agentId?: string; provider?: AgentProvider; workdir?: string }, @@ -447,15 +468,15 @@ describe("target coalesced behavior", () => { } }); - test("bounds tool output while hydrating provider history", async () => { + test("bounds tool output while importing provider history", async () => { const harness = createHarness(); try { - const { agentId, session } = await createManagedSession(harness); const output = `${"a".repeat(512 * 1024)}${"z".repeat(512 * 1024)}`; const expectedItem = limitAgentTimelineItemContent(toolCall({ status: "completed", output })); - session.setHistory([timelineEvent(toolCall({ status: "completed", output }))]); - await harness.manager.hydrateTimelineFromProvider(agentId, { force: true }); + const { agentId } = await resumeWithProviderHistory(harness, [ + timelineEvent(toolCall({ status: "completed", output })), + ]); expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([ expectedItem, @@ -1341,14 +1362,16 @@ describe("target coalesced behavior", () => { } }); - test("history hydration projects immediately without waiting for a streaming window", async () => { + test("history import projects immediately without waiting for a streaming window", async () => { vi.useFakeTimers(); const harness = createHarness(); try { - const { agentId, session } = await createManagedSession(harness); - session.setHistory([assistant("a1"), assistant("a2"), reasoning("r1"), reasoning("r2")]); - - await harness.manager.hydrateTimelineFromProvider(agentId, { force: true }); + const { agentId } = await resumeWithProviderHistory(harness, [ + assistant("a1"), + assistant("a2"), + reasoning("r1"), + reasoning("r2"), + ]); const rows = await harness.manager.getTimelineRows(agentId); expect(getTimelineItems(rows)).toEqual([ diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 7d014ef122..e5aa5c6a3e 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -733,20 +733,21 @@ test("an accepted prompt the timeline fails to commit surfaces a history error", } }); -test("refreshing an agent replaces its timeline instead of appending a second copy", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-rehydrate-")); +/** What the Reload agent action does for a loaded agent. */ +async function reloadAgent(manager: AgentManager, agentId: string) { + await manager.reloadAgentSession(agentId); + return manager.reconcileProviderHistory(agentId); +} + +test("Reload agent keeps the history and its identity, and adds only what the provider continued", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-reload-")); + const providerHistory: AgentTimelineItem[] = [ + { type: "user_message", text: "continue", messageId: "provider-1" }, + { type: "assistant_message", text: "done" }, + ]; class HistorySession extends TestAgentSession { override async *streamHistory(): AsyncGenerator { - yield { - type: "timeline", - provider: "codex", - item: { type: "user_message", text: "continue" }, - }; - yield { - type: "timeline", - provider: "codex", - item: { type: "assistant_message", text: "done" }, - }; + for (const item of providerHistory) yield { type: "timeline", provider: "codex", item }; } } class HistoryClient extends TestAgentClient { @@ -772,43 +773,58 @@ test("refreshing an agent replaces its timeline instead of appending a second co workspaceId: undefined, }); agentId = agent.id; + const created = await manager.fetchTimeline(agent.id); + const published: AgentManagerEvent[] = []; + manager.subscribe((event) => published.push(event), { + agentId: agent.id, + replayState: false, + }); + const timelineEvents = () => + published.filter((event) => event.type === "agent_stream" && event.event.type === "timeline"); - // What session.handleRefreshAgentRequest does for a loaded agent, twice over. - for (let refresh = 0; refresh < 2; refresh += 1) { - await manager.reloadAgentSession(agent.id, undefined, { rehydrateFromDisk: true }); - await manager.hydrateTimelineFromProvider(agent.id, { broadcast: true }); - await manager.flush(); - } + // The provider has a turn this new agent does not. + expect(await reloadAgent(manager, agent.id)).toEqual({ kind: "appended" }); + await manager.flush(); + expect(timelineEvents()).toHaveLength(2); - const rows = await manager.getTimelineRows(agent.id); - expect(rows.map((row) => row.item)).toEqual([ - { type: "user_message", text: "continue" }, - { type: "assistant_message", text: "done" }, - ]); + // Nothing changed at the provider: nothing is written or sent. + expect(await reloadAgent(manager, agent.id)).toEqual({ kind: "unchanged" }); + await manager.flush(); + expect(timelineEvents()).toHaveLength(2); + + // The conversation was continued outside Paseo: only the new turn arrives. + providerHistory.push( + { type: "user_message", text: "next", messageId: "provider-2" }, + { type: "assistant_message", text: "more" }, + ); + expect(await reloadAgent(manager, agent.id)).toEqual({ kind: "appended" }); + await manager.flush(); + expect(timelineEvents()).toHaveLength(4); + + const page = await manager.fetchTimeline(agent.id); + expect(page.rows.map((row) => row.item)).toEqual(providerHistory); + expect(page.rows.map((row) => row.seqStart)).toEqual([1, 2, 3, 4]); + expect(page).toMatchObject({ epoch: created.epoch, retirementCount: 0 }); } finally { if (agentId) await manager.closeAgent(agentId).catch(() => undefined); rmSync(workdir, { recursive: true, force: true }); } }); -test("a failed history replay commits none of the replay", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-replay-fail-")); - let failReplay = false; +test("Reload agent keeps the history when the provider history differs or cannot be read", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-reload-fail-")); + let providerHistory: AgentTimelineItem[] = [ + { type: "user_message", text: "continue", messageId: "provider-1" }, + { type: "assistant_message", text: "done" }, + ]; + let failRead = false; class FlakyHistorySession extends TestAgentSession { override async *streamHistory(): AsyncGenerator { - yield { - type: "timeline", - provider: "codex", - item: { type: "user_message", text: "continue" }, - }; - if (failReplay) { - throw new Error("history stream closed"); + yield { type: "timeline", provider: "codex", item: providerHistory[0]! }; + if (failRead) throw new Error("history stream closed"); + for (const item of providerHistory.slice(1)) { + yield { type: "timeline", provider: "codex", item }; } - yield { - type: "timeline", - provider: "codex", - item: { type: "assistant_message", text: "done" }, - }; } } class FlakyHistoryClient extends TestAgentClient { @@ -834,46 +850,38 @@ test("a failed history replay commits none of the replay", async () => { workspaceId: undefined, }); agentId = agent.id; - await manager.reloadAgentSession(agent.id, undefined, { rehydrateFromDisk: true }); - await manager.hydrateTimelineFromProvider(agent.id, { broadcast: true }); - await manager.flush(); - const committed = await manager.getTimelineRows(agent.id); + await reloadAgent(manager, agent.id); + const committed = await manager.fetchTimeline(agent.id); - failReplay = true; - await manager.reloadAgentSession(agent.id, undefined, { rehydrateFromDisk: true }); - await expect( - manager.hydrateTimelineFromProvider(agent.id, { broadcast: true }), - ).rejects.toThrow("history stream closed"); - await manager.flush(); + failRead = true; + await expect(reloadAgent(manager, agent.id)).rejects.toThrow("history stream closed"); + expect(await manager.fetchTimeline(agent.id)).toEqual(committed); - // Reload with rehydration discards the active history before the replay (phase 1 keeps - // that); the failed replay must not leave the rows it streamed before failing. - expect(committed.map((row) => row.item)).toEqual([ - { type: "user_message", text: "continue" }, - { type: "assistant_message", text: "done" }, - ]); - expect(await manager.getTimelineRows(agent.id)).toEqual([]); - - // A retry once the provider recovers replaces that history rather than stacking on it. - failReplay = false; - await manager.hydrateTimelineFromProvider(agent.id, { broadcast: true }); - await manager.flush(); - expect((await manager.getTimelineRows(agent.id)).map((row) => row.item)).toEqual([ - { type: "user_message", text: "continue" }, - { type: "assistant_message", text: "done" }, - ]); + // Edited outside Paseo under the same prompt: the owned answer stays. + failRead = false; + providerHistory = [providerHistory[0]!, { type: "assistant_message", text: "rewritten" }]; + expect(await reloadAgent(manager, agent.id)).toEqual({ + kind: "diverged", + reason: "The provider history has a different last answer", + }); + expect(await manager.fetchTimeline(agent.id)).toEqual(committed); } 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 () => { +test("resuming retries the history import after a stream failure and imports it once", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-retry-")); let attempts = 0; class RetryingHistorySession extends TestAgentSession { override async *streamHistory(): AsyncGenerator { attempts += 1; + yield { + type: "timeline", + provider: "codex", + item: { type: "user_message", text: "hello", messageId: "provider-1" }, + }; if (attempts === 1) throw new Error("history interrupted"); yield { type: "timeline", @@ -894,27 +902,29 @@ test("retries provider history hydration after a stream failure", async () => { })(), }, logger, + idFactory: () => "00000000-0000-4000-8000-000000000120", }); - let agentId: string | null = null; + const handle = { provider: "codex", sessionId: "retry-history" } as const; try { - const agent = await manager.resumeAgentFromPersistence( - { provider: "codex", sessionId: "retry-history" }, - { cwd: workdir }, - ); - agentId = agent.id; - await expect(manager.hydrateTimelineFromProvider(agent.id)).rejects.toThrow( + await expect(manager.resumeAgentFromPersistence(handle, { cwd: workdir })).rejects.toThrow( "history interrupted", ); - expect(manager.getAgent(agent.id)?.historyPrimed).toBe(false); + // A failed import leaves no history behind that a later read would trust. + expect(manager.getAgent("00000000-0000-4000-8000-000000000120")).toBeNull(); + expect(await manager.hasStoredHistory("00000000-0000-4000-8000-000000000120")).toBe(false); - await manager.hydrateTimelineFromProvider(agent.id); + const agent = await manager.resumeAgentFromPersistence(handle, { cwd: workdir }); + await manager.closeAgent(agent.id); + await manager.resumeAgentFromPersistence(handle, { cwd: workdir }, agent.id); + + // The second resume found the history and did not read the provider's again. expect(attempts).toBe(2); - expect(await manager.getTimeline(agent.id)).toContainEqual({ - type: "assistant_message", - text: "provider history", - }); + expect(await manager.getTimeline(agent.id)).toEqual([ + { type: "user_message", text: "hello", messageId: "provider-1" }, + { type: "assistant_message", text: "provider history" }, + ]); } finally { - if (agentId) await manager.closeAgent(agentId).catch(() => undefined); + await manager.closeAgent("00000000-0000-4000-8000-000000000120").catch(() => undefined); rmSync(workdir, { recursive: true, force: true }); } }); @@ -1866,12 +1876,20 @@ function fakeCodexEmitting(args: FakeCodexEmitterArgs): AgentClient { async createSession(config: AgentSessionConfig) { return new FakeCodexSession(config); }, - async resumeSession() { - throw new Error("unused"); + async resumeSession(_handle: AgentPersistenceHandle, config?: Partial) { + return new FakeCodexSession({ provider: "codex", cwd: config?.cwd ?? tmpdir() }); }, }; } +/** Resumes a codex agent whose history Paseo does not hold, so it imports the provider's. */ +function resumeWithProviderHistory(manager: AgentManager, cwd: string) { + return manager.resumeAgentFromPersistence( + { provider: "codex", sessionId: "provider-history" }, + { cwd }, + ); +} + const logger = createTestLogger(); test("does not register a session that finishes starting after shutdown begins", async () => { @@ -2672,9 +2690,7 @@ test("reload releases the original writer before resuming the same session", asy workspaceId: undefined, }); for (let i = 0; i < 3; i++) { - const reloaded = await manager.reloadAgentSession(created.id, undefined, { - rehydrateFromDisk: true, - }); + const reloaded = await manager.reloadAgentSession(created.id); expect(reloaded.id).toBe(created.id); expect(reloaded.persistence?.sessionId).toBe(created.persistence?.sessionId); } @@ -4629,7 +4645,6 @@ test("importProviderSession imports the selected session without listing and pub }, }); expect(imported.lifecycle).toBe("idle"); - expect(imported.historyPrimed).toBe(true); expect(await manager.getTimeline(imported.id)).toEqual([ { type: "user_message", text: "Trace provider imports" }, { type: "assistant_message", text: "Done" }, @@ -4819,7 +4834,6 @@ test("reloadAgentSession preserves timeline and does not force history replay", type: "assistant_message", text: "keep this timeline in memory", }); - await manager.hydrateTimelineFromProvider(snapshot.id); const beforeReload = await manager.getTimeline(snapshot.id); expect(beforeReload).toHaveLength(1); @@ -4829,13 +4843,11 @@ test("reloadAgentSession preserves timeline and does not force history replay", const afterReload = await manager.getTimeline(snapshot.id); expect(afterReload).toEqual(beforeReload); - // If reload resets historyPrimed, this would replay provider history and append another item. - await manager.hydrateTimelineFromProvider(snapshot.id); - const afterHydrate = await manager.getTimeline(snapshot.id); - expect(afterHydrate).toEqual(beforeReload); + expect(await manager.reconcileProviderHistory(snapshot.id)).toEqual({ kind: "unchanged" }); + expect(await manager.getTimeline(snapshot.id)).toEqual(beforeReload); }); -test("reloadAgentSession clears provider children before rehydrating from disk", async () => { +test("Reload agent keeps provider children and ends the running ones", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-child-reload-")); const storage = new AgentStorage(join(workdir, "agents"), logger); let activeSession: TestAgentSession | null = null; @@ -4871,9 +4883,11 @@ test("reloadAgentSession clears provider children before rehydrating from disk", }); await vi.waitFor(() => expect(manager.listProviderSubagents(snapshot.id)).toHaveLength(1)); - await manager.reloadAgentSession(snapshot.id, undefined, { rehydrateFromDisk: true }); + await reloadAgent(manager, snapshot.id); - expect(manager.listProviderSubagents(snapshot.id)).toEqual([]); + expect(manager.listProviderSubagents(snapshot.id)).toEqual([ + expect.objectContaining({ id: "stale-child", status: "canceled" }), + ]); }); test("reloadAgentSession terminalizes running provider children when preserving history", async () => { @@ -4919,7 +4933,7 @@ test("reloadAgentSession terminalizes running provider children when preserving ]); }); -test("hydrateTimelineFromProvider restores and broadcasts provider children from session history", async () => { +test("resuming without stored history restores provider children from the provider history", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-child-history-")); const storage = new AgentStorage(join(workdir, "agents"), logger); class ProviderChildHistorySession extends TestAgentSession { @@ -4937,8 +4951,11 @@ test("hydrateTimelineFromProvider restores and broadcasts provider children from } } class ProviderChildHistoryClient extends TestAgentClient { - override async createSession(config: AgentSessionConfig): Promise { - return new ProviderChildHistorySession(config); + override async resumeSession( + _handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + return new ProviderChildHistorySession({ provider: "codex", cwd: config?.cwd ?? workdir }); } } const manager = new AgentManager({ @@ -4947,16 +4964,8 @@ test("hydrateTimelineFromProvider restores and broadcasts provider children from logger, idFactory: () => "00000000-0000-4000-8000-000000000117", }); - const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - const events: AgentManagerEvent[] = []; - manager.subscribe((event) => events.push(event), { - agentId: snapshot.id, - replayState: false, - }); - await manager.hydrateTimelineFromProvider(snapshot.id, { force: true, broadcast: true }); + const snapshot = await resumeWithProviderHistory(manager, workdir); expect(manager.listProviderSubagents(snapshot.id)).toEqual([ expect.objectContaining({ @@ -4972,60 +4981,6 @@ test("hydrateTimelineFromProvider restores and broadcasts provider children from parentAgentId: snapshot.id, }), ]); - expect(events).toContainEqual({ - type: "provider_subagent", - event: { - type: "upsert", - subagent: expect.objectContaining({ - id: "restored-child", - parentAgentId: snapshot.id, - }), - }, - }); -}); - -test("force provider hydration removes children absent from current history", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-child-force-")); - const storage = new AgentStorage(join(workdir, "agents"), logger); - let session: TestAgentSession | null = null; - class ProviderChildForceClient extends TestAgentClient { - override async createSession(config: AgentSessionConfig): Promise { - session = new TestAgentSession(config); - return session; - } - } - const manager = new AgentManager({ - clients: { codex: new ProviderChildForceClient() }, - registry: storage, - logger, - idFactory: () => "00000000-0000-4000-8000-000000000118", - }); - const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - session?.pushEvent({ - type: "provider_subagent", - provider: "codex", - event: { type: "upsert", id: "removed-by-rewind", status: "completed" }, - }); - await vi.waitFor(() => expect(manager.listProviderSubagents(snapshot.id)).toHaveLength(1)); - const events: AgentManagerEvent[] = []; - manager.subscribe((event) => events.push(event), { - agentId: snapshot.id, - replayState: false, - }); - - await manager.hydrateTimelineFromProvider(snapshot.id, { force: true, broadcast: true }); - - expect(manager.listProviderSubagents(snapshot.id)).toEqual([]); - expect(events).toContainEqual({ - type: "provider_subagent", - event: { - type: "remove", - parentAgentId: snapshot.id, - subagentId: "removed-by-rewind", - }, - }); }); test("reloadAgentSession preserves current title when config title is unset", async () => { @@ -6352,7 +6307,7 @@ test("does not trim committed history", async () => { expect(fetched.window.maxSeq).toBe(3); }); -test("hydrateTimeline preserves assistant chunk, reasoning, and tool timeline history", async () => { +test("history import preserves assistant chunk, reasoning, and tool timeline history", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-canonical-assistant-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -6411,8 +6366,14 @@ test("hydrateTimeline preserves assistant chunk, reasoning, and tool timeline hi return new ChunkedAssistantHistorySession(config); } - async resumeSession(): Promise { - throw new Error("Not used in this test"); + async resumeSession( + _handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + return new ChunkedAssistantHistorySession({ + provider: "codex", + cwd: config?.cwd ?? tmpdir(), + }); } } @@ -6425,16 +6386,7 @@ test("hydrateTimeline preserves assistant chunk, reasoning, and tool timeline hi idFactory: () => "00000000-0000-4000-8000-000000000121", }); - const snapshot = await manager.createAgent( - { - provider: "codex", - cwd: workdir, - }, - undefined, - { workspaceId: undefined }, - ); - - await manager.hydrateTimelineFromProvider(snapshot.id, { force: true }); + const snapshot = await resumeWithProviderHistory(manager, workdir); expect(await manager.getTimeline(snapshot.id)).toEqual([ { type: "assistant_message", text: "chunk one chunk two" }, @@ -6456,7 +6408,7 @@ test("hydrateTimeline preserves assistant chunk, reasoning, and tool timeline hi ]); }); -test("hydrateTimeline preserves reasoning between assistant chunks", async () => { +test("history import preserves reasoning between assistant chunks", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-reasoning-interleave-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -6493,8 +6445,14 @@ test("hydrateTimeline preserves reasoning between assistant chunks", async () => return new ReasoningInterleavedHistorySession(config); } - async resumeSession(): Promise { - throw new Error("Not used in this test"); + async resumeSession( + _handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + return new ReasoningInterleavedHistorySession({ + provider: "codex", + cwd: config?.cwd ?? tmpdir(), + }); } } @@ -6507,16 +6465,7 @@ test("hydrateTimeline preserves reasoning between assistant chunks", async () => idFactory: () => "00000000-0000-4000-8000-000000000122", }); - const snapshot = await manager.createAgent( - { - provider: "codex", - cwd: workdir, - }, - undefined, - { workspaceId: undefined }, - ); - - await manager.hydrateTimelineFromProvider(snapshot.id, { force: true }); + const snapshot = await resumeWithProviderHistory(manager, workdir); expect(await manager.getTimeline(snapshot.id)).toEqual([ { @@ -10619,11 +10568,12 @@ test("a stored-only archive closes a shared resume before a later protected load } }); -test("a shared agent load upgrades provider history hydration to broadcast", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-shared-load-broadcast-")); +test("concurrent agent loads share one provider history import", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-shared-load-import-")); const storage = new AgentStorage(join(workdir, "agents"), logger); const historyStarted = deferred(); const historyAllowed = deferred(); + let historyReads = 0; const client = new (class extends TestAgentClient { override async resumeSession( _handle: AgentPersistenceHandle, @@ -10631,6 +10581,7 @@ test("a shared agent load upgrades provider history hydration to broadcast", asy ): Promise { return new (class extends TestAgentSession { override async *streamHistory(): AsyncGenerator { + historyReads += 1; historyStarted.resolve(); await historyAllowed.promise; yield { @@ -10653,31 +10604,26 @@ test("a shared agent load upgrades provider history hydration to broadcast", asy const events: AgentManagerEvent[] = []; manager.subscribe((event) => events.push(event), { agentId: agent.id, replayState: false }); - const quietLoad = ensureAgentLoaded(agent.id, { + const firstLoad = ensureAgentLoaded(agent.id, { agentManager: manager, agentStorage: storage, logger, }); await historyStarted.promise; - const broadcastingLoad = ensureAgentLoaded(agent.id, { + const secondLoad = ensureAgentLoaded(agent.id, { agentManager: manager, agentStorage: storage, - broadcastTimeline: true, logger, }); historyAllowed.resolve(); - await Promise.all([quietLoad, broadcastingLoad]); + await Promise.all([firstLoad, secondLoad]); - expect(events).toContainEqual( - expect.objectContaining({ - type: "agent_stream", - agentId: agent.id, - event: expect.objectContaining({ - type: "timeline", - item: { type: "assistant_message", text: "Recovered history" }, - }), - }), - ); + expect(historyReads).toBe(1); + expect(await manager.getTimeline(agent.id)).toEqual([ + { type: "assistant_message", text: "Recovered history" }, + ]); + // The import is the agent's starting history, which clients fetch; it is not live output. + expect(events.filter((event) => event.type === "agent_stream")).toEqual([]); } finally { historyAllowed.resolve(); await manager.flush().catch(() => undefined); @@ -10880,7 +10826,7 @@ test("provider close failure retains the runtime for cleanup instead of allowing } }); -test("hydrateTimeline keeps provider user_message items when no canonical user history exists", async () => { +test("history import keeps provider user_message items when no canonical user history exists", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-keep-user-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -10922,8 +10868,14 @@ test("hydrateTimeline keeps provider user_message items when no canonical user h return new HistoryWithUserMessagesSession(config); } - async resumeSession(): Promise { - throw new Error("Not used in this test"); + async resumeSession( + _handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + return new HistoryWithUserMessagesSession({ + provider: "codex", + cwd: config?.cwd ?? tmpdir(), + }); } } @@ -10936,16 +10888,7 @@ test("hydrateTimeline keeps provider user_message items when no canonical user h idFactory: () => "00000000-0000-4000-8000-000000000203", }); - const snapshot = await manager.createAgent( - { - provider: "codex", - cwd: workdir, - }, - undefined, - { workspaceId: undefined }, - ); - - await manager.hydrateTimelineFromProvider(snapshot.id, { force: true }); + const snapshot = await resumeWithProviderHistory(manager, workdir); const timeline = await manager.getTimeline(snapshot.id); const userMessages = timeline.filter((item) => item.type === "user_message"); @@ -10954,7 +10897,7 @@ test("hydrateTimeline keeps provider user_message items when no canonical user h expect(assistantMessages).toHaveLength(2); }); -test("hydrateTimeline preserves provider replay timestamps and marks missing ones untrusted", async () => { +test("history import preserves provider replay timestamps and marks missing ones untrusted", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-timestamps-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -10987,8 +10930,11 @@ test("hydrateTimeline preserves provider replay timestamps and marks missing one return new TimestampedHistorySession(config); } - async resumeSession(): Promise { - throw new Error("Not used in this test"); + async resumeSession( + _handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + return new TimestampedHistorySession({ provider: "codex", cwd: config?.cwd ?? tmpdir() }); } } @@ -11001,16 +10947,7 @@ test("hydrateTimeline preserves provider replay timestamps and marks missing one idFactory: () => "00000000-0000-4000-8000-000000000204", }); - const snapshot = await manager.createAgent( - { - provider: "codex", - cwd: workdir, - }, - undefined, - { workspaceId: undefined }, - ); - - await manager.hydrateTimelineFromProvider(snapshot.id, { force: true }); + const snapshot = await resumeWithProviderHistory(manager, workdir); const timeline = (await manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0 })).rows; expect(timeline).toHaveLength(2); @@ -11771,11 +11708,7 @@ test("system-injected prompts appear as trigger tools during history replay", as idFactory: () => "00000000-0000-4000-8000-0000000005a2", }); - const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); - - await manager.hydrateTimelineFromProvider(snapshot.id, { force: true }); + const snapshot = await resumeWithProviderHistory(manager, workdir); const timeline = await manager.getTimeline(snapshot.id); expect(timeline).toContainEqual( @@ -12040,13 +11973,6 @@ test("commits startup notices once on create and after restored history", async metadata: { cwd: workdir }, }); ids.push(resumed.id); - await manager.hydrateTimelineFromProvider(resumed.id); - expect((await manager.getTimelineRows(resumed.id)).map((row) => row.item)).toEqual([ - history, - limitedToolOutput, - notice, - ]); - await manager.hydrateTimelineFromProvider(resumed.id, { force: true }); expect((await manager.getTimelineRows(resumed.id)).map((row) => row.item)).toEqual([ history, limitedToolOutput, @@ -12146,11 +12072,13 @@ test.each(["live", "history"])( path === "live" ? { turnItems: [item] } : { historyItems: [item] }, ); const manager = new AgentManager({ clients: { codex }, registry: storage, logger }); - const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); + const agent = + path === "live" + ? await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }) + : await resumeWithProviderHistory(manager, workdir); if (path === "live") await manager.runAgent(agent.id, { text: "start" }); - else await manager.hydrateTimelineFromProvider(agent.id, { force: true }); expect(await manager.getTimeline(agent.id)).toContainEqual( expect.objectContaining({ type: "tool_call", @@ -12430,13 +12358,14 @@ test.each(["live", "history", "accepted"] as const)( ingestion === "history" ? { historyItems: [item] } : { turnItems: [item] }, ); const manager = new AgentManager({ clients: { codex }, logger }); - const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { - workspaceId: undefined, - }); + const agent = + ingestion === "history" + ? await resumeWithProviderHistory(manager, workdir) + : await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); try { - if (ingestion === "history") - await manager.hydrateTimelineFromProvider(agent.id, { force: true }); - else if (ingestion === "accepted") { + if (ingestion === "accepted") { for await (const _event of manager.streamAgent(agent.id, prompt, { clientMessageId: "report", })) { diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 6125e3fc64..9b89c32908 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -282,12 +282,20 @@ export interface SubscribeOptions { includeInternal?: boolean; } -interface HydrateTimelineOptions { - force?: boolean; - broadcast?: boolean | (() => boolean); - broadcastTimeline?: boolean; +type ProviderSubagentStreamEvent = Extract; + +interface ProviderHistory { + entries: TimelineEntry[]; + subagents: ProviderSubagentStreamEvent[]; } +/** What Reload agent found in the provider's history; see `reconcileProviderHistory`. */ +export type ProviderHistoryReconcile = + | { kind: "unchanged" } + | { kind: "appended" } + | { kind: "diverged"; reason: string } + | { kind: "rewind_open" }; + export type ImportablePersistedAgentQueryOptions = ListImportableSessionsOptions & { /** * When set, only providers in this set are scanned, in addition to the @@ -462,7 +470,6 @@ interface ManagedAgentBase { inFlightPermissionResponses: Set; pendingReplacement: boolean; persistence: AgentPersistenceHandle | null; - historyPrimed: boolean; lastUserMessageAt: Date | null; activeTurnId: string | null; activeTurnStartedAt: Date | null; @@ -703,6 +710,11 @@ function toTimelineEntry(event: Extract) return event.timestamp ? { item: event.item, timestamp: event.timestamp } : { item: event.item }; } +/** The live event that publishes a row committed from provider history. */ +function historyEvent(provider: AgentProvider, commit: TimelineCommit): AgentStreamEvent { + return { type: "timeline", provider, item: commit.row.item }; +} + function buildImportedTimelineRows(entries: readonly ImportedTimelineEntry[]): AgentTimelineRow[] { const rows: AgentTimelineRow[] = []; for (const entry of entries) { @@ -1329,13 +1341,13 @@ export class AgentManager { } async getTimeline(id: string): Promise { - this.requireAgent(id); + await this.requireHistory(id); const rows = await this.timeline.readRows({ agentId: id }); return rows.map((row) => row.item); } async getTimelineRows(id: string): Promise { - this.requireAgent(id); + await this.requireHistory(id); return await this.timeline.readRows({ agentId: id }); } @@ -1343,12 +1355,12 @@ export class AgentManager { id: string, options?: AgentTimelineFetchOptions, ): Promise { - this.requireAgent(id); + await this.requireHistory(id); return await this.timeline.fetch({ agentId: id }, options); } listProviderSubagents(parentAgentId: string): ProviderSubagentDescriptor[] { - this.requirePublicAgent(parentAgentId); + this.requirePublicParent(parentAgentId); return this.providerSubagents.list(parentAgentId); } @@ -1367,7 +1379,7 @@ export class AgentManager { parentAgentId: string, subagentId: string, ): ProviderSubagentDescriptor | null { - this.requirePublicAgent(parentAgentId); + this.requirePublicParent(parentAgentId); return this.providerSubagents.get(parentAgentId, subagentId); } @@ -1376,7 +1388,8 @@ export class AgentManager { subagentId: string, options?: AgentTimelineFetchOptions, ): Promise { - this.requirePublicAgent(parentAgentId); + this.requirePublicParent(parentAgentId); + await this.requireHistory(parentAgentId); return this.providerSubagents.fetchTimeline(parentAgentId, subagentId, options); } @@ -1438,7 +1451,6 @@ export class AgentManager { workspaceId: options.workspaceId, owner: options.owner, launchEnv: options.env ? { ...options.env } : undefined, - historyPrimed: true, }); if (!agent.internal) { this.pluginLifecycle?.emit("agent.create.after", { @@ -1567,6 +1579,7 @@ export class AgentManager { launchEnv, persistence: handle, restoring: true, + importMissingHistory: true, }); } @@ -1641,7 +1654,6 @@ export class AgentManager { timelineRows, timelineNextSeq: timelineRows.length + 1, persistence: imported.persistence, - historyPrimed: true, initialTitle, publishWhenReady: true, }); @@ -1657,20 +1669,18 @@ export class AgentManager { } } - // Hot-reload an active agent session with config overrides. By default the - // in-memory timeline is preserved (used for voice-mode toggles and similar - // config swaps). When `rehydrateFromDisk` is set, the timeline is wiped so a - // new epoch is minted and provider history is re-streamed — this is what the - // user-facing "Reload agent" action wants when the on-disk session was - // mutated outside Paseo. + /** + * Hot-reloads an active agent session, optionally with config overrides (voice-mode toggles and + * similar config swaps). The timeline is kept as it is. Reload agent then brings in what the + * provider's history adds with `reconcileProviderHistory`. + */ reloadAgentSession( agentId: string, overrides?: Partial, - options?: { rehydrateFromDisk?: boolean }, ): Promise { return this.trackAgentRegistrationOperation( this.runLifecycleMutation(agentId, async () => { - return (await this.reloadSession(agentId, overrides, options)).agent; + return (await this.reloadSession(agentId, overrides)).agent; }), ); } @@ -1678,7 +1688,6 @@ export class AgentManager { private async reloadSession( agentId: string, overrides?: Partial, - options?: { rehydrateFromDisk?: boolean }, ): Promise<{ agent: ManagedAgent; rewind: RewindSettlement }> { this.assertAcceptingAgentRegistrations(); let existing = this.requireSessionAgent(agentId); @@ -1686,8 +1695,8 @@ export class AgentManager { await this.cancelAgentRunBefore(agentId, "reload"); existing = this.requireSessionAgent(agentId); } - const { rehydrateFromDisk, settleRewind } = await this.planReloadHistory(agentId, options); - const preservedHistoryPrimed = existing.historyPrimed; + // An open rewind is judged from the resumed session's history. + const settleRewind = ((await this.timeline.describe({ agentId }))?.rewind ?? null) !== null; const preservedLastUsage = existing.lastUsage; const preservedLastError = existing.lastError; const preservedAttention = existing.attention; @@ -1747,8 +1756,6 @@ export class AgentManager { await this.requireExternalMcpSupport(session, providerLaunchConfig, launchContext); this.assertAcceptingAgentRegistrations(); - if (rehydrateFromDisk) await this.discardTimelineForRehydrate(agentId); - // Preserve existing labels and timeline during reload. handedToRegistration = true; const registered = this.registerSession(session, storedConfig, agentId, { @@ -1759,7 +1766,6 @@ export class AgentManager { createdAt: existing.createdAt, updatedAt: existing.updatedAt, lastUserMessageAt: existing.lastUserMessageAt, - historyPrimed: rehydrateFromDisk ? false : preservedHistoryPrimed, lastUsage: preservedLastUsage, lastError: preservedLastError, attention: preservedAttention, @@ -1793,27 +1799,43 @@ export class AgentManager { } /** - * An open rewind is judged from the resumed session's history, so that history must not replace - * the timeline first. + * Brings in what the provider's own history adds to the agent's owned history, which is never + * changed: rows the provider has after the owned conversation are appended as ordinary live rows. + * Any other difference, including an answer Paseo saw only part of, is reported as `diverged` + * and changes nothing. */ - private async planReloadHistory( - agentId: string, - options: { rehydrateFromDisk?: boolean } | undefined, - ): Promise<{ rehydrateFromDisk: boolean; settleRewind: boolean }> { - const openRewind = (await this.timeline.describe({ agentId }))?.rewind ?? null; - return { - rehydrateFromDisk: (options?.rehydrateFromDisk ?? false) && openRewind === null, - settleRewind: openRewind !== null, - }; + reconcileProviderHistory(agentId: string): Promise { + return this.runLifecycleMutation(agentId, async () => { + const agent = this.requireSessionAgent(agentId); + const history = await this.readProviderHistory(agent.session.streamHistory()); + const result = await this.timeline.reconcileProviderHistory({ agentId }, history.entries, { + publish: (commits) => { + for (const commit of commits) { + this.publishTimelineEvent(agentId, historyEvent(agent.provider, commit), commit); + } + }, + }); + this.logger.info({ agentId, result: result.kind }, "agent.provider_history.reconciled"); + if (result.kind === "diverged" || result.kind === "rewind_open") { + if (result.kind === "diverged") { + this.logger.warn({ agentId, reason: result.reason }, "agent.provider_history.diverged"); + } + return result; + } + if (result.kind === "unchanged") return result; + if (result.kind === "unpublished") { + this.logger.error({ err: result.error, agentId }, "Failed to publish provider history"); + } + await this.applyProviderSubagentHistory(agentId, history.subagents, { dispatch: true }); + this.touchUpdatedAt(agent); + this.emitState(agent); + return { kind: "appended" }; + }); } - // Wipes the in-memory timeline so registerSession mints a new epoch and - // hydrateTimelineFromProvider re-streams the freshly read provider history. - private async discardTimelineForRehydrate(agentId: string): Promise { - await this.timeline.delete({ agentId }); - for (const event of await this.providerSubagents.deleteParent(agentId)) { - this.dispatch({ type: "provider_subagent", event }); - } + /** Whether the timeline holds the agent's history, so reads need no provider session. */ + async hasStoredHistory(agentId: string): Promise { + return (await this.timeline.describe({ agentId })) !== null; } private async closeReloadedSession(session: AgentSession, agentId: string): Promise { @@ -2084,7 +2106,8 @@ export class AgentManager { agent.updatedAt = new Date(archivedAt); await this.closeAgentRuntime(agentId); await this.syncNativeArchiveState(stored.provider, stored.persistence, "archive"); - await this.discardRetainedAgentState(agentId); + // The history and provider children stay: an archived chat reads them without its provider. + this.paseoToolPolicies.delete(agentId); await this.cascadeArchiveChildren(stored); @@ -2229,7 +2252,6 @@ export class AgentManager { finalizedForegroundTurnIds: new Set(), unsubscribeSession: null, persistence: record.persistence ?? null, - historyPrimed: true, lastUserMessageAt: record.lastUserMessageAt ? new Date(record.lastUserMessageAt) : null, lastUsage: undefined, lastError: record.lastError ?? undefined, @@ -2519,7 +2541,8 @@ export class AgentManager { await this.syncNativeArchiveState(record.provider, record.persistence, "archive"); - await this.discardRetainedAgentState(agentId); + this.paseoToolPolicies.delete(agentId); + this.detachPluginTools(agentId); if (!nextRecord.internal) this.dispatchStoredAgentState(nextRecord); await this.fireAgentArchived(agentId); @@ -3525,17 +3548,6 @@ export class AgentManager { return iterator.done ? null : iterator.value; } - /** - * Hydrates the runtime timeline from provider history. No-ops if already hydrated. - */ - async hydrateTimelineFromProvider( - agentId: string, - options?: HydrateTimelineOptions, - ): Promise { - const agent = this.requireSessionAgent(agentId); - await this.hydrateTimelineFromLegacyProviderHistory(agent, options); - } - async rewind(agentId: string, messageId: string, mode: RewindMode): Promise { const agent = this.requireSessionAgent(agentId); // Without an active run, the rewind holds the run from its call, so no prompt starts @@ -3950,7 +3962,11 @@ export class AgentManager { timelineRows?: AgentTimelineRow[]; timelineNextSeq?: number; persistence?: AgentPersistenceHandle; - historyPrimed?: boolean; + /** + * Without history in the timeline, import the provider's own history into it before the + * agent is registered. A failed read leaves the agent unregistered and the timeline absent. + */ + importMissingHistory?: boolean; lastUsage?: AgentUsage; lastError?: string; attention?: AttentionState; @@ -3981,8 +3997,11 @@ export class AgentManager { ); const now = new Date(); - const { timelineWasPresent } = await this.initializeAgentTimelineForRegister({ + // Read history before publishing the agent: a provider failure must leave the + // session unregistered so the registration catch closes it. + const imported = await this.initializeAgentTimelineForRegister({ agentId: resolvedAgentId, + session, now, options, }); @@ -3992,29 +4011,22 @@ export class AgentManager { session, config, now, - timelineWasPresent, options, }); - // Read history before publishing the agent: a provider failure must leave the - // session unregistered so the registration catch closes it. - const startupHistory: AgentStreamEvent[] = []; - if (session.initialTimeline?.length && !managed.historyPrimed) { - for await (const event of session.streamHistory()) { - startupHistory.push(limitAgentStreamEventContent(event)); - } - } - this.assertAcceptingAgentRegistrations(); this.agents.set(resolvedAgentId, managed); registered = true; // Initialize previousStatus to track transitions this.previousStatuses.set(resolvedAgentId, managed.lifecycle); + if (imported) { + await this.applyProviderSubagentHistory(resolvedAgentId, imported.subagents, { + dispatch: false, + }); + } if (session.initialTimeline?.length) { - if (!managed.historyPrimed) { - // Legacy/imported chats need their existing history before startup rows. - await this.primeTimelineFromLegacyProviderHistory(managed, false, startupHistory); - } else { + // An imported history already holds the startup rows. + if (!imported) { await this.timeline.append( { agentId: managed.id }, buildImportedTimelineRows(session.initialTimeline).map((row) => ({ @@ -4100,8 +4112,10 @@ export class AgentManager { return new Error(`Provider '${config.provider}' does not support MCP servers${source}`); } + /** Returns the imported provider history, or null when the timeline was seeded or kept. */ private async initializeAgentTimelineForRegister(params: { agentId: string; + session: AgentSession; now: Date; options: | { @@ -4109,24 +4123,61 @@ export class AgentManager { timelineRows?: AgentTimelineRow[]; timelineNextSeq?: number; persistence?: AgentPersistenceHandle; + importMissingHistory?: boolean; createdAt?: Date; updatedAt?: Date; } | undefined; - }): Promise<{ timelineWasPresent: boolean }> { - const { agentId, now, options } = params; + }): Promise { + const { agentId, session, now, options } = params; const scope: TimelineScope = { agentId }; const explicitTimelineSeed = buildExplicitTimelineSeedForRegister(now, options); if (explicitTimelineSeed) { - const timelineWasPresent = (await this.timeline.describe(scope)) !== null; await this.timeline.create(scope, explicitTimelineSeed); - return { timelineWasPresent }; + return null; } - const existing = await this.timeline.describe(scope); - if (!existing) { + if (await this.timeline.describe(scope)) return null; + if (!options?.importMissingHistory) { await this.timeline.create(scope, { timestamp: now.toISOString() }); + return null; + } + const history = await this.readProviderHistory(session.streamHistory()); + const result = await this.timeline.importHistory(scope, history.entries, { + timestamp: now.toISOString(), + }); + return result.kind === "imported" ? history : null; + } + + /** + * Reads a provider history to its end before anything is written, so a stream that fails + * halfway changes nothing. + */ + private async readProviderHistory( + stream: AsyncIterable, + ): Promise { + const history: ProviderHistory = { entries: [], subagents: [] }; + for await (const rawEvent of stream) { + const event = limitAgentStreamEventContent(rawEvent); + if (event.type === "provider_subagent") { + history.subagents.push(event); + } else if (event.type === "timeline") { + for (const item of projectAgentMessages(event.item)) { + history.entries.push(toTimelineEntry({ ...event, item })); + } + } + } + return history; + } + + private async applyProviderSubagentHistory( + agentId: string, + events: readonly ProviderSubagentStreamEvent[], + options: { dispatch: boolean }, + ): Promise { + for (const event of events) { + const update = await this.providerSubagents.apply(agentId, event.provider, event.event); + if (update && options.dispatch) this.dispatch({ type: "provider_subagent", event: update }); } - return { timelineWasPresent: existing !== null }; } private buildManagedAgentForRegister(params: { @@ -4134,14 +4185,12 @@ export class AgentManager { session: AgentSession; config: AgentSessionConfig; now: Date; - timelineWasPresent: boolean; options: | { createdAt?: Date; updatedAt?: Date; lastUserMessageAt?: Date | null; labels?: Record; - historyPrimed?: boolean; lastUsage?: AgentUsage; lastError?: string; attention?: AttentionState; @@ -4152,7 +4201,7 @@ export class AgentManager { } | undefined; }): ActiveManagedAgent { - const { resolvedAgentId, session, config, now, timelineWasPresent, options } = params; + const { resolvedAgentId, session, config, now, options } = params; return { id: resolvedAgentId, provider: config.provider, @@ -4183,7 +4232,6 @@ export class AgentManager { options?.persistence ?? session.describePersistence(), config, ), - historyPrimed: options?.historyPrimed ?? timelineWasPresent, lastUserMessageAt: options?.lastUserMessageAt ?? null, lastUsage: options?.lastUsage, lastError: options?.lastError, @@ -4488,147 +4536,6 @@ export class AgentManager { } } - private async hydrateTimelineFromLegacyProviderHistory( - agent: ActiveManagedAgent, - options?: HydrateTimelineOptions, - ): Promise { - if (agent.historyPrimed && !options?.force) { - return; - } - - const broadcast = options?.broadcast ?? false; - const broadcastTimeline = options?.broadcastTimeline ?? broadcast; - - if (options?.force) { - await this.forceHydrateTimelineFromLegacyProviderHistory( - agent, - typeof broadcast === "function" ? broadcast() : broadcast, - typeof broadcastTimeline === "function" ? broadcastTimeline() : broadcastTimeline, - ); - return; - } - - await this.primeTimelineFromLegacyProviderHistory(agent, broadcast); - } - - private async forceHydrateTimelineFromLegacyProviderHistory( - agent: ActiveManagedAgent, - broadcast: boolean, - broadcastTimeline: boolean, - ): Promise { - const historyEvents: Extract[] = []; - const providerSubagentEvents: Extract[] = []; - for await (const rawEvent of agent.session.streamHistory()) { - const event = limitAgentStreamEventContent(rawEvent); - if (event.type === "timeline") { - for (const item of projectAgentMessages(event.item)) { - historyEvents.push({ ...event, item }); - } - } else if (event.type === "provider_subagent") { - providerSubagentEvents.push(event); - } - } - - this.agentStreamCoalescer.flushAndDiscard(agent.id); - const scope: TimelineScope = { agentId: agent.id }; - // A new history identity: cursors from the replaced history read as stale. - await this.timeline.create(scope, { timestamp: new Date().toISOString() }); - agent.historyPrimed = true; - - for (const event of await this.providerSubagents.deleteParent(agent.id)) { - if (broadcast) { - this.dispatch({ type: "provider_subagent", event }); - } - } - for (const event of providerSubagentEvents) { - const update = await this.providerSubagents.apply(agent.id, event.provider, event.event); - if (update && broadcast) { - this.dispatch({ type: "provider_subagent", event: update }); - } - } - const replayed = await this.timeline.append(scope, historyEvents.map(toTimelineEntry), { - publish: (commits) => { - if (!broadcastTimeline) return; - commits.forEach((commit, index) => - this.publishTimelineEvent(agent.id, historyEvents[index], commit), - ); - }, - }); - this.reportUnpublished(agent.id, replayed); - this.touchUpdatedAt(agent); - this.emitState(agent); - } - - private async primeTimelineFromLegacyProviderHistory( - agent: ActiveManagedAgent, - broadcast: boolean | (() => boolean), - history: - | AsyncIterable - | Iterable = agent.session.streamHistory(), - ): Promise { - const deferredBroadcast = typeof broadcast === "function"; - const historyEvents: Extract[] = []; - const historySubagentEvents: Extract[] = []; - agent.historyPrimed = false; - try { - // Collect the whole replay before touching either store. A stream that fails - // halfway then leaves the committed timeline as it was, instead of a partial - // copy the next attempt would append to. - for await (const rawEvent of history) { - const event = limitAgentStreamEventContent(rawEvent); - if (event.type === "provider_subagent") { - historySubagentEvents.push(event); - continue; - } - if (event.type !== "timeline") { - continue; - } - for (const item of projectAgentMessages(event.item)) { - historyEvents.push({ ...event, item }); - } - } - } catch (error) { - this.logger.warn({ err: error, agentId: agent.id }, "Failed to hydrate provider history"); - throw error; - } - - const providerSubagentEvents: AgentManagerEvent[] = []; - for (const event of historySubagentEvents) { - const update = await this.providerSubagents.apply(agent.id, event.provider, event.event); - if (!update) continue; - const managerEvent: AgentManagerEvent = { type: "provider_subagent", event: update }; - if (deferredBroadcast) { - providerSubagentEvents.push(managerEvent); - } else if (broadcast) { - this.dispatch(managerEvent); - } - } - const primed = await this.timeline.append( - { agentId: agent.id }, - historyEvents.map(toTimelineEntry), - { - publish: (committed) => { - if (deferredBroadcast || !broadcast) return; - committed.forEach((commit, index) => - this.publishTimelineEvent(agent.id, historyEvents[index], commit), - ); - }, - }, - ); - this.reportUnpublished(agent.id, primed); - agent.historyPrimed = true; - - if (typeof broadcast !== "function" || !broadcast()) { - return; - } - for (const event of providerSubagentEvents) { - this.dispatch(event); - } - primed.commits.forEach((commit, index) => - this.publishTimelineEvent(agent.id, historyEvents[index], commit), - ); - } - private notifyForegroundTurnWaiters(agentId: string, event: AgentStreamEvent): void { const turnId = getAgentStreamEventTurnId(event); if (turnId == null) { @@ -5883,6 +5790,15 @@ export class AgentManager { return agent; } + /** A loaded agent, or one whose history the timeline holds without a runtime. */ + private async requireHistory(id: string): Promise { + const normalizedId = validateAgentId(id, "requireHistory"); + if (this.agents.has(normalizedId)) return; + if (!(await this.timeline.describe({ agentId: normalizedId }))) { + throw new Error(`Unknown agent '${normalizedId}'`); + } + } + private requireSessionAgent(id: string): ActiveManagedAgent { const agent = this.requireAgent(id); if (agent.session === null) { @@ -5891,6 +5807,14 @@ export class AgentManager { return agent; } + /** + * Provider children are read with their parent's history, so a parent that is not loaded is + * read from what is stored. A loaded parent must be public. + */ + private requirePublicParent(id: string): void { + if (this.agents.has(id)) this.requirePublicAgent(id); + } + private requirePublicAgent(id: string): LiveManagedAgent { const agent = this.requireAgent(id); if (agent.internal) { diff --git a/packages/server/src/server/agent/agent-projections.test.ts b/packages/server/src/server/agent/agent-projections.test.ts index ec4186b022..26452c6606 100644 --- a/packages/server/src/server/agent/agent-projections.test.ts +++ b/packages/server/src/server/agent/agent-projections.test.ts @@ -92,7 +92,6 @@ function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent persistence: { ...basePersistence }, lastUsage: undefined, lastError: lastErrorValue, - historyPrimed: true, lastUserMessageAt: now, attention: { requiresAttention: false }, }; diff --git a/packages/server/src/server/agent/agent-storage.test.ts b/packages/server/src/server/agent/agent-storage.test.ts index 723f612085..0ebb008199 100644 --- a/packages/server/src/server/agent/agent-storage.test.ts +++ b/packages/server/src/server/agent/agent-storage.test.ts @@ -125,7 +125,6 @@ function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent sessionId: overrides.sessionId ?? "session-123", }), persistence: overrides.persistence ?? null, - historyPrimed: overrides.historyPrimed ?? true, lastUserMessageAt: overrides.lastUserMessageAt ?? core.now, lastUsage: overrides.lastUsage, lastError: overrides.lastError, diff --git a/packages/server/src/server/agent/import-sessions.test.ts b/packages/server/src/server/agent/import-sessions.test.ts index a9b4e1800c..823784a2f9 100644 --- a/packages/server/src/server/agent/import-sessions.test.ts +++ b/packages/server/src/server/agent/import-sessions.test.ts @@ -100,7 +100,6 @@ function makeManagedAgent(args: { ...(args.nativeHandle ? { nativeHandle: args.nativeHandle } : {}), metadata: { provider, cwd: args.cwd }, }, - historyPrimed: true, lastUserMessageAt: null, attention: { requiresAttention: false }, foregroundTurnWaiters: new Set(), @@ -678,7 +677,6 @@ class ProviderImportHarness { this.activeAgent = this.snapshot; return this.snapshot; }, - hydrateTimelineFromProvider: async () => {}, getTimeline: async () => this.timeline, closeAgent: async (agentId: string) => { this.closedAgentIds.push(agentId); diff --git a/packages/server/src/server/agent/mcp-server.test.ts b/packages/server/src/server/agent/mcp-server.test.ts index ecf2d912a5..b706157f07 100644 --- a/packages/server/src/server/agent/mcp-server.test.ts +++ b/packages/server/src/server/agent/mcp-server.test.ts @@ -267,7 +267,7 @@ function buildAgentManagerSpies() { listAgents: vi.fn().mockReturnValue([]), getTimeline: vi.fn().mockReturnValue([]), resumeAgentFromPersistence: vi.fn(), - hydrateTimelineFromProvider: vi.fn().mockResolvedValue(undefined), + hasStoredHistory: vi.fn().mockResolvedValue(false), appendTimelineItem: vi.fn().mockResolvedValue(undefined), emitLiveTimelineItem: vi.fn().mockResolvedValue(undefined), getPluginTools: () => [], @@ -6677,12 +6677,13 @@ describe("agent snapshot MCP serialization", () => { id: "archived-activity-agent", currentModeId: "default", } as ManagedAgent; - spies.agentManager.getAgent - .mockReturnValueOnce(null) - .mockReturnValue(snapshot) - .mockReturnValue(snapshot); + let resumed = false; + spies.agentManager.getAgent.mockImplementation(() => (resumed ? snapshot : null)); spies.agentStorage.get.mockResolvedValue(record); - spies.agentManager.resumeAgentFromPersistence.mockResolvedValue(snapshot); + spies.agentManager.resumeAgentFromPersistence.mockImplementation(async () => { + resumed = true; + return snapshot; + }); spies.agentManager.getTimeline.mockReturnValue([ { kind: "status", @@ -6708,10 +6709,35 @@ describe("agent snapshot MCP serialization", () => { }), ); expect(spies.agentManager.resumeAgentFromPersistence).toHaveBeenCalled(); - expect(spies.agentManager.hydrateTimelineFromProvider).toHaveBeenCalledWith( - "archived-activity-agent", - { broadcast: expect.any(Function) }, + }); + + it("reads get_agent_activity from stored history without resuming the provider", async () => { + const { agentManager, agentStorage, spies } = createTestDeps(); + const record = createStoredRecord({ id: "stored-activity-agent", lastModeId: "plan" }); + spies.agentManager.getAgent.mockReturnValue(null); + spies.agentStorage.get.mockResolvedValue(record); + spies.agentManager.hasStoredHistory.mockResolvedValue(true); + spies.agentManager.getTimeline.mockReturnValue([ + { type: "assistant_message", text: "Stored reply" }, + ]); + + const server = await createAgentMcpServer({ + agentManager, + agentStorage, + logger, + providerSnapshotManager: createClaudeOnlyManager(), + }); + const tool = registeredTool(server, "get_agent_activity"); + const response = await tool.handler({ agentId: "stored-activity-agent" }); + + expect(response.structuredContent).toEqual( + expect.objectContaining({ + agentId: "stored-activity-agent", + updateCount: 1, + currentModeId: "plan", + }), ); + expect(spies.agentManager.resumeAgentFromPersistence).not.toHaveBeenCalled(); }); it("get_agent_activity limit counts projected messages, not raw deltas", async () => { diff --git a/packages/server/src/server/agent/provider-snapshot-manager.test.ts b/packages/server/src/server/agent/provider-snapshot-manager.test.ts index 66d86f8931..f663ccf55b 100644 --- a/packages/server/src/server/agent/provider-snapshot-manager.test.ts +++ b/packages/server/src/server/agent/provider-snapshot-manager.test.ts @@ -50,7 +50,6 @@ const CLOSED_PARENT: ManagedAgent = { inFlightPermissionResponses: new Set(), pendingReplacement: false, persistence: null, - historyPrimed: false, lastUserMessageAt: null, activeTurnId: null, activeTurnStartedAt: null, diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts index 0469d9fbe1..78c41bcc39 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts @@ -244,9 +244,8 @@ describe("Codex app-server provider (local e2e)", () => { const threadId = manager.getAgent(agentId)?.persistence?.sessionId; expect(threadId).toBeTruthy(); for (let i = 0; i < 3; i++) { - const reloaded = await manager.reloadAgentSession(agentId, undefined, { - rehydrateFromDisk: true, - }); + const reloaded = await manager.reloadAgentSession(agentId); + await manager.reconcileProviderHistory(agentId); expect(reloaded.persistence?.sessionId).toBe(threadId); expect(reloaded.lifecycle).toBe("idle"); } diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 7366d91fca..c7f45cda7f 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -610,7 +610,6 @@ function createReopenableCodexAgent(threads: Record) { async function open(): Promise { await manager.resumeAgentFromPersistence(handle, config, agentId); - await manager.hydrateTimelineFromProvider(agentId); } return { diff --git a/packages/server/src/server/agent/providers/codex-async-questions.test.ts b/packages/server/src/server/agent/providers/codex-async-questions.test.ts index a794720614..ca608a6cc7 100644 --- a/packages/server/src/server/agent/providers/codex-async-questions.test.ts +++ b/packages/server/src/server/agent/providers/codex-async-questions.test.ts @@ -185,8 +185,8 @@ test("failed rewind preserves question state", async () => { } }); -async function manage(session: CodexAppServerAgentSession) { - const manager = new AgentManager({ +function managerFor(session: CodexAppServerAgentSession): AgentManager { + return new AgentManager({ clients: { codex: { provider: "codex", @@ -199,6 +199,10 @@ async function manage(session: CodexAppServerAgentSession) { }, logger: createTestLogger(), }); +} + +async function manage(session: CodexAppServerAgentSession) { + const manager = managerFor(session); const agent = await manager.createAgent( { provider: "codex", cwd: tmpdir(), model: "gpt-5.4" }, undefined, @@ -209,10 +213,13 @@ async function manage(session: CodexAppServerAgentSession) { test("manager publishes and saves the provider state after rewind", async () => { const { session } = await setupRewind(); - const { manager, agent } = await manage(session); + const manager = managerFor(session); + // A resumed agent imports the provider history it can rewind. + const agent = await manager.resumeAgentFromPersistence(session.describePersistence(), { + cwd: tmpdir(), + model: "gpt-5.4", + }); try { - // A resumed agent's conversation holds the provider history it can rewind. - await manager.hydrateTimelineFromProvider(agent.id, { force: true }); await manager.rewind(agent.id, "rewind-here", "conversation"); expect(session.describePersistence().sessionId).toBe("forked-thread"); expect( diff --git a/packages/server/src/server/agent/rewind/rewind.test.ts b/packages/server/src/server/agent/rewind/rewind.test.ts index f2e0ec79f3..f66bd41c53 100644 --- a/packages/server/src/server/agent/rewind/rewind.test.ts +++ b/packages/server/src/server/agent/rewind/rewind.test.ts @@ -4,6 +4,7 @@ import { join } from "node:path"; import { afterEach, describe, expect, test } from "vitest"; import { createTestLogger } from "../../../test-utils/test-logger.js"; +import { ensureAgentLoaded, loadAgentHistory } from "../agent-loading.js"; import { AgentManager, type AgentManagerEvent } from "../agent-manager.js"; import { RewindNotAppliedError, @@ -87,6 +88,12 @@ async function createRewindHarness( return { manager, session, client, backend, agentId: agent.id }; } +/** What the Reload agent action does for a loaded agent. */ +async function reloadAgent(manager: AgentManager, agentId: string) { + await manager.reloadAgentSession(agentId); + return manager.reconcileProviderHistory(agentId); +} + /** A conversation the provider answers, with no history before Paseo's first prompt. */ async function createConversationHarness(options: { registry?: AgentStorage } = {}) { const session = new FakeRewindSession(); @@ -279,6 +286,48 @@ describe("AgentManager conversation rewind", () => { ]); }); + test("archiving and unarchiving after A, B, rewind B, C keeps C and leaves B retired", async () => { + const registry = await createFlakyStorage(); + const harness = await createConversationHarness({ registry }); + const { manager, agentId, session, client } = harness; + const loader = { agentManager: manager, agentStorage: registry, logger: createTestLogger() }; + await harness.prompt("A"); + await harness.prompt("B"); + await manager.rewind(agentId, "B", "conversation"); + await harness.prompt("C"); + const before = await manager.fetchTimeline(agentId, { limit: 0 }); + + await manager.archiveAgent(agentId); + + // The archived chat reads from its stored history, without starting the provider. + expect(manager.getAgent(agentId)).toBeNull(); + expect(await loadAgentHistory(agentId, loader)).toMatchObject({ kind: "stored" }); + expect(await manager.fetchTimeline(agentId, { limit: 0 })).toEqual(before); + expect(client.resumedSessionIds).toEqual([]); + + // Unarchive, as the refresh request runs it: load the agent, then reconcile its history. + await manager.unarchiveSnapshot(agentId); + await ensureAgentLoaded(agentId, loader); + expect(await manager.reconcileProviderHistory(agentId)).toEqual({ kind: "unchanged" }); + await harness.prompt("D"); + + expect(client.resumedSessionIds).toHaveLength(1); + expect(await harness.activeTexts()).toEqual([ + "A", + "reply to A", + "C", + "reply to C", + "D", + "reply to D", + ]); + expect(await manager.fetchTimeline(agentId, { limit: 0 })).toMatchObject({ + epoch: before.epoch, + retirementCount: 1, + }); + expect(harness.retirements()).toHaveLength(1); + expect(session.recordedRewinds).toHaveLength(1); + }); + test("reloading settles an unknown outcome the provider history proves applied", async () => { const harness = await createConversationHarness(); const { manager, agentId, session } = harness; @@ -287,7 +336,7 @@ describe("AgentManager conversation rewind", () => { session.nextRewind = { kind: "unknown", applied: true }; await expect(manager.rewind(agentId, "B", "conversation")).rejects.toThrow(); - await manager.reloadAgentSession(agentId, undefined, { rehydrateFromDisk: true }); + await reloadAgent(manager, agentId); await harness.prompt("C"); expect(session.recordedRewinds).toHaveLength(1); @@ -303,7 +352,7 @@ describe("AgentManager conversation rewind", () => { session.nextRewind = { kind: "unknown", applied: false }; await expect(manager.rewind(agentId, "B", "conversation")).rejects.toThrow(); - await manager.reloadAgentSession(agentId, undefined, { rehydrateFromDisk: true }); + await reloadAgent(manager, agentId); await harness.prompt("C"); expect(harness.retirements()).toEqual([]); @@ -327,7 +376,7 @@ describe("AgentManager conversation rewind", () => { // A provider that replays no history leaves nothing to compare. session.history = []; - await manager.reloadAgentSession(agentId, undefined, { rehydrateFromDisk: true }); + await reloadAgent(manager, agentId); await expect(harness.prompt("C")).rejects.toThrow( "Paseo could not confirm whether the conversation was rewound.", @@ -353,7 +402,7 @@ describe("AgentManager conversation rewind", () => { // Refused before reaching the provider, so the provider history has no C. await expect(harness.prompt("C")).rejects.toThrow(); - await manager.reloadAgentSession(agentId, undefined, { rehydrateFromDisk: true }); + await reloadAgent(manager, agentId); await harness.prompt("D"); expect(harness.retirements()).toHaveLength(1); @@ -369,7 +418,7 @@ describe("AgentManager conversation rewind", () => { await expect(manager.rewind(agentId, "B", "conversation")).rejects.toThrow(); await expect(harness.prompt("C")).rejects.toThrow(); - await manager.reloadAgentSession(agentId, undefined, { rehydrateFromDisk: true }); + await reloadAgent(manager, agentId); await harness.prompt("D"); expect(harness.retirements()).toEqual([]); @@ -400,14 +449,15 @@ describe("AgentManager conversation rewind", () => { ); expect(harness.retirements()).toEqual([]); - await manager.reloadAgentSession(agentId, undefined, { rehydrateFromDisk: true }); + await reloadAgent(manager, agentId); await harness.prompt("C"); expect(client.resumedSessionIds).toEqual(["rewound-session"]); expect((await registry.get(agentId))?.persistence?.sessionId).toBe("rewound-session"); - // The reload committed the recorded cut on the same timeline; it did not replay history. + // The reload committed the recorded cut on the same timeline, without judging the replay: + // its one history read is the reconcile after the rewind settled. expect(session.recordedRewinds).toHaveLength(1); - expect(session.historyReadCount).toBe(0); + expect(session.historyReadCount).toBe(1); expect(harness.retirements()).toEqual([ expect.objectContaining({ epoch: before.epoch, retireFromSeq: before.rows[2]!.seqStart }), ]); diff --git a/packages/server/src/server/agent/timeline/agent-timeline.test.ts b/packages/server/src/server/agent/timeline/agent-timeline.test.ts index 6ce6a121a2..777528761b 100644 --- a/packages/server/src/server/agent/timeline/agent-timeline.test.ts +++ b/packages/server/src/server/agent/timeline/agent-timeline.test.ts @@ -1034,3 +1034,388 @@ describe("conversation rewind", () => { expect(pages).toBe(20); }); }); + +describe("provider history", () => { + const scope = root("agent"); + + function prompt(messageId: string): TimelineEntry { + return { item: { type: "user_message", text: messageId, messageId } }; + } + + function submitted(clientMessageId: string): TimelineEntry { + return { + item: { + type: "user_message", + text: clientMessageId, + messageId: clientMessageId, + clientMessageId, + }, + }; + } + + function reply(text: string): TimelineEntry { + return { item: { type: "assistant_message", text } }; + } + + function texts(rows: readonly { item: AgentTimelineItem }[]): string[] { + return rows.map((row) => ("text" in row.item ? row.item.text : row.item.type)); + } + + function committedTexts(commits: readonly { row: { item: AgentTimelineItem } }[]): string[] { + return texts(commits.map((commit) => commit.row)); + } + + async function owned(entries: TimelineEntry[]): Promise<{ + backend: DeferredTimelineBackend; + timeline: AgentTimeline; + }> { + const backend = new DeferredTimelineBackend(); + const timeline = new AgentTimeline(backend); + await timeline.create(scope, { epoch: "e" }); + await timeline.append(scope, entries); + return { backend, timeline }; + } + + it("imports history into a scope that has none, once", async () => { + const timeline = new AgentTimeline(new DeferredTimelineBackend()); + const history = [prompt("a"), reply("A"), prompt("b"), reply("B")]; + + const first = await timeline.importHistory(scope, history); + const repeat = await timeline.importHistory(scope, history); + + expect(first).toMatchObject({ kind: "imported", head: { size: 4, retirementCount: 0 } }); + expect(repeat).toEqual({ kind: "present", head: first.head }); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "b", "B"]); + const tail = await timeline.fetch(scope); + expect(tail).toMatchObject({ epoch: first.head.epoch, startSeq: 1, endSeq: 4 }); + }); + + it("leaves existing history alone when asked to import", async () => { + const { timeline } = await owned([prompt("a"), reply("A")]); + + const result = await timeline.importHistory(scope, [prompt("x"), reply("X")]); + + expect(result).toMatchObject({ kind: "present", head: { epoch: "e", size: 2 } }); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A"]); + }); + + function tool(callId: string, output: string, status: "running" | "completed" = "completed") { + const detail = { type: "shell" as const, command: "npm test", output }; + const item: AgentTimelineItem = + status === "running" + ? { type: "tool_call", callId, name: "shell", status, error: null, detail } + : { type: "tool_call", callId, name: "shell", status, error: null, detail }; + return { item }; + } + + it("changes nothing when the provider holds the same conversation", async () => { + const { backend, timeline } = await owned([ + prompt("a"), + tool("t1", "3 passed"), + reply("Tests "), + reply("pass."), + // Paseo writes notifications itself; the provider has none. + { item: { type: "notification", level: "info", message: "Agent resumed" } }, + prompt("b"), + reply("B"), + ]); + const appendsBefore = backend.appends.length; + const published: unknown[] = []; + + const result = await timeline.reconcileProviderHistory( + scope, + [ + prompt("a"), + // The same tool call, its fields in another order. + { + item: { + detail: { output: "3 passed", command: "npm test", type: "shell" }, + error: null, + status: "completed", + name: "shell", + callId: "t1", + type: "tool_call", + }, + }, + // The provider stores the answer whole; Paseo streamed it in chunks. + reply("Tests pass."), + prompt("b"), + reply("B"), + ], + { publish: (commits) => published.push(commits) }, + ); + + expect(result).toEqual({ kind: "unchanged" }); + expect(published).toEqual([]); + expect(backend.appends.length).toBe(appendsBefore); + expect(await timeline.describe(scope)).toMatchObject({ epoch: "e", retirementCount: 0 }); + }); + + it("appends the turns the provider has after the owned conversation", async () => { + const { timeline } = await owned([prompt("a"), reply("A")]); + const published: TimelineCommit[][] = []; + + const result = await timeline.reconcileProviderHistory( + scope, + [prompt("a"), reply("A"), prompt("b"), reply("B")], + { publish: (commits) => published.push(commits) }, + ); + + expect(result).toMatchObject({ kind: "appended" }); + expect(published.map(committedTexts)).toEqual([["b", "B"]]); + expect(published[0].map((commit) => commit.row.seq)).toEqual([3, 4]); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "b", "B"]); + expect(await timeline.describe(scope)).toMatchObject({ epoch: "e", retirementCount: 0 }); + }); + + it.each([ + { + case: "the turn has no output yet", + owned: [prompt("a"), reply("A"), prompt("b")], + appended: ["tool_call", "B"], + }, + { + case: "the turn ends on a finished tool call", + owned: [prompt("a"), reply("A"), prompt("b"), tool("t1", "3 passed")], + appended: ["B"], + }, + ])("appends the rows the provider has after the last turn when $case", async (example) => { + const { timeline } = await owned(example.owned); + const before = await timeline.readRows(scope); + + const result = await timeline.reconcileProviderHistory(scope, [ + prompt("a"), + reply("A"), + prompt("b"), + tool("t1", "3 passed"), + reply("B"), + ]); + + expect(result).toMatchObject({ kind: "appended" }); + if (result.kind !== "appended") throw new Error("expected an append"); + expect(committedTexts(result.commits)).toEqual(example.appended); + // The rows Paseo already showed are the same rows, unchanged. + expect((await timeline.readRows(scope)).slice(0, before.length)).toEqual(before); + expect(await timeline.describe(scope)).toMatchObject({ epoch: "e", retirementCount: 0 }); + }); + + it.each([ + { + case: "an earlier answer differs under the same prompts", + provider: [prompt("a"), reply("edited"), prompt("b"), reply("B"), prompt("c"), reply("C")], + reason: "The provider history has a different answer", + }, + { + case: "the last answer differs under the same prompts", + provider: [prompt("a"), reply("A"), prompt("b"), reply("something else"), prompt("c")], + reason: "The provider history has a different last answer", + }, + { + case: "the provider's last answer is shorter than Paseo's", + provider: [prompt("a"), reply("A"), prompt("b"), reply("B"), prompt("c")], + owned: [prompt("a"), reply("A"), prompt("b"), reply("B and more")], + reason: "The provider history has a different last answer", + }, + { + case: "the provider finished an answer Paseo saw only part of", + owned: [prompt("a"), reply("A"), prompt("b"), reply("The answer is")], + provider: [prompt("a"), reply("A"), prompt("b"), reply("The answer is 42."), prompt("c")], + reason: "The provider history has a different last answer", + }, + { + case: "a tool call Paseo saw running has finished", + owned: [prompt("a"), reply("A"), prompt("b"), tool("t1", "", "running")], + provider: [prompt("a"), reply("A"), prompt("b"), tool("t1", "3 passed"), reply("B")], + reason: "The provider history has a different last answer", + }, + { + case: "only the indentation of an answer differs", + owned: [prompt("a"), reply("def f():\n return 1"), prompt("b"), reply("B")], + provider: [prompt("a"), reply("def f():\n return 1"), prompt("b"), reply("B")], + reason: "The provider history has a different answer", + }, + { + case: "only the spacing inside a string differs", + owned: [prompt("a"), reply("A"), prompt("b"), reply('print("a b")')], + provider: [prompt("a"), reply("A"), prompt("b"), reply('print("a b")'), prompt("c")], + reason: "The provider history has a different last answer", + }, + { + case: "the answer is the same but a tool result differs", + owned: [prompt("a"), tool("t1", "3 passed"), reply("Done"), prompt("b"), reply("B")], + provider: [prompt("a"), tool("t1", "1 failed"), reply("Done"), prompt("b"), reply("B")], + reason: "The provider history has a different answer", + }, + { + case: "the prompts differ", + provider: [prompt("a"), reply("A"), prompt("x"), reply("B"), prompt("c")], + reason: "The provider history has different prompts", + }, + { + case: "the provider dropped a turn Paseo shows", + provider: [prompt("a"), reply("A")], + reason: "The provider history is missing turns Paseo shows", + }, + ])("keeps the owned history when $case", async (example) => { + const { backend, timeline } = await owned( + example.owned ?? [prompt("a"), reply("A"), prompt("b"), reply("B")], + ); + const before = await timeline.readRows(scope); + const appendsBefore = backend.appends.length; + + const result = await timeline.reconcileProviderHistory(scope, example.provider); + + expect(result).toEqual({ kind: "diverged", reason: example.reason }); + expect(await timeline.readRows(scope)).toEqual(before); + expect(backend.appends.length).toBe(appendsBefore); + expect(await timeline.describe(scope)).toMatchObject({ epoch: "e", retirementCount: 0 }); + }); + + function promptText(messageId: string, text: string): TimelineEntry { + return { item: { type: "user_message", text, messageId } }; + } + + it.each([ + { + case: "an earlier prompt was edited under the same id", + provider: [promptText("a", "a, edited"), reply("A"), prompt("b"), reply("B"), prompt("c")], + }, + { + case: "the last prompt was edited under the same id", + provider: [prompt("a"), reply("A"), promptText("b", "b, edited"), reply("B"), prompt("c")], + }, + { + case: "only the spacing inside a prompt differs", + owned: [promptText("a", "fix it"), reply("A"), prompt("b"), reply("B")], + provider: [promptText("a", "fix it"), reply("A"), prompt("b"), reply("B")], + }, + { + case: "only a prompt's trailing whitespace differs", + provider: [promptText("a", "a "), reply("A"), prompt("b"), reply("B")], + }, + ])("keeps the owned history when $case", async (example) => { + const { backend, timeline } = await owned( + example.owned ?? [prompt("a"), reply("A"), prompt("b"), reply("B")], + ); + const before = await timeline.readRows(scope); + const appendsBefore = backend.appends.length; + + const result = await timeline.reconcileProviderHistory(scope, example.provider); + + expect(result).toEqual({ + kind: "diverged", + reason: "The provider history has different prompts", + }); + expect(await timeline.readRows(scope)).toEqual(before); + expect(backend.appends.length).toBe(appendsBefore); + }); + + it("matches a prompt Paseo submitted to the provider's stored copy of it", async () => { + const typed = "Fix the login bug.\n\n Keep the API as it is."; + const { timeline } = await owned([ + { item: { type: "user_message", text: typed, messageId: "c-1", clientMessageId: "c-1" } }, + ]); + await timeline.attachProviderMessageId(scope, "c-1", "p-1"); + await timeline.append(scope, [reply("Done.")]); + + const result = await timeline.reconcileProviderHistory(scope, [ + promptText("p-1", typed), + reply("Done."), + prompt("b"), + reply("B"), + ]); + + expect(result).toMatchObject({ kind: "appended" }); + expect(texts(await timeline.readRows(scope))).toEqual([typed, "Done.", "b", "B"]); + }); + + it("reads a prompt the provider repeats unchanged as one turn", async () => { + const { timeline } = await owned([prompt("a"), reply("A")]); + + const result = await timeline.reconcileProviderHistory(scope, [ + prompt("a"), + prompt("a"), + reply("A"), + ]); + + expect(result).toEqual({ kind: "unchanged" }); + }); + + it("keeps the owned history when the provider repeats a prompt id with other text", async () => { + const { backend, timeline } = await owned([prompt("a")]); + const appendsBefore = backend.appends.length; + + const result = await timeline.reconcileProviderHistory(scope, [ + prompt("a"), + promptText("a", "something else"), + reply("A"), + ]); + + expect(result).toEqual({ + kind: "diverged", + reason: "The provider history has two prompts with one id", + }); + expect(texts(await timeline.readRows(scope))).toEqual(["a"]); + expect(backend.appends.length).toBe(appendsBefore); + }); + + it("does not append provider turns after a prompt the provider never received", async () => { + const { timeline } = await owned([prompt("a"), reply("A"), submitted("refused")]); + + const result = await timeline.reconcileProviderHistory(scope, [ + prompt("a"), + reply("A"), + prompt("b"), + reply("B"), + ]); + + expect(result).toEqual({ + kind: "diverged", + reason: "A prompt the provider did not receive is last", + }); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "refused"]); + }); + + it("keeps a rewound turn retired while appending what the provider added after it", async () => { + const { timeline } = await owned([prompt("a"), reply("A"), prompt("b"), reply("B")]); + const intent = await timeline.beginRewind(scope, "b"); + await timeline.recordRewindCut(scope, intent.id, { kind: "at_target" }); + await timeline.completeRewind(scope, intent.id); + await timeline.append(scope, [prompt("c"), reply("C")]); + + const stale = await timeline.reconcileProviderHistory(scope, [ + prompt("a"), + reply("A"), + prompt("b"), + reply("B"), + prompt("c"), + reply("C"), + ]); + const continued = await timeline.reconcileProviderHistory(scope, [ + prompt("a"), + reply("A"), + prompt("c"), + reply("C"), + prompt("d"), + reply("D"), + ]); + + expect(stale).toEqual({ + kind: "diverged", + reason: "The provider history has different prompts", + }); + expect(continued).toMatchObject({ kind: "appended" }); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "c", "C", "d", "D"]); + expect(await timeline.describe(scope)).toMatchObject({ epoch: "e", retirementCount: 1 }); + }); + + it("leaves history alone while a rewind is open", async () => { + const { timeline } = await owned([prompt("a"), reply("A"), prompt("b"), reply("B")]); + await timeline.beginRewind(scope, "b"); + + const result = await timeline.reconcileProviderHistory(scope, [prompt("a"), reply("A")]); + + expect(result).toEqual({ kind: "rewind_open" }); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "b", "B"]); + }); +}); diff --git a/packages/server/src/server/agent/timeline/index.ts b/packages/server/src/server/agent/timeline/index.ts index 294797bd34..a9992da4cb 100644 --- a/packages/server/src/server/agent/timeline/index.ts +++ b/packages/server/src/server/agent/timeline/index.ts @@ -7,7 +7,9 @@ export { type TimelineCommit, type TimelineEntry, type TimelineHead, + type TimelineImportResult, type TimelineIngestResult, + type TimelineReconcileResult, type TimelineRetirement, type TimelineRetirementCommit, type TimelineRewindCompletion, diff --git a/packages/server/src/server/agent/timeline/internal/agent-timeline.ts b/packages/server/src/server/agent/timeline/internal/agent-timeline.ts index 43f54b26cc..beea974a65 100644 --- a/packages/server/src/server/agent/timeline/internal/agent-timeline.ts +++ b/packages/server/src/server/agent/timeline/internal/agent-timeline.ts @@ -18,6 +18,7 @@ import type { } from "./backend.js"; import { UnknownTimelineError } from "./backend.js"; import { MemoryTimelineBackend } from "./memory-backend.js"; +import { planProviderHistory } from "./provider-history.js"; export type { TimelineEntry, @@ -102,6 +103,23 @@ export type TimelineAppendResult = /** The rows are committed but `publish` threw; see `TimelineIngestResult`. */ | { kind: "unpublished"; commits: TimelineCommit[]; error: unknown }; +export type TimelineImportResult = + | { kind: "imported"; head: TimelineHead } + /** The scope already has history; nothing changed. */ + | { kind: "present"; head: TimelineHead }; + +export type TimelineReconcileResult = + /** The provider's history holds what the owned history holds. Nothing changed. */ + | { kind: "unchanged" } + /** The provider continued the conversation; its turns were appended. */ + | { kind: "appended"; commits: TimelineCommit[] } + /** Committed, but `publish` threw; see `TimelineIngestResult`. */ + | { kind: "unpublished"; commits: TimelineCommit[]; error: unknown } + /** The provider's history disagrees with the owned history, which is kept unchanged. */ + | { kind: "diverged"; reason: string } + /** A rewind is open; it settles first. Nothing changed. */ + | { kind: "rewind_open" }; + export interface TimelineIngestOptions { /** * Runs right after the commit, before any later operation on the agent's timeline starts, so @@ -146,6 +164,53 @@ export class AgentTimeline { }); } + /** + * Creates the scope's history from the provider's, in one commit, when the scope has none. + * Existing history is left as it is, so repeating an import adds nothing. + */ + importHistory( + scope: TimelineScope, + entries: readonly TimelineEntry[], + options: { timestamp?: string } = {}, + ): Promise { + return this.run(scope, async () => { + const existing = await this.backend.head(scope); + if (existing) return { kind: "present", head: existing }; + const head = await this.backend.create(scope, { + entries: entries.map(boundEntry), + timestamp: options.timestamp, + }); + return { kind: "imported", head: this.track(scope, head) }; + }); + } + + /** + * Appends what the provider's history has after the owned history, which is never changed. When + * the histories differ in any compared content the result is `diverged` and nothing is written; + * see `planProviderHistory` for what is compared and what continuation is supported. + */ + reconcileProviderHistory( + scope: TimelineScope, + entries: readonly TimelineEntry[], + options?: { publish?: (commits: TimelineCommit[]) => void }, + ): Promise { + return this.run(scope, async () => { + const head = await this.requireHead(scope); + if (head.rewind) return { kind: "rewind_open" }; + const plan = planProviderHistory({ + rows: await this.backend.readRows(scope), + prompts: await this.backend.prompts(scope), + provider: entries.map(boundEntry), + }); + if (plan.kind !== "append") return plan; + const commits = await this.commit(scope, plan.entries); + const failure = publishCommitted(options?.publish, commits); + return failure + ? { kind: "unpublished", commits, error: failure.error } + : { kind: "appended", commits }; + }); + } + delete(scope: TimelineScope): Promise { return this.run(scope, async () => { await this.backend.delete(scope); @@ -346,11 +411,7 @@ export class AgentTimeline { scope: TimelineScope, entries: readonly TimelineEntry[], ): Promise { - const bounded = entries.map((entry) => ({ - ...entry, - item: limitAgentTimelineItemContent(entry.item), - })); - const { rows, head } = await this.backend.append(scope, bounded); + const { rows, head } = await this.backend.append(scope, entries.map(boundEntry)); this.track(scope, head); return rows.map((row) => ({ row, @@ -385,6 +446,10 @@ export class AgentTimeline { function settledOperation(): void {} +function boundEntry(entry: TimelineEntry): TimelineEntry { + return { ...entry, item: limitAgentTimelineItemContent(entry.item) }; +} + interface PublishFailure { error: unknown; } diff --git a/packages/server/src/server/agent/timeline/internal/provider-history.ts b/packages/server/src/server/agent/timeline/internal/provider-history.ts new file mode 100644 index 0000000000..0cabf1d253 --- /dev/null +++ b/packages/server/src/server/agent/timeline/internal/provider-history.ts @@ -0,0 +1,280 @@ +import type { AgentTimelineItem } from "../../agent-sdk-types.js"; +import { mergeToolCallItems, type ProjectedTimelineRow } from "../../timeline-projection.js"; +import type { TimelineEntry, TimelinePrompt } from "./backend.js"; + +/** + * What the provider's history adds to the owned history. Owned rows are never changed: the plan + * only appends rows the provider has after them. + */ +export type ProviderHistoryPlan = + | { kind: "unchanged" } + | { kind: "append"; entries: TimelineEntry[] } + | { kind: "diverged"; reason: string }; + +type ToolCallItem = Extract; + +/** + * A run of turn content compared as a whole. Consecutive assistant text joins into one text + * segment, as the projection joins streamed chunks; a tool call folds its updates by `callId`. + */ +type Segment = + | { kind: "text" | "reasoning"; text: string } + | { kind: "tool_call"; item: ToolCallItem } + | { kind: "item"; item: AgentTimelineItem }; + +interface Turn { + /** The provider's id for the prompt that opens the turn. */ + promptId: string | undefined; + /** The opening prompt's text, as the timeline shows it. */ + promptText: string; + segments: Segment[]; +} + +interface OwnedTurn extends Turn { + /** A prompt Paseo submitted that the provider never acknowledged follows the opening prompt. */ + hasUndeliveredPrompt: boolean; +} + +interface ProviderTurn extends Turn { + /** Index of the opening prompt in the provider entries. */ + start: number; + /** Index of the first and of the last provider entry of each segment. */ + segmentStarts: number[]; + segmentEnds: number[]; +} + +/** + * Turns are aligned by the provider's prompt ids and compared exactly, prompt text included. A + * provider prompt id repeated with other text is divergence. Every owned turn but the + * last must equal the provider's. The last may be a whole-segment prefix of the provider's, whose + * further rows are appended as new rows. A segment that differs at all, including an answer Paseo + * saw only part of or a tool call that finished since, is divergence, and nothing is written. + * + * Compared: assistant and reasoning text exactly, tool calls by `callId`, `name`, `status`, + * `error` and `detail`, and every other item except the ones Paseo writes itself (errors, + * notifications, plugin rows). Tool call `metadata` is not compared. + */ +export function planProviderHistory(input: { + rows: readonly ProjectedTimelineRow[]; + prompts: readonly TimelinePrompt[]; + provider: readonly TimelineEntry[]; +}): ProviderHistoryPlan { + const owned = ownedTurns(input.rows, input.prompts); + const provider = providerTurns(input.provider); + if ([...owned.turns, ...provider].some((turn) => turn.promptId === undefined)) { + return { kind: "diverged", reason: "A prompt has no provider id" }; + } + if (owned.turns.length > provider.length) { + return { kind: "diverged", reason: "The provider history is missing turns Paseo shows" }; + } + if (provider.some((turn, index) => index > 0 && turn.promptId === provider[index - 1].promptId)) { + return { kind: "diverged", reason: "The provider history has two prompts with one id" }; + } + const promptsDiffer = owned.turns.some( + (turn, index) => + turn.promptId !== provider[index].promptId || turn.promptText !== provider[index].promptText, + ); + if (promptsDiffer) { + return { kind: "diverged", reason: "The provider history has different prompts" }; + } + const last = owned.turns.length - 1; + const earlierDiffers = owned.turns + .slice(0, Math.max(last, 0)) + .some((turn, index) => !sameSegments(turn.segments, provider[index].segments)); + if (earlierDiffers) { + return { kind: "diverged", reason: "The provider history has a different answer" }; + } + + let appendFrom: number; + if (last < 0) { + appendFrom = input.rows.length === 0 ? 0 : (provider[0]?.start ?? input.provider.length); + } else { + const from = continuationStart(owned.turns[last], provider[last]); + if (from === null) { + return { kind: "diverged", reason: "The provider history has a different last answer" }; + } + appendFrom = from ?? provider[last + 1]?.start ?? input.provider.length; + } + + const entries = input.provider.slice(appendFrom); + if (entries.length === 0) return { kind: "unchanged" }; + const trailing = last >= 0 ? owned.turns[last] : owned.preamble; + if (trailing.hasUndeliveredPrompt) { + // Appending would put provider turns after a prompt the provider never received. + return { kind: "diverged", reason: "A prompt the provider did not receive is last" }; + } + return { kind: "append", entries }; +} + +/** + * Where the provider's last-turn entries continue past the owned turn: an entry index, undefined + * when the turns are equal, or null when the owned turn is not a whole-segment prefix of the + * provider's. Appended entries must not update a segment Paseo already shows, such as a later + * update to a tool call. + */ +function continuationStart(owned: OwnedTurn, provider: ProviderTurn): number | undefined | null { + const matched = owned.segments.length; + if (!sameSegments(owned.segments, provider.segments.slice(0, matched))) return null; + if (matched === provider.segments.length) return undefined; + const from = provider.segmentStarts[matched]; + const matchedEnd = Math.max(provider.start, ...provider.segmentEnds.slice(0, matched)); + return matchedEnd < from ? from : null; +} + +function ownedTurns( + rows: readonly ProjectedTimelineRow[], + prompts: readonly TimelinePrompt[], +): { preamble: OwnedTurn; turns: OwnedTurn[] } { + const promptsBySeq = new Map(prompts.map((prompt) => [prompt.seqStart, prompt])); + const preamble: OwnedTurn = { + promptId: undefined, + promptText: "", + segments: [], + hasUndeliveredPrompt: false, + }; + const turns: OwnedTurn[] = []; + let current = preamble; + for (const row of rows) { + const prompt = row.item.type === "user_message" ? promptsBySeq.get(row.seqStart) : undefined; + if ( + prompt && + row.item.type === "user_message" && + (!prompt.submitted || prompt.providerMessageId) + ) { + current = { + promptId: prompt.providerMessageId ?? prompt.messageId, + promptText: row.item.text, + segments: [], + hasUndeliveredPrompt: false, + }; + turns.push(current); + } else if (prompt) { + current.hasUndeliveredPrompt = true; + } else { + addToSegments(current.segments, row.item); + } + } + return { preamble, turns }; +} + +function providerTurns(entries: readonly TimelineEntry[]): ProviderTurn[] { + const turns: ProviderTurn[] = []; + entries.forEach((entry, index) => { + const current = turns.at(-1); + if (entry.item.type === "user_message") { + // Some providers replay a prompt twice in a row; it is one turn. The same id with other + // text opens its own turn, which the plan reports as divergence. + const repeated = + entry.item.messageId !== undefined && + current?.promptId === entry.item.messageId && + current.promptText === entry.item.text; + if (!repeated) { + turns.push({ + promptId: entry.item.messageId, + promptText: entry.item.text, + segments: [], + start: index, + segmentStarts: [], + segmentEnds: [], + }); + } + return; + } + if (!current) return; + const segment = addToSegments(current.segments, entry.item); + if (segment === null) return; + if (segment === current.segmentStarts.length) current.segmentStarts.push(index); + current.segmentEnds[segment] = index; + }); + return turns; +} + +/** Adds an item to its turn's segments; returns the segment's index, or null if not compared. */ +function addToSegments(segments: Segment[], item: AgentTimelineItem): number | null { + const previous = segments.at(-1); + switch (item.type) { + case "assistant_message": + case "reasoning": { + const kind = item.type === "assistant_message" ? "text" : "reasoning"; + if (previous?.kind === kind) { + previous.text += item.text; + } else { + segments.push({ kind, text: item.text }); + } + return segments.length - 1; + } + case "tool_call": { + const index = segments.findIndex( + (segment) => segment.kind === "tool_call" && segment.item.callId === item.callId, + ); + const existing = segments[index]; + if (existing?.kind !== "tool_call") { + segments.push({ kind: "tool_call", item }); + return segments.length - 1; + } + existing.item = mergeToolCallItems(existing.item, item); + return index; + } + case "user_message": + case "error": + case "notification": + case "plugin": + return null; + default: + segments.push({ kind: "item", item }); + return segments.length - 1; + } +} + +function sameSegments(a: readonly Segment[], b: readonly Segment[]): boolean { + return a.length === b.length && a.every((segment, index) => sameSegment(segment, b[index])); +} + +function sameSegment(a: Segment, b: Segment): boolean { + if (a.kind === "tool_call" || b.kind === "tool_call") { + return ( + a.kind === "tool_call" && + b.kind === "tool_call" && + sameValue(toolCallContent(a.item), toolCallContent(b.item)) + ); + } + if (a.kind === "item" || b.kind === "item") { + return a.kind === "item" && b.kind === "item" && sameValue(a.item, b.item); + } + return a.kind === b.kind && a.text === b.text; +} + +function toolCallContent(item: ToolCallItem): unknown { + return { + callId: item.callId, + name: item.name, + status: item.status, + error: item.error, + detail: item.detail, + }; +} + +/** Structural equality: object key order and absent-versus-undefined keys are not content. */ +function sameValue(a: unknown, b: unknown): boolean { + if (a === b) return true; + if (Array.isArray(a) || Array.isArray(b)) { + return ( + Array.isArray(a) && + Array.isArray(b) && + a.length === b.length && + a.every((value, index) => sameValue(value, b[index])) + ); + } + if (!isRecord(a) || !isRecord(b)) return false; + const aKeys = definedKeys(a); + const bKeys = definedKeys(b); + return aKeys.length === bKeys.length && aKeys.every((key) => sameValue(a[key], b[key])); +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} + +function definedKeys(value: Record): string[] { + return Object.keys(value).filter((key) => value[key] !== undefined); +} diff --git a/packages/server/src/server/agent/tools/paseo-tools.ts b/packages/server/src/server/agent/tools/paseo-tools.ts index feb73ab34f..e306f93a43 100644 --- a/packages/server/src/server/agent/tools/paseo-tools.ts +++ b/packages/server/src/server/agent/tools/paseo-tools.ts @@ -33,7 +33,7 @@ import { } from "../agent-projections.js"; import { summarizeAgentActivity } from "../activity-curator.js"; import type { AgentStorage } from "../agent-storage.js"; -import { AgentNotFoundError, ensureAgentLoaded } from "../agent-loading.js"; +import { AgentNotFoundError, loadAgentHistory } from "../agent-loading.js"; import { isStoredAgentProviderAvailable } from "../../persistence-hooks.js"; import { archiveByScope, @@ -3168,20 +3168,23 @@ function createToolCatalog( }, }, async ({ agentId, limit }) => { - await ensureAgentLoaded(agentId, { + const source = await loadAgentHistory(agentId, { agentManager, agentStorage, logger: childLogger, }); const timeline = await agentManager.getTimeline(agentId); - const snapshot = agentManager.getAgent(agentId); + const currentModeId = + source.kind === "live" + ? (agentManager.getAgent(agentId) ?? source.agent).currentModeId + : (source.record.lastModeId ?? null); return { content: [], structuredContent: ensureValidJson({ agentId, ...summarizeAgentActivity(timeline, limit), - currentModeId: snapshot?.currentModeId ?? null, + currentModeId: currentModeId ?? null, }), }; }, diff --git a/packages/server/src/server/cli-agent-logs.test.ts b/packages/server/src/server/cli-agent-logs.test.ts index 634d0615c0..597610bad6 100644 --- a/packages/server/src/server/cli-agent-logs.test.ts +++ b/packages/server/src/server/cli-agent-logs.test.ts @@ -114,8 +114,10 @@ test("logs --since filters history and live output through the CLI and daemon", title: "logs since", }); const manager = ctx.daemon.daemon.agentManager; - await manager.appendTimelineItem(agent.id, { type: "user_message", text: "OLD_USER" }); - await manager.appendTimelineItem(agent.id, { type: "assistant_message", text: "OLD_REPLY" }); + const oldUser = { type: "user_message", text: "OLD_USER", messageId: "old-user" } as const; + const oldReply = { type: "assistant_message", text: "OLD_REPLY" } as const; + await manager.appendTimelineItem(agent.id, oldUser); + await manager.appendTimelineItem(agent.id, oldReply); const host = `127.0.0.1:${ctx.daemon.port}`; const logs = (args: string[]) => runLocalPaseo(["--host", host, ...args], {}, ctx.daemon.paseoHome); @@ -218,16 +220,25 @@ test("logs --since filters history and live output through the CLI and daemon", await Promise.all([ waitForCliOutput(futureFollow, "AFTER_CUTOFF"), (async () => { - await manager.appendTimelineItem(agent.id, { + const liveUser = { type: "user_message", text: "LIVE_BEFORE_CUTOFF", - }); - historyEvents.push({ - type: "timeline", - provider: "codex", - timestamp: "2110-01-01T00:00:00Z", - item: { type: "user_message", text: "AFTER_CUTOFF" }, - }); + messageId: "live-user", + } as const; + await manager.appendTimelineItem(agent.id, liveUser); + // The provider's history continues the chat with a turn Paseo has not seen, which + // Reload agent appends with the provider's timestamp. + historyEvents.push( + { type: "timeline", provider: "codex", item: oldUser }, + { type: "timeline", provider: "codex", item: oldReply }, + { type: "timeline", provider: "codex", item: liveUser }, + { + type: "timeline", + provider: "codex", + timestamp: "2110-01-01T00:00:00Z", + item: { type: "user_message", text: "AFTER_CUTOFF", messageId: "after-cutoff" }, + }, + ); await ctx.client.refreshAgent(agent.id); })(), ]); diff --git a/packages/server/src/server/daemon-e2e/agent-refresh-rehydrates-timeline.e2e.test.ts b/packages/server/src/server/daemon-e2e/agent-refresh-rehydrates-timeline.e2e.test.ts index 301df35e4b..9999a41833 100644 --- a/packages/server/src/server/daemon-e2e/agent-refresh-rehydrates-timeline.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/agent-refresh-rehydrates-timeline.e2e.test.ts @@ -4,7 +4,17 @@ import { tmpdir } from "node:os"; import path from "node:path"; import pino from "pino"; +import type { + AgentPersistenceHandle, + AgentSession, + AgentSessionConfig, + AgentStreamEvent, +} from "../agent/agent-sdk-types.js"; import { ClaudeAgentClient } from "../agent/providers/claude/agent.js"; +import { + MockLoadTestAgentClient, + MockLoadTestAgentSession, +} from "../agent/providers/mock-load-test-agent.js"; import { DaemonClient } from "../test-utils/daemon-client.js"; import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js"; @@ -54,7 +64,7 @@ function timelineText(entries: ReadonlyArray<{ item: { type: string; text?: stri .join("\n"); } -describe("daemon E2E - refresh rehydrates timeline from on-disk session", () => { +describe("daemon E2E - Reload agent brings in what the on-disk session added", () => { let claudeConfigDir: string; let prevClaudeConfigDir: string | undefined; let cwd: string; @@ -99,7 +109,7 @@ describe("daemon E2E - refresh rehydrates timeline from on-disk session", () => } }, 60_000); - test("refresh picks up entries appended externally and advances the epoch", async () => { + test("Reload agent appends entries added externally and keeps the history", async () => { const logger = pino({ level: "silent" }); daemon = await createTestPaseoDaemon({ agentClients: { @@ -151,6 +161,150 @@ describe("daemon E2E - refresh rehydrates timeline from on-disk session", () => expect(afterText).toContain("second hello"); expect(afterText).toContain("second reply"); expect(after.entries.length).toBeGreaterThan(countBefore); - expect(after.epoch).not.toBe(epochBefore); + // The kept rows are the same rows, so cursors clients hold stay valid. + expect(after.epoch).toBe(epochBefore); + expect(after.entries.slice(0, countBefore)).toEqual(before.entries); + + // Reloading again finds nothing new. + await client.refreshAgent(imported.id); + const again = await client.fetchAgentTimeline(imported.id, { + direction: "tail", + limit: 0, + projection: "canonical", + }); + expect(again.epoch).toBe(epochBefore); + expect(again.entries).toEqual(after.entries); + }, 30_000); + + test("Reload agent reports an external edit to an earlier answer and keeps the history", async () => { + const logger = pino({ level: "silent" }); + daemon = await createTestPaseoDaemon({ + agentClients: { + claude: new ClaudeAgentClient({ + logger, + resolveBinary: async () => "/test/claude/bin", + }), + }, + logger, + }); + client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + await client.connect(); + await client.fetchAgents({ subscribe: {} }); + const imported = await client.importAgent({ provider: "claude", sessionId, cwd }); + const before = await client.fetchAgentTimeline(imported.id, { + direction: "tail", + limit: 0, + projection: "canonical", + }); + + // The session file now holds a different first answer, with only its spacing changed. + const edited: ClaudeJsonlEntry[] = [ + userEntry(sessionId, cwd, "first hello", "user-uuid-1"), + assistantEntry(sessionId, cwd, "first reply"), + userEntry(sessionId, cwd, "second hello", "user-uuid-2"), + assistantEntry(sessionId, cwd, "second reply"), + ]; + writeFileSync(sessionFile, `${edited.map((entry) => JSON.stringify(entry)).join("\n")}\n`); + + const refreshed = await client.refreshAgent(imported.id); + + expect(refreshed.historyDiverged).toBe(true); + const after = await client.fetchAgentTimeline(imported.id, { + direction: "tail", + limit: 0, + projection: "canonical", + }); + expect(after.epoch).toBe(before.epoch); + expect(after.entries).toEqual(before.entries); + }, 30_000); +}); + +// A provider whose sessions report child agents, and that counts every session it resumes. +class ChildReportingSession extends MockLoadTestAgentSession { + private readonly observers = new Set<(event: AgentStreamEvent) => void>(); + override subscribe(callback: (event: AgentStreamEvent) => void): () => void { + const unsubscribe = super.subscribe(callback); + this.observers.add(callback); + return () => { + unsubscribe(); + this.observers.delete(callback); + }; + } + push(event: AgentStreamEvent): void { + for (const callback of this.observers) callback(event); + } +} + +class ChildReportingProvider extends MockLoadTestAgentClient { + session!: ChildReportingSession; + resumes = 0; + override async createSession(config: AgentSessionConfig): Promise { + this.session = new ChildReportingSession({ config, sessionId: "child-reporting-session" }); + return this.session; + } + override async resumeSession( + handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + this.resumes += 1; + return super.resumeSession(handle, config); + } +} + +describe("daemon E2E - an archived agent's history reads without its provider", () => { + let daemon: TestPaseoDaemon | undefined; + let client: DaemonClient | undefined; + let cwd: string; + + beforeEach(() => { + cwd = mkdtempSync(path.join(tmpdir(), "archived-history-")); + }); + + afterEach(async () => { + await client?.close().catch(() => undefined); + await daemon?.close().catch(() => undefined); + client = undefined; + daemon = undefined; + rmSync(cwd, { recursive: true, force: true }); + }, 60_000); + + test("the provider children of an archived agent open without resuming the provider", async () => { + const provider = new ChildReportingProvider(); + daemon = await createTestPaseoDaemon({ isDev: true, agentClients: { mock: provider } }); + client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + await client.connect(); + await client.fetchAgents({ subscribe: {} }); + const agent = await client.createAgent({ provider: "mock", cwd, model: "ten-second-stream" }); + provider.session.push({ + type: "provider_subagent", + provider: "mock", + event: { type: "upsert", id: "child", title: "Child", status: "completed" }, + }); + provider.session.push({ + type: "provider_subagent", + provider: "mock", + event: { + type: "timeline", + id: "child", + item: { type: "assistant_message", text: "Child result" }, + }, + }); + const live = await client.fetchProviderSubagentTimeline(agent.id, "child"); + expect(live.rows.map((row) => row.item)).toEqual([ + { type: "assistant_message", text: "Child result" }, + ]); + + await client.archiveAgent(agent.id); + + const children = await client.listProviderSubagents(agent.id); + expect(children.subagents).toEqual([ + expect.objectContaining({ id: "child", title: "Child", status: "completed" }), + ]); + const stored = await client.fetchProviderSubagentTimeline(agent.id, "child"); + expect(stored.rows.map((row) => row.item)).toEqual([ + { type: "assistant_message", text: "Child result" }, + ]); + expect(stored.epoch).toBe(live.epoch); + expect(provider.resumes).toBe(0); }, 30_000); }); diff --git a/packages/server/src/server/schedule/service.ts b/packages/server/src/server/schedule/service.ts index 6121b7e6bd..411bb3084e 100644 --- a/packages/server/src/server/schedule/service.ts +++ b/packages/server/src/server/schedule/service.ts @@ -237,7 +237,6 @@ type ScheduleAgentManager = Pick< AgentManager, | "createAgent" | "getRegisteredProviderIds" - | "hydrateTimelineFromProvider" | "resumeAgentFromPersistence" | "runAgent" | "waitForAgentEvent" diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index c55fe9f469..84c05d7f1b 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -62,7 +62,11 @@ import { resolveStoredAgentUpdatedAt, toAgentPersistenceHandle, } from "./persistence-hooks.js"; -import { ensureAgentLoaded, ensureUnarchivedAgentLoaded } from "./agent/agent-loading.js"; +import { + ensureAgentLoaded, + ensureUnarchivedAgentLoaded, + loadAgentHistory, +} from "./agent/agent-loading.js"; import { sendPromptToAgent, waitForAgentRunStartWithTimeout, @@ -2185,6 +2189,18 @@ export class Session { return this.enrichAgentPayload(toAgentPayload(agent)); } + /** The agent whose history a read serves, without opening its provider when it is stored. */ + private async loadAgentHistoryPayload(agentId: string): Promise { + const source = await loadAgentHistory(agentId, { + agentManager: this.agentManager, + agentStorage: this.agentStorage, + logger: this.sessionLogger, + }); + return source.kind === "live" + ? await this.buildAgentPayload(source.agent) + : this.buildStoredAgentPayload(source.record); + } + private buildStoredAgentPayload( record: StoredAgentRecord, registeredProviderIds = new Set(this.providerSnapshotManager.listRegisteredProviderIds()), @@ -4674,7 +4690,6 @@ export class Session { throw error; } await unarchiveAgentState(this.agentStorage, this.agentManager, snapshot.id); - await this.agentManager.hydrateTimelineFromProvider(snapshot.id); await this.agentUpdates.forwardLiveAgent(snapshot); const timelineSize = (await this.agentManager.getTimeline(snapshot.id)).length; if (requestId) { @@ -4794,13 +4809,14 @@ export class Session { try { await this.restoreOwningWorkspaceForLegacyAgentRefresh(agentId); await unarchiveAgentState(this.agentStorage, this.agentManager, agentId); + // Without stored history, loading the agent imports the provider's; with it, Reload agent + // keeps the history and brings in what the provider added. + const historyPresent = await this.agentManager.hasStoredHistory(agentId); let snapshot: ManagedAgent; const existing = this.agentManager.getAgent(agentId); if (existing) { await this.interruptAgentIfRunning(agentId); - snapshot = await this.agentManager.reloadAgentSession(agentId, undefined, { - rehydrateFromDisk: true, - }); + snapshot = await this.agentManager.reloadAgentSession(agentId); } else { const record = await this.agentStorage.get(agentId); if (!record) { @@ -4819,12 +4835,13 @@ export class Session { snapshot = await ensureAgentLoaded(agentId, { agentManager: this.agentManager, agentStorage: this.agentStorage, - broadcastTimeline: true, logger: this.sessionLogger, }); } - await this.agentManager.hydrateTimelineFromProvider(agentId, { broadcast: true }); - await this.agentUpdates.forwardLiveAgent(snapshot); + const reconciled = historyPresent + ? await this.agentManager.reconcileProviderHistory(agentId) + : null; + await this.agentUpdates.forwardLiveAgent(this.agentManager.getAgent(agentId) ?? snapshot); const timelineSize = (await this.agentManager.getTimeline(agentId)).length; if (requestId) { this.emit({ @@ -4834,6 +4851,7 @@ export class Session { agentId, requestId, timelineSize, + ...(reconciled?.kind === "diverged" ? { historyDiverged: true } : {}), }, }); } @@ -7891,12 +7909,7 @@ export class Session { const cursor = msg.cursor ? this.readWireCursor(msg.cursor, source) : undefined; try { - const snapshot = await ensureAgentLoaded(msg.agentId, { - agentManager: this.agentManager, - agentStorage: this.agentStorage, - logger: this.sessionLogger, - }); - const agentPayload = await this.buildAgentPayload(snapshot); + const agentPayload = await this.loadAgentHistoryPayload(msg.agentId); const fetchedControlTimeline = await this.agentManager.fetchTimeline(msg.agentId, { direction, @@ -7949,7 +7962,7 @@ export class Session { : {}), entries: entries.map((entry) => { const payloadEntry = { - provider: snapshot.provider, + provider: agentPayload.provider, item: entry.item, timestamp: entry.timestamp, seqStart: entry.seqStart, @@ -8026,11 +8039,7 @@ export class Session { source?: object, ): Promise { try { - await ensureAgentLoaded(msg.agentId, { - agentManager: this.agentManager, - agentStorage: this.agentStorage, - logger: this.sessionLogger, - }); + await this.loadAgentHistoryPayload(msg.agentId); const rows = await this.agentManager.getTimelineRows(msg.agentId); const before = await this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", @@ -8081,11 +8090,7 @@ export class Session { source?: object, ): Promise { try { - await ensureAgentLoaded(msg.agentId, { - agentManager: this.agentManager, - agentStorage: this.agentStorage, - logger: this.sessionLogger, - }); + await this.loadAgentHistoryPayload(msg.agentId); const rows = await this.agentManager.getTimelineRows(msg.agentId); const timeline = await this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", @@ -8128,15 +8133,26 @@ export class Session { } } + /** + * Provider children are part of their parent's history: a stored parent, archived or not, is + * read without resuming its provider. + */ + private async loadProviderSubagentParent(parentAgentId: string): Promise { + const source = await loadAgentHistory(parentAgentId, { + agentManager: this.agentManager, + agentStorage: this.agentStorage, + logger: this.sessionLogger, + }); + if (source.kind === "stored" && source.record.internal) { + throw new Error(`Unknown agent '${parentAgentId}'`); + } + } + private async handleProviderSubagentListRequest( msg: Extract, ): Promise { try { - await ensureUnarchivedAgentLoaded(msg.parentAgentId, { - agentManager: this.agentManager, - agentStorage: this.agentStorage, - logger: this.sessionLogger, - }); + await this.loadProviderSubagentParent(msg.parentAgentId); this.emit({ type: "agent.provider_subagents.list.response", payload: { @@ -8165,11 +8181,7 @@ export class Session { ): Promise { const direction: AgentTimelineFetchDirection = msg.direction ?? (msg.cursor ? "after" : "tail"); try { - await ensureUnarchivedAgentLoaded(msg.parentAgentId, { - agentManager: this.agentManager, - agentStorage: this.agentStorage, - logger: this.sessionLogger, - }); + await this.loadProviderSubagentParent(msg.parentAgentId); const descriptor = this.agentManager.getProviderSubagent(msg.parentAgentId, msg.subagentId); if (!descriptor) { throw new Error("Provider subagent not found"); @@ -8263,12 +8275,7 @@ export class Session { msg: Extract, ): Promise { try { - const snapshot = await ensureAgentLoaded(msg.agentId, { - agentManager: this.agentManager, - agentStorage: this.agentStorage, - logger: this.sessionLogger, - }); - const agentPayload = await this.buildAgentPayload(snapshot); + const agentPayload = await this.loadAgentHistoryPayload(msg.agentId); const timeline = await this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", limit: 0, @@ -8283,7 +8290,7 @@ export class Session { : null, boundaryMessageId: msg.boundaryMessageId, agentTitle: agentPayload.title, - cwd: snapshot.cwd, + cwd: agentPayload.cwd, }); this.emit({ diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 9e73c9cad4..7e0208f45e 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -144,7 +144,8 @@ interface SessionTestAccess { preferredId?: string, extras?: unknown, ): Promise; - hydrateTimelineFromProvider(agentId: string): Promise; + hasStoredHistory(agentId: string): Promise; + reconcileProviderHistory(agentId: string): Promise; getTimeline(agentId: string): readonly unknown[]; setTitle(agentId: string, title: string): Promise; }; @@ -358,7 +359,6 @@ function makeManagedAgent(input: { inFlightPermissionResponses: new Set(), pendingReplacement: false, persistence: null, - historyPrimed: true, lastUserMessageAt: null, attention: { requiresAttention: false, @@ -5403,7 +5403,8 @@ test("refresh_agent_request leaves workspace archival independent when its direc session.agentManager.getAgent = () => managed; session.interruptAgentIfRunning = async () => undefined; session.agentManager.reloadAgentSession = async () => managed; - session.agentManager.hydrateTimelineFromProvider = async () => undefined; + session.agentManager.hasStoredHistory = async () => true; + session.agentManager.reconcileProviderHistory = async () => ({ kind: "unchanged" }); session.agentManager.getTimeline = () => []; session.agentUpdates.forwardLiveAgent = async () => undefined; @@ -5501,7 +5502,8 @@ test("refresh_agent_request leaves workspace archival independent when its direc session.agentManager.getAgent = () => managed; session.interruptAgentIfRunning = async () => undefined; session.agentManager.reloadAgentSession = async () => managed; - session.agentManager.hydrateTimelineFromProvider = async () => undefined; + session.agentManager.hasStoredHistory = async () => true; + session.agentManager.reconcileProviderHistory = async () => ({ kind: "unchanged" }); session.agentManager.getTimeline = () => []; session.agentUpdates.forwardLiveAgent = async () => undefined; @@ -5592,7 +5594,8 @@ test("refresh_agent_request does not recreate or unarchive a deleted worktree", session.agentManager.getAgent = () => managed; session.interruptAgentIfRunning = async () => undefined; session.agentManager.reloadAgentSession = async () => managed; - session.agentManager.hydrateTimelineFromProvider = async () => undefined; + session.agentManager.hasStoredHistory = async () => true; + session.agentManager.reconcileProviderHistory = async () => ({ kind: "unchanged" }); session.agentManager.getTimeline = () => []; session.agentUpdates.forwardLiveAgent = async () => undefined; @@ -5691,7 +5694,8 @@ test("refresh_agent_request does not inspect an archived worktree branch", async session.agentManager.getAgent = () => managed; session.interruptAgentIfRunning = async () => undefined; session.agentManager.reloadAgentSession = async () => managed; - session.agentManager.hydrateTimelineFromProvider = async () => undefined; + session.agentManager.hasStoredHistory = async () => true; + session.agentManager.reconcileProviderHistory = async () => ({ kind: "unchanged" }); session.agentManager.getTimeline = () => []; session.agentUpdates.forwardLiveAgent = async () => undefined; @@ -5820,7 +5824,8 @@ test("legacy refresh_agent_request restores a real deleted worktree", async () = session.agentManager.getAgent = () => managed; session.interruptAgentIfRunning = async () => undefined; session.agentManager.reloadAgentSession = async () => managed; - session.agentManager.hydrateTimelineFromProvider = async () => undefined; + session.agentManager.hasStoredHistory = async () => true; + session.agentManager.reconcileProviderHistory = async () => ({ kind: "unchanged" }); session.agentManager.getTimeline = () => []; session.agentUpdates.forwardLiveAgent = async () => undefined; diff --git a/packages/server/src/server/wire-compat.test.ts b/packages/server/src/server/wire-compat.test.ts index 9328f1f05c..de89963169 100644 --- a/packages/server/src/server/wire-compat.test.ts +++ b/packages/server/src/server/wire-compat.test.ts @@ -105,7 +105,6 @@ class InMemoryAgentManager { inFlightPermissionResponses: new Set(), pendingReplacement: false, persistence: null, - historyPrimed: true, lastUsage: undefined, lastError: undefined, attention: { requiresAttention: false, attentionReason: null, attentionTimestamp: null },