fix(server): share send-prompt path between MCP and Session

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.
This commit is contained in:
Mohamed Boudra
2026-04-16 18:58:31 +07:00
parent 2e95003154
commit a11c246bdc
3 changed files with 136 additions and 88 deletions

View File

@@ -40,6 +40,7 @@ import {
parseDurationString, parseDurationString,
resolveProviderAndModel, resolveProviderAndModel,
sanitizePermissionRequest, sanitizePermissionRequest,
sendPromptToAgent,
setupFinishNotification, setupFinishNotification,
serializeSnapshotWithMetadata, serializeSnapshotWithMetadata,
startAgentRun, startAgentRun,
@@ -768,30 +769,20 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom
}, },
}, },
async ({ agentId, prompt, sessionMode, background = false, notifyOnFinish = false }) => { 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)) { if (agentManager.hasInFlightRun(agentId)) {
waitTracker.cancel(agentId, "Agent run interrupted by new prompt"); waitTracker.cancel(agentId, "Agent run interrupted by new prompt");
} }
if (sessionMode) { await sendPromptToAgent({
await agentManager.setAgentMode(agentId, sessionMode); agentManager,
} agentStorage,
agentId,
try { userMessageText: prompt,
agentManager.recordUserMessage(agentId, prompt, { prompt,
emitState: false, sessionMode,
logger: childLogger,
}); });
} catch (error) {
childLogger.error({ err: error, agentId }, "Failed to record user message");
}
startAgentRun(agentManager, agentId, prompt, childLogger, {
replaceRunning: true,
});
if (notifyOnFinish && callerAgentId) { if (notifyOnFinish && callerAgentId) {
setupFinishNotification({ setupFinishNotification({
agentManager, agentManager,

View File

@@ -1,10 +1,15 @@
import { z } from "zod"; import { z } from "zod";
import type { Logger } from "pino"; 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 type { AgentManager, ManagedAgent, WaitForAgentResult } from "./agent-manager.js";
import { curateAgentActivity } from "./activity-curator.js"; import { curateAgentActivity } from "./activity-curator.js";
import type { AgentStorage } from "./agent-storage.js"; import type { AgentStorage } from "./agent-storage.js";
import { ensureAgentLoaded } from "./agent-loading.js";
import { serializeAgentSnapshot } from "../messages.js"; import { serializeAgentSnapshot } from "../messages.js";
import { StoredScheduleSchema } from "../schedule/types.js"; import { StoredScheduleSchema } from "../schedule/types.js";
import type { AgentProvider } from "./agent-sdk-types.js"; import type { AgentProvider } from "./agent-sdk-types.js";
@@ -94,6 +99,7 @@ export function resolveProviderAndModel(params: {
export type StartAgentRunOptions = { export type StartAgentRunOptions = {
replaceRunning?: boolean; replaceRunning?: boolean;
runOptions?: AgentRunOptions;
}; };
/** /**
@@ -171,9 +177,10 @@ export function startAgentRun(
options?: StartAgentRunOptions, options?: StartAgentRunOptions,
): void { ): void {
const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId)); const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId));
const runOptions = options?.runOptions;
const iterator = shouldReplace const iterator = shouldReplace
? agentManager.replaceAgentRun(agentId, prompt) ? agentManager.replaceAgentRun(agentId, prompt, runOptions)
: agentManager.streamAgent(agentId, prompt); : agentManager.streamAgent(agentId, prompt, runOptions);
void (async () => { void (async () => {
try { try {
for await (const _ of iterator) { 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<boolean> {
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<void> {
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 { interface SetupFinishNotificationParams {
agentManager: AgentManager; agentManager: AgentManager;
agentStorage: AgentStorage; agentStorage: AgentStorage;

View File

@@ -65,6 +65,7 @@ import {
toAgentPersistenceHandle, toAgentPersistenceHandle,
} from "./persistence-hooks.js"; } from "./persistence-hooks.js";
import { ensureAgentLoaded } from "./agent/agent-loading.js"; import { ensureAgentLoaded } from "./agent/agent-loading.js";
import { sendPromptToAgent, unarchiveAgentState } from "./agent/mcp-shared.js";
import { experimental_createMCPClient } from "ai"; import { experimental_createMCPClient } from "ai";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import type { VoiceCallerContext, VoiceSpeakHandler } from "./voice-types.js"; import type { VoiceCallerContext, VoiceSpeakHandler } from "./voice-types.js";
@@ -2188,21 +2189,6 @@ export class Session {
}); });
} }
private async unarchiveAgentState(agentId: string): Promise<boolean> {
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<void> { private async unarchiveAgentByHandle(handle: AgentPersistenceHandle): Promise<void> {
const records = await this.agentStorage.list(); const records = await this.agentStorage.list();
const matched = records.find( const matched = records.find(
@@ -2213,7 +2199,7 @@ export class Session {
if (!matched) { if (!matched) {
return; return;
} }
await this.unarchiveAgentState(matched.id); await unarchiveAgentState(this.agentStorage, this.agentManager, matched.id);
} }
private async handleUpdateAgentRequest( 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)` : ""}`, `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 { try {
await ensureAgentLoaded(agentId, { await sendPromptToAgent({
agentManager: this.agentManager, agentManager: this.agentManager,
agentStorage: this.agentStorage, agentStorage: this.agentStorage,
agentId,
userMessageText: text,
prompt,
messageId,
runOptions,
logger: this.sessionLogger, logger: this.sessionLogger,
}); });
return { ok: true };
} catch (error) { } catch (error) {
this.handleAgentRunError(agentId, error, "Failed to initialize agent before sending prompt"); this.handleAgentRunError(agentId, error, "Failed to send agent message");
return { return {
ok: false, ok: false,
error: error instanceof Error ? error.message : String(error), 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 { try {
await this.unarchiveAgentByHandle(handle); await this.unarchiveAgentByHandle(handle);
const snapshot = await this.agentManager.resumeAgentFromPersistence(handle, overrides); 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.agentManager.hydrateTimelineFromProvider(snapshot.id);
await this.forwardAgentUpdate(snapshot); await this.forwardAgentUpdate(snapshot);
const timelineSize = this.agentManager.getTimeline(snapshot.id).length; const timelineSize = this.agentManager.getTimeline(snapshot.id).length;
@@ -2943,7 +2919,7 @@ export class Session {
this.sessionLogger.info({ agentId }, `Refreshing agent ${agentId} from persistence`); this.sessionLogger.info({ agentId }, `Refreshing agent ${agentId} from persistence`);
try { try {
await this.unarchiveAgentState(agentId); await unarchiveAgentState(this.agentStorage, this.agentManager, agentId);
let snapshot: ManagedAgent; let snapshot: ManagedAgent;
const existing = this.agentManager.getAgent(agentId); const existing = this.agentManager.getAgent(agentId);
if (existing) { if (existing) {
@@ -6396,44 +6372,32 @@ export class Session {
try { try {
const agentId = resolved.agentId; 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); const prompt = this.buildAgentPrompt(msg.text, msg.images);
this.sessionLogger.trace( this.sessionLogger.trace(
{ agentId, messageId: msg.messageId }, { agentId, messageId: msg.messageId, textPrefix: msg.text.slice(0, 80) },
"send_agent_message_request: starting agent stream", "send_agent_message_request: dispatching shared sendPromptToAgent",
); );
const started = this.startAgentStream(agentId, prompt); try {
if (!started.ok) { 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({ this.emit({
type: "send_agent_message_response", type: "send_agent_message_response",
payload: { payload: {
requestId: msg.requestId, requestId: msg.requestId,
agentId, agentId,
accepted: false, accepted: false,
error: started.error, error: message,
}, },
}); });
return; return;