fix(工作台): 改进 MCP 降级与会话/运行恢复

MCP 不可用时跳过并记录 unavailableResources，避免整次 run 失败；会话记忆与前端恢复为失败 run 补齐孤儿 user 消息占位，并统一 run forwardedProps 与 token 用量汇总。

Author: wangjk9527

apps/api/src/conversation-memory.ts

@@ -252,11 +252,9 @@ export const buildConversationMemoryMessagesWithReport = (
   });
   const currentUserMessageId = lastUserMessage(input.runInput)?.id ?? `${input.runId}:user`;
   const messages: Message[] = selected.summary ? [summaryToMessage(selected.summary, policy.maxMessageChars)] : [];
-  messages.push(...selected.history.map((record) => ({
-    id: `memory:${record.id}`,
-    role: record.role,
-    content: boundText(record.content_text, policy.maxMessageChars)
-  })));
+  messages.push(
+    ...normalizeHistoryMessagePairs(selected.history, policy.maxMessageChars)
+  );
 
   messages.push({
     id: currentUserMessageId,
@@ -490,6 +488,50 @@ const generateSummaryText = async (input: {
   });
 };
 
+/**
+ * Normalises orphaned user messages in conversation history into valid
+ * user/assistant pairs by injecting a synthetic assistant error reply.
+ *
+ * An orphaned user message is a user-role record that is NOT followed by an
+ * assistant reply – either it sits at the tail of the history or it is
+ * directly followed by another user message.  This happens when a run fails
+ * before the agent produces any output: `persistCurrentUserMessage` has
+ * already written the record but no corresponding assistant record is stored.
+ *
+ * Simply discarding these records loses conversation context.  Instead we
+ * inject a placeholder assistant message so the LLM:
+ *   1. Sees the full history of what the user asked.
+ *   2. Understands that the previous attempt failed to produce a response.
+ *   3. Receives a strictly alternating user/assistant sequence, which all
+ *      major providers (Anthropic, DashScope, …) require.
+ */
+const ORPHANED_USER_MESSAGE_PLACEHOLDER =
+  "Previous request failed before the assistant produced a response.";
+
+const normalizeHistoryMessagePairs = (
+  records: ConversationMessageRecord[],
+  maxMessageChars: number
+): Message[] => {
+  const result: Message[] = [];
+  for (let i = 0; i < records.length; i++) {
+    const cur = records[i]!;
+    const next = records[i + 1] as ConversationMessageRecord | undefined;
+    result.push({
+      id: `memory:${cur.id}`,
+      role: cur.role,
+      content: boundText(cur.content_text, maxMessageChars)
+    });
+    if (cur.role === "user" && (!next || next.role === "user")) {
+      result.push({
+        id: `memory:${cur.id}:error-placeholder`,
+        role: "assistant",
+        content: ORPHANED_USER_MESSAGE_PLACEHOLDER
+      });
+    }
+  }
+  return result;
+};
+
 const selectBudgetedHistory = (input: {
   currentUserText: string;
   history: ConversationMessageRecord[];

apps/api/src/run-config-resolver.ts

@@ -36,6 +36,7 @@ export type ResolvedRunConfig = {
 export type McpRuntime = {
   servers: PolicyMcpClientConfig[];
   toolNames: string[];
+  skipped?: { id: string; reason: string }[];
 };
 
 type ResolveRunConfigInput = {
@@ -110,6 +111,20 @@ export const resolveRunConfig = (input: ResolveRunConfigInput): ResolvedRunConfi
     input.userId,
     input.workspaceId
   );
+  if (mcpRuntime.skipped && mcpRuntime.skipped.length > 0) {
+    const skippedIds = new Set(mcpRuntime.skipped.map((entry) => entry.id));
+    effectiveRunConfig.enabledMcpServerIds = effectiveRunConfig.enabledMcpServerIds.filter(
+      (id) => !skippedIds.has(id)
+    );
+    effectiveRunConfig.unavailableResources = [
+      ...(effectiveRunConfig.unavailableResources ?? []),
+      ...mcpRuntime.skipped.map((entry) => ({
+        kind: "mcp-server" as const,
+        id: entry.id,
+        reason: entry.reason
+      }))
+    ];
+  }
   enforceSkillMcpPolicy(skillSelection.effectiveToolPolicy.allowedTools, mcpRuntime.toolNames);
 
   return {
@@ -441,78 +456,112 @@ const resolveMcpRuntime = (
 ): McpRuntime => {
   const usedNames = new Set<string>();
   const toolNames: string[] = [];
-  const servers = serverIds.map((id): PolicyMcpClientConfig => {
-    const resource = metadataStore.configResources.get({
-      id,
-      workspace_id: workspaceId,
-      user_id: userId,
-      kind: "mcp-server"
-    });
-    const transport = stringRecordValue(resource.payload, "transport") ?? "streamable-http";
-    if (transport !== "streamable-http" && transport !== "sse" && transport !== "stdio") {
-      throw new Error(`MCP_TRANSPORT_UNSUPPORTED:${id}:${transport}`);
+  const servers: PolicyMcpClientConfig[] = [];
+  const skipped: { id: string; reason: string }[] = [];
+  for (const id of serverIds) {
+    try {
+      servers.push(
+        buildMcpServerConfig({
+          id,
+          metadataStore,
+          usedNames,
+          toolNames,
+          userId,
+          workspaceId
+        })
+      );
+    } catch (error) {
+      skipped.push({
+        id,
+        reason: error instanceof Error ? error.message : String(error)
+      });
     }
-    const urlOrCommand = stringRecordValue(resource.payload, "serverUrl") ?? stringRecordValue(resource.payload, "url");
-    if (!urlOrCommand) {
-      throw new Error(`MCP_SERVER_URL_REQUIRED:${id}`);
+  }
+  return {
+    servers,
+    toolNames,
+    ...(skipped.length > 0 ? { skipped } : {})
+  };
+};
+
+const buildMcpServerConfig = (input: {
+  id: string;
+  metadataStore: MetadataStore;
+  usedNames: Set<string>;
+  toolNames: string[];
+  userId: string;
+  workspaceId: string;
+}): PolicyMcpClientConfig => {
+  const { id, metadataStore, usedNames, toolNames, userId, workspaceId } = input;
+  const resource = metadataStore.configResources.get({
+    id,
+    workspace_id: workspaceId,
+    user_id: userId,
+    kind: "mcp-server"
+  });
+  const transport = stringRecordValue(resource.payload, "transport") ?? "streamable-http";
+  if (transport !== "streamable-http" && transport !== "sse" && transport !== "stdio") {
+    throw new Error(`MCP_TRANSPORT_UNSUPPORTED:${id}:${transport}`);
+  }
+  const urlOrCommand = stringRecordValue(resource.payload, "serverUrl") ?? stringRecordValue(resource.payload, "url");
+  if (!urlOrCommand) {
+    throw new Error(`MCP_SERVER_URL_REQUIRED:${id}`);
+  }
+  const manifest = resource.payload.toolManifest;
+  if (!Array.isArray(manifest)) {
+    throw new Error(`MCP_TOOL_MANIFEST_REQUIRED:${id}`);
+  }
+  const toolAllowlist = stringArrayRecordValue(resource.payload, "toolAllowlist")
+    ?? csvRecordValue(resource.payload, "toolAllowlist");
+  manifest.forEach((tool) => {
+    if (!isRecord(tool) || typeof tool.name !== "string") {
+      throw new Error(`MCP_TOOL_MANIFEST_INVALID:${id}`);
     }
-    const manifest = resource.payload.toolManifest;
-    if (!Array.isArray(manifest)) {
-      throw new Error(`MCP_TOOL_MANIFEST_REQUIRED:${id}`);
+    if (!matchesMcpToolAllowlist(id, tool.name, toolAllowlist)) {
+      return;
     }
-    const toolAllowlist = stringArrayRecordValue(resource.payload, "toolAllowlist")
-      ?? csvRecordValue(resource.payload, "toolAllowlist");
-    manifest.forEach((tool) => {
-      if (!isRecord(tool) || typeof tool.name !== "string") {
-        throw new Error(`MCP_TOOL_MANIFEST_INVALID:${id}`);
-      }
-      if (!matchesMcpToolAllowlist(id, tool.name, toolAllowlist)) {
-        return;
-      }
-      const baseName = `mcp__${sanitizeMcpName(id)}__${sanitizeMcpName(tool.name)}`;
-      let resolved = baseName.slice(0, 64);
-      let suffix = 1;
-      while (usedNames.has(resolved)) {
-        const marker = `_${suffix}`;
-        resolved = `${baseName.slice(0, 64 - marker.length)}${marker}`;
-        suffix += 1;
-      }
-      usedNames.add(resolved);
-      toolNames.push(resolved);
-    });
-    const secret = resource.secret_ref
-      ? metadataStore.secrets.get({ ref: resource.secret_ref, workspace_id: workspaceId, user_id: userId })
-      : {};
-    const headers = resolveMcpHeaders(resource.payload, secret);
-    const timeoutMs = numericRecordValue(resource.payload, "timeoutMs") ?? numericRecordValue(resource.payload, "timeout_ms");
-    const common = {
-      serverId: id,
-      ...(toolAllowlist && toolAllowlist.length > 0 ? { toolAllowlist } : {}),
-      ...(timeoutMs !== undefined ? { timeoutMs: Math.max(1000, Math.min(10 * 60 * 1000, Math.floor(timeoutMs))) } : {})
-    };
-    if (transport === "stdio") {
-      const stdio = resolveStdioCommand(resource.payload, urlOrCommand);
-      const cwd = stringRecordValue(resource.payload, "cwd");
-      const env = recordStringMapValue(resource.payload.env);
-      return {
-        ...common,
-        type: "stdio",
-        command: stdio.command,
-        ...(stdio.args.length > 0 ? { args: stdio.args } : {}),
-        ...(cwd ? { cwd } : {}),
-        ...(env ? { env } : {})
-      };
+    const baseName = `mcp__${sanitizeMcpName(id)}__${sanitizeMcpName(tool.name)}`;
+    let resolved = baseName.slice(0, 64);
+    let suffix = 1;
+    while (usedNames.has(resolved)) {
+      const marker = `_${suffix}`;
+      resolved = `${baseName.slice(0, 64 - marker.length)}${marker}`;
+      suffix += 1;
     }
+    usedNames.add(resolved);
+    toolNames.push(resolved);
+  });
+  const secret = resource.secret_ref
+    ? metadataStore.secrets.get({ ref: resource.secret_ref, workspace_id: workspaceId, user_id: userId })
+    : {};
+  const headers = resolveMcpHeaders(resource.payload, secret);
+  const timeoutMs = numericRecordValue(resource.payload, "timeoutMs") ?? numericRecordValue(resource.payload, "timeout_ms");
+  const common = {
+    serverId: id,
+    ...(toolAllowlist && toolAllowlist.length > 0 ? { toolAllowlist } : {}),
+    ...(timeoutMs !== undefined ? { timeoutMs: Math.max(1000, Math.min(10 * 60 * 1000, Math.floor(timeoutMs))) } : {})
+  };
+  if (transport === "stdio") {
+    const stdio = resolveStdioCommand(resource.payload, urlOrCommand);
+    const cwd = stringRecordValue(resource.payload, "cwd");
+    const env = recordStringMapValue(resource.payload.env);
     return {
-      type: transport === "streamable-http" ? "http" : "sse",
-      url: urlOrCommand,
-      serverId: id,
-      ...(toolAllowlist && toolAllowlist.length > 0 ? { toolAllowlist } : {}),
-      ...(timeoutMs !== undefined ? { timeoutMs: Math.max(1000, Math.min(10 * 60 * 1000, Math.floor(timeoutMs))) } : {}),
-      ...(Object.keys(headers).length > 0 ? { headers } : {})
+      ...common,
+      type: "stdio",
+      command: stdio.command,
+      ...(stdio.args.length > 0 ? { args: stdio.args } : {}),
+      ...(cwd ? { cwd } : {}),
+      ...(env ? { env } : {})
     };
-  });
-  return { servers, toolNames };
+  }
+  return {
+    type: transport === "streamable-http" ? "http" : "sse",
+    url: urlOrCommand,
+    serverId: id,
+    ...(toolAllowlist && toolAllowlist.length > 0 ? { toolAllowlist } : {}),
+    ...(timeoutMs !== undefined ? { timeoutMs: Math.max(1000, Math.min(10 * 60 * 1000, Math.floor(timeoutMs))) } : {}),
+    ...(Object.keys(headers).length > 0 ? { headers } : {})
+  };
 };
 
 const resolveMcpHeaders = (

apps/api/src/run-finalizer.ts

@@ -194,7 +194,7 @@ export const createRunStatusDelta = (
 ): BaseEvent => ({
   type: EventType.STATE_DELTA,
   delta: [
-    { op: "replace", path: "/runStatus", value: status },
+    { op: "add", path: "/runStatus", value: status },
     ...(errorMessage ? [{ op: "add", path: "/errorMessage", value: errorMessage }] : [])
   ],
   timestamp: Date.now()

apps/api/src/run-input.ts

@@ -50,6 +50,11 @@ export type EffectiveRunConfig = {
    * The run continues; this list is surfaced in `run.config.resolved` for diagnostics.
    */
   disabledByPolicy?: { kind: "knowledge-base" | "mcp-server" | "model-profile"; id: string }[];
+  /**
+   * Resources silently dropped because runtime configuration is invalid (missing manifest,
+   * URL, etc.). The run continues without those MCP tools.
+   */
+  unavailableResources?: { kind: "mcp-server"; id: string; reason: string }[];
 };
 
 export type PerRunMentionKind = "db" | "kb" | "mcp" | "skill";