From 269aa53c27fe5d7caa0ace8c66fd8ffe8de50403 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Thu, 2 Jul 2026 14:48:32 +0200 Subject: [PATCH] fix(paseo-agent): retain prompts and required tools Emit submitted prompts into the daemon timeline so refresh and reload can replay the full Paseo Agent conversation. Keep nonpersistent providers from getting synthetic resume handles, and gate Paseo tool injection through provider capabilities. --- docs/paseo-agent.md | 12 + .../server/src/server/agent/agent-manager.ts | 46 ++- .../agent/providers/paseo-agent/agent.test.ts | 14 +- .../agent/providers/paseo-agent/agent.ts | 17 +- packages/server/src/server/bootstrap.ts | 5 +- .../paseo-agent-persistence.e2e.test.ts | 263 ++++++++++++++++++ .../src/server/test-utils/paseo-daemon.ts | 2 + 7 files changed, 338 insertions(+), 21 deletions(-) create mode 100644 packages/server/src/server/daemon-e2e/paseo-agent-persistence.e2e.test.ts diff --git a/docs/paseo-agent.md b/docs/paseo-agent.md index faa1b18cd4..8863ba26b3 100644 --- a/docs/paseo-agent.md +++ b/docs/paseo-agent.md @@ -19,6 +19,15 @@ instance and the `openai/gpt-4o-mini` model exposed by that instance. > `--host ` to CLI smoke commands so they hit your isolated daemon, not the real > daemon on `:6767`. +## Timeline and persistence + +Paseo Agent does not write or resume Pi CLI session files such as +`~/.pi/agent/sessions/*`; it runs Pi's harness in process and currently reports no +provider-owned session persistence. App refreshes and daemon-side agent reloads therefore +depend on Paseo's own timeline rows, not on a Pi history replay. Submitted prompts must be +emitted as `user_message` timeline rows when the turn starts; assistant text, reasoning, +and tool calls arrive later from Pi events and are mapped into the same Paseo timeline. + ## Provider catalog Paseo Agent model providers are catalog-driven, but Paseo does not copy Pi's provider @@ -255,6 +264,9 @@ Transports: HTTP (streamable) is the primary path (the injected `paseo` server i SSE and stdio transports are also wired via the MCP SDK. No extra config is needed: MCP servers come from Paseo's normal injection/config, not from `agents.paseo`. +The internal `paseo` MCP server is required for Paseo Agent sessions and is injected even +when the global `mcp.injectIntoAgents` setting is disabled for other providers. + ## Agent definitions Paseo Agent can load a Paseo-owned agent definition from `$PASEO_HOME/agents/*.md`. diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 720edbfcbb..d5f44832bc 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -1085,7 +1085,8 @@ export class AgentManager { this.foregroundRuns.clearAgent(agentId, existing); await this.closeReloadedSession(existing.session, agentId); - if (rehydrateFromDisk) { + const canRehydrateFromProviderPersistence = rehydrateFromDisk && Boolean(handle); + if (canRehydrateFromProviderPersistence) { // Wipe both durable and in-memory timeline so registerSession mints a // new epoch and hydrateTimelineFromProvider re-streams the freshly read // provider history into an empty timeline. @@ -1100,7 +1101,7 @@ export class AgentManager { createdAt: existing.createdAt, updatedAt: existing.updatedAt, lastUserMessageAt: existing.lastUserMessageAt, - historyPrimed: rehydrateFromDisk ? false : preservedHistoryPrimed, + historyPrimed: canRehydrateFromProviderPersistence ? false : preservedHistoryPrimed, lastUsage: preservedLastUsage, lastError: preservedLastError, attention: preservedAttention, @@ -1812,11 +1813,7 @@ export class AgentManager { nextLifecycle = "idle"; } mutableAgent.lifecycle = nextLifecycle; - const persistenceHandle = - mutableAgent.session.describePersistence() ?? - (mutableAgent.runtimeInfo?.sessionId - ? { provider: mutableAgent.provider, sessionId: mutableAgent.runtimeInfo.sessionId } - : null); + const persistenceHandle = this.describePersistenceHandle(mutableAgent); if (persistenceHandle) { mutableAgent.persistence = attachPersistenceCwd(persistenceHandle, mutableAgent.cwd); } @@ -2832,11 +2829,11 @@ export class AgentManager { newInfo.sessionId !== agent.runtimeInfo?.sessionId || newInfo.modeId !== agent.runtimeInfo?.modeId; agent.runtimeInfo = newInfo; - if (!agent.persistence && newInfo.sessionId) { - agent.persistence = attachPersistenceCwd( - { provider: agent.provider, sessionId: newInfo.sessionId }, - agent.cwd, - ); + if (!agent.persistence) { + const persistenceHandle = this.describePersistenceHandle(agent); + if (persistenceHandle) { + agent.persistence = attachPersistenceCwd(persistenceHandle, agent.cwd); + } } // Emit state if runtimeInfo changed so clients get the updated model if (changed && options?.emit !== false) { @@ -3128,7 +3125,7 @@ export class AgentManager { private onStreamThreadStarted(agent: ActiveManagedAgent): void { const previousSessionId = agent.persistence?.sessionId ?? null; - const handle = agent.session.describePersistence(); + const handle = this.describePersistenceHandle(agent); if (handle) { agent.persistence = attachPersistenceCwd(handle, agent.cwd); if (agent.persistence?.sessionId !== previousSessionId) { @@ -3138,6 +3135,19 @@ export class AgentManager { void this.refreshRuntimeInfo(agent); } + private describePersistenceHandle(agent: ActiveManagedAgent): AgentPersistenceHandle | null { + const handle = agent.session.describePersistence(); + if (handle) { + return handle; + } + if (!agent.capabilities.supportsSessionPersistence) { + return null; + } + return agent.runtimeInfo?.sessionId + ? { provider: agent.provider, sessionId: agent.runtimeInfo.sessionId } + : null; + } + private async onStreamTimelineEvent(params: { agent: ActiveManagedAgent; event: Extract; @@ -3719,11 +3729,13 @@ export class AgentManager { agentId: string, ): Promise { const storedConfig = await this.normalizeConfig(stripInternalPaseoMcpServer(config)); + const client = this.clients.get(storedConfig.provider); + const mcpBaseUrl = client && this.shouldEnablePaseoTools(client) ? this.mcpBaseUrl : null; const launchConfig = this.applyDaemonAppendSystemPrompt( withRuntimePaseoMcpServer({ config: storedConfig, agentId, - mcpBaseUrl: this.mcpBaseUrl, + mcpBaseUrl, mcpAuthToken: this.mcpAuthToken, }), ); @@ -3756,7 +3768,7 @@ export class AgentManager { }, }; if ( - this.paseoToolsEnabled && + this.shouldEnablePaseoTools(client) && client.capabilities.supportsNativePaseoTools && this.paseoToolCatalogFactory ) { @@ -3765,6 +3777,10 @@ export class AgentManager { return context; } + private shouldEnablePaseoTools(client: AgentClient): boolean { + return this.paseoToolsEnabled || client.capabilities.requiresPaseoTools === true; + } + private resolveProviderLaunchConfig( launchConfig: AgentSessionConfig, launchContext: AgentLaunchContext, diff --git a/packages/server/src/server/agent/providers/paseo-agent/agent.test.ts b/packages/server/src/server/agent/providers/paseo-agent/agent.test.ts index c245528557..f801b55689 100644 --- a/packages/server/src/server/agent/providers/paseo-agent/agent.test.ts +++ b/packages/server/src/server/agent/providers/paseo-agent/agent.test.ts @@ -483,6 +483,11 @@ describe("PaseoAgentSession runtime events", () => { contextWindowUsedTokens: 1234, }, timeline: [ + { + type: "user_message", + text: "hello", + messageId: expect.any(String), + }, { type: "reasoning", text: "thinking" }, { type: "tool_call", @@ -504,6 +509,7 @@ describe("PaseoAgentSession runtime events", () => { ], }); expect(events.map((event) => event.type)).toEqual([ + "timeline", "thread_started", "turn_started", "timeline", @@ -554,7 +560,13 @@ describe("PaseoAgentSession runtime events", () => { await expect(resultPromise).resolves.toMatchObject({ sessionId: "pi-session-1", finalText: "", - timeline: [], + timeline: [ + { + type: "user_message", + text: "cancel me", + messageId: expect.any(String), + }, + ], }); expect(fakePi.abortCalls).toBe(1); expect(events).toContainEqual({ diff --git a/packages/server/src/server/agent/providers/paseo-agent/agent.ts b/packages/server/src/server/agent/providers/paseo-agent/agent.ts index d42a3110f7..e5ca7f352f 100644 --- a/packages/server/src/server/agent/providers/paseo-agent/agent.ts +++ b/packages/server/src/server/agent/providers/paseo-agent/agent.ts @@ -67,6 +67,7 @@ const PASEO_AGENT_CAPABILITIES: AgentCapabilityFlags = { supportsDynamicModes: false, // MCP servers from AgentSessionConfig.mcpServers are bridged to Pi custom tools. supportsMcpServers: true, + requiresPaseoTools: true, supportsReasoningStream: true, supportsToolInvocations: true, }; @@ -315,6 +316,18 @@ export class PaseoAgentSession implements AgentSession { }); } + private emitSubmittedUserMessage(text: string, messageId: string, turnId: string): void { + if (text.trim().length === 0) { + return; + } + this.emit({ + type: "timeline", + provider: PASEO_AGENT_PROVIDER, + turnId, + item: { type: "user_message", text, messageId }, + }); + } + async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise { return runProviderTurn({ prompt, @@ -328,14 +341,16 @@ export class PaseoAgentSession implements AgentSession { async startTurn( prompt: AgentPromptInput, - _options?: AgentRunOptions, + options?: AgentRunOptions, ): Promise<{ turnId: string }> { if (this.activeTurnId) { throw new Error("A Paseo Agent turn is already active"); } const payload = convertPromptInput(prompt); const turnId = randomUUID(); + const messageId = options?.messageId ?? randomUUID(); this.activeTurnId = turnId; + this.emitSubmittedUserMessage(payload.text, messageId, turnId); void this.piSession .prompt(payload.text, payload.images ? { images: payload.images } : undefined) diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index ac60813e09..916a5f314e 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1000,7 +1000,6 @@ export async function createPaseoDaemon( agentManager.setPaseoToolsEnabled(config.mcpInjectIntoAgents !== false); const mcpEnabled = config.mcpEnabled ?? true; - let agentMcpBaseUrl: string | null = null; if (mcpEnabled) { const agentMcpRoute = "/mcp/agents"; @@ -1159,11 +1158,9 @@ export async function createPaseoDaemon( const logAndResolve = async () => { boundListenTarget = resolveBoundListenTarget(listenTarget, httpServer); const mcpBaseUrl = mcpEnabled ? createAgentMcpBaseUrl(boundListenTarget) : null; - agentMcpBaseUrl = config.mcpInjectIntoAgents === false ? null : mcpBaseUrl; - agentManager.setMcpBaseUrl(agentMcpBaseUrl); + agentManager.setMcpBaseUrl(mcpBaseUrl); agentManager.setPaseoToolsEnabled(config.mcpInjectIntoAgents !== false); daemonConfigStore.onFieldChange("mcp.injectIntoAgents", (value) => { - agentManager.setMcpBaseUrl(value ? mcpBaseUrl : null); agentManager.setPaseoToolsEnabled(value !== false); }); daemonConfigStore.onFieldChange("appendSystemPrompt", (value) => { diff --git a/packages/server/src/server/daemon-e2e/paseo-agent-persistence.e2e.test.ts b/packages/server/src/server/daemon-e2e/paseo-agent-persistence.e2e.test.ts new file mode 100644 index 0000000000..7979179e81 --- /dev/null +++ b/packages/server/src/server/daemon-e2e/paseo-agent-persistence.e2e.test.ts @@ -0,0 +1,263 @@ +import { afterEach, describe, expect, test } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import type { AgentSessionEvent } from "@earendil-works/pi-coding-agent"; +import pino from "pino"; + +import { PaseoAgentSession } from "../agent/providers/paseo-agent/agent.js"; +import type { PaseoAgentSessionHandle } from "../agent/providers/paseo-agent/pi-services.js"; +import { DaemonClient } from "../test-utils/daemon-client.js"; +import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js"; +import type { + AgentCapabilityFlags, + AgentClient, + AgentMode, + AgentModelDefinition, + AgentSession, + AgentSessionConfig, + FetchCatalogOptions, +} from "../agent/agent-sdk-types.js"; + +const PASEO_AGENT_TEST_CAPABILITIES: AgentCapabilityFlags = { + supportsStreaming: true, + supportsSessionPersistence: false, + supportsDynamicModes: false, + supportsMcpServers: true, + requiresPaseoTools: true, + supportsReasoningStream: true, + supportsToolInvocations: true, + supportsRewindConversation: false, + supportsRewindFiles: false, + supportsRewindBoth: false, +}; + +class FakeInProcessPiSession { + readonly sessionId: string; + readonly thinkingLevel = "medium"; + readonly model = { provider: "openrouter-main", id: "test-model" }; + readonly messages: Array<{ role: string; content: unknown; toolCallId?: string }> = []; + readonly agent = { state: { errorMessage: "" } }; + abortCalls = 0; + disposeCalls = 0; + promptCalls: Array<{ text: string; options: unknown }> = []; + private readonly subscribers = new Set<(event: AgentSessionEvent) => void>(); + + constructor(sessionId: string) { + this.sessionId = sessionId; + } + + subscribe(callback: (event: AgentSessionEvent) => void): () => void { + this.subscribers.add(callback); + return () => { + this.subscribers.delete(callback); + }; + } + + async prompt(text: string, options?: unknown): Promise { + this.promptCalls.push({ text, options }); + this.messages.push({ role: "user", content: text }); + setTimeout(() => { + this.emit({ type: "agent_start" } as AgentSessionEvent); + this.emit({ type: "turn_start" } as AgentSessionEvent); + this.messages.push({ + role: "assistant", + content: [{ type: "text", text: `ack: ${text}` }], + }); + this.emit({ + type: "message_update", + message: { role: "assistant", content: [] }, + assistantMessageEvent: { type: "text_delta", delta: `ack: ${text}` }, + } as AgentSessionEvent); + this.emit({ + type: "agent_end", + messages: this.messages, + willRetry: false, + } as AgentSessionEvent); + }, 0); + } + + async abort(): Promise { + this.abortCalls += 1; + } + + dispose(): void { + this.disposeCalls += 1; + } + + getSessionStats() { + return { + sessionFile: undefined, + sessionId: this.sessionId, + userMessages: 1, + assistantMessages: 1, + toolCalls: 0, + toolResults: 0, + totalMessages: 2, + tokens: { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, total: 2 }, + cost: 0, + contextUsage: { contextWindow: 200000, tokens: 2, percentage: 0.01 }, + }; + } + + setThinkingLevel(): void {} + + async setModel(): Promise {} + + private emit(event: AgentSessionEvent): void { + for (const subscriber of this.subscribers) { + subscriber(event); + } + } +} + +class RecordingPaseoAgentClient implements AgentClient { + readonly provider = "paseo"; + readonly capabilities = PASEO_AGENT_TEST_CAPABILITIES; + readonly createdConfigs: AgentSessionConfig[] = []; + private nextSessionOrdinal = 0; + + async createSession(config: AgentSessionConfig): Promise { + this.createdConfigs.push(config); + this.nextSessionOrdinal += 1; + const fakePi = new FakeInProcessPiSession(`fake-paseo-session-${this.nextSessionOrdinal}`); + const handle = { + session: fakePi, + modelRegistry: { find: () => fakePi.model }, + resourceLoader: {}, + sessionManager: {}, + } as unknown as PaseoAgentSessionHandle; + const mcpBridge = { + tools: [], + async close() {}, + }; + return new PaseoAgentSession(handle, config, mcpBridge, null, []); + } + + async resumeSession(): Promise { + throw new Error("test Paseo Agent client does not support resume"); + } + + async fetchCatalog(_options: FetchCatalogOptions): Promise<{ + models: AgentModelDefinition[]; + modes: AgentMode[]; + }> { + return { + models: [ + { + provider: "paseo", + id: "openrouter-main/test-model", + label: "Test Model", + isDefault: true, + }, + ], + modes: [], + }; + } + + async isAvailable(): Promise { + return true; + } +} + +function textTimelineItems( + entries: ReadonlyArray<{ item: { type: string; text?: string; messageId?: string } }>, +) { + return entries + .filter( + (entry) => entry.item.type === "user_message" || entry.item.type === "assistant_message", + ) + .map((entry) => entry.item); +} + +describe("daemon E2E - Paseo Agent persistence", () => { + const tempDirs: string[] = []; + let daemon: TestPaseoDaemon | undefined; + let client: DaemonClient | undefined; + + afterEach(async () => { + await client?.close().catch(() => undefined); + await daemon?.close().catch(() => undefined); + client = undefined; + daemon = undefined; + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } + }, 60_000); + + test("Paseo Agent user messages survive a client refresh and agent reload", async () => { + const cwd = mkdtempSync(path.join(tmpdir(), "paseo-agent-persist-cwd-")); + tempDirs.push(cwd); + const paseo = new RecordingPaseoAgentClient(); + daemon = await createTestPaseoDaemon({ + agentClients: { paseo }, + logger: pino({ level: "silent" }), + }); + client = new DaemonClient({ + url: `ws://127.0.0.1:${daemon.port}/ws`, + appVersion: "0.1.70", + }); + await client.connect(); + await client.fetchAgents({ subscribe: { subscriptionId: "paseo-agent-persistence-test" } }); + + const agent = await client.createAgent({ provider: "paseo", cwd }); + await client.sendMessage(agent.id, "hello from the user", { messageId: "client-message-1" }); + await client.waitForFinish(agent.id, 5_000); + + await client.close(); + client = new DaemonClient({ + url: `ws://127.0.0.1:${daemon.port}/ws`, + appVersion: "0.1.70", + }); + await client.connect(); + await client.fetchAgents({ + subscribe: { subscriptionId: "paseo-agent-persistence-refresh-test" }, + }); + + const afterClientRefresh = await client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 0, + projection: "canonical", + }); + expect(textTimelineItems(afterClientRefresh.entries)).toEqual([ + { type: "user_message", text: "hello from the user", messageId: "client-message-1" }, + { type: "assistant_message", text: "ack: hello from the user" }, + ]); + + await client.refreshAgent(agent.id); + const afterAgentReload = await client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 0, + projection: "canonical", + }); + expect(textTimelineItems(afterAgentReload.entries)).toEqual([ + { type: "user_message", text: "hello from the user", messageId: "client-message-1" }, + { type: "assistant_message", text: "ack: hello from the user" }, + ]); + }, 30_000); + + test("Paseo Agent gets the internal Paseo MCP server when global tool injection is off", async () => { + const cwd = mkdtempSync(path.join(tmpdir(), "paseo-agent-tools-cwd-")); + tempDirs.push(cwd); + const paseo = new RecordingPaseoAgentClient(); + daemon = await createTestPaseoDaemon({ + agentClients: { paseo }, + mcpInjectIntoAgents: false, + logger: pino({ level: "silent" }), + }); + client = new DaemonClient({ + url: `ws://127.0.0.1:${daemon.port}/ws`, + appVersion: "0.1.70", + }); + await client.connect(); + await client.fetchAgents({ subscribe: { subscriptionId: "paseo-agent-tools-test" } }); + + const agent = await client.createAgent({ provider: "paseo", cwd }); + + expect(paseo.createdConfigs.at(-1)?.mcpServers?.paseo).toMatchObject({ + type: "http", + url: `http://127.0.0.1:${daemon.port}/mcp/agents?callerAgentId=${agent.id}`, + }); + expect(daemon.daemon.agentManager.getAgent(agent.id)?.config.mcpServers?.paseo).toBeUndefined(); + }, 30_000); +}); diff --git a/packages/server/src/server/test-utils/paseo-daemon.ts b/packages/server/src/server/test-utils/paseo-daemon.ts index 58c2ebd9b5..b371d20d46 100644 --- a/packages/server/src/server/test-utils/paseo-daemon.ts +++ b/packages/server/src/server/test-utils/paseo-daemon.ts @@ -19,6 +19,7 @@ interface TestPaseoDaemonOptions { listen?: string; logger?: Parameters[1]; mcpEnabled?: boolean; + mcpInjectIntoAgents?: boolean; mcpDebug?: boolean; isDev?: boolean; relayEnabled?: boolean; @@ -158,6 +159,7 @@ async function prepareTestDaemonConfig( corsAllowedOrigins: options.corsAllowedOrigins ?? [], hostnames: true, mcpEnabled: options.mcpEnabled ?? true, + mcpInjectIntoAgents: options.mcpInjectIntoAgents, staticDir, mcpDebug: options.mcpDebug ?? false, isDev: options.isDev,