feat(trajectory): 重试/切换/传输事件写入轨迹账本——LLM 可观测性补齐

crates/agent-gui/src/lib/chat/runner/agentRunner.ts

@@ -29,12 +29,12 @@ import {
   createStreamingTextReconciler,
   describeProviderCacheShape,
   finalizeProviderStreamOptions,
+  llm,
   normalizeErrorMessage,
   type ProviderRuntimeConfig,
   prepareProviderRequest,
   resolveProviderCacheRetention,
   type StreamOptionsEx,
-  streamSimpleByApi,
   type ToolChoice,
   toSimpleStreamReasoning,
 } from "../../providers/llm";
@@ -52,7 +52,12 @@ import {
   type ProviderFailoverCandidate,
   withProviderFailover,
 } from "../../providers/runtime/providerFailover";
+import { resolveStreamRetryConfig } from "../../providers/runtime/retryPolicy";
 import type { RetryAttemptRecord } from "../../providers/runtime/streamRetry";
+import {
+  captureTransportSnapshot,
+  type TransportSnapshot,
+} from "../../providers/runtime/transportSnapshot";
 import type { RuntimePlatform } from "../../runtimePlatform";
 import type { ProviderId, ReasoningLevel, SelectedModel } from "../../settings";
 import { createSubagentScheduler, type SubagentScheduler } from "../../subagents/scheduler";
@@ -462,6 +467,22 @@ export async function runAssistantWithTools(params: {
   } | null>;
   onToolStatus?: (status: string | null) => void;
   onRetryAttempts?: (round: number, attempts: RetryAttemptRecord[]) => void;
+  /** 每次跨供应商切换（含跳过熔断打开的主选）。targetIndex 是稳定候选下标（0 = 主选）。 */
+  onFailoverAttempt?: (
+    round: number,
+    event: {
+      attempt: number;
+      fromLabel: string;
+      toLabel: string;
+      targetIndex: number;
+      errorMessage: string;
+    },
+  ) => void;
+  /** 每个实际尝试的候选各fire一次：脱敏后的传输装配快照（只含头名，不含值）。 */
+  onTransportAttempt?: (
+    round: number,
+    snapshot: TransportSnapshot & { providerLabel: string },
+  ) => void;
   signal?: AbortSignal;
   debugLogger?: StreamDebugLogger;
   subagentScheduler?: SubagentScheduler;
@@ -1200,6 +1221,7 @@ export async function runAssistantWithTools(params: {
     ) => {
       const round = ++streamRound;
       const retryAttemptsForRound: RetryAttemptRecord[] = [];
+      let failoverAttemptsForRound = 0;
       params.onRetryAttempts?.(round, retryAttemptsForRound);
       const streamTools =
         streamContext.tools ?? (agent?.state.tools as Context["tools"] | undefined) ?? llmTools;
@@ -1324,11 +1346,18 @@ export async function runAssistantWithTools(params: {
           reasoning: normalizeStreamReasoning(options?.reasoning) ?? fallbackReasoning,
           workdir: params.workdir,
           streamRetry: {
-            onRetry: (attempt, maxAttempts, errorMessage) => {
+            ...resolveStreamRetryConfig(target.runtime.retryPolicy),
+            onRetry: (attempt, maxAttempts, errorMessage, plannedDelayMs) => {
               params.onToolStatus?.(
                 `第 ${round} 轮：连接已断开，正在重试 (${attempt}/${maxAttempts})...`,
               );
-              retryAttemptsForRound.push({ attempt, maxAttempts, errorMessage });
+              retryAttemptsForRound.push({
+                attempt,
+                maxAttempts,
+                errorMessage,
+                ...(plannedDelayMs === undefined ? {} : { plannedDelayMs }),
+                providerLabel: target.label,
+              });
               params.onRetryAttempts?.(round, retryAttemptsForRound.slice());
             },
             onRetryRecovered: () => {
@@ -1354,6 +1383,17 @@ export async function runAssistantWithTools(params: {
           },
         });
 
+        try {
+          // 逐候选独立采样：failover 各目标的装配头集互不泄漏是核心正确性
+          // 要求，快照按实际尝试的目标各记一份，观察失败不影响请求。
+          params.onTransportAttempt?.(round, {
+            ...captureTransportSnapshot(streamOptions.headers),
+            providerLabel: target.label,
+          });
+        } catch (error) {
+          console.warn("[agent-runner] transport observer threw; request is unaffected", error);
+        }
+
         // A discarded failover attempt for this round may have left a live
         // probe/aggregator behind; finish it quietly and drop its blocks so
         // the winning attempt starts from a clean slate.
@@ -1405,10 +1445,14 @@ export async function runAssistantWithTools(params: {
           }),
         );
 
-        return streamSimpleByApi(targetModel, effectiveContext, streamOptions);
+        return llm.stream({
+          model: targetModel,
+          context: effectiveContext,
+          options: streamOptions,
+        });
       };
 
-      const wrapWithGuard = (stream: ReturnType<typeof streamSimpleByApi>) =>
+      const wrapWithGuard = (stream: ReturnType<typeof llm.stream>) =>
         wrapStreamWithToolCallArgumentGuard(stream, (toolCall, reason) => {
           incompleteToolCallArguments.set(toolCall.id, reason);
         });
@@ -1453,8 +1497,18 @@ export async function runAssistantWithTools(params: {
       const failoverStream = withProviderFailover(candidates, {
         config: failoverParams.config,
         signal: options?.signal,
-        onFailover: ({ fromLabel, toLabel, errorMessage }) => {
+        onFailover: ({ fromLabel, toLabel, toIndex, errorMessage }) => {
           lastFailoverErrorMessage = errorMessage;
+          failoverAttemptsForRound += 1;
+          params.onFailoverAttempt?.(round, {
+            attempt: failoverAttemptsForRound,
+            fromLabel,
+            toLabel,
+            // toIndex 是本轮 candidates 数组下标；映射回稳定候选下标（0 = 主选），
+            // sticky 重排后账本里的目标身份才不随轮次漂移。
+            targetIndex: targetOrder[toIndex] ?? toIndex,
+            errorMessage,
+          });
           params.onToolStatus?.(`第 ${round} 轮：${fromLabel} 不可用，正在切换到 ${toLabel}...`);
         },
         onCommitted: (candidateIndex) => {

crates/agent-gui/src/lib/providers/llm.ts

@@ -48,3 +48,5 @@ export type {
   StreamOptionsEx,
   ToolChoice,
 } from "./runtime/types";
+export { llm, llmStream } from "./service/llmService";
+export type { LlmAdapter, LlmStreamRequest } from "./service/types";

crates/agent-gui/src/lib/providers/runtime/payloadPipeline.ts

@@ -7,6 +7,11 @@ import {
   attachOpenAICompletionsNativeAttachments,
   attachOpenAIResponsesNativeAttachments,
 } from "../nativeResponsesAttachments";
+import {
+  composePayloadInterceptorChain,
+  installDefaultPayloadInterceptors,
+  type PayloadInterceptor,
+} from "../service/interceptors";
 import { attachAnthropicAutomaticCaching } from "./anthropicCache";
 import { attachAnthropicLongContextBeta } from "./anthropicLongContext";
 import { attachCodexPromptCacheHint } from "./codexPromptCache";
@@ -83,76 +88,117 @@ export function attachPayloadDebugLogging(
   };
 }
 
-const finalizePayloadMiddlewares = composePayloadMiddlewares([
-  (options, params) => attachAnthropicAutomaticCaching(params.providerId, params.baseUrl, options),
-  (options, params) =>
-    attachAnthropicLongContextBeta(options, {
-      providerId: params.providerId,
-      baseUrl: params.baseUrl,
-      model: params.model,
-      context: params.context,
-    }),
-  (options, params) => attachCodexResponsesStorage(params.providerId, options),
-  (options, params) =>
-    attachCodexPromptCacheHint(
-      params.providerId,
-      params.baseUrl,
-      params.promptCacheHintMode,
-      params.model,
-      options,
-    ),
-  (options, params) =>
-    attachProviderNativeWebSearch(params.providerId, options, params.nativeWebSearch, {
-      baseUrl: params.baseUrl,
-    }),
-  (options, params) =>
-    attachXaiResponsesPayloadCompat(options, {
-      providerId: params.providerId,
-      baseUrl: params.baseUrl,
-    }),
-  (options, params) =>
-    attachDeepSeekResponsesPayloadCompat(options, {
-      providerId: params.providerId,
-      model: params.model,
-      context: params.context,
-    }),
-  (options, params) => {
-    if (!params.context || !params.model) return options;
-    let nextOptions = attachOpenAIResponsesNativeAttachments(options, {
-      context: params.context,
-      model: params.model,
-      providerId: params.providerId,
-      workdir: params.workdir,
-    });
-    nextOptions = attachOpenAICompletionsNativeAttachments(nextOptions, {
-      context: params.context,
-      model: params.model,
-      providerId: params.providerId,
-      workdir: params.workdir,
-    });
-    nextOptions = attachAnthropicMessagesNativeAttachments(nextOptions, {
-      context: params.context,
-      model: params.model,
-      providerId: params.providerId,
-      workdir: params.workdir,
-    });
-    return attachGeminiGenerativeAINativeAttachments(nextOptions, {
-      context: params.context,
-      model: params.model,
-      providerId: params.providerId,
-      workdir: params.workdir,
-    });
+/**
+ * 现有 10 个中间件的具名默认拦截器（PR-3 注册化）。顺序与注册化前的
+ * finalizePayloadMiddlewares 数组逐项一致——顺序即协议正确性的一部分
+ * （如 native attachments 必须先于 gemini thought guard），由顺序快照
+ * 测试锁定。payload-debug-logging 是钉住的链尾：自定义拦截器插入在默认
+ * 拦截器之后、它之前，保证自定义改动仍被调试日志观测到。
+ */
+const DEFAULT_PAYLOAD_INTERCEPTORS: readonly PayloadInterceptor[] = [
+  {
+    name: "anthropic-automatic-caching",
+    intercept: (options, params) =>
+      attachAnthropicAutomaticCaching(params.providerId, params.baseUrl, options),
+  },
+  {
+    name: "anthropic-long-context-beta",
+    intercept: (options, params) =>
+      attachAnthropicLongContextBeta(options, {
+        providerId: params.providerId,
+        baseUrl: params.baseUrl,
+        model: params.model,
+        context: params.context,
+      }),
+  },
+  {
+    name: "codex-responses-storage",
+    intercept: (options, params) => attachCodexResponsesStorage(params.providerId, options),
+  },
+  {
+    name: "codex-prompt-cache-hint",
+    intercept: (options, params) =>
+      attachCodexPromptCacheHint(
+        params.providerId,
+        params.baseUrl,
+        params.promptCacheHintMode,
+        params.model,
+        options,
+      ),
+  },
+  {
+    name: "provider-native-web-search",
+    intercept: (options, params) =>
+      attachProviderNativeWebSearch(params.providerId, options, params.nativeWebSearch, {
+        baseUrl: params.baseUrl,
+      }),
+  },
+  {
+    name: "xai-responses-payload-compat",
+    intercept: (options, params) =>
+      attachXaiResponsesPayloadCompat(options, {
+        providerId: params.providerId,
+        baseUrl: params.baseUrl,
+      }),
+  },
+  {
+    name: "deepseek-responses-payload-compat",
+    intercept: (options, params) =>
+      attachDeepSeekResponsesPayloadCompat(options, {
+        providerId: params.providerId,
+        model: params.model,
+        context: params.context,
+      }),
+  },
+  {
+    name: "native-attachments",
+    intercept: (options, params) => {
+      if (!params.context || !params.model) return options;
+      let nextOptions = attachOpenAIResponsesNativeAttachments(options, {
+        context: params.context,
+        model: params.model,
+        providerId: params.providerId,
+        workdir: params.workdir,
+      });
+      nextOptions = attachOpenAICompletionsNativeAttachments(nextOptions, {
+        context: params.context,
+        model: params.model,
+        providerId: params.providerId,
+        workdir: params.workdir,
+      });
+      nextOptions = attachAnthropicMessagesNativeAttachments(nextOptions, {
+        context: params.context,
+        model: params.model,
+        providerId: params.providerId,
+        workdir: params.workdir,
+      });
+      return attachGeminiGenerativeAINativeAttachments(nextOptions, {
+        context: params.context,
+        model: params.model,
+        providerId: params.providerId,
+        workdir: params.workdir,
+      });
+    },
+  },
+  {
+    name: "gemini-thought-signature-guard",
+    intercept: (options, params) =>
+      attachGeminiThoughtSignatureGuard(options, {
+        providerId: params.providerId,
+        baseUrl: params.baseUrl,
+      }),
   },
-  (options, params) =>
-    attachGeminiThoughtSignatureGuard(options, {
-      providerId: params.providerId,
-      baseUrl: params.baseUrl,
-    }),
-  (options, params) => attachPayloadDebugLogging(options, params.debugLogger, params.extra),
-]);
+  {
+    name: "payload-debug-logging",
+    intercept: (options, params) =>
+      attachPayloadDebugLogging(options, params.debugLogger, params.extra),
+  },
+];
+
+installDefaultPayloadInterceptors(DEFAULT_PAYLOAD_INTERCEPTORS);
 
 export function finalizeProviderStreamOptions(
   params: FinalizeProviderStreamOptionsParams,
 ): StreamOptionsEx {
-  return finalizePayloadMiddlewares(params.options, params);
+  return composePayloadInterceptorChain()(params.options, params);
 }

crates/agent-gui/src/lib/providers/runtime/providerRuntimeConfig.ts

@@ -39,6 +39,7 @@ export function createProviderRuntimeConfig(
     promptCacheRetention: provider.promptCacheRetention,
     nativeWebSearchEnabled: controls.nativeWebSearchEnabled,
     useSystemProxy: provider.useSystemProxy,
+    ...(provider.retryPolicy ? { retryPolicy: provider.retryPolicy } : {}),
     modelConfig: findProviderModelConfig(provider, model),
   } as ProviderRuntimeConfig;
 }

crates/agent-gui/src/lib/providers/runtime/retryPolicy.ts

@@ -0,0 +1,24 @@
+import type { ProviderRetryPolicy } from "../../settings";
+import type { StreamRetryConfig } from "./streamRetry";
+
+/**
+ * 供应商级重试策略 → withStreamRetry 选项的解析（PR-2 策略归属权反转）。
+ *
+ * - 缺省（undefined）→ 空对象：不带 maxAttempts/disabled，withStreamRetry
+ *   落到全局 DEFAULT_STREAM_RETRY_MAX_ATTEMPTS——与反转前行为逐字节一致；
+ * - off → { disabled: true }：禁用流内重试（不影响跨供应商 failover）；
+ * - custom → { maxAttempts: maxRetries + 1 }：设置里存的是"首次失败后的
+ *   重试次数"（用户口径，不含首次请求），withStreamRetry 的 maxAttempts
+ *   是总尝试数，两者相差恰好 1。
+ *
+ * 返回值供消费方与自己的 onRetry/onRetryRecovered 回调展开合并；回调语义
+ * 与 buffer-until-commit 缓冲不受策略影响。failover 场景下每个候选用各自
+ * runtime 的策略调用本函数——策略跟着目标供应商走，与传输配置同口径。
+ */
+export function resolveStreamRetryConfig(
+  retryPolicy: ProviderRetryPolicy | undefined,
+): Pick<StreamRetryConfig, "maxAttempts" | "disabled"> {
+  if (!retryPolicy) return {};
+  if (retryPolicy.mode === "off") return { disabled: true };
+  return { maxAttempts: retryPolicy.maxRetries + 1 };
+}