puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-10 18:35:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(timeline): preserve history across reload and archive

Browse Source 
Reload agent used to delete the timeline and replay the provider's
history under a new epoch. That brought back turns the user had rewound
and dropped anything the provider no longer had. Archive deleted the
timeline too.

Paseo's timeline is now the owner of the conversation. Reload agent
keeps it and reconciles the provider's history append-only: turns are
aligned by provider prompt id and compared exactly (prompt text,
assistant and reasoning text, tool calls). When the owned conversation
is a whole-segment prefix of the provider's, the rows after it are
appended as new live rows. Any other difference changes nothing and
agent_refreshed reports historyDiverged, which the app shows as a
warning on Reload agent and Unarchive.

Divergence is deliberately conservative:
- An answer or tool call Paseo saw only partly can never be completed by
  a reload. History rows carry no turn id to update a live row in place.
- A prompt sent with attachments diverges on every reload. Providers
  store the rendered attachment text, and the owned row keeps only the
  typed text. Recording the sent prompt is left to provider handoff work.

Archive now only closes the runtime. Stored history, including provider
children, is read without resuming or even requiring the provider.
Forced provider replacement is removed.

Tradeoff: an archived chat's history stays in memory until the daemon
exits, measured at about 1.1-1.2x its JSON size (about 0.5 MB for a
typical session, about 16 MB for a heavy tool session). Moving
retained history to disk belongs to the persisted backend, not eviction,
which would drop the rewinds this change keeps.
This commit is contained in:
Mohamed Boudra
2026-11-10 11:43:10 +01:00
parent 5ed1f8d108
commit dc85c2bfbc
38 changed files with 1753 additions and 644 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/e2e/browser/agent-message-rewind.spec.ts
+104 -1
View File
@@ -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<void> {
await expect(messages).toContainText(prompts);
}
async function reloadAgentFromTabMenu(page: Page, agentId: string): Promise<void> {
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<ReturnType<typeof seedMockAgentWorkspace>>,
): Promise<void> {
await closeWorkspaceAgentTab(page, agent.agentId);
await expect.poll(() => fetchAgentArchivedAt(agent.client, agent.agentId)).not.toBeNull();
}
async function unarchiveAgent(page: Page): Promise<void> {
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<void> {
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<void> {
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<void> {
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.";
packages/app/src/components/archived-agent-callout.tsx
+7 -2
View File
@@ -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<string | null>(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 (
<KeyboardTranslateView style={styles.container}>
packages/app/src/i18n/resources/ar.ts
+2
View File
@@ -699,6 +699,8 @@ export const ar: TranslationResources = {
resumeCommandUnavailable: "أمر الاستئناف غير متوفر",
reloadingAgent: "وكيل إعادة التحميل...",
reloadedAgent: "وكيل إعادة تحميل",
reloadedAgentHistoryDiverged:
"تمت إعادة تحميل الوكيل، لكن سجل المزوّد لا يطابق هذه المحادثة، لذا لم تتم مزامنته",
failedToReloadAgent: "فشل في إعادة تحميل الوكيل",
failedToCloseAgent: "فشل في إغلاق الوكيل",
},
packages/app/src/i18n/resources/en.ts
+2
View File
@@ -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",
},
packages/app/src/i18n/resources/es.ts
+2
View File
@@ -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",
},
packages/app/src/i18n/resources/fr.ts
+2
View File
@@ -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",
},
packages/app/src/i18n/resources/ja.ts
+2
View File
@@ -706,6 +706,8 @@ export const ja: TranslationResources = {
resumeCommandUnavailable: "再開コマンドが利用できません",
reloadingAgent: "エージェントを再読み込み中...",
reloadedAgent: "エージェントを再読み込みしました",
reloadedAgentHistoryDiverged:
"エージェントを再読み込みしましたが、プロバイダーの履歴がこのチャットと一致しないため同期しませんでした",
failedToReloadAgent: "エージェントの再読み込みに失敗しました",
failedToCloseAgent: "エージェントを閉じられませんでした",
},
packages/app/src/i18n/resources/ko.ts
+2
View File
@@ -701,6 +701,8 @@ export const ko: TranslationResources = {
resumeCommandUnavailable: "재개 명령을 사용할 수 없습니다",
reloadingAgent: "에이전트 다시 로드 중...",
reloadedAgent: "에이전트를 다시 로드했습니다",
reloadedAgentHistoryDiverged:
"에이전트를 다시 로드했지만 제공자 기록이 이 채팅과 일치하지 않아 동기화하지 않았습니다",
failedToReloadAgent: "에이전트를 다시 로드하지 못했습니다",
failedToCloseAgent: "에이전트를 닫지 못했습니다",
},
packages/app/src/i18n/resources/pt-BR.ts
+2
View File
@@ -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",
},
packages/app/src/i18n/resources/ru.ts
+2
View File
@@ -706,6 +706,8 @@ export const ru: TranslationResources = {
resumeCommandUnavailable: "Команда возобновления недоступна",
reloadingAgent: "Перезагрузка агента...",
reloadedAgent: "Агент перезагружен",
reloadedAgentHistoryDiverged:
"Агент перезагружен, но история провайдера не совпадает с этим чатом, поэтому она не синхронизирована",
failedToReloadAgent: "Не удалось перезагрузить агента",
failedToCloseAgent: "Не удалось закрыть агента",
},
packages/app/src/i18n/resources/zh-CN.ts
+2
View File
@@ -699,6 +699,8 @@ export const zhCN: TranslationResources = {
resumeCommandUnavailable: "恢复命令不可用",
reloadingAgent: "正在重新加载 Agent...",
reloadedAgent: "已重新加载 Agent",
reloadedAgentHistoryDiverged:
"已重新加载 Agent，但提供方的历史记录与此对话不一致，因此未同步",
failedToReloadAgent: "重新加载 Agent 失败",
failedToCloseAgent: "关闭 Agent 失败",
},
packages/app/src/screens/workspace/workspace-screen.tsx
+11 -6
View File
@@ -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"),
packages/protocol/src/messages.ts
+2
View File
@@ -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);
packages/server/src/server/agent/agent-loading.test.ts
+108 -11
View File
@@ -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<AgentSessionConfig>,
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 });
}
});
packages/server/src/server/agent/agent-loading.ts
+52 -35
View File
@@ -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<ManagedAgent>;
options: { broadcastTimeline: boolean };
}
const pendingAgentInitializations = new Map<string, PendingAgentInitialization>();
const pendingAgentInitializations = new Map<string, Promise<ManagedAgent>>();
export type AgentLoaderManager = Pick<
AgentManager,
| "createAgent"
| "getAgent"
| "getRegisteredProviderIds"
| "hydrateTimelineFromProvider"
| "resumeAgentFromPersistence"
"createAgent" | "getAgent" | "getRegisteredProviderIds" | "resumeAgentFromPersistence"
> &
Partial<Pick<AgentManager, "waitForAgentClose">>;
@@ -40,10 +31,53 @@ export interface EnsureAgentLoadedDeps {
agentManager: AgentLoaderManager;
agentStorage: AgentStorage;
validProviders?: Iterable<AgentProvider>;
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<AgentManager, "hasStoredHistory">;
}
/**
* 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<AgentHistorySource> {
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<StoredAgentRecord> {
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);
}
}
packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts
+32 -9
View File
@@ -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<string, TestAgentSession>();
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([
packages/server/src/server/agent/agent-manager.test.ts
+178 -249
View File
@@ -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<AgentStreamEvent> {
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<AgentStreamEvent> {
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<AgentStreamEvent> {
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<AgentSessionConfig>) {
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<AgentSession> {
return new ProviderChildHistorySession(config);
override async resumeSession(
_handle: AgentPersistenceHandle,
config?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
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<AgentSession> {
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<AgentSession> {
throw new Error("Not used in this test");
async resumeSession(
_handle: AgentPersistenceHandle,
config?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
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<AgentSession> {
throw new Error("Not used in this test");
async resumeSession(
_handle: AgentPersistenceHandle,
config?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
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<void>();
const historyAllowed = deferred<void>();
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<AgentSession> {
return new (class extends TestAgentSession {
override async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
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<AgentSession> {
throw new Error("Not used in this test");
async resumeSession(
_handle: AgentPersistenceHandle,
config?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
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<AgentSession> {
throw new Error("Not used in this test");
async resumeSession(
_handle: AgentPersistenceHandle,
config?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
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",
})) {
packages/server/src/server/agent/agent-manager.ts
+151 -227
View File
@@ -282,12 +282,20 @@ export interface SubscribeOptions {
includeInternal?: boolean;
}
interface HydrateTimelineOptions {
force?: boolean;
broadcast?: boolean | (() => boolean);
broadcastTimeline?: boolean;
type ProviderSubagentStreamEvent = Extract<AgentStreamEvent, { type: "provider_subagent" }>;
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<string>;
pendingReplacement: boolean;
persistence: AgentPersistenceHandle | null;
historyPrimed: boolean;
lastUserMessageAt: Date | null;
activeTurnId: string | null;
activeTurnStartedAt: Date | null;
@@ -703,6 +710,11 @@ function toTimelineEntry(event: Extract<AgentStreamEvent, { type: "timeline" }>)
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<AgentTimelineItem[]> {
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<AgentTimelineRow[]> {
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<AgentTimelineFetchResult> {
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<AgentTimelineFetchResult> {
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<AgentSessionConfig>,
options?: { rehydrateFromDisk?: boolean },
): Promise<ManagedAgent> {
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<AgentSessionConfig>,
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<ProviderHistoryReconcile> {
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<void> {
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<boolean> {
return (await this.timeline.describe({ agentId })) !== null;
}
private async closeReloadedSession(session: AgentSession, agentId: string): Promise<void> {
@@ -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<void> {
const agent = this.requireSessionAgent(agentId);
await this.hydrateTimelineFromLegacyProviderHistory(agent, options);
}
async rewind(agentId: string, messageId: string, mode: RewindMode): Promise<void> {
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<ProviderHistory | null> {
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<AgentStreamEvent>,
): Promise<ProviderHistory> {
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<void> {
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<string, string>;
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<void> {
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<void> {
const historyEvents: Extract<AgentStreamEvent, { type: "timeline" }>[] = [];
const providerSubagentEvents: Extract<AgentStreamEvent, { type: "provider_subagent" }>[] = [];
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<AgentStreamEvent>
| Iterable<AgentStreamEvent> = agent.session.streamHistory(),
): Promise<void> {
const deferredBroadcast = typeof broadcast === "function";
const historyEvents: Extract<AgentStreamEvent, { type: "timeline" }>[] = [];
const historySubagentEvents: Extract<AgentStreamEvent, { type: "provider_subagent" }>[] = [];
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<void> {
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) {
packages/server/src/server/agent/agent-projections.test.ts
-1
View File
@@ -92,7 +92,6 @@ function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent
persistence: { ...basePersistence },
lastUsage: undefined,
lastError: lastErrorValue,
historyPrimed: true,
lastUserMessageAt: now,
attention: { requiresAttention: false },
};
packages/server/src/server/agent/agent-storage.test.ts
-1
View File
@@ -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,
packages/server/src/server/agent/import-sessions.test.ts
-2
View File
@@ -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);
packages/server/src/server/agent/mcp-server.test.ts
+35 -9
View File
@@ -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 () => {
packages/server/src/server/agent/provider-snapshot-manager.test.ts
-1
View File
@@ -50,7 +50,6 @@ const CLOSED_PARENT: ManagedAgent = {
inFlightPermissionResponses: new Set(),
pendingReplacement: false,
persistence: null,
historyPrimed: false,
lastUserMessageAt: null,
activeTurnId: null,
activeTurnStartedAt: null,
packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts
+2 -3
View File
@@ -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");
}
packages/server/src/server/agent/providers/codex-app-server-agent.test.ts
-1
View File
@@ -610,7 +610,6 @@ function createReopenableCodexAgent(threads: Record<string, unknown[]>) {
async function open(): Promise<void> {
await manager.resumeAgentFromPersistence(handle, config, agentId);
await manager.hydrateTimelineFromProvider(agentId);
}
return {
packages/server/src/server/agent/providers/codex-async-questions.test.ts
+12 -5
View File
@@ -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(
packages/server/src/server/agent/rewind/rewind.test.ts
+58 -8
View File
@@ -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 }),
]);
packages/server/src/server/agent/timeline/agent-timeline.test.ts
+385
View File
@@ -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"]);
});
});
packages/server/src/server/agent/timeline/index.ts
+2
View File
@@ -7,7 +7,9 @@ export {
type TimelineCommit,
type TimelineEntry,
type TimelineHead,
type TimelineImportResult,
type TimelineIngestResult,
type TimelineReconcileResult,
type TimelineRetirement,
type TimelineRetirementCommit,
type TimelineRewindCompletion,
packages/server/src/server/agent/timeline/internal/agent-timeline.ts
+70 -5
View File
@@ -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<TimelineImportResult> {
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<TimelineReconcileResult> {
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<void> {
return this.run(scope, async () => {
await this.backend.delete(scope);
@@ -346,11 +411,7 @@ export class AgentTimeline {
scope: TimelineScope,
entries: readonly TimelineEntry[],
): Promise<TimelineCommit[]> {
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;
}
packages/server/src/server/agent/timeline/internal/provider-history.ts
+280
View File
@@ -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<AgentTimelineItem, { type: "tool_call" }>;
/**
* 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<string, unknown> {
return typeof value === "object" && value !== null;
}
function definedKeys(value: Record<string, unknown>): string[] {
return Object.keys(value).filter((key) => value[key] !== undefined);
}
packages/server/src/server/agent/tools/paseo-tools.ts
+7 -4
View File
@@ -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,
}),
};
},
packages/server/src/server/cli-agent-logs.test.ts
+21 -10
View File
@@ -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);
})(),
]);
packages/server/src/server/daemon-e2e/agent-refresh-rehydrates-timeline.e2e.test.ts
+157 -3
View File
@@ -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<ChildReportingSession> {
this.session = new ChildReportingSession({ config, sessionId: "child-reporting-session" });
return this.session;
}
override async resumeSession(
handle: AgentPersistenceHandle,
config?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
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);
});
packages/server/src/server/schedule/service.ts
-1
View File
@@ -237,7 +237,6 @@ type ScheduleAgentManager = Pick<
AgentManager,
| "createAgent"
| "getRegisteredProviderIds"
| "hydrateTimelineFromProvider"
| "resumeAgentFromPersistence"
| "runAgent"
| "waitForAgentEvent"
packages/server/src/server/session.ts
+49 -42
View File
@@ -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<AgentSnapshotPayload> {
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<void> {
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<void> {
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<void> {
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<SessionInboundMessage, { type: "agent.provider_subagents.list.request" }>,
): Promise<void> {
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<void> {
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<SessionInboundMessage, { type: "agent.fork_context.request" }>,
): Promise<void> {
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({
packages/server/src/server/session.workspaces.test.ts
+12 -7
View File
@@ -144,7 +144,8 @@ interface SessionTestAccess {
preferredId?: string,
extras?: unknown,
): Promise<unknown>;
hydrateTimelineFromProvider(agentId: string): Promise<unknown>;
hasStoredHistory(agentId: string): Promise<boolean>;
reconcileProviderHistory(agentId: string): Promise<unknown>;
getTimeline(agentId: string): readonly unknown[];
setTitle(agentId: string, title: string): Promise<unknown>;
};
@@ -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;
packages/server/src/server/wire-compat.test.ts
-1
View File
@@ -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 },