mirror of
https://github.com/getpaseo/paseo.git
synced 2026-07-29 12:01:31 +00:00
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:
@@ -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,
|
||||
|
||||
@@ -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<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 {
|
||||
agentManager: AgentManager;
|
||||
agentStorage: AgentStorage;
|
||||
|
||||
@@ -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<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> {
|
||||
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;
|
||||
|
||||
Reference in New Issue
Block a user