feat(providers): 供应商级流式重试策略——default/off/custom 三态反转到设置层

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,6 +52,7 @@ import {
   type ProviderFailoverCandidate,
   withProviderFailover,
 } from "../../providers/runtime/providerFailover";
+import { resolveStreamRetryConfig } from "../../providers/runtime/retryPolicy";
 import type { RetryAttemptRecord } from "../../providers/runtime/streamRetry";
 import type { RuntimePlatform } from "../../runtimePlatform";
 import type { ProviderId, ReasoningLevel, SelectedModel } from "../../settings";
@@ -1324,6 +1325,7 @@ export async function runAssistantWithTools(params: {
           reasoning: normalizeStreamReasoning(options?.reasoning) ?? fallbackReasoning,
           workdir: params.workdir,
           streamRetry: {
+            ...resolveStreamRetryConfig(target.runtime.retryPolicy),
             onRetry: (attempt, maxAttempts, errorMessage) => {
               params.onToolStatus?.(
                 `第 ${round} 轮：连接已断开，正在重试 (${attempt}/${maxAttempts})...`,
@@ -1405,10 +1407,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);
         });

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/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 };
+}

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

@@ -1,165 +1,23 @@
 import type { Api, Context, Model } from "@earendil-works/pi-ai";
-import { stream as streamAnthropic } from "@earendil-works/pi-ai/api/anthropic-messages";
-import {
-  type GoogleOptions,
-  stream as streamGoogle,
-} from "@earendil-works/pi-ai/api/google-generative-ai";
-import {
-  type OpenAICompletionsOptions,
-  stream as streamOpenAICompletions,
-} from "@earendil-works/pi-ai/api/openai-completions";
-import {
-  type OpenAIResponsesOptions,
-  stream as streamOpenAIResponses,
-} from "@earendil-works/pi-ai/api/openai-responses";
-import { DEEPSEEK_RESPONSES_API, streamDeepSeekResponses } from "../deepSeekNative";
-import { resolveMaxTokens } from "./common";
-import { rejectEmptyOpenAICompletionsResponse } from "./openAICompletionsStream";
-import { withStreamRetry } from "./streamRetry";
-import {
-  clampOpenAIReasoningEffort,
-  resolveAnthropicThinkingRuntime,
-  resolveGeminiThinkingRuntime,
-} from "./thinkingLevels";
-import type { StreamOptionsEx, ToolChoice } from "./types";
+import { ensureDefaultLlmAdapters } from "../service/defaultAdapters";
+import { resolveAdapter } from "../service/registry";
+import type { StreamOptionsEx } from "./types";
 
-function mapToolChoiceToOpenAI(
-  toolChoice: ToolChoice | undefined,
-): OpenAICompletionsOptions["toolChoice"] | undefined {
-  if (!toolChoice) return undefined;
-  if (toolChoice === "any") return "required";
-  if (toolChoice === "auto" || toolChoice === "none") return toolChoice;
-  return {
-    type: "function",
-    function: {
-      name: toolChoice.name,
-    },
-  };
-}
-
-function mapToolChoiceToGoogle(
-  toolChoice: ToolChoice | undefined,
-): GoogleOptions["toolChoice"] | undefined {
-  if (!toolChoice) return undefined;
-  if (toolChoice === "auto" || toolChoice === "none" || toolChoice === "any") {
-    return toolChoice;
-  }
-  return "auto";
-}
-
-function buildOpenAIBaseOptions(model: Model<Api>, options: StreamOptionsEx) {
-  return {
-    temperature: options.temperature,
-    maxTokens: resolveMaxTokens(options.maxTokens, model.maxTokens),
-    signal: options.signal,
-    apiKey: options.apiKey,
-    cacheRetention: options.cacheRetention,
-    sessionId: options.sessionId,
-    headers: options.headers,
-    onPayload: options.onPayload,
-    maxRetryDelayMs: options.maxRetryDelayMs,
-    metadata: options.metadata,
-  };
-}
+// 保证经本模块的任何调用（含被测试按路径 mock 后又还原的场景）注册表已就绪。
+ensureDefaultLlmAdapters();
 
+/**
+ * 协议分发针孔（PR-1 seam 骨架）。
+ *
+ * 原五协议 switch 已搬移到 ../service/：pi-ai 四协议在 service/piAiAdapter.ts，
+ * DeepSeek 原生在 service/deepSeekAdapter.ts；本函数只剩注册表路由一行，
+ * 保留原签名、语义与 "Unsupported model API: ..." 错误文案。
+ *
+ * 统一入口 llm.stream()（service/llmService.ts）也经由本模块出站——传输
+ * golden 与 failover 测试按本模块路径 mock 即可截获全部出站流，这一
+ * 可观测点是 seam 的公开契约，后续 PR 不得绕开。
+ */
 export function streamSimpleByApi(model: Model<Api>, context: Context, options: StreamOptionsEx) {
-  switch (model.api) {
-    case "anthropic-messages": {
-      // Anthropic：需要我们自己调用 streamAnthropic()，以便显式传 toolChoice（以及启用/禁用 thinking）。
-      const anthropicThinking = resolveAnthropicThinkingRuntime(model, options);
-      // Anthropic 拒绝 extended thinking 与强制工具（"any"/{type:"tool"}）同请求
-      // （400）。降级为 auto：有界强制的调用方（plan mode 补提交轮）同时注入了
-      // 消息级提醒，语义仍然成立；直接 400 反而会进重试/failover 循环。
-      const requestedToolChoice = options.toolChoice ?? "none";
-      const anthropicToolChoice =
-        anthropicThinking.thinkingEnabled &&
-        requestedToolChoice !== "none" &&
-        requestedToolChoice !== "auto"
-          ? "auto"
-          : requestedToolChoice;
-      return withStreamRetry(
-        () => {
-          return streamAnthropic(model as Model<"anthropic-messages">, context, {
-            temperature: options.temperature,
-            maxTokens: anthropicThinking.maxTokens,
-            signal: options.signal,
-            apiKey: options.apiKey,
-            cacheRetention: options.cacheRetention,
-            sessionId: options.sessionId,
-            headers: options.headers,
-            onPayload: options.onPayload,
-            maxRetryDelayMs: options.maxRetryDelayMs,
-            metadata: options.metadata,
-            thinkingEnabled: anthropicThinking.thinkingEnabled,
-            ...(anthropicThinking.effort ? { effort: anthropicThinking.effort } : {}),
-            ...(anthropicThinking.thinkingBudgetTokens !== undefined
-              ? { thinkingBudgetTokens: anthropicThinking.thinkingBudgetTokens }
-              : {}),
-            toolChoice: anthropicToolChoice,
-          });
-        },
-        { signal: options.signal, ...options.streamRetry },
-      );
-    }
-    case "openai-completions": {
-      // 严格校验的 OpenAI 兼容端点（xAI/各类中转网关）对「带 tool_choice 但没带
-      // tools」的请求直接 400（"A tool_choice was set on the request but no tools
-      // were specified"）——compaction 摘要、标题生成等 text-only 请求没有工具，
-      // 会踩中。tool_choice 在无工具时本就无意义，只在请求真正携带 tools 时下发。
-      const openAIOptions: OpenAICompletionsOptions = {
-        ...buildOpenAIBaseOptions(model, options),
-        reasoningEffort: clampOpenAIReasoningEffort(model, options.reasoning),
-        toolChoice: context.tools?.length ? mapToolChoiceToOpenAI(options.toolChoice) : undefined,
-      };
-      return withStreamRetry(
-        () => {
-          return rejectEmptyOpenAICompletionsResponse(
-            streamOpenAICompletions(model as Model<"openai-completions">, context, openAIOptions),
-          );
-        },
-        { signal: options.signal, ...options.streamRetry },
-      );
-    }
-    case DEEPSEEK_RESPONSES_API:
-      return withStreamRetry(() => streamDeepSeekResponses(model, context, options), {
-        signal: options.signal,
-        ...options.streamRetry,
-      });
-    case "openai-responses": {
-      const openAIOptions: OpenAIResponsesOptions = {
-        ...buildOpenAIBaseOptions(model, options),
-        reasoningEffort: clampOpenAIReasoningEffort(model, options.reasoning),
-      };
-      return withStreamRetry(
-        () => streamOpenAIResponses(model as Model<"openai-responses">, context, openAIOptions),
-        {
-          signal: options.signal,
-          ...options.streamRetry,
-        },
-      );
-    }
-    case "google-generative-ai": {
-      const googleOptions: GoogleOptions = {
-        temperature: options.temperature,
-        maxTokens: resolveMaxTokens(options.maxTokens, model.maxTokens),
-        signal: options.signal,
-        apiKey: options.apiKey,
-        headers: options.headers,
-        onPayload: options.onPayload,
-        maxRetryDelayMs: options.maxRetryDelayMs,
-        metadata: options.metadata,
-        thinking: resolveGeminiThinkingRuntime(model, options.reasoning),
-        toolChoice: mapToolChoiceToGoogle(options.toolChoice) ?? "none",
-      };
-      return withStreamRetry(
-        () => streamGoogle(model as Model<"google-generative-ai">, context, googleOptions),
-        {
-          signal: options.signal,
-          ...options.streamRetry,
-        },
-      );
-    }
-    default:
-      throw new Error(`Unsupported model API: ${model.api}`);
-  }
+  ensureDefaultLlmAdapters();
+  return resolveAdapter(model.api).stream(model, context, options);
 }

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

@@ -15,6 +15,7 @@ import {
   withHostedSearchProbeHeader,
 } from "../hostedSearchEvents";
 import { providerSupportsNativeWebSearch } from "../nativeWebSearch";
+import { llm } from "../service/llmService";
 import { appendSystemPrompt, normalizeSessionId } from "./common";
 import { normalizeErrorMessage } from "./errors";
 import { createStreamingTextReconciler } from "./messageUtils";
@@ -32,7 +33,7 @@ import {
   resolveProviderCacheRetention,
   toSimpleStreamReasoning,
 } from "./requestOptions";
-import { streamSimpleByApi } from "./streamByApi";
+import { resolveStreamRetryConfig } from "./retryPolicy";
 import { buildTextModeToolResultsForAssistant } from "./textModeToolRecovery";
 import type { ProviderRuntimeConfig, StreamOptionsEx } from "./types";
 
@@ -115,6 +116,7 @@ function buildTextOnlyStreamOptions(params: {
     // hosted by the upstream provider, so it can stay on auto when explicitly enabled.
     toolChoice: usesOpenAIChatNativeWebSearch ? undefined : nativeWebSearch ? "auto" : "none",
     streamRetry: {
+      ...resolveStreamRetryConfig(params.runtime.retryPolicy),
       onRetry: params.onRetryStatus,
       onRetryRecovered: params.onRetryRecovered,
     },
@@ -331,7 +333,7 @@ export async function streamAssistantMessage(params: {
 
   const startAttemptStream = (activeContext: Context) => {
     if (!failover || failover.fallbacks.length === 0) {
-      return streamSimpleByApi(m, activeContext, options);
+      return llm.stream({ model: m, context: activeContext, options });
     }
     // Candidate order: sticky active target first, then the rest in
     // primary→queue order. Breaker-open targets are skipped inside
@@ -360,7 +362,7 @@ export async function streamAssistantMessage(params: {
             : fallbackTargetIdentity(targetIndex),
         start: async () => {
           if (targetIndex === 0 || !fallback) {
-            return streamSimpleByApi(m, activeContext, options);
+            return llm.stream({ model: m, context: activeContext, options });
           }
           const prepared = await prepareFallbackTarget(targetIndex);
           params.debugLogger?.logRequest(
@@ -370,7 +372,11 @@ export async function streamAssistantMessage(params: {
               options: prepared.options,
             }),
           );
-          return streamSimpleByApi(prepared.model, activeContext, prepared.options);
+          return llm.stream({
+            model: prepared.model,
+            context: activeContext,
+            options: prepared.options,
+          });
         },
       } satisfies ProviderFailoverCandidate;
     });
@@ -575,7 +581,7 @@ export async function completeAssistantMessage(params: {
 
   return withPowerActivity("assistant-complete", `${params.providerId}:${modelId}`, async () => {
     try {
-      const s = streamSimpleByApi(m, callContext, options);
+      const s = llm.stream({ model: m, context: callContext, options });
       const final = await s.result();
 
       if (final.stopReason === "error" || final.stopReason === "aborted") {

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

@@ -6,6 +6,7 @@ import type {
   PromptCacheHintMode,
   ProviderId,
   ProviderModelConfig,
+  ProviderRetryPolicy,
   ReasoningLevel,
 } from "../../settings";
 import type { StreamRetryConfig } from "./streamRetry";
@@ -36,6 +37,8 @@ export type ProviderRuntimeConfig = {
   promptCacheRetention?: "short" | "long";
   nativeWebSearchEnabled?: boolean;
   useSystemProxy?: boolean;
+  /** 供应商级流内重试策略；缺省 = 全局默认。failover 逐候选独立携带。 */
+  retryPolicy?: ProviderRetryPolicy;
   modelConfig?: ProviderModelConfig;
 };
 

crates/agent-gui/src/lib/providers/service/deepSeekAdapter.ts

@@ -0,0 +1,19 @@
+import { DEEPSEEK_RESPONSES_API, streamDeepSeekResponses } from "../deepSeekNative";
+import { withStreamRetry } from "../runtime/streamRetry";
+import type { LlmAdapter } from "./types";
+
+/**
+ * DeepSeek 原生协议适配器。
+ *
+ * streamByApi.ts 中 DEEPSEEK_RESPONSES_API 分支的原样搬移（PR-1 行为等价
+ * 不变量）：withStreamRetry 包装位置与参数逐字保持。
+ */
+export const deepSeekAdapter: LlmAdapter = {
+  apis: [DEEPSEEK_RESPONSES_API] as const,
+  stream(model, context, options) {
+    return withStreamRetry(() => streamDeepSeekResponses(model, context, options), {
+      signal: options.signal,
+      ...options.streamRetry,
+    });
+  },
+};

crates/agent-gui/src/lib/providers/service/defaultAdapters.ts

@@ -0,0 +1,18 @@
+import { deepSeekAdapter } from "./deepSeekAdapter";
+import { piAiAdapter } from "./piAiAdapter";
+import { registerAdapter } from "./registry";
+
+let installed = false;
+
+/**
+ * 安装默认适配器（幂等）。
+ *
+ * 由分发针孔（runtime/streamByApi.ts 兼容壳）与 llm.stream() 各自在模块加载
+ * 时调用：无论消费方从哪个入口进来，注册表都已就绪；重复调用零开销。
+ */
+export function ensureDefaultLlmAdapters(): void {
+  if (installed) return;
+  installed = true;
+  registerAdapter(piAiAdapter);
+  registerAdapter(deepSeekAdapter);
+}

crates/agent-gui/src/lib/providers/service/index.ts

@@ -0,0 +1,6 @@
+export { deepSeekAdapter } from "./deepSeekAdapter";
+export { ensureDefaultLlmAdapters } from "./defaultAdapters";
+export { llm, llmStream, setLlmServiceDevModeForTest } from "./llmService";
+export { piAiAdapter } from "./piAiAdapter";
+export { registerAdapter, registeredApis, resolveAdapter } from "./registry";
+export type { LlmAdapter, LlmStreamRequest } from "./types";