From a11c246bdceca3dff819959161a0d004a515fd25 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Thu, 16 Apr 2026 18:58:31 +0700 Subject: [PATCH] fix(server): share send-prompt path between MCP and Session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract `unarchiveAgentState` and `sendPromptToAgent` into `mcp-shared` so every surface (app/WS, MCP, CLI-through-MCP) runs the same sequence: unarchive → ensureAgentLoaded → optional mode change → recordUserMessage → startAgentRun. MCP `send_agent_prompt` previously skipped unarchive and cold-agent rehydration, so sending to an archived agent over MCP left it hidden from `paseo ls` and failed entirely when the agent wasn't already in memory. Session now delegates to the shared function and its private `unarchiveAgentState` is gone. --- .../server/src/server/agent/mcp-server.ts | 29 ++---- .../server/src/server/agent/mcp-shared.ts | 99 ++++++++++++++++++- packages/server/src/server/session.ts | 96 ++++++------------ 3 files changed, 136 insertions(+), 88 deletions(-) diff --git a/packages/server/src/server/agent/mcp-server.ts b/packages/server/src/server/agent/mcp-server.ts index a71134b25..8279386cf 100644 --- a/packages/server/src/server/agent/mcp-server.ts +++ b/packages/server/src/server/agent/mcp-server.ts @@ -40,6 +40,7 @@ import { parseDurationString, resolveProviderAndModel, sanitizePermissionRequest, + sendPromptToAgent, setupFinishNotification, serializeSnapshotWithMetadata, startAgentRun, @@ -768,30 +769,20 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom }, }, async ({ agentId, prompt, sessionMode, background = false, notifyOnFinish = false }) => { - const snapshot = agentManager.getAgent(agentId); - if (!snapshot) { - throw new Error(`Agent ${agentId} not found`); - } - if (agentManager.hasInFlightRun(agentId)) { waitTracker.cancel(agentId, "Agent run interrupted by new prompt"); } - if (sessionMode) { - await agentManager.setAgentMode(agentId, sessionMode); - } - - try { - agentManager.recordUserMessage(agentId, prompt, { - emitState: false, - }); - } catch (error) { - childLogger.error({ err: error, agentId }, "Failed to record user message"); - } - - startAgentRun(agentManager, agentId, prompt, childLogger, { - replaceRunning: true, + await sendPromptToAgent({ + agentManager, + agentStorage, + agentId, + userMessageText: prompt, + prompt, + sessionMode, + logger: childLogger, }); + if (notifyOnFinish && callerAgentId) { setupFinishNotification({ agentManager, diff --git a/packages/server/src/server/agent/mcp-shared.ts b/packages/server/src/server/agent/mcp-shared.ts index 3edf8f6c7..e3e959985 100644 --- a/packages/server/src/server/agent/mcp-shared.ts +++ b/packages/server/src/server/agent/mcp-shared.ts @@ -1,10 +1,15 @@ import { z } from "zod"; import type { Logger } from "pino"; -import type { AgentPromptInput, AgentPermissionRequest } from "./agent-sdk-types.js"; +import type { + AgentPromptInput, + AgentPermissionRequest, + AgentRunOptions, +} from "./agent-sdk-types.js"; import type { AgentManager, ManagedAgent, WaitForAgentResult } from "./agent-manager.js"; import { curateAgentActivity } from "./activity-curator.js"; import type { AgentStorage } from "./agent-storage.js"; +import { ensureAgentLoaded } from "./agent-loading.js"; import { serializeAgentSnapshot } from "../messages.js"; import { StoredScheduleSchema } from "../schedule/types.js"; import type { AgentProvider } from "./agent-sdk-types.js"; @@ -94,6 +99,7 @@ export function resolveProviderAndModel(params: { export type StartAgentRunOptions = { replaceRunning?: boolean; + runOptions?: AgentRunOptions; }; /** @@ -171,9 +177,10 @@ export function startAgentRun( options?: StartAgentRunOptions, ): void { const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId)); + const runOptions = options?.runOptions; const iterator = shouldReplace - ? agentManager.replaceAgentRun(agentId, prompt) - : agentManager.streamAgent(agentId, prompt); + ? agentManager.replaceAgentRun(agentId, prompt, runOptions) + : agentManager.streamAgent(agentId, prompt, runOptions); void (async () => { try { for await (const _ of iterator) { @@ -185,6 +192,92 @@ export function startAgentRun( })(); } +/** + * Clear the archived flag from a stored agent record. + * Shared across Session (app/WS), MCP, and CLI so every surface that acts on + * an archived agent unarchives it the same way. + */ +export async function unarchiveAgentState( + agentStorage: AgentStorage, + agentManager: AgentManager, + agentId: string, +): Promise { + const record = await agentStorage.get(agentId); + if (!record || !record.archivedAt) { + return false; + } + const updatedAt = new Date().toISOString(); + await agentStorage.upsert({ + ...record, + archivedAt: null, + updatedAt, + }); + agentManager.notifyAgentState(agentId); + return true; +} + +export interface SendPromptToAgentParams { + agentManager: AgentManager; + agentStorage: AgentStorage; + agentId: string; + /** Raw user text to record in the timeline. */ + userMessageText: string; + /** Prompt to dispatch to the provider (may include image blocks or wrapped text). */ + prompt: AgentPromptInput; + messageId?: string; + runOptions?: AgentRunOptions; + /** Optional mode to set on the agent before the run starts. */ + sessionMode?: string; + logger: Logger; +} + +/** + * Full send-prompt orchestration: unarchive → load → (optional mode change) → + * record user message → start run. + * + * Every surface that sends a prompt to an agent (Session/WS, MCP, CLI-through-MCP) + * MUST go through this so behavior can never drift between them. + */ +export async function sendPromptToAgent(params: SendPromptToAgentParams): Promise { + const { + agentManager, + agentStorage, + agentId, + userMessageText, + prompt, + messageId, + runOptions, + sessionMode, + logger, + } = params; + + await unarchiveAgentState(agentStorage, agentManager, agentId); + + await ensureAgentLoaded(agentId, { + agentManager, + agentStorage, + logger, + }); + + if (sessionMode) { + await agentManager.setAgentMode(agentId, sessionMode); + } + + try { + agentManager.recordUserMessage(agentId, userMessageText, { + messageId, + emitState: false, + }); + } catch (error) { + logger.error({ err: error, agentId }, "Failed to record user message"); + } + + startAgentRun(agentManager, agentId, prompt, logger, { + replaceRunning: true, + runOptions, + }); +} + interface SetupFinishNotificationParams { agentManager: AgentManager; agentStorage: AgentStorage; diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 039e678fe..f85641ae4 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -65,6 +65,7 @@ import { toAgentPersistenceHandle, } from "./persistence-hooks.js"; import { ensureAgentLoaded } from "./agent/agent-loading.js"; +import { sendPromptToAgent, unarchiveAgentState } from "./agent/mcp-shared.js"; import { experimental_createMCPClient } from "ai"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import type { VoiceCallerContext, VoiceSpeakHandler } from "./voice-types.js"; @@ -2188,21 +2189,6 @@ export class Session { }); } - private async unarchiveAgentState(agentId: string): Promise { - const record = await this.agentStorage.get(agentId); - if (!record || !record.archivedAt) { - return false; - } - const updatedAt = new Date().toISOString(); - await this.agentStorage.upsert({ - ...record, - archivedAt: null, - updatedAt, - }); - this.agentManager.notifyAgentState(agentId); - return true; - } - private async unarchiveAgentByHandle(handle: AgentPersistenceHandle): Promise { const records = await this.agentStorage.list(); const matched = records.find( @@ -2213,7 +2199,7 @@ export class Session { if (!matched) { return; } - await this.unarchiveAgentState(matched.id); + await unarchiveAgentState(this.agentStorage, this.agentManager, matched.id); } private async handleUpdateAgentRequest( @@ -2711,38 +2697,28 @@ export class Session { `Sending text to agent ${agentId}${images && images.length > 0 ? ` with ${images.length} image attachment(s)` : ""}`, ); - await this.unarchiveAgentState(agentId); + const promptText = options?.spokenInput ? wrapSpokenInput(text) : text; + const prompt = this.buildAgentPrompt(promptText, images); try { - await ensureAgentLoaded(agentId, { + await sendPromptToAgent({ agentManager: this.agentManager, agentStorage: this.agentStorage, + agentId, + userMessageText: text, + prompt, + messageId, + runOptions, logger: this.sessionLogger, }); + return { ok: true }; } catch (error) { - this.handleAgentRunError(agentId, error, "Failed to initialize agent before sending prompt"); + this.handleAgentRunError(agentId, error, "Failed to send agent message"); return { ok: false, error: error instanceof Error ? error.message : String(error), }; } - - try { - this.agentManager.recordUserMessage(agentId, text, { - messageId, - emitState: false, - }); - } catch (error) { - this.sessionLogger.error( - { err: error, agentId }, - `Failed to record user message for agent ${agentId}`, - ); - } - - const promptText = options?.spokenInput ? wrapSpokenInput(text) : text; - const prompt = this.buildAgentPrompt(promptText, images); - - return this.startAgentStream(agentId, prompt, runOptions); } /** @@ -2902,7 +2878,7 @@ export class Session { try { await this.unarchiveAgentByHandle(handle); const snapshot = await this.agentManager.resumeAgentFromPersistence(handle, overrides); - await this.unarchiveAgentState(snapshot.id); + await unarchiveAgentState(this.agentStorage, this.agentManager, snapshot.id); await this.agentManager.hydrateTimelineFromProvider(snapshot.id); await this.forwardAgentUpdate(snapshot); const timelineSize = this.agentManager.getTimeline(snapshot.id).length; @@ -2943,7 +2919,7 @@ export class Session { this.sessionLogger.info({ agentId }, `Refreshing agent ${agentId} from persistence`); try { - await this.unarchiveAgentState(agentId); + await unarchiveAgentState(this.agentStorage, this.agentManager, agentId); let snapshot: ManagedAgent; const existing = this.agentManager.getAgent(agentId); if (existing) { @@ -6396,44 +6372,32 @@ export class Session { try { const agentId = resolved.agentId; - await this.unarchiveAgentState(agentId); - - await ensureAgentLoaded(agentId, { - agentManager: this.agentManager, - agentStorage: this.agentStorage, - logger: this.sessionLogger, - }); - - this.sessionLogger.trace( - { agentId, messageId: msg.messageId, textPrefix: msg.text.slice(0, 80) }, - "send_agent_message_request: recording user message", - ); - try { - this.agentManager.recordUserMessage(agentId, msg.text, { - messageId: msg.messageId, - emitState: false, - }); - } catch (error) { - this.sessionLogger.error( - { err: error, agentId }, - "Failed to record user message for send_agent_message_request", - ); - } const prompt = this.buildAgentPrompt(msg.text, msg.images); this.sessionLogger.trace( - { agentId, messageId: msg.messageId }, - "send_agent_message_request: starting agent stream", + { agentId, messageId: msg.messageId, textPrefix: msg.text.slice(0, 80) }, + "send_agent_message_request: dispatching shared sendPromptToAgent", ); - const started = this.startAgentStream(agentId, prompt); - if (!started.ok) { + try { + await sendPromptToAgent({ + agentManager: this.agentManager, + agentStorage: this.agentStorage, + agentId, + userMessageText: msg.text, + prompt, + messageId: msg.messageId, + logger: this.sessionLogger, + }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + this.handleAgentRunError(agentId, error, "Failed to send agent message"); this.emit({ type: "send_agent_message_response", payload: { requestId: msg.requestId, agentId, accepted: false, - error: started.error, + error: message, }, }); return;