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,