diff --git a/crates/agent-gui/src/lib/chat/runner/agentRunner.ts b/crates/agent-gui/src/lib/chat/runner/agentRunner.ts index 6218434b6..f57c4529e 100644 --- a/crates/agent-gui/src/lib/chat/runner/agentRunner.ts +++ b/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) => + const wrapWithGuard = (stream: ReturnType) => 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) => { diff --git a/crates/agent-gui/src/lib/providers/llm.ts b/crates/agent-gui/src/lib/providers/llm.ts index 6813b73de..74c042b4b 100644 --- a/crates/agent-gui/src/lib/providers/llm.ts +++ b/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"; diff --git a/crates/agent-gui/src/lib/providers/runtime/payloadPipeline.ts b/crates/agent-gui/src/lib/providers/runtime/payloadPipeline.ts index dfc50f68f..091a9d4a7 100644 --- a/crates/agent-gui/src/lib/providers/runtime/payloadPipeline.ts +++ b/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); } diff --git a/crates/agent-gui/src/lib/providers/runtime/providerRuntimeConfig.ts b/crates/agent-gui/src/lib/providers/runtime/providerRuntimeConfig.ts index bca6b3e5c..2544c9a5e 100644 --- a/crates/agent-gui/src/lib/providers/runtime/providerRuntimeConfig.ts +++ b/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; } diff --git a/crates/agent-gui/src/lib/providers/runtime/retryPolicy.ts b/crates/agent-gui/src/lib/providers/runtime/retryPolicy.ts new file mode 100644 index 000000000..0aba0f12b --- /dev/null +++ b/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 { + if (!retryPolicy) return {}; + if (retryPolicy.mode === "off") return { disabled: true }; + return { maxAttempts: retryPolicy.maxRetries + 1 }; +} diff --git a/crates/agent-gui/src/lib/providers/runtime/streamByApi.ts b/crates/agent-gui/src/lib/providers/runtime/streamByApi.ts index ad45c1a33..39f2e7ec2 100644 --- a/crates/agent-gui/src/lib/providers/runtime/streamByApi.ts +++ b/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, 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, 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); } diff --git a/crates/agent-gui/src/lib/providers/runtime/streamRetry.ts b/crates/agent-gui/src/lib/providers/runtime/streamRetry.ts index aef4299d6..c0c3887b1 100644 --- a/crates/agent-gui/src/lib/providers/runtime/streamRetry.ts +++ b/crates/agent-gui/src/lib/providers/runtime/streamRetry.ts @@ -19,9 +19,15 @@ export type StreamRetryConfig = { disabled?: boolean; /** * Retry ordinal (1..maxRetries) about to be attempted, invoked before the - * backoff sleep. `errorMessage` is the failure that triggered this retry. + * backoff sleep. `errorMessage` is the failure that triggered this retry; + * `plannedDelayMs` is the backoff about to be slept (PR-4 audit field). */ - onRetry?: (attempt: number, maxAttempts: number, errorMessage: string) => void; + onRetry?: ( + attempt: number, + maxAttempts: number, + errorMessage: string, + plannedDelayMs?: number, + ) => void; /** Invoked once a retried attempt commits its first content-bearing event. */ onRetryRecovered?: () => void; }; @@ -147,10 +153,17 @@ export function withStreamRetry( if (isRetryableAssistantError(terminalMessage(terminal))) { const errorMessage = terminalMessage(terminal)?.errorMessage || "Unknown error"; attempt += 1; - options?.onRetry?.(attempt - 1, maxAttempts - 1, errorMessage); + // Computed before the callback so the audit trail records the exact + // backoff about to be slept. Rounded to whole milliseconds: setTimeout + // is ms-granular anyway, and a fractional float drifts by 1 ulp per + // trajectory persistence merge (serde_json best-effort float parse), + // which would give the same retry two identities in the converged + // ledger — duplicated rows and an inflated retry count. + const plannedDelayMs = Math.round(computeStreamRetryBackoffMs(attempt - 1)); + options?.onRetry?.(attempt - 1, maxAttempts - 1, errorMessage, plannedDelayMs); hasRetried = true; try { - await sleepWithAbort(computeStreamRetryBackoffMs(attempt - 1), signal); + await sleepWithAbort(plannedDelayMs, signal); source = factory(); continue; } catch { diff --git a/crates/agent-gui/src/lib/providers/runtime/textOnlyRuntime.ts b/crates/agent-gui/src/lib/providers/runtime/textOnlyRuntime.ts index a7b2de98f..666c3b13e 100644 --- a/crates/agent-gui/src/lib/providers/runtime/textOnlyRuntime.ts +++ b/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,8 +33,9 @@ import { resolveProviderCacheRetention, toSimpleStreamReasoning, } from "./requestOptions"; -import { streamSimpleByApi } from "./streamByApi"; +import { resolveStreamRetryConfig } from "./retryPolicy"; import { buildTextModeToolResultsForAssistant } from "./textModeToolRecovery"; +import { captureTransportSnapshot, type TransportSnapshot } from "./transportSnapshot"; import type { ProviderRuntimeConfig, StreamOptionsEx } from "./types"; function buildTextOnlySystemSuffix(allowJsonOutput = false) { @@ -75,7 +77,12 @@ function buildTextOnlyStreamOptions(params: { cacheRetention?: CacheRetention; nativeWebSearch?: boolean; debugLogger?: StreamDebugLogger; - onRetryStatus?: (attempt: number, maxAttempts: number, errorMessage: string) => void; + onRetryStatus?: ( + attempt: number, + maxAttempts: number, + errorMessage: string, + plannedDelayMs?: number, + ) => void; onRetryRecovered?: () => void; }): StreamOptionsEx { const sessionId = normalizeSessionId(params.sessionId); @@ -115,6 +122,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, }, @@ -153,7 +161,13 @@ export type TextStreamFailoverParams = { /** Fired when an attempt commits on a different target than the previous ones. */ onSwitched?: (event: { target: TextStreamFailoverTarget | null; errorMessage: string }) => void; /** Fired before each switch, including a skip of an open-breaker primary. */ - onFailover?: (event: { fromLabel: string; toLabel: string; errorMessage: string }) => void; + onFailover?: (event: { + fromLabel: string; + toLabel: string; + /** Stable candidate index of the switch target (0 = primary). */ + targetIndex: number; + errorMessage: string; + }) => void; }; export async function streamAssistantMessage(params: { @@ -170,8 +184,15 @@ export async function streamAssistantMessage(params: { allowJsonOutput?: boolean; nativeWebSearch?: boolean; onHostedSearch?: (block: HostedSearchBlock) => void; - onRetryStatus?: (attempt: number, maxAttempts: number, errorMessage: string) => void; + onRetryStatus?: ( + attempt: number, + maxAttempts: number, + errorMessage: string, + plannedDelayMs?: number, + ) => void; onRetryRecovered?: () => void; + /** 每个实际尝试的候选各 fire 一次:脱敏后的传输装配快照(只含头名,不含值)。 */ + onTransportAttempt?: (snapshot: TransportSnapshot & { providerLabel: string }) => void; /** Exact text-only provider boundary after its mandatory system suffix is appended. */ onRequestStart?: (info: { context: Context; systemSuffix: string }) => void; failover?: TextStreamFailoverParams; @@ -329,9 +350,25 @@ export async function streamAssistantMessage(params: { let activeFailoverTargetIndex = 0; let lastFailoverErrorMessage = ""; + /** 逐候选独立采样;观察失败不影响请求。 */ + const noteTransportAttempt = ( + label: string, + attemptOptions: StreamOptionsEx | undefined, + ): void => { + try { + params.onTransportAttempt?.({ + ...captureTransportSnapshot(attemptOptions?.headers), + providerLabel: label, + }); + } catch (error) { + console.warn("text-only transport observer failed; continuing without diagnostics", error); + } + }; + const startAttemptStream = (activeContext: Context) => { if (!failover || failover.fallbacks.length === 0) { - return streamSimpleByApi(m, activeContext, options); + noteTransportAttempt(primaryFailoverLabel, 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 +397,8 @@ export async function streamAssistantMessage(params: { : fallbackTargetIdentity(targetIndex), start: async () => { if (targetIndex === 0 || !fallback) { - return streamSimpleByApi(m, activeContext, options); + noteTransportAttempt(primaryFailoverLabel, options); + return llm.stream({ model: m, context: activeContext, options }); } const prepared = await prepareFallbackTarget(targetIndex); params.debugLogger?.logRequest( @@ -370,7 +408,12 @@ export async function streamAssistantMessage(params: { options: prepared.options, }), ); - return streamSimpleByApi(prepared.model, activeContext, prepared.options); + noteTransportAttempt(fallback.label, prepared.options); + return llm.stream({ + model: prepared.model, + context: activeContext, + options: prepared.options, + }); }, } satisfies ProviderFailoverCandidate; }); @@ -382,6 +425,9 @@ export async function streamAssistantMessage(params: { failover.onFailover?.({ fromLabel: event.fromLabel, toLabel: event.toLabel, + // Map the per-call candidates index back to the stable target index + // (0 = primary) so sticky reordering can't skew the audit trail. + targetIndex: targetOrder[event.toIndex] ?? event.toIndex, errorMessage: event.errorMessage, }); }, @@ -575,7 +621,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") { diff --git a/crates/agent-gui/src/lib/providers/runtime/transportSnapshot.ts b/crates/agent-gui/src/lib/providers/runtime/transportSnapshot.ts new file mode 100644 index 000000000..62e090ce6 --- /dev/null +++ b/crates/agent-gui/src/lib/providers/runtime/transportSnapshot.ts @@ -0,0 +1,40 @@ +import { + LIVEAGENT_UPSTREAM_ORIGIN_HEADER, + LIVEAGENT_UPSTREAM_URL_HEADER, + LIVEAGENT_USE_SYSTEM_PROXY_HEADER, +} from "@liveagent/ui/lib/providers/proxy"; + +/** + * 一次实际出站尝试的传输装配摘要,供轨迹账本审计逐候选独立性 + * (主选带 use-system-proxy 头、备选不带,互不泄漏)。 + * + * 脱敏不变量:只读头**名**与路由标记,绝不读头值——鉴权头(authorization/ + * x-api-key/x-goog-api-key)、代理 token、base64 覆盖包的取值全部不进快照。 + * upstream origin 是 scheme+host(与 step_end 已落盘的 provider/model 同 + * 敏感级),fullUrl 模式下完整 URL 可能含 query 凭据,因此只记布尔标记。 + */ +export type TransportSnapshot = { + upstreamOrigin?: string; + useSystemProxy: boolean; + fullUrl: boolean; + /** 全部头名,小写去重后按字典序;值一律不采集。 */ + headerNames: readonly string[]; +}; + +export function captureTransportSnapshot( + headers: Record | undefined, +): TransportSnapshot { + const byLowerName = new Map(); + for (const [name, value] of Object.entries(headers ?? {})) { + // null 是"删除该头"标记(pi-ai ProviderHeaders 语义),不会出现在出站请求里。 + if (value === null) continue; + byLowerName.set(name.toLowerCase(), value); + } + const upstreamOrigin = byLowerName.get(LIVEAGENT_UPSTREAM_ORIGIN_HEADER)?.trim(); + return { + ...(upstreamOrigin === undefined || upstreamOrigin === "" ? {} : { upstreamOrigin }), + useSystemProxy: byLowerName.get(LIVEAGENT_USE_SYSTEM_PROXY_HEADER) === "1", + fullUrl: byLowerName.has(LIVEAGENT_UPSTREAM_URL_HEADER), + headerNames: [...byLowerName.keys()].sort(), + }; +} diff --git a/crates/agent-gui/src/lib/providers/runtime/types.ts b/crates/agent-gui/src/lib/providers/runtime/types.ts index 305ff11bc..837441483 100644 --- a/crates/agent-gui/src/lib/providers/runtime/types.ts +++ b/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; }; diff --git a/crates/agent-gui/src/lib/providers/service/deepSeekAdapter.ts b/crates/agent-gui/src/lib/providers/service/deepSeekAdapter.ts new file mode 100644 index 000000000..828963edf --- /dev/null +++ b/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, + }); + }, +}; diff --git a/crates/agent-gui/src/lib/providers/service/defaultAdapters.ts b/crates/agent-gui/src/lib/providers/service/defaultAdapters.ts new file mode 100644 index 000000000..51c987956 --- /dev/null +++ b/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); +} diff --git a/crates/agent-gui/src/lib/providers/service/index.ts b/crates/agent-gui/src/lib/providers/service/index.ts new file mode 100644 index 000000000..8a4f2a8f7 --- /dev/null +++ b/crates/agent-gui/src/lib/providers/service/index.ts @@ -0,0 +1,11 @@ +export { deepSeekAdapter } from "./deepSeekAdapter"; +export { ensureDefaultLlmAdapters } from "./defaultAdapters"; +export { + listPayloadInterceptorNames, + type PayloadInterceptor, + usePayloadInterceptor, +} from "./interceptors"; +export { llm, llmStream, setLlmServiceDevModeForTest } from "./llmService"; +export { piAiAdapter } from "./piAiAdapter"; +export { registerAdapter, registeredApis, resolveAdapter } from "./registry"; +export type { LlmAdapter, LlmStreamRequest } from "./types"; diff --git a/crates/agent-gui/src/lib/providers/service/interceptors.ts b/crates/agent-gui/src/lib/providers/service/interceptors.ts new file mode 100644 index 000000000..8bda3713e --- /dev/null +++ b/crates/agent-gui/src/lib/providers/service/interceptors.ts @@ -0,0 +1,113 @@ +import type { ProviderPayloadMiddleware } from "../runtime/payloadPipeline"; + +/** + * 具名 payload 拦截器(PR-3 拦截器注册化)。 + * + * name 是注册表内的唯一身份:顺序快照测试按名字逐一断言,同名重复注册 + * 抛错,dispose 按名字移除。intercept 即原 ProviderPayloadMiddleware—— + * 纯函数 (options, params) => options,注册化不改变中间件本身的契约。 + */ +export type PayloadInterceptor = { + readonly name: string; + readonly intercept: ProviderPayloadMiddleware; +}; + +/** + * 默认拦截器(现有 10 个中间件的具名包装,payloadPipeline.ts 模块初始化 + * 时一次性安装,顺序即原 finalizePayloadMiddlewares 数组顺序)。最后一个 + * 是钉住链尾的 payload-debug-logging。 + */ +let defaultInterceptors: readonly PayloadInterceptor[] = []; + +/** 自定义拦截器,按注册先后排列;执行位置在默认拦截器之后、链尾之前。 */ +const customInterceptors: PayloadInterceptor[] = []; + +/** + * 组合结果缓存。finalizeProviderStreamOptions 在热路径上(agentRunner 每 + * 轮、textOnly 每次调用),注册/移除时失效重建,调用时零分配。 + */ +let composedChain: ProviderPayloadMiddleware | undefined; + +function invalidateComposedChain(): void { + composedChain = undefined; +} + +function hasInterceptorName(name: string): boolean { + return ( + defaultInterceptors.some((entry) => entry.name === name) || + customInterceptors.some((entry) => entry.name === name) + ); +} + +/** + * 安装默认拦截器链。仅供 payloadPipeline.ts 模块初始化调用一次;重复安装 + * 抛错(防止测试加载器或 HMR 下的双重初始化悄悄改变链序)。 + */ +export function installDefaultPayloadInterceptors( + interceptors: readonly PayloadInterceptor[], +): void { + if (defaultInterceptors.length > 0) { + throw new Error("Default payload interceptors were already installed"); + } + const seen = new Set(); + for (const entry of interceptors) { + if (seen.has(entry.name)) { + throw new Error(`Duplicate default payload interceptor name: ${entry.name}`); + } + seen.add(entry.name); + } + defaultInterceptors = [...interceptors]; + invalidateComposedChain(); +} + +/** + * 注册自定义拦截器,返回 dispose(幂等)。 + * + * 执行顺序不变量:默认拦截器(除链尾)→ 自定义(按注册先后)→ + * payload-debug-logging 链尾。自定义改动因此仍被调试日志观测到。 + */ +export function usePayloadInterceptor(interceptor: PayloadInterceptor): () => void { + if (!interceptor.name) { + throw new Error("PayloadInterceptor requires a non-empty name"); + } + if (typeof interceptor.intercept !== "function") { + throw new Error(`PayloadInterceptor "${interceptor.name}" requires an intercept function`); + } + if (hasInterceptorName(interceptor.name)) { + throw new Error(`Payload interceptor is already registered: ${interceptor.name}`); + } + customInterceptors.push(interceptor); + invalidateComposedChain(); + let disposed = false; + return () => { + if (disposed) return; + disposed = true; + const index = customInterceptors.indexOf(interceptor); + if (index >= 0) { + customInterceptors.splice(index, 1); + invalidateComposedChain(); + } + }; +} + +/** 当前生效的拦截器名字序列(执行顺序),供顺序快照测试与诊断。 */ +export function listPayloadInterceptorNames(): readonly string[] { + return orderedInterceptors().map((entry) => entry.name); +} + +function orderedInterceptors(): readonly PayloadInterceptor[] { + if (defaultInterceptors.length === 0) return [...customInterceptors]; + const head = defaultInterceptors.slice(0, -1); + const tail = defaultInterceptors[defaultInterceptors.length - 1]; + return [...head, ...customInterceptors, tail]; +} + +/** 按执行顺序组合出的单一中间件;结果缓存至下一次注册/移除。 */ +export function composePayloadInterceptorChain(): ProviderPayloadMiddleware { + if (!composedChain) { + const chain = orderedInterceptors(); + composedChain = (options, params) => + chain.reduce((next, entry) => entry.intercept(next, params), options); + } + return composedChain; +} diff --git a/crates/agent-gui/src/lib/providers/service/llmService.ts b/crates/agent-gui/src/lib/providers/service/llmService.ts new file mode 100644 index 000000000..f560ae17f --- /dev/null +++ b/crates/agent-gui/src/lib/providers/service/llmService.ts @@ -0,0 +1,68 @@ +import type { AssistantMessageEventStream } from "@earendil-works/pi-ai"; +import { streamSimpleByApi } from "../runtime/streamByApi"; +import { usePayloadInterceptor } from "./interceptors"; +import type { LlmStreamRequest } from "./types"; + +/** + * dev 构建探测。 + * + * Vite 下 import.meta.env.DEV 为真;生产构建被静态替换为 false。Node 测试 + * 加载器(esbuild CJS 转译)中 import.meta 是空壳,可选链安全落到 false—— + * 测试如需覆盖冻结路径用 setLlmServiceDevModeForTest。 + */ +function detectDevBuild(): boolean { + try { + return Boolean((import.meta as ImportMeta & { env?: { DEV?: boolean } }).env?.DEV); + } catch { + return false; + } +} + +let devModeOverride: boolean | undefined; + +/** 测试专用:强制指定 dev 冻结开关(undefined 恢复自动探测)。 */ +export function setLlmServiceDevModeForTest(value: boolean | undefined): void { + devModeOverride = value; +} + +function isDevBuild(): boolean { + return devModeOverride ?? detectDevBuild(); +} + +/** 已分发过的请求信封——一次性分发不变量的记账。 */ +const dispatchedRequests = new WeakSet(); + +/** + * LLM 统一流式入口。 + * + * 职责被刻意压到最小(PR-1 行为等价不变量): + * 1. 一次性分发——同一请求信封重复分发抛错,杜绝"复用上一轮请求信封"这类 + * 隐性共享(重试/failover 的重放语义在适配器内部与调用方,不经此层); + * 2. dev 冻结——仅 dev 构建把信封冻住,让越过 seam 之后的信封突变当场以 + * TypeError 暴露(ESM 严格模式);生产构建零开销; + * 3. 经 runtime/streamByApi.ts 协议分发针孔路由到注册表适配器。针孔是 seam + * 的公开可观测点(传输 golden 与 failover 测试按该模块路径 mock 截获 + * 全部出站流),不得绕开。 + * + * 传输路由字段(headers 内 x-liveagent-* 等)不透明透传:不读取、不判断、 + * 不缓存。 + */ +export function llmStream(request: LlmStreamRequest): AssistantMessageEventStream { + if (dispatchedRequests.has(request)) { + throw new Error("LlmStreamRequest was already dispatched; build a fresh request per stream"); + } + dispatchedRequests.add(request); + if (isDevBuild()) { + Object.freeze(request); + } + return streamSimpleByApi(request.model, request.context, request.options); +} + +export const llm = { + stream: llmStream, + /** + * 注册自定义 payload 拦截器(PR-3),返回幂等 dispose。执行位置在默认 + * 拦截器之后、payload-debug-logging 链尾之前。 + */ + use: usePayloadInterceptor, +}; diff --git a/crates/agent-gui/src/lib/providers/service/piAiAdapter.ts b/crates/agent-gui/src/lib/providers/service/piAiAdapter.ts new file mode 100644 index 000000000..cc6b5bb97 --- /dev/null +++ b/crates/agent-gui/src/lib/providers/service/piAiAdapter.ts @@ -0,0 +1,189 @@ +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 { resolveMaxTokens } from "../runtime/common"; +import { rejectEmptyOpenAICompletionsResponse } from "../runtime/openAICompletionsStream"; +import { withStreamRetry } from "../runtime/streamRetry"; +import { + clampOpenAIReasoningEffort, + resolveAnthropicThinkingRuntime, + resolveGeminiThinkingRuntime, +} from "../runtime/thinkingLevels"; +import type { StreamOptionsEx, ToolChoice } from "../runtime/types"; +import type { LlmAdapter } from "./types"; + +// ============================================================================ +// pi-ai 四协议适配器。 +// +// 各分支为 streamByApi.ts 原实现的原样搬移(PR-1 行为等价不变量):分支内的 +// withStreamRetry 包装位置、toolChoice 映射、thinking runtime 解析、注释 +// 一并保留,不做任何重写。判定基准是 PR-0 golden 快照零修改通过。 +// ============================================================================ + +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, 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, + }; +} + +function streamAnthropicMessages(model: Model, context: Context, options: StreamOptionsEx) { + // 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 }, + ); +} + +function streamOpenAICompletionsApi(model: Model, context: Context, options: StreamOptionsEx) { + // 严格校验的 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 }, + ); +} + +function streamOpenAIResponsesApi(model: Model, context: Context, options: StreamOptionsEx) { + 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, + }, + ); +} + +function streamGoogleGenerativeAi(model: Model, context: Context, options: StreamOptionsEx) { + 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, + }, + ); +} + +export const piAiAdapter: LlmAdapter = { + apis: [ + "anthropic-messages", + "openai-completions", + "openai-responses", + "google-generative-ai", + ] as const, + stream(model, context, options) { + switch (model.api) { + case "anthropic-messages": + return streamAnthropicMessages(model, context, options); + case "openai-completions": + return streamOpenAICompletionsApi(model, context, options); + case "openai-responses": + return streamOpenAIResponsesApi(model, context, options); + case "google-generative-ai": + return streamGoogleGenerativeAi(model, context, options); + default: + // 注册表按 apis 路由到这里,正常不可达;防御分支保持同一错误文案。 + throw new Error(`Unsupported model API: ${model.api}`); + } + }, +}; diff --git a/crates/agent-gui/src/lib/providers/service/registry.ts b/crates/agent-gui/src/lib/providers/service/registry.ts new file mode 100644 index 000000000..94214bb19 --- /dev/null +++ b/crates/agent-gui/src/lib/providers/service/registry.ts @@ -0,0 +1,38 @@ +import type { LlmAdapter } from "./types"; + +/** + * api → adapter 注册表。 + * + * PR-1 的注册表是模块内静态注册(registerAdapter 仅供本目录的默认装配与 + * 测试使用),不提供运行期动态卸载——那是 PR-3 拦截器注册化的范畴。 + */ +const adaptersByApi = new Map(); + +export function registerAdapter(adapter: LlmAdapter): void { + for (const api of adapter.apis) { + const existing = adaptersByApi.get(api); + if (existing && existing !== adapter) { + throw new Error(`Duplicate LLM adapter registration for API: ${api}`); + } + adaptersByApi.set(api, adapter); + } +} + +/** + * 解析一个 wire 协议的适配器。 + * + * 未注册协议的错误文案与重构前 streamByApi.ts 的 default 分支逐字保持一致 + * ("Unsupported model API: ..."),错误路径也不漂移。 + */ +export function resolveAdapter(api: string): LlmAdapter { + const adapter = adaptersByApi.get(api); + if (!adapter) { + throw new Error(`Unsupported model API: ${api}`); + } + return adapter; +} + +/** 已注册协议列表(测试用,按注册顺序)。 */ +export function registeredApis(): string[] { + return [...adaptersByApi.keys()]; +} diff --git a/crates/agent-gui/src/lib/providers/service/types.ts b/crates/agent-gui/src/lib/providers/service/types.ts new file mode 100644 index 000000000..8c0ca6b50 --- /dev/null +++ b/crates/agent-gui/src/lib/providers/service/types.ts @@ -0,0 +1,38 @@ +import type { Api, AssistantMessageEventStream, Context, Model } from "@earendil-works/pi-ai"; +import type { StreamRetryConfig } from "../runtime/streamRetry"; +import type { StreamOptionsEx } from "../runtime/types"; + +/** + * 一次 LLM 流式请求的完整信封。 + * + * `model.api` 决定路由到哪个适配器;`context` 与 `options` 不做任何解释, + * 原样交给适配器——传输路由字段(headers 里的 x-liveagent-*、useSystemProxy + * 派生头)对 seam 不透明透传,seam 不读取、不判断、不缓存。 + */ +export type LlmStreamRequest = { + model: Model; + context: Context; + options: StreamOptionsEx; +}; + +/** + * LLM 适配器:把一组 wire 协议接到统一分发入口 llm.stream() 上。 + * + * PR-1 只要求 stream()(行为与被包装的原实现逐行等价);resolveModel / + * retryPolicy 是 PR-2(策略归属权反转)预留的可选能力,当前没有实现者, + * 重试策略仍由调用方通过 options.streamRetry 携带。 + */ +export type LlmAdapter = { + /** 本适配器承接的 wire 协议 id 集合(即 model.api 的取值)。 */ + readonly apis: readonly string[]; + /** 发起一次流式请求。必须保持被包装实现的语义,包括流内重试的包装位置。 */ + stream( + model: Model, + context: Context, + options: StreamOptionsEx, + ): AssistantMessageEventStream; + /** PR-2 预留:路由时机的模型解析。 */ + resolveModel?(model: Model): Model; + /** PR-2 预留:供应商级重试策略查询。 */ + retryPolicy?(model: Model): StreamRetryConfig | undefined; +}; diff --git a/crates/agent-gui/src/lib/trajectory/recorder.ts b/crates/agent-gui/src/lib/trajectory/recorder.ts index e77399b4c..ffe9a4148 100644 --- a/crates/agent-gui/src/lib/trajectory/recorder.ts +++ b/crates/agent-gui/src/lib/trajectory/recorder.ts @@ -20,6 +20,7 @@ import type { TrajectoryUsage, } from "@liveagent/ui/lib/trajectory/types"; import { createTrajectoryPersistenceQueue } from "./persistenceQueue"; +import { scrubSecretsFromErrorText } from "./scrub"; /** 工具参数在事件里的截断长度:实时通道要小,详情由正文索引另行提供。 */ const TOOL_ARGS_PREVIEW_CHARS = 200; @@ -72,7 +73,39 @@ export type TrajectoryRecorder = { stepEnd: (step: number, info: TrajectoryStepEndInfo) => void; noteRetry: ( step: number, - info: { attempt: number; maxRetries?: number; delayMs?: number; error?: string }, + info: { + attempt: number; + maxRetries?: number; + delayMs?: number; + error?: string; + /** 候选标签("Provider · model");failover 下区分各候选自己的重试。 */ + provider?: string; + }, + ) => void; + /** 跨供应商切换。`attempt` 是本次请求内的切换序号(1 起)。 */ + noteFailover: ( + step: number, + info: { + attempt: number; + fromLabel?: string; + toLabel?: string; + targetIndex?: number; + error?: string; + }, + ) => void; + /** + * 一次实际尝试的传输装配快照。调用方必须只传头名(不传值)—— + * recorder 不做二次脱敏,线格式里根本没有放头值的字段。 + */ + noteTransport: ( + step: number, + info: { + provider?: string; + upstreamOrigin?: string; + useSystemProxy?: boolean; + fullUrl?: boolean; + headerNames?: readonly string[]; + }, ) => void; toolStart: (step: number, toolCall: { id: string; name: string; arguments?: unknown }) => void; toolEnd: ( @@ -124,6 +157,8 @@ export const NOOP_TRAJECTORY_RECORDER: TrajectoryRecorder = { firstToken: () => {}, stepEnd: () => {}, noteRetry: () => {}, + noteFailover: () => {}, + noteTransport: () => {}, toolStart: () => {}, toolEnd: () => {}, compactionStart: () => {}, @@ -164,6 +199,10 @@ export function createTrajectoryRecorder(params: { }; const queue = createTrajectoryPersistenceQueue({ conversationId, ports, warn }); + /** err 字段的统一入口:供应商报错可能回显带密钥的 URL/头,落盘前洗掉。 */ + const scrubError = (error: string | undefined): string | undefined => + error === undefined ? undefined : scrubSecretsFromErrorText(error); + const scheduleFlush = () => { if (timer !== null || disposed) return; timer = setTimeout(() => { @@ -209,6 +248,7 @@ export function createTrajectoryRecorder(params: { const emitStepEnd = (turn: number, step: number, info: TrajectoryStepEndInfo) => { openSteps.delete(stepKey(turn, step)); + const error = scrubError(info.error); emit({ k: "step_end", t: turn, @@ -220,7 +260,7 @@ export function createTrajectoryRecorder(params: { ...(info.model === undefined ? {} : { m: info.model }), ...(info.api === undefined ? {} : { api: info.api }), ...(info.stopReason === undefined ? {} : { sr: info.stopReason }), - ...(info.error === undefined ? {} : { err: info.error }), + ...(error === undefined ? {} : { err: error }), }); }; @@ -289,6 +329,7 @@ export function createTrajectoryRecorder(params: { emitStepEnd(currentTurn, step, info); }, noteRetry: (step, info) => { + const error = scrubError(info.error); emit({ k: "retry", t: currentTurn, @@ -297,7 +338,37 @@ export function createTrajectoryRecorder(params: { n: info.attempt, ...(info.maxRetries === undefined ? {} : { max: info.maxRetries }), ...(info.delayMs === undefined ? {} : { delay: info.delayMs }), - ...(info.error === undefined ? {} : { err: info.error }), + ...(error === undefined ? {} : { err: error }), + ...(info.provider === undefined ? {} : { p: info.provider }), + }); + }, + noteFailover: (step, info) => { + const error = scrubError(info.error); + emit({ + k: "failover", + t: currentTurn, + s: step, + at: Date.now(), + n: info.attempt, + ...(info.fromLabel === undefined ? {} : { from: info.fromLabel }), + ...(info.toLabel === undefined ? {} : { to: info.toLabel }), + ...(info.targetIndex === undefined ? {} : { ti: info.targetIndex }), + ...(error === undefined ? {} : { err: error }), + }); + }, + noteTransport: (step, info) => { + emit({ + k: "transport", + t: currentTurn, + s: step, + at: Date.now(), + ...(info.provider === undefined ? {} : { p: info.provider }), + ...(info.upstreamOrigin === undefined ? {} : { o: info.upstreamOrigin }), + ...(info.useSystemProxy === undefined ? {} : { sp: info.useSystemProxy }), + ...(info.fullUrl === undefined ? {} : { fu: info.fullUrl }), + ...(info.headerNames === undefined || info.headerNames.length === 0 + ? {} + : { hn: [...info.headerNames] }), }); }, toolStart: (step, toolCall) => { @@ -336,6 +407,7 @@ export function createTrajectoryRecorder(params: { }); }, compactionEnd: (info) => { + const error = scrubError(info.error); emit({ k: "compaction_end", t: info.standalone === true ? null : currentTurn, @@ -343,12 +415,13 @@ export function createTrajectoryRecorder(params: { st: info.status, ...(info.tokensBefore === undefined ? {} : { before: info.tokensBefore }), ...(info.tokensAfter === undefined ? {} : { after: info.tokensAfter }), - ...(info.error === undefined ? {} : { err: info.error }), + ...(error === undefined ? {} : { err: error }), }); }, endTurn: (info) => { if (!turnOpen) return; turnOpen = false; + const error = scrubError(info.error); const unfinishedSteps = [...openSteps] .map((key) => { const [turnText, stepText] = key.split(" "); @@ -373,7 +446,7 @@ export function createTrajectoryRecorder(params: { t: currentTurn, at: Date.now(), st: info.status, - ...(info.error === undefined ? {} : { err: info.error }), + ...(error === undefined ? {} : { err: error }), }); }, flush, diff --git a/crates/agent-gui/src/lib/trajectory/scrub.ts b/crates/agent-gui/src/lib/trajectory/scrub.ts new file mode 100644 index 000000000..624300eee --- /dev/null +++ b/crates/agent-gui/src/lib/trajectory/scrub.ts @@ -0,0 +1,24 @@ +/** + * 轨迹事件错误文本的密钥洗涤。 + * + * 供应商报错可能回显完整请求 URL(Gemini 的 key 走 query 参数)或鉴权头 + * (Bearer token)。轨迹账本会落盘并跨端下发,任何进入 err 字段的文本都 + * 必须先过这一层。只做模式级替换,不改动正常报错文本。 + */ + +/** 取值形似密钥的 query 参数名(含 URL 编码变体场景由参数名匹配兜底)。 */ +const SENSITIVE_QUERY_PARAM_PATTERN = + /([?&](?:key|api[-_]?key|apikey|token|access[-_]?token|secret)=)[^&\s"']+/gi; + +/** Authorization: Bearer 回显。 */ +const BEARER_TOKEN_PATTERN = /(bearer\s+)[a-z0-9._~+/-]{8,}=*/gi; + +/** 常见密钥前缀(OpenAI/Anthropic sk-、Google AIza)。 */ +const KNOWN_KEY_SHAPE_PATTERN = /\b(?:sk|AIza)[A-Za-z0-9_-]{16,}\b/g; + +export function scrubSecretsFromErrorText(text: string): string { + return text + .replace(SENSITIVE_QUERY_PARAM_PATTERN, "$1[redacted]") + .replace(BEARER_TOKEN_PATTERN, "$1[redacted]") + .replace(KNOWN_KEY_SHAPE_PATTERN, "[redacted]"); +} diff --git a/crates/agent-gui/src/pages/chat/turns/runAgentConversationTurn.ts b/crates/agent-gui/src/pages/chat/turns/runAgentConversationTurn.ts index c691d5e00..add098a29 100644 --- a/crates/agent-gui/src/pages/chat/turns/runAgentConversationTurn.ts +++ b/crates/agent-gui/src/pages/chat/turns/runAgentConversationTurn.ts @@ -1307,11 +1307,33 @@ export async function runAgentConversationTurn(params: RunAgentConversationTurnP attempt: latest.attempt, // maxAttempts 含首次尝试,重试上限要减去它。 maxRetries: Math.max(0, latest.maxAttempts - 1), + ...(latest.plannedDelayMs === undefined ? {} : { delayMs: latest.plannedDelayMs }), ...(latest.errorMessage === "" ? {} : { error: latest.errorMessage }), + ...(latest.providerLabel === undefined ? {} : { provider: latest.providerLabel }), }); } updateRetryAttempts(attempts, transcriptStore); }, + onFailoverAttempt: (_round, event) => { + trajectory.noteFailover(activeAgentRound, { + attempt: event.attempt, + fromLabel: event.fromLabel, + toLabel: event.toLabel, + targetIndex: event.targetIndex, + ...(event.errorMessage === "" ? {} : { error: event.errorMessage }), + }); + }, + onTransportAttempt: (_round, snapshot) => { + trajectory.noteTransport(activeAgentRound, { + provider: snapshot.providerLabel, + ...(snapshot.upstreamOrigin === undefined + ? {} + : { upstreamOrigin: snapshot.upstreamOrigin }), + useSystemProxy: snapshot.useSystemProxy, + fullUrl: snapshot.fullUrl, + headerNames: snapshot.headerNames, + }); + }, onBeforeNextTurn: async ({ round, assistant, toolResults, emittedMessages }) => { publishPersistableAgentProgress(round, assistant, toolResults); latestAgentEmittedMessages = emittedMessages.slice(); diff --git a/crates/agent-gui/src/pages/chat/turns/runTextConversationTurn.ts b/crates/agent-gui/src/pages/chat/turns/runTextConversationTurn.ts index 9d49f1bca..e90cb8288 100644 --- a/crates/agent-gui/src/pages/chat/turns/runTextConversationTurn.ts +++ b/crates/agent-gui/src/pages/chat/turns/runTextConversationTurn.ts @@ -376,11 +376,13 @@ export async function runTextConversationTurn(params: RunTextConversationTurnPar onSwitched: ({ target, errorMessage }) => { failover.onSwitched?.({ target, round: textRound, errorMessage }); }, - onFailover: ({ fromLabel, toLabel, errorMessage }) => { + onFailover: ({ fromLabel, toLabel, targetIndex, errorMessage }) => { trajectoryFailoverAttempt += 1; - trajectory.noteRetry(textRound, { + trajectory.noteFailover(textRound, { attempt: trajectoryFailoverAttempt, - maxRetries: failover.fallbacks.length, + fromLabel, + toLabel, + targetIndex, ...(errorMessage === "" ? {} : { error: errorMessage }), }); failoverStatusVisible = true; @@ -394,6 +396,17 @@ export async function runTextConversationTurn(params: RunTextConversationTurnPar workdir: conversationCwd, sessionId, nativeWebSearch: nativeWebSearchEnabled, + onTransportAttempt: (snapshot) => { + trajectory.noteTransport(textRound, { + provider: snapshot.providerLabel, + ...(snapshot.upstreamOrigin === undefined + ? {} + : { upstreamOrigin: snapshot.upstreamOrigin }), + useSystemProxy: snapshot.useSystemProxy, + fullUrl: snapshot.fullUrl, + headerNames: snapshot.headerNames, + }); + }, onRequestStart: ({ context, systemSuffix }) => { recordTextRequestStart(context, systemSuffix); }, @@ -438,14 +451,20 @@ export async function runTextConversationTurn(params: RunTextConversationTurnPar }, signal: scope.controller.signal, debugLogger: streamAttempt === 0 ? conversationDebugLogger : recoveryDebugLogger, - onRetryStatus: (attempt, maxAttempts, errorMessage) => { + onRetryStatus: (attempt, maxAttempts, errorMessage, plannedDelayMs) => { trajectory.noteRetry(textRound, { attempt, maxRetries: Math.max(0, maxAttempts - 1), + ...(plannedDelayMs === undefined ? {} : { delayMs: plannedDelayMs }), ...(errorMessage === "" ? {} : { error: errorMessage }), }); updateGatewayBridgeToolStatus(`连接已断开,正在重试 (${attempt}/${maxAttempts})...`); - retryAttemptsForAttempt.push({ attempt, maxAttempts, errorMessage }); + retryAttemptsForAttempt.push({ + attempt, + maxAttempts, + errorMessage, + ...(plannedDelayMs === undefined ? {} : { plannedDelayMs }), + }); updateRetryAttempts(retryAttemptsForAttempt.slice(), transcriptStore); }, onRetryRecovered: () => { diff --git a/crates/agent-gui/test/chat/agent-runner.test.mjs b/crates/agent-gui/test/chat/agent-runner.test.mjs index 1e1d0e203..fae61a5f4 100644 --- a/crates/agent-gui/test/chat/agent-runner.test.mjs +++ b/crates/agent-gui/test/chat/agent-runner.test.mjs @@ -374,6 +374,13 @@ const llmMock = { }, }; }, + // runner 经统一入口 llm.stream() 出站;mock 同构转发到上面的 + // streamSimpleByApi,两个入口共享同一份请求记账。 + llm: { + stream(request) { + return llmMock.streamSimpleByApi(request.model, request.context, request.options); + }, + }, }; const loader = createTsModuleLoader({ diff --git a/crates/agent-gui/test/providers/llm-interceptors.test.mjs b/crates/agent-gui/test/providers/llm-interceptors.test.mjs new file mode 100644 index 000000000..0ee38920f --- /dev/null +++ b/crates/agent-gui/test/providers/llm-interceptors.test.mjs @@ -0,0 +1,278 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { createTsModuleLoader } from "../helpers/load-ts-module.mjs"; + +/** + * PR-3 payload 拦截器注册化(feat-llm-interceptors)单元测试: + * + * 1. 默认注册顺序快照——10 个具名拦截器逐一断言,顺序即协议正确性的 + * 一部分,任何重排都必须显式改这里; + * 2. 自定义拦截器语义——params 可见、options 可变换、执行位置在默认 + * 拦截器之后且在 payload-debug-logging 链尾之前; + * 3. dispose 移除且幂等、同名重复注册抛错; + * 4. 行为等价——空注册态下 finalizeProviderStreamOptions 的输出与 + * 直接用旧数组顺序 compose 的结果逐字段一致(配合 golden 两套件 + * 零修改通过构成 PR-3 的等价证据)。 + */ + +const loader = createTsModuleLoader(); +// 先加载 payloadPipeline(安装默认拦截器),再取注册表 API。 +const { + finalizeProviderStreamOptions, + composePayloadMiddlewares, + attachPayloadDebugLogging, +} = loader.loadModule("src/lib/providers/runtime/payloadPipeline.ts"); +const { listPayloadInterceptorNames, usePayloadInterceptor } = loader.loadModule( + "src/lib/providers/service/interceptors.ts", +); +const { llm } = loader.loadModule("src/lib/providers/service/llmService.ts"); + +const EXPECTED_DEFAULT_ORDER = [ + "anthropic-automatic-caching", + "anthropic-long-context-beta", + "codex-responses-storage", + "codex-prompt-cache-hint", + "provider-native-web-search", + "xai-responses-payload-compat", + "deepseek-responses-payload-compat", + "native-attachments", + "gemini-thought-signature-guard", + "payload-debug-logging", +]; + +function baseParams(overrides = {}) { + return { + providerId: "claude_code", + baseUrl: "https://relay.example/v1", + options: {}, + ...overrides, + }; +} + +test("默认注册顺序快照:10 个具名拦截器逐一一致", () => { + assert.deepEqual(listPayloadInterceptorNames(), EXPECTED_DEFAULT_ORDER); +}); + +test("llm.use 暴露注册入口且与 usePayloadInterceptor 同源", () => { + assert.equal(llm.use, usePayloadInterceptor); +}); + +test("自定义拦截器:params 可见、options 可变换、dispose 后恢复原样", () => { + const seen = []; + const dispose = usePayloadInterceptor({ + name: "test-marker", + intercept: (options, params) => { + seen.push(params.providerId); + return { ...options, headers: { ...(options.headers ?? {}), "x-test-marker": "1" } }; + }, + }); + try { + const withMarker = finalizeProviderStreamOptions(baseParams()); + assert.equal(withMarker.headers["x-test-marker"], "1"); + assert.deepEqual(seen, ["claude_code"]); + } finally { + dispose(); + } + const withoutMarker = finalizeProviderStreamOptions(baseParams()); + assert.equal(withoutMarker.headers?.["x-test-marker"], undefined); +}); + +test("自定义拦截器插入默认之后、payload-debug-logging 链尾之前", () => { + const dispose = usePayloadInterceptor({ + name: "test-order", + intercept: (options) => options, + }); + try { + const names = listPayloadInterceptorNames(); + assert.equal(names[names.length - 1], "payload-debug-logging"); + assert.equal(names[names.length - 2], "test-order"); + assert.deepEqual( + names.slice(0, EXPECTED_DEFAULT_ORDER.length - 1), + EXPECTED_DEFAULT_ORDER.slice(0, -1), + ); + } finally { + dispose(); + } + assert.deepEqual(listPayloadInterceptorNames(), EXPECTED_DEFAULT_ORDER); +}); + +test("链尾观测不变量:自定义追加的 onPayload 变换仍被 debug logging 看到", async () => { + const dispose = usePayloadInterceptor({ + name: "test-payload-mutator", + intercept: (options) => ({ + ...options, + onPayload: async (payload) => ({ ...payload, injected: true }), + }), + }); + const logged = []; + try { + const options = finalizeProviderStreamOptions( + baseParams({ + debugLogger: { logRequest: (entry) => logged.push(entry) }, + }), + ); + const result = await options.onPayload({ base: true }, { api: "anthropic-messages", provider: "anthropic" }); + assert.deepEqual(result, { base: true, injected: true }); + assert.equal(logged.length, 1); + assert.deepEqual(logged[0].payload, { base: true, injected: true }); + } finally { + dispose(); + } +}); + +test("dispose 幂等:重复调用不影响其他注册", () => { + const disposeA = usePayloadInterceptor({ name: "test-a", intercept: (o) => o }); + const disposeB = usePayloadInterceptor({ name: "test-b", intercept: (o) => o }); + disposeA(); + disposeA(); + const names = listPayloadInterceptorNames(); + assert.ok(!names.includes("test-a")); + assert.ok(names.includes("test-b")); + disposeB(); + assert.deepEqual(listPayloadInterceptorNames(), EXPECTED_DEFAULT_ORDER); +}); + +test("同名重复注册抛错(含与默认拦截器同名)", () => { + const dispose = usePayloadInterceptor({ name: "test-dup", intercept: (o) => o }); + try { + assert.throws( + () => usePayloadInterceptor({ name: "test-dup", intercept: (o) => o }), + /already registered: test-dup/, + ); + } finally { + dispose(); + } + assert.throws( + () => usePayloadInterceptor({ name: "anthropic-automatic-caching", intercept: (o) => o }), + /already registered: anthropic-automatic-caching/, + ); + assert.throws(() => usePayloadInterceptor({ name: "", intercept: (o) => o }), /non-empty name/); + assert.throws( + () => usePayloadInterceptor({ name: "test-no-fn" }), + /requires an intercept function/, + ); +}); + +test("行为等价:空注册态 finalize 输出与旧数组组合逐字段一致", async () => { + // 按注册化前的 finalizePayloadMiddlewares 数组原样重建旧组合链 + // (同一批 attach* 实现、同一顺序),对非平凡参数逐字段对比输出。 + const { attachAnthropicAutomaticCaching } = loader.loadModule( + "src/lib/providers/runtime/anthropicCache.ts", + ); + const { attachAnthropicLongContextBeta } = loader.loadModule( + "src/lib/providers/runtime/anthropicLongContext.ts", + ); + const { attachCodexResponsesStorage } = loader.loadModule( + "src/lib/providers/runtime/codexStorage.ts", + ); + const { attachCodexPromptCacheHint } = loader.loadModule( + "src/lib/providers/runtime/codexPromptCache.ts", + ); + const { attachProviderNativeWebSearch } = loader.loadModule( + "src/lib/providers/runtime/nativeSearchPayload.ts", + ); + const { attachXaiResponsesPayloadCompat } = loader.loadModule( + "src/lib/providers/runtime/xaiResponsesPayload.ts", + ); + const { attachDeepSeekResponsesPayloadCompat } = loader.loadModule( + "src/lib/providers/runtime/deepSeekResponsesPayload.ts", + ); + const attachments = loader.loadModule("src/lib/providers/nativeResponsesAttachments.ts"); + const { attachGeminiThoughtSignatureGuard } = loader.loadModule( + "src/lib/providers/runtime/geminiToolPayload.ts", + ); + + const legacyChain = 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 = attachments.attachOpenAIResponsesNativeAttachments(options, { + context: params.context, + model: params.model, + providerId: params.providerId, + workdir: params.workdir, + }); + nextOptions = attachments.attachOpenAICompletionsNativeAttachments(nextOptions, { + context: params.context, + model: params.model, + providerId: params.providerId, + workdir: params.workdir, + }); + nextOptions = attachments.attachAnthropicMessagesNativeAttachments(nextOptions, { + context: params.context, + model: params.model, + providerId: params.providerId, + workdir: params.workdir, + }); + return attachments.attachGeminiGenerativeAINativeAttachments(nextOptions, { + context: params.context, + model: params.model, + providerId: params.providerId, + workdir: params.workdir, + }); + }, + (options, params) => + attachGeminiThoughtSignatureGuard(options, { + providerId: params.providerId, + baseUrl: params.baseUrl, + }), + (options, params) => attachPayloadDebugLogging(options, params.debugLogger, params.extra), + ]); + + // 覆盖多形态参数:anthropic 缓存路径、codex cache hint 路径、debug 链尾。 + const paramMatrix = [ + baseParams(), + baseParams({ providerId: "claude_code", options: { headers: { "x-a": "1" } } }), + baseParams({ + providerId: "codex", + baseUrl: "https://chatgpt.com/backend-api/codex", + promptCacheHintMode: "auto", + }), + baseParams({ providerId: "gemini", baseUrl: "https://generativelanguage.googleapis.com" }), + ]; + for (const params of paramMatrix) { + const viaRegistry = finalizeProviderStreamOptions(params); + const viaLegacy = legacyChain(params.options, params); + // onPayload 是闭包无法 deepEqual;先断言存在性一致,再剥离比较其余字段。 + assert.equal( + typeof viaRegistry.onPayload, + typeof viaLegacy.onPayload, + `onPayload presence must match for ${params.providerId}`, + ); + const { onPayload: _a, ...restRegistry } = viaRegistry; + const { onPayload: _b, ...restLegacy } = viaLegacy; + assert.deepEqual(restRegistry, restLegacy, `options must match for ${params.providerId}`); + } +}); diff --git a/crates/agent-gui/test/providers/llm-service-seam.test.mjs b/crates/agent-gui/test/providers/llm-service-seam.test.mjs new file mode 100644 index 000000000..0b46e0f2f --- /dev/null +++ b/crates/agent-gui/test/providers/llm-service-seam.test.mjs @@ -0,0 +1,263 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { createTsModuleLoader } from "../helpers/load-ts-module.mjs"; + +// ============================================================================ +// PR-1 seam 骨架单元测试:注册表分发、未知协议错误文案等价、一次性分发、 +// dev 冻结/生产不冻结、兼容壳与统一入口 llm.stream() 的 wire payload 等价。 +// +// 行为等价的总判定基准是 PR-0 golden 两套件零修改通过(见 +// wire-payload-golden.test.mjs / transport-golden.test.mjs);本文件补充 +// seam 自身的新契约。 +// ============================================================================ + +const realAnthropic = await import( + new URL( + "../../node_modules/@earendil-works/pi-ai/dist/api/anthropic-messages.js", + import.meta.url, + ).href +); +const realCompletions = await import( + new URL( + "../../node_modules/@earendil-works/pi-ai/dist/api/openai-completions.js", + import.meta.url, + ).href +); +const realResponses = await import( + new URL( + "../../node_modules/@earendil-works/pi-ai/dist/api/openai-responses.js", + import.meta.url, + ).href +); +const realGoogle = await import( + new URL( + "../../node_modules/@earendil-works/pi-ai/dist/api/google-generative-ai.js", + import.meta.url, + ).href +); + +const loader = createTsModuleLoader({ + mocks: { + "@earendil-works/pi-ai/api/anthropic-messages": { stream: realAnthropic.stream }, + "@earendil-works/pi-ai/api/openai-completions": { stream: realCompletions.stream }, + "@earendil-works/pi-ai/api/openai-responses": { stream: realResponses.stream }, + "@earendil-works/pi-ai/api/google-generative-ai": { stream: realGoogle.stream }, + }, +}); + +const { streamSimpleByApi } = loader.loadModule("src/lib/providers/runtime/streamByApi.ts"); +const { llm, llmStream, setLlmServiceDevModeForTest } = loader.loadModule( + "src/lib/providers/service/llmService.ts", +); +const { registeredApis, resolveAdapter, registerAdapter } = loader.loadModule( + "src/lib/providers/service/registry.ts", +); +const { piAiAdapter } = loader.loadModule("src/lib/providers/service/piAiAdapter.ts"); +const { deepSeekAdapter } = loader.loadModule("src/lib/providers/service/deepSeekAdapter.ts"); +const { DEEPSEEK_RESPONSES_API } = loader.loadModule("src/lib/providers/deepSeekNative.ts"); + +function buildModel(api, overrides = {}) { + return { + id: "test-model", + provider: "openai", + api, + baseUrl: "https://example.com/v1", + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 128000, + maxTokens: 4096, + ...overrides, + }; +} + +function buildContext() { + return { + systemPrompt: "You are a precise assistant.", + messages: [{ role: "user", content: "hello world", timestamp: 1 }], + }; +} + +/** 在 onPayload 截获 wire payload 后中断请求(同 golden 的捕获通道)。 */ +async function captureViaEntry(entry, model, context, options = {}) { + let captured; + const stream = entry(model, context, { + apiKey: "sk-test", + ...options, + onPayload: async (payload) => { + captured = payload; + throw new Error("__capture_stop__"); + }, + }); + try { + await stream.result(); + } catch { + // onPayload 抛错中断请求属预期。 + } + assert.ok(captured, "expected wire payload capture"); + return JSON.parse(JSON.stringify(captured)); +} + +test.afterEach(() => { + setLlmServiceDevModeForTest(undefined); +}); + +// --------------------------------------------------------------------------- +// 注册表分发 +// --------------------------------------------------------------------------- + +test("seam/registry: 五协议各归其所(4×pi-ai + deepseek 原生)", () => { + // 触发默认装配(llmService 模块加载即注册,此处显式断言注册表内容)。 + assert.deepEqual(registeredApis().sort(), [ + "anthropic-messages", + DEEPSEEK_RESPONSES_API, + "google-generative-ai", + "openai-completions", + "openai-responses", + ].sort()); + + for (const api of [ + "anthropic-messages", + "openai-completions", + "openai-responses", + "google-generative-ai", + ]) { + assert.equal(resolveAdapter(api), piAiAdapter, `${api} should route to piAiAdapter`); + } + assert.equal(resolveAdapter(DEEPSEEK_RESPONSES_API), deepSeekAdapter); +}); + +test("seam/registry: 未注册协议错误文案与重构前逐字一致", () => { + assert.throws(() => resolveAdapter("mock-api"), /^Error: Unsupported model API: mock-api$/); + // 经兼容壳走同一路径、同一文案。 + assert.throws( + () => streamSimpleByApi(buildModel("mock-api"), buildContext(), { apiKey: "k" }), + /^Error: Unsupported model API: mock-api$/, + ); +}); + +test("seam/registry: 同一协议重复注册不同适配器立即抛错", () => { + const rogue = { apis: ["anthropic-messages"], stream: () => {} }; + assert.throws( + () => registerAdapter(rogue), + /Duplicate LLM adapter registration for API: anthropic-messages/, + ); + // 同一适配器重复注册幂等(默认装配的 ensure 语义依赖它)。 + registerAdapter(piAiAdapter); +}); + +// --------------------------------------------------------------------------- +// llm.stream() 信封语义 +// --------------------------------------------------------------------------- + +test("seam/llm.stream: 同一请求信封二次分发抛错(一次性分发)", async () => { + const request = { + model: buildModel("openai-completions"), + context: buildContext(), + options: { + apiKey: "sk-test", + onPayload: async () => { + throw new Error("__capture_stop__"); + }, + }, + }; + const first = llm.stream(request); + try { + await first.result(); + } catch { + // 中断属预期。 + } + assert.throws(() => llm.stream(request), /already dispatched/); +}); + +test("seam/llm.stream: dev 构建冻结请求信封,生产构建不冻结", async () => { + setLlmServiceDevModeForTest(true); + const devRequest = { + model: buildModel("openai-completions"), + context: buildContext(), + options: { + apiKey: "sk-test", + onPayload: async () => { + throw new Error("__capture_stop__"); + }, + }, + }; + const devStream = llm.stream(devRequest); + try { + await devStream.result(); + } catch { + // 中断属预期。 + } + assert.ok(Object.isFrozen(devRequest), "dev build must freeze the request envelope"); + + setLlmServiceDevModeForTest(false); + const prodRequest = { + model: buildModel("openai-completions"), + context: buildContext(), + options: { + apiKey: "sk-test", + onPayload: async () => { + throw new Error("__capture_stop__"); + }, + }, + }; + const prodStream = llm.stream(prodRequest); + try { + await prodStream.result(); + } catch { + // 中断属预期。 + } + assert.equal(Object.isFrozen(prodRequest), false, "prod build must not freeze"); +}); + +test("seam/llm.stream: 测试加载器环境自动探测落到不冻结(import.meta 空壳)", async () => { + // 不设 override:detectDevBuild 在 esbuild CJS 转译下 import.meta.env 不存在。 + const request = { + model: buildModel("openai-completions"), + context: buildContext(), + options: { + apiKey: "sk-test", + onPayload: async () => { + throw new Error("__capture_stop__"); + }, + }, + }; + const stream = llmStream(request); + try { + await stream.result(); + } catch { + // 中断属预期。 + } + assert.equal(Object.isFrozen(request), false); +}); + +// --------------------------------------------------------------------------- +// 兼容壳与统一入口等价 +// --------------------------------------------------------------------------- + +test("seam/equivalence: 兼容壳与 llm.stream() 产出同一 wire payload", async () => { + const context = buildContext(); + const viaShim = await captureViaEntry( + streamSimpleByApi, + buildModel("openai-completions"), + context, + ); + const viaService = await captureViaEntry( + (model, ctx, options) => llm.stream({ model, context: ctx, options }), + buildModel("openai-completions"), + context, + ); + assert.deepEqual(viaService, viaShim); +}); + +test("seam/equivalence: deepseek 原生协议经两个入口同样等价", async () => { + const context = buildContext(); + const model = buildModel(DEEPSEEK_RESPONSES_API, { provider: "deepseek" }); + const viaShim = await captureViaEntry(streamSimpleByApi, model, context); + const viaService = await captureViaEntry( + (m, ctx, options) => llm.stream({ model: m, context: ctx, options }), + model, + context, + ); + assert.deepEqual(viaService, viaShim); +}); diff --git a/crates/agent-gui/test/providers/provider-retry-policy.test.mjs b/crates/agent-gui/test/providers/provider-retry-policy.test.mjs new file mode 100644 index 000000000..f1a836914 --- /dev/null +++ b/crates/agent-gui/test/providers/provider-retry-policy.test.mjs @@ -0,0 +1,232 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { createTsModuleLoader } from "../helpers/load-ts-module.mjs"; + +/** + * PR-2 供应商级重试策略(feat-llm-retry-policy)单元测试: + * + * 1. normalizeProviderRetryPolicy 归一化矩阵——非法/缺省一律落 default + * (不落字段),custom 的 maxRetries(不含首次请求)钳位 1..10,旧配置 + * 零迁移; + * 2. createProviderRuntimeConfig 唯一构造点透传 retryPolicy; + * 3. resolveStreamRetryConfig 三种 mode 的消费方合并语义——缺省时不带 + * maxAttempts/disabled(等价于反转前的全局默认行为),custom 时把用户 + * 口径的重试次数换算为 withStreamRetry 的总尝试数(+1); + * 4. failover 逐候选策略独立:每个候选按各自 runtime 解析出不同的 + * streamRetry 配置; + * 5. UI 展示镜像常量与 streamRetry.ts 运行时真源一致(重试数 = 总尝试数-1)。 + */ + +const loader = createTsModuleLoader(); +const settings = loader.loadModule("src/lib/settings/index.ts"); +const { normalizeProviderRetryPolicy, normalizeCustomProvider } = settings; +const { createProviderRuntimeConfig } = loader.loadModule( + "src/lib/providers/runtime/providerRuntimeConfig.ts", +); +const { resolveStreamRetryConfig } = loader.loadModule("src/lib/providers/runtime/retryPolicy.ts"); +const { DEFAULT_STREAM_RETRY_MAX_ATTEMPTS } = loader.loadModule( + "src/lib/providers/runtime/streamRetry.ts", +); + +// --------------------------------------------------------------------------- +// 1. 归一化矩阵 +// --------------------------------------------------------------------------- + +test("normalizeProviderRetryPolicy: default/非法输入一律返回 undefined(不落字段)", () => { + for (const input of [ + undefined, + null, + {}, + "off", + 42, + { mode: "default" }, + { mode: "always" }, + { mode: "custom" }, + { mode: "custom", maxRetries: "3" }, + { mode: "custom", maxRetries: Number.NaN }, + { mode: "custom", maxRetries: Number.POSITIVE_INFINITY }, + ]) { + assert.equal( + normalizeProviderRetryPolicy(input), + undefined, + `input ${JSON.stringify(input)} must normalize to undefined`, + ); + } +}); + +test("normalizeProviderRetryPolicy: off 与 custom 的合法形态", () => { + assert.deepEqual(normalizeProviderRetryPolicy({ mode: "off" }), { mode: "off" }); + assert.deepEqual(normalizeProviderRetryPolicy({ mode: "off", maxRetries: 5 }), { mode: "off" }); + assert.deepEqual(normalizeProviderRetryPolicy({ mode: "custom", maxRetries: 3 }), { + mode: "custom", + maxRetries: 3, + }); +}); + +test("normalizeProviderRetryPolicy: custom maxRetries 钳位 1..10 且取整", () => { + assert.deepEqual(normalizeProviderRetryPolicy({ mode: "custom", maxRetries: 0 }), { + mode: "custom", + maxRetries: 1, + }); + assert.deepEqual(normalizeProviderRetryPolicy({ mode: "custom", maxRetries: -5 }), { + mode: "custom", + maxRetries: 1, + }); + assert.deepEqual(normalizeProviderRetryPolicy({ mode: "custom", maxRetries: 99 }), { + mode: "custom", + maxRetries: 10, + }); + assert.deepEqual(normalizeProviderRetryPolicy({ mode: "custom", maxRetries: 2.6 }), { + mode: "custom", + maxRetries: 3, + }); +}); + +test("normalizeCustomProvider: 旧配置(无 retryPolicy)零迁移——归一化结果不含该字段", () => { + const provider = normalizeCustomProvider({ + id: "legacy-1", + name: "Legacy", + type: "claude_code", + baseUrl: "https://relay.example/v1", + apiKey: "k", + models: [], + activeModels: [], + }); + assert.ok(!("retryPolicy" in provider), "legacy provider must not gain a retryPolicy field"); +}); + +test("normalizeCustomProvider: 配置了 retryPolicy 时原样保留", () => { + const provider = normalizeCustomProvider({ + id: "p-1", + name: "P", + type: "claude_code", + baseUrl: "https://relay.example/v1", + apiKey: "k", + models: [], + activeModels: [], + retryPolicy: { mode: "custom", maxRetries: 2 }, + }); + assert.deepEqual(provider.retryPolicy, { mode: "custom", maxRetries: 2 }); + + const offProvider = normalizeCustomProvider({ + id: "p-2", + name: "P2", + type: "claude_code", + baseUrl: "https://relay.example/v1", + apiKey: "k", + models: [], + activeModels: [], + retryPolicy: { mode: "off" }, + }); + assert.deepEqual(offProvider.retryPolicy, { mode: "off" }); +}); + +// --------------------------------------------------------------------------- +// 2. 唯一构造点透传 +// --------------------------------------------------------------------------- + +function createProvider(overrides = {}) { + return { + id: "provider-1", + name: "Relay", + type: "claude_code", + baseUrl: "https://relay.example/v1", + isFullUrl: true, + apiKey: "test-key", + models: [], + activeModels: [], + promptCachingEnabled: true, + useSystemProxy: false, + ...overrides, + }; +} + +test("createProviderRuntimeConfig: retryPolicy 经唯一构造点透传", () => { + const runtime = createProviderRuntimeConfig( + createProvider({ retryPolicy: { mode: "custom", maxRetries: 2 } }), + "claude-sonnet-4-6", + settings.DEFAULT_CHAT_RUNTIME_CONTROLS, + ); + assert.deepEqual(runtime.retryPolicy, { mode: "custom", maxRetries: 2 }); +}); + +test("createProviderRuntimeConfig: 未配置 retryPolicy 时 runtime 不含该字段", () => { + const runtime = createProviderRuntimeConfig( + createProvider(), + "claude-sonnet-4-6", + settings.DEFAULT_CHAT_RUNTIME_CONTROLS, + ); + assert.ok(!("retryPolicy" in runtime), "unset policy must not appear on the runtime config"); +}); + +// --------------------------------------------------------------------------- +// 3. 消费方合并语义 +// --------------------------------------------------------------------------- + +test("resolveStreamRetryConfig: default(缺省)返回空对象——withStreamRetry 落全局默认", () => { + assert.deepEqual(resolveStreamRetryConfig(undefined), {}); +}); + +test("resolveStreamRetryConfig: off 返回 disabled:true", () => { + assert.deepEqual(resolveStreamRetryConfig({ mode: "off" }), { disabled: true }); +}); + +test("resolveStreamRetryConfig: custom 把重试次数换算为总尝试数(maxRetries+1)", () => { + assert.deepEqual(resolveStreamRetryConfig({ mode: "custom", maxRetries: 2 }), { + maxAttempts: 3, + }); + assert.deepEqual(resolveStreamRetryConfig({ mode: "custom", maxRetries: 1 }), { + maxAttempts: 2, + }); +}); + +test("resolveStreamRetryConfig: 与消费方回调展开合并后互不覆盖", () => { + const onRetry = () => {}; + const onRetryRecovered = () => {}; + const merged = { + ...resolveStreamRetryConfig({ mode: "custom", maxRetries: 4 }), + onRetry, + onRetryRecovered, + }; + assert.equal(merged.maxAttempts, 5); + assert.equal(merged.onRetry, onRetry); + assert.equal(merged.onRetryRecovered, onRetryRecovered); + assert.ok(!("disabled" in merged)); + + const mergedDefault = { ...resolveStreamRetryConfig(undefined), onRetry, onRetryRecovered }; + assert.deepEqual(Object.keys(mergedDefault).sort(), ["onRetry", "onRetryRecovered"]); +}); + +// --------------------------------------------------------------------------- +// 4. failover 逐候选策略独立 +// --------------------------------------------------------------------------- + +test("failover 候选按各自 runtime 解析出独立的重试配置", () => { + const primary = createProviderRuntimeConfig( + createProvider({ id: "primary", retryPolicy: { mode: "custom", maxRetries: 2 } }), + "claude-sonnet-4-6", + settings.DEFAULT_CHAT_RUNTIME_CONTROLS, + ); + const fallbackOff = createProviderRuntimeConfig( + createProvider({ id: "fallback-off", retryPolicy: { mode: "off" } }), + "claude-sonnet-4-6", + settings.DEFAULT_CHAT_RUNTIME_CONTROLS, + ); + const fallbackDefault = createProviderRuntimeConfig( + createProvider({ id: "fallback-default" }), + "claude-sonnet-4-6", + settings.DEFAULT_CHAT_RUNTIME_CONTROLS, + ); + + assert.deepEqual(resolveStreamRetryConfig(primary.retryPolicy), { maxAttempts: 3 }); + assert.deepEqual(resolveStreamRetryConfig(fallbackOff.retryPolicy), { disabled: true }); + assert.deepEqual(resolveStreamRetryConfig(fallbackDefault.retryPolicy), {}); +}); + +// --------------------------------------------------------------------------- +// 5. UI 展示镜像常量与运行时真源一致 +// --------------------------------------------------------------------------- + +test("PROVIDER_RETRY_DEFAULT_MAX_RETRIES 与 DEFAULT_STREAM_RETRY_MAX_ATTEMPTS-1 一致", () => { + assert.equal(settings.PROVIDER_RETRY_DEFAULT_MAX_RETRIES, DEFAULT_STREAM_RETRY_MAX_ATTEMPTS - 1); +}); diff --git a/crates/agent-gui/test/providers/stream-retry.test.mjs b/crates/agent-gui/test/providers/stream-retry.test.mjs index bab095560..ae379ef28 100644 --- a/crates/agent-gui/test/providers/stream-retry.test.mjs +++ b/crates/agent-gui/test/providers/stream-retry.test.mjs @@ -169,6 +169,29 @@ test("withStreamRetry passes the failing attempt's error message as onRetry's th ]); }); +test("withStreamRetry reports the exact backoff about to be slept as onRetry's fourth argument", async () => { + let calls = 0; + const plannedDelays = []; + const wrapped = withStreamRetry( + () => { + calls += 1; + if (calls < 3) return createErrorStream("503 service unavailable"); + return createSuccessStream("final answer"); + }, + { + maxAttempts: 5, + onRetry: (_attempt, _maxAttempts, _errorMessage, plannedDelayMs) => + plannedDelays.push(plannedDelayMs), + }, + ); + + await collectEvents(wrapped); + assert.equal(plannedDelays.length, 2); + // codex-style backoff: base 200ms * 2^(n-1) * uniform(0.9, 1.1). + assert.ok(plannedDelays[0] >= 180 && plannedDelays[0] <= 220, `attempt 1: ${plannedDelays[0]}`); + assert.ok(plannedDelays[1] >= 360 && plannedDelays[1] <= 440, `attempt 2: ${plannedDelays[1]}`); +}); + test("withStreamRetry never calls onRetryRecovered when no retry occurred", async () => { let recoveredCalls = 0; const wrapped = withStreamRetry(() => createSuccessStream("first try"), { diff --git a/crates/agent-gui/test/providers/transport-golden.test.mjs b/crates/agent-gui/test/providers/transport-golden.test.mjs new file mode 100644 index 000000000..d13406f0b --- /dev/null +++ b/crates/agent-gui/test/providers/transport-golden.test.mjs @@ -0,0 +1,402 @@ +import assert from "node:assert/strict"; +import path from "node:path"; +import test from "node:test"; +import { fileURLToPath } from "node:url"; +import { createTsModuleLoader } from "../helpers/load-ts-module.mjs"; + +// ============================================================================ +// LLM seam 改造(golden 基线之二):传输装配整体快照。 +// +// custom-headers-propagation.test.mjs 断言"自定义头能抵达";本文件把 +// prepareProviderRequest 的完整输出(反代 URL + 全量头集 + base64 覆盖包 +// 解码内容)逐字段锁死,并锁定 failover 场景下逐候选传输配置的独立性—— +// 供应商级 useSystemProxy 是"网络可达性属于每个目标"这一公理的载体, +// seam 重构绝不允许把主选的传输事实泄漏给备选。 +// ============================================================================ + +const rootDir = path.resolve(fileURLToPath(new URL("../..", import.meta.url))); +const abs = (rel) => path.join(rootDir, rel); + +const piAiEventStream = await import( + new URL( + "../../node_modules/@earendil-works/pi-ai/dist/utils/event-stream.js", + import.meta.url, + ).href +); + +const PROXY_SERVER_INFO = { baseUrl: "http://127.0.0.1:18080", token: "proxy-token" }; +const SESSION_ID = "00000000-0000-4000-8000-000000000001"; + +function decodeOverrides(headers) { + const encoded = headers["x-liveagent-upstream-headers"]; + if (encoded === undefined) return undefined; + return JSON.parse(Buffer.from(encoded, "base64").toString("utf8")); +} + +/** 覆盖包单独解码断言,其余头逐字段断言。 */ +function splitPrepared(prepared) { + const { "x-liveagent-upstream-headers": _encoded, ...headers } = prepared.headers; + return { baseUrl: prepared.baseUrl, headers, overrides: decodeOverrides(prepared.headers) }; +} + +// --------------------------------------------------------------------------- +// 第一部分:prepareProviderRequest 完整输出快照(真实实现,仅 mock tauri invoke) +// --------------------------------------------------------------------------- + +const transportLoader = createTsModuleLoader({ + mocks: { + "@tauri-apps/api/core": { + async invoke(command) { + if (command === "proxy_get_server_info") return PROXY_SERVER_INFO; + throw new Error(`unexpected tauri invoke: ${command}`); + }, + }, + }, +}); +const { prepareProviderRequest } = transportLoader.loadModule( + "src/lib/providers/runtime/requestOptions.ts", +); +const { ANTHROPIC_DEFAULT_REQUEST_HEADERS } = transportLoader.loadModule( + "@liveagent/ui/lib/providers/customHeaders.ts", +); + +test("golden/transport: anthropic 全量头集(内置默认头 + 自定义头 + 覆盖包 + use-system-proxy)", async () => { + const prepared = await prepareProviderRequest( + "claude_code", + { + baseUrl: "https://api.anthropic.com/v1", + apiKey: "sk-ant-test", + customHeaders: [ + { key: "X-Relay-Channel", value: "vip" }, + // 浏览器禁止头名:常规通道会被 WebView 丢弃,只能靠覆盖包送达。 + { key: "Cookie", value: "session=abc" }, + ], + useSystemProxy: true, + }, + { sessionId: SESSION_ID }, + ); + const { baseUrl, headers, overrides } = splitPrepared(prepared); + + assert.equal(baseUrl, "http://127.0.0.1:18080/proxy/claude_code/v1"); + assert.deepEqual(headers, { + "x-api-key": "sk-ant-test", + ...ANTHROPIC_DEFAULT_REQUEST_HEADERS, + "X-Relay-Channel": "vip", + Cookie: "session=abc", + "x-liveagent-upstream-origin": "https://api.anthropic.com", + "x-liveagent-proxy-token": "proxy-token", + "x-liveagent-use-system-proxy": "1", + }); + // 覆盖包 = 内置默认头 + 自定义头;鉴权头(x-api-key)按排除集绝不进包。 + assert.deepEqual(overrides, { + ...ANTHROPIC_DEFAULT_REQUEST_HEADERS, + "X-Relay-Channel": "vip", + Cookie: "session=abc", + }); +}); + +test("golden/transport: codex Responses 链路带 session/conversation 头;直连时无 use-system-proxy", async () => { + const prepared = await prepareProviderRequest( + "codex", + { baseUrl: "https://chatgpt.com/backend-api/codex", apiKey: "sk-codex-test" }, + { sessionId: SESSION_ID }, + ); + const { baseUrl, headers, overrides } = splitPrepared(prepared); + + assert.equal(baseUrl, "http://127.0.0.1:18080/proxy/codex/backend-api/codex"); + assert.deepEqual(headers, { + Authorization: "Bearer sk-codex-test", + session_id: SESSION_ID, + conversation_id: SESSION_ID, + "x-liveagent-upstream-origin": "https://chatgpt.com", + "x-liveagent-proxy-token": "proxy-token", + }); + assert.deepEqual(overrides, { + session_id: SESSION_ID, + conversation_id: SESSION_ID, + }); +}); + +test("golden/transport: codex Completions 格式绝不泄漏 session/conversation 头", async () => { + const prepared = await prepareProviderRequest( + "codex", + { + baseUrl: "https://relay.example.com/v1", + apiKey: "sk-relay-test", + requestFormat: "openai-completions", + }, + { sessionId: SESSION_ID }, + ); + const { baseUrl, headers, overrides } = splitPrepared(prepared); + + assert.equal(baseUrl, "http://127.0.0.1:18080/proxy/codex/v1"); + // 无状态协议仅 Bearer;头集不含任何需要覆盖包的条目。 + assert.deepEqual(headers, { + Authorization: "Bearer sk-relay-test", + "x-liveagent-upstream-origin": "https://relay.example.com", + "x-liveagent-proxy-token": "proxy-token", + }); + assert.equal(overrides, undefined); +}); + +test("golden/transport: gemini 用 x-goog-api-key 单头鉴权", async () => { + const prepared = await prepareProviderRequest( + "gemini", + { baseUrl: "https://generativelanguage.googleapis.com", apiKey: "g-test-key" }, + { sessionId: SESSION_ID }, + ); + const { baseUrl, headers, overrides } = splitPrepared(prepared); + + assert.equal(baseUrl, "http://127.0.0.1:18080/proxy/gemini"); + assert.deepEqual(headers, { + "x-goog-api-key": "g-test-key", + "x-liveagent-upstream-origin": "https://generativelanguage.googleapis.com", + "x-liveagent-proxy-token": "proxy-token", + }); + assert.equal(overrides, undefined); +}); + +test("golden/transport: deepseek full URL 模式保留完整上游 URL(含查询参数)", async () => { + const prepared = await prepareProviderRequest( + "deepseek", + { + baseUrl: "https://relay.example.com/openai/v1/responses?alt=x", + apiKey: "sk-ds-test", + isFullUrl: true, + }, + { sessionId: SESSION_ID }, + ); + const { baseUrl, headers, overrides } = splitPrepared(prepared); + + assert.equal(baseUrl, "http://127.0.0.1:18080/proxy/deepseek"); + assert.deepEqual(headers, { + Authorization: "Bearer sk-ds-test", + "x-liveagent-upstream-origin": "https://relay.example.com", + "x-liveagent-upstream-url": "https://relay.example.com/openai/v1/responses?alt=x", + "x-liveagent-proxy-token": "proxy-token", + }); + assert.equal(overrides, undefined); +}); + +// --------------------------------------------------------------------------- +// 第二部分:failover 逐候选传输配置独立性。 +// 场景来自网络拓扑用户故事:主选是走应用代理的国外供应商,备选是直连的国内 +// 中转。断言两个目标各自独立装配(use-system-proxy 头互不泄漏),主选未提交 +// 失败后备选以自己的传输配置接管。 +// --------------------------------------------------------------------------- + +/** 捕获经 streamSimpleByApi 发出的每次调用(model + options.headers)。 */ +const streamCalls = []; +let streamImpl = () => { + throw new Error("streamImpl was not configured"); +}; + +const failoverLoader = createTsModuleLoader({ + mocks: { + "@tauri-apps/api/core": { + async invoke(command) { + if (command === "proxy_get_server_info") return PROXY_SERVER_INFO; + throw new Error(`unexpected tauri invoke: ${command}`); + }, + }, + [abs("src/lib/providers/runtime/streamByApi.ts")]: { + streamSimpleByApi: (model, context, options) => { + streamCalls.push({ model, options }); + return streamImpl(model, context, options); + }, + }, + [abs("src/lib/system/powerActivity.ts")]: { + withPowerActivity: (_scope, _reason, run) => run(), + }, + [abs("src/lib/debug/agentDebug.ts")]: { + buildStreamRequestDebugPayload: () => ({}), + }, + [abs("src/lib/providers/hostedSearchEvents.ts")]: { + createHostedSearchProbeId: () => undefined, + withHostedSearchProbeHeader: (headers) => headers ?? {}, + startHostedSearchFetchProbe: () => ({ finish: async () => {} }), + createHostedSearchEventAggregator: () => ({ + accept: () => {}, + complete: () => [], + fail: () => {}, + dispose: () => {}, + getBlocks: () => [], + }), + }, + }, +}); + +const { streamAssistantMessage } = failoverLoader.loadModule( + "src/lib/providers/runtime/textOnlyRuntime.ts", +); +const { resetFailoverBreakers } = failoverLoader.loadModule( + "src/lib/providers/runtime/providerFailover.ts", +); + +function makeAssistantMessage(overrides = {}) { + return { + role: "assistant", + content: [], + api: "anthropic-messages", + provider: "claude_code", + model: "claude-x", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: 1, + }; +} + +function makeSourceStream(events) { + const stream = piAiEventStream.createAssistantMessageEventStream(); + for (const event of events) stream.push(event); + return stream; +} + +function successStream(text) { + const message = { ...makeAssistantMessage(), content: [{ type: "text", text }] }; + return makeSourceStream([ + { type: "start", partial: message }, + { type: "text_delta", contentIndex: 0, delta: text, partial: message }, + { type: "done", reason: "stop", message }, + ]); +} + +function uncommittedErrorStream(errorMessage) { + const message = { ...makeAssistantMessage(), stopReason: "error", errorMessage }; + return makeSourceStream([ + { type: "start", partial: message }, + { type: "error", reason: "error", error: message }, + ]); +} + +test.beforeEach(() => { + resetFailoverBreakers(); + streamCalls.length = 0; +}); + +test("golden/transport-failover: 主选走代理 + 备选直连,逐候选传输配置互不泄漏", async () => { + // 主选:国外供应商,勾选走应用代理。 + const primaryRuntime = { + baseUrl: "https://api.anthropic.com/v1", + apiKey: "sk-primary", + promptCachingEnabled: false, + useSystemProxy: true, + }; + // 备选:国内中转,直连(不带 useSystemProxy)。 + const fallbackRuntime = { + baseUrl: "https://relay.cn.example/v1", + apiKey: "sk-fallback", + promptCachingEnabled: false, + }; + + streamImpl = (_model, _context, options) => + options.headers["x-liveagent-use-system-proxy"] === "1" + ? uncommittedErrorStream("502 upstream proxy unavailable") + : successStream("fallback-answer"); + + const final = await streamAssistantMessage({ + providerId: "claude_code", + model: "claude-x", + runtime: primaryRuntime, + context: { messages: [{ role: "user", content: "hi", timestamp: 1 }] }, + sessionId: SESSION_ID, + onTextDelta: () => {}, + failover: { + config: { maxSwitches: 3, failureThreshold: 3, cooldownSeconds: 60 }, + primary: { + selectedModel: { customProviderId: "p-abroad", model: "claude-x" }, + label: "国外官方 · claude-x", + }, + fallbacks: [ + { + selectedModel: { customProviderId: "p-cn-relay", model: "claude-x" }, + providerId: "claude_code", + model: "claude-x", + label: "国内中转 · claude-x", + runtime: fallbackRuntime, + }, + ], + }, + }); + + assert.equal(final.content[0].text, "fallback-answer"); + assert.equal(streamCalls.length, 2); + + // 候选 1(主选):真实 prepareProviderRequest 输出,带 use-system-proxy。 + const primaryCall = streamCalls[0]; + assert.equal(primaryCall.model.baseUrl, "http://127.0.0.1:18080/proxy/claude_code/v1"); + assert.equal(primaryCall.options.headers["x-liveagent-use-system-proxy"], "1"); + assert.equal(primaryCall.options.headers["x-api-key"], "sk-primary"); + assert.equal( + primaryCall.options.headers["x-liveagent-upstream-origin"], + "https://api.anthropic.com", + ); + + // 候选 2(备选):独立装配,绝不继承主选的 use-system-proxy 与凭据。 + const fallbackCall = streamCalls[1]; + assert.equal(fallbackCall.model.baseUrl, "http://127.0.0.1:18080/proxy/claude_code/v1"); + assert.equal(fallbackCall.options.headers["x-liveagent-use-system-proxy"], undefined); + assert.equal(fallbackCall.options.headers["x-api-key"], "sk-fallback"); + assert.equal( + fallbackCall.options.headers["x-liveagent-upstream-origin"], + "https://relay.cn.example", + ); +}); + +test("golden/transport-failover: 反向拓扑(主选直连 + 备选走代理)同样逐候选独立", async () => { + const primaryRuntime = { + baseUrl: "https://relay.cn.example/v1", + apiKey: "sk-primary-direct", + promptCachingEnabled: false, + }; + const fallbackRuntime = { + baseUrl: "https://api.anthropic.com/v1", + apiKey: "sk-fallback-proxied", + promptCachingEnabled: false, + useSystemProxy: true, + }; + + streamImpl = (_model, _context, options) => + options.headers["x-liveagent-use-system-proxy"] === "1" + ? successStream("proxied-answer") + : uncommittedErrorStream("503 relay unavailable"); + + const final = await streamAssistantMessage({ + providerId: "claude_code", + model: "claude-x", + runtime: primaryRuntime, + context: { messages: [{ role: "user", content: "hi", timestamp: 1 }] }, + sessionId: SESSION_ID, + onTextDelta: () => {}, + failover: { + config: { maxSwitches: 3, failureThreshold: 3, cooldownSeconds: 60 }, + primary: { + selectedModel: { customProviderId: "p-cn-relay", model: "claude-x" }, + label: "国内中转 · claude-x", + }, + fallbacks: [ + { + selectedModel: { customProviderId: "p-abroad", model: "claude-x" }, + providerId: "claude_code", + model: "claude-x", + label: "国外官方 · claude-x", + runtime: fallbackRuntime, + }, + ], + }, + }); + + assert.equal(final.content[0].text, "proxied-answer"); + assert.equal(streamCalls.length, 2); + assert.equal(streamCalls[0].options.headers["x-liveagent-use-system-proxy"], undefined); + assert.equal(streamCalls[0].options.headers["x-api-key"], "sk-primary-direct"); + assert.equal(streamCalls[1].options.headers["x-liveagent-use-system-proxy"], "1"); + assert.equal(streamCalls[1].options.headers["x-api-key"], "sk-fallback-proxied"); +}); diff --git a/crates/agent-gui/test/providers/wire-payload-golden.test.mjs b/crates/agent-gui/test/providers/wire-payload-golden.test.mjs new file mode 100644 index 000000000..11fce5b6d --- /dev/null +++ b/crates/agent-gui/test/providers/wire-payload-golden.test.mjs @@ -0,0 +1,348 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { createTsModuleLoader } from "../helpers/load-ts-module.mjs"; + +// ============================================================================ +// LLM seam 改造(golden 基线之一):五协议最终 wire payload 整体快照。 +// +// 现有 provider 测试逐字段断言单个行为(tool_choice、缓存断点、thinking 档位…); +// 本文件的职责不同——把每条协议"固定输入 → 完整请求体"逐字段锁死,作为后续 +// seam 重构(PR-1 适配器包装 / PR-3 拦截器注册化)"行为等价"的判定基准。 +// 快照有意写成显式对象字面量而非 .snapshot 文件:diff 直接可读,且杜绝 +// 无意 re-record。 +// +// 捕获通道:走真实 pi-ai stream()(连同 finalizeProviderStreamOptions 全部 +// payload 中间件),用 onPayload 截获最终线格式后抛错中断,网络零触碰。 +// ============================================================================ + +const realAnthropic = await import( + new URL( + "../../node_modules/@earendil-works/pi-ai/dist/api/anthropic-messages.js", + import.meta.url, + ).href +); +const realCompletions = await import( + new URL( + "../../node_modules/@earendil-works/pi-ai/dist/api/openai-completions.js", + import.meta.url, + ).href +); +const realResponses = await import( + new URL( + "../../node_modules/@earendil-works/pi-ai/dist/api/openai-responses.js", + import.meta.url, + ).href +); +const realGoogle = await import( + new URL( + "../../node_modules/@earendil-works/pi-ai/dist/api/google-generative-ai.js", + import.meta.url, + ).href +); + +const loader = createTsModuleLoader({ + mocks: { + "@earendil-works/pi-ai/api/anthropic-messages": { stream: realAnthropic.stream }, + "@earendil-works/pi-ai/api/openai-completions": { stream: realCompletions.stream }, + "@earendil-works/pi-ai/api/openai-responses": { stream: realResponses.stream }, + "@earendil-works/pi-ai/api/google-generative-ai": { stream: realGoogle.stream }, + }, +}); + +const { createModelFromConfig } = loader.loadModule("src/lib/providers/runtime/modelFactory.ts"); +const { streamSimpleByApi } = loader.loadModule("src/lib/providers/runtime/streamByApi.ts"); +const { finalizeProviderStreamOptions } = loader.loadModule( + "src/lib/providers/runtime/payloadPipeline.ts", +); + +// 固定 session id:payload 中所有会话关联字段(prompt_cache_key/metadata.user_id) +// 由它派生,保证快照确定性。 +const SESSION_ID = "00000000-0000-4000-8000-000000000001"; + +const TOOLS = [ + { + name: "read_file", + description: "Read a file", + parameters: { + type: "object", + properties: { path: { type: "string" } }, + required: ["path"], + }, + }, +]; + +function buildContext({ withTools = true } = {}) { + return { + systemPrompt: "You are a precise assistant.", + messages: [{ role: "user", content: "hello world", timestamp: 1 }], + ...(withTools ? { tools: TOOLS } : {}), + }; +} + +/** + * 走真实装配链(finalizeProviderStreamOptions → streamSimpleByApi → 真实 + * pi-ai stream),在 onPayload 链尾截获最终 wire payload 后中断请求。 + */ +async function captureWirePayload(providerId, model, context, baseOptions) { + let captured; + const finalized = finalizeProviderStreamOptions({ + providerId, + baseUrl: model.baseUrl, + options: baseOptions, + context, + model, + }); + const prevOnPayload = finalized.onPayload; + const stream = streamSimpleByApi(model, context, { + ...finalized, + onPayload: async (payload, m) => { + captured = prevOnPayload ? ((await prevOnPayload(payload, m)) ?? payload) : payload; + throw new Error("__capture_stop__"); + }, + }); + try { + await stream.result(); + } catch { + // onPayload 抛错中断请求属预期。 + } + assert.ok(captured, `expected wire payload capture for ${model.id}`); + // JSON 往返归一化:golden 锁定的是线上 JSON 形态;值为 undefined 的键 + // (如 responses 链路的 prompt_cache_retention)序列化后不存在,不入快照。 + return JSON.parse(JSON.stringify(captured)); +} + +const WIRE_TOOL_SCHEMA = { + type: "object", + properties: { path: { type: "string" } }, + required: ["path"], +}; + +test("golden/anthropic-messages: 官方端点完整请求体(adaptive thinking + 缓存断点 + metadata)", async () => { + const baseUrl = "https://api.anthropic.com/v1"; + const model = createModelFromConfig( + "claude_code", + "claude-sonnet-4-6", + baseUrl, + undefined, + undefined, + baseUrl, + ); + const payload = await captureWirePayload("claude_code", model, buildContext(), { + apiKey: "sk-test", + reasoning: "high", + toolChoice: "auto", + sessionId: SESSION_ID, + cacheRetention: "short", + metadata: { user_id: SESSION_ID }, + }); + + assert.deepEqual(payload, { + model: "claude-sonnet-4-6", + messages: [{ role: "user", content: [{ type: "text", text: "hello world" }] }], + max_tokens: 128000, + stream: true, + system: [{ type: "text", text: "You are a precise assistant." }], + tools: [ + { + name: "read_file", + description: "Read a file", + eager_input_streaming: true, + input_schema: WIRE_TOOL_SCHEMA, + }, + ], + thinking: { type: "adaptive", display: "summarized" }, + output_config: { effort: "high" }, + metadata: { user_id: SESSION_ID }, + tool_choice: { type: "auto" }, + cache_control: { type: "ephemeral" }, + }); +}); + +test("golden/openai-completions: 中转端点完整请求体(带工具 + reasoning_effort)", async () => { + const baseUrl = "https://relay.example.com/v1"; + const model = createModelFromConfig( + "codex", + "gpt-5.2", + baseUrl, + "openai-completions", + undefined, + baseUrl, + ); + const payload = await captureWirePayload("codex", model, buildContext(), { + apiKey: "sk-test", + reasoning: "high", + toolChoice: "auto", + sessionId: SESSION_ID, + }); + + assert.deepEqual(payload, { + model: "gpt-5.2", + messages: [ + { role: "system", content: "You are a precise assistant." }, + { role: "user", content: "hello world" }, + ], + stream: true, + stream_options: { include_usage: true }, + max_completion_tokens: 128000, + tools: [ + { + type: "function", + function: { + name: "read_file", + description: "Read a file", + parameters: WIRE_TOOL_SCHEMA, + strict: false, + }, + }, + ], + tool_choice: "auto", + reasoning_effort: "high", + }); +}); + +test("golden/openai-completions: text-only 请求既不带 tools 也不带 tool_choice(严格网关 400 回归)", async () => { + const baseUrl = "https://relay.example.com/v1"; + const model = createModelFromConfig( + "codex", + "gpt-5.2", + baseUrl, + "openai-completions", + undefined, + baseUrl, + ); + const payload = await captureWirePayload( + "codex", + model, + buildContext({ withTools: false }), + { apiKey: "sk-test", toolChoice: "auto", sessionId: SESSION_ID }, + ); + + assert.deepEqual(payload, { + model: "gpt-5.2", + messages: [ + { role: "system", content: "You are a precise assistant." }, + { role: "user", content: "hello world" }, + ], + stream: true, + stream_options: { include_usage: true }, + max_completion_tokens: 128000, + }); +}); + +test("golden/openai-responses: codex 官方端点完整请求体(store + prompt_cache_key + encrypted reasoning)", async () => { + const baseUrl = "https://chatgpt.com/backend-api/codex"; + const model = createModelFromConfig( + "codex", + "gpt-5.2-codex", + baseUrl, + undefined, + undefined, + baseUrl, + ); + const payload = await captureWirePayload("codex", model, buildContext(), { + apiKey: "sk-test", + reasoning: "high", + sessionId: SESSION_ID, + cacheRetention: "short", + }); + + assert.deepEqual(payload, { + model: "gpt-5.2-codex", + input: [ + { role: "system", content: "You are a precise assistant." }, + { role: "user", content: [{ type: "input_text", text: "hello world" }] }, + ], + stream: true, + prompt_cache_key: SESSION_ID, + store: true, + max_output_tokens: 142000, + tools: [ + { + type: "function", + name: "read_file", + description: "Read a file", + parameters: WIRE_TOOL_SCHEMA, + }, + ], + reasoning: { effort: "high", summary: "auto" }, + include: ["reasoning.encrypted_content"], + }); +}); + +test("golden/google-generative-ai: 官方端点完整请求体(thinkingLevel + functionCallingConfig)", async () => { + const baseUrl = "https://generativelanguage.googleapis.com"; + const model = createModelFromConfig( + "gemini", + "gemini-3-pro-preview", + baseUrl, + undefined, + undefined, + baseUrl, + ); + const payload = await captureWirePayload("gemini", model, buildContext(), { + apiKey: "test-key", + reasoning: "high", + toolChoice: "auto", + sessionId: SESSION_ID, + }); + + assert.deepEqual(payload, { + model: "gemini-3-pro-preview", + contents: [{ role: "user", parts: [{ text: "hello world" }] }], + config: { + maxOutputTokens: 65536, + systemInstruction: "You are a precise assistant.", + tools: [ + { + functionDeclarations: [ + { + name: "read_file", + description: "Read a file", + parametersJsonSchema: WIRE_TOOL_SCHEMA, + }, + ], + }, + ], + toolConfig: { functionCallingConfig: { mode: "AUTO" } }, + thinkingConfig: { includeThoughts: true, thinkingLevel: "HIGH" }, + }, + }); +}); + +test("golden/deepseek-responses: 原生适配器完整请求体(developer role + reasoning effort 直通)", async () => { + const baseUrl = "https://api.deepseek.com"; + const model = createModelFromConfig( + "deepseek", + "deepseek-v4-flash", + baseUrl, + undefined, + undefined, + baseUrl, + ); + const payload = await captureWirePayload("deepseek", model, buildContext(), { + apiKey: "sk-test", + reasoning: "high", + toolChoice: "auto", + sessionId: SESSION_ID, + }); + + assert.deepEqual(payload, { + model: "deepseek-v4-flash", + input: [ + { role: "developer", content: "You are a precise assistant." }, + { role: "user", content: [{ type: "input_text", text: "hello world" }] }, + ], + stream: true, + max_output_tokens: 384000, + tools: [ + { + type: "function", + name: "read_file", + description: "Read a file", + parameters: WIRE_TOOL_SCHEMA, + }, + ], + tool_choice: "auto", + reasoning: { effort: "high" }, + }); +}); diff --git a/crates/agent-gui/test/trajectory/event-log.test.mjs b/crates/agent-gui/test/trajectory/event-log.test.mjs index 969462d06..30fb8e53e 100644 --- a/crates/agent-gui/test/trajectory/event-log.test.mjs +++ b/crates/agent-gui/test/trajectory/event-log.test.mjs @@ -140,6 +140,111 @@ test("retries are deduplicated by attempt and sorted", () => { ); }); +test("retry provider labels survive into the ledger", () => { + const ledger = buildTrajectoryLedger([ + ...turnEvents(), + { k: "retry", t: 1, s: 1, at: BASE + 50, n: 1, max: 3, p: "P1 · claude-x", delay: 200 }, + ]); + const retry = ledger.turns[0].steps[0].retries[0]; + assert.equal(retry.provider, "P1 · claude-x"); + assert.equal(retry.delayMs, 200); +}); + +test("failover events converge into the owning step sorted by attempt", () => { + const ledger = buildTrajectoryLedger([ + ...turnEvents(), + { + k: "failover", + t: 1, + s: 1, + at: BASE + 60, + n: 2, + from: "P2 · claude-x", + to: "P3 · gpt-y", + ti: 2, + }, + { + k: "failover", + t: 1, + s: 1, + at: BASE + 30, + n: 1, + from: "P1 · claude-x", + to: "P2 · claude-x", + ti: 1, + err: "503 from primary", + }, + // 重复投递(实时+落盘)收敛为一条。 + { + k: "failover", + t: 1, + s: 1, + at: BASE + 30, + n: 1, + from: "P1 · claude-x", + to: "P2 · claude-x", + ti: 1, + err: "503 from primary", + }, + ]); + const failovers = ledger.turns[0].steps[0].failovers; + assert.equal(failovers.length, 2); + assert.deepEqual( + failovers.map((entry) => entry.attempt), + [1, 2], + ); + assert.equal(failovers[0].fromLabel, "P1 · claude-x"); + assert.equal(failovers[0].toLabel, "P2 · claude-x"); + assert.equal(failovers[0].targetIndex, 1); + assert.equal(failovers[0].error, "503 from primary"); + assert.equal(failovers[1].targetIndex, 2); +}); + +test("transport snapshots keep per-candidate independence in the ledger", () => { + const ledger = buildTrajectoryLedger([ + ...turnEvents(), + { + k: "transport", + t: 1, + s: 1, + at: BASE + 15, + p: "P1 · claude-x", + o: "https://api.primary.example", + sp: true, + fu: false, + hn: ["x-liveagent-proxy-token", "x-liveagent-upstream-origin", "x-liveagent-use-system-proxy"], + }, + { + k: "transport", + t: 1, + s: 1, + at: BASE + 35, + p: "P2 · claude-x", + o: "https://api.fallback.example", + sp: false, + fu: false, + hn: ["x-liveagent-proxy-token", "x-liveagent-upstream-origin"], + }, + ]); + const transports = ledger.turns[0].steps[0].transports; + assert.equal(transports.length, 2); + assert.equal(transports[0].useSystemProxy, true); + assert.equal(transports[1].useSystemProxy, false); + assert.ok(transports[0].headerNames.includes("x-liveagent-use-system-proxy")); + assert.ok(!transports[1].headerNames.includes("x-liveagent-use-system-proxy")); +}); + +test("failover and transport events converge order-independently like all others", () => { + const events = [ + ...turnEvents(), + { k: "transport", t: 1, s: 1, at: BASE + 15, p: "P1", sp: true, fu: false, hn: ["a"] }, + { k: "failover", t: 1, s: 1, at: BASE + 30, n: 1, from: "P1", to: "P2", ti: 1 }, + ]; + const ordered = buildTrajectoryLedger(events); + const shuffled = buildTrajectoryLedger([...events].reverse()); + assert.deepEqual(shuffled, ordered); +}); + test("hasTiming stays false when no operation carried a timestamp", () => { const ledger = buildTrajectoryLedger([{ k: "user", t: 1, at: Number.NaN }]); assert.equal(ledger.hasTiming, false); diff --git a/crates/agent-gui/test/trajectory/recorder.test.mjs b/crates/agent-gui/test/trajectory/recorder.test.mjs index d42f65464..12324e57a 100644 --- a/crates/agent-gui/test/trajectory/recorder.test.mjs +++ b/crates/agent-gui/test/trajectory/recorder.test.mjs @@ -167,6 +167,89 @@ test("first token is recorded once per step", () => { assert.equal(published.filter((event) => event.k === "first_token").length, 2); }); +test("noteRetry carries the provider label and planned delay for the audit trail", () => { + const { recorder, published } = harness(); + recorder.beginTurn({ turn: 2 }); + recorder.noteRetry(1, { + attempt: 1, + maxRetries: 5, + delayMs: 200, + error: "503 service unavailable", + provider: "P1 · claude-x", + }); + const retry = published.find((event) => event.k === "retry"); + assert.equal(retry.t, 2); + assert.equal(retry.s, 1); + assert.equal(retry.n, 1); + assert.equal(retry.max, 5); + assert.equal(retry.delay, 200); + assert.equal(retry.err, "503 service unavailable"); + assert.equal(retry.p, "P1 · claude-x"); +}); + +test("noteFailover emits a failover event with switch identity", () => { + const { recorder, published } = harness(); + recorder.beginTurn({ turn: 3 }); + recorder.noteFailover(2, { + attempt: 1, + fromLabel: "P1 · claude-x", + toLabel: "P2 · claude-x", + targetIndex: 1, + error: "503 from primary", + }); + const failover = published.find((event) => event.k === "failover"); + assert.equal(failover.t, 3); + assert.equal(failover.s, 2); + assert.equal(failover.n, 1); + assert.equal(failover.from, "P1 · claude-x"); + assert.equal(failover.to, "P2 · claude-x"); + assert.equal(failover.ti, 1); + assert.equal(failover.err, "503 from primary"); +}); + +test("noteTransport records header names and routing flags without values", () => { + const { recorder, published } = harness(); + recorder.beginTurn({ turn: 1 }); + recorder.noteTransport(1, { + provider: "P1 · claude-x", + upstreamOrigin: "https://api.example.com", + useSystemProxy: true, + fullUrl: false, + headerNames: ["x-liveagent-proxy-token", "x-liveagent-upstream-origin"], + }); + const transport = published.find((event) => event.k === "transport"); + assert.equal(transport.p, "P1 · claude-x"); + assert.equal(transport.o, "https://api.example.com"); + assert.equal(transport.sp, true); + assert.equal(transport.fu, false); + assert.deepEqual(transport.hn, ["x-liveagent-proxy-token", "x-liveagent-upstream-origin"]); + const serialized = JSON.stringify(transport); + assert.ok(!serialized.includes("Bearer"), "transport events must never carry header values"); +}); + +test("error text entering the ledger is scrubbed of URL keys and bearer tokens", () => { + const { recorder, published } = harness(); + recorder.beginTurn({ turn: 1 }); + recorder.noteRetry(1, { + attempt: 1, + error: + "fetch failed: https://api.example.com/v1?key=AIzaSyC-secret-sample-0123456789012 Authorization: Bearer sk-proj-abcdef1234567890abcdef", + }); + recorder.stepEnd(1, { status: "error", error: "401 x-goog-api-key=AIzaSyD_other_key_0123456789" }); + recorder.endTurn({ status: "error", error: "Bearer sk-ant-api03-verysecretvalue12345" }); + + for (const event of published) { + const serialized = JSON.stringify(event); + assert.ok(!serialized.includes("AIzaSy"), `leaked google key in ${event.k}`); + assert.ok(!serialized.includes("sk-proj-"), `leaked openai key in ${event.k}`); + assert.ok(!serialized.includes("sk-ant-"), `leaked anthropic key in ${event.k}`); + } + const retry = published.find((event) => event.k === "retry"); + assert.ok(retry.err.includes("[redacted]")); + assert.ok(retry.err.includes("fetch failed"), "non-secret prose survives scrubbing"); +}); + + test("context previews are bounded while preserving their source", () => { const { recorder, published } = harness(); recorder.noteContext({ source: "parent-message-bus", text: "x".repeat(2_000) }); diff --git a/crates/agent-gui/test/trajectory/scrub-transport.test.mjs b/crates/agent-gui/test/trajectory/scrub-transport.test.mjs new file mode 100644 index 000000000..94b243ebc --- /dev/null +++ b/crates/agent-gui/test/trajectory/scrub-transport.test.mjs @@ -0,0 +1,116 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { createTsModuleLoader } from "../helpers/load-ts-module.mjs"; + +const loader = createTsModuleLoader(); +const { scrubSecretsFromErrorText } = loader.loadModule("src/lib/trajectory/scrub.ts"); +const { captureTransportSnapshot } = loader.loadModule( + "src/lib/providers/runtime/transportSnapshot.ts", +); + +// --------------------------------------------------------------------------- +// scrubSecretsFromErrorText +// --------------------------------------------------------------------------- + +test("scrubs API keys from URL query parameters", () => { + const scrubbed = scrubSecretsFromErrorText( + "400 from https://generativelanguage.googleapis.com/v1beta/models?key=AIzaSyC0123456789abcdefghijklmnopqrstu&alt=sse", + ); + assert.ok(!scrubbed.includes("AIzaSyC0123456789abcdefghijklmnopqrstu")); + assert.ok(scrubbed.includes("?key=[redacted]")); + assert.ok(scrubbed.includes("&alt=sse"), "non-secret query params survive"); +}); + +test("scrubs bearer tokens wherever they are echoed", () => { + const scrubbed = scrubSecretsFromErrorText( + 'fetch failed: header Authorization: Bearer sk-proj-secretsecretsecret1234 rejected', + ); + assert.ok(!scrubbed.includes("sk-proj-secretsecretsecret1234")); + assert.match(scrubbed, /Bearer \[redacted\]/i); +}); + +test("scrubs bare known-shape keys outside URLs and headers", () => { + const scrubbed = scrubSecretsFromErrorText( + "provider rejected key sk-ant-api03-0123456789abcdef0123 (expired)", + ); + assert.ok(!scrubbed.includes("sk-ant-api03-0123456789abcdef0123")); + assert.ok(scrubbed.includes("(expired)")); +}); + +test("does not mangle ordinary provider errors", () => { + const text = + "503 service unavailable: upstream connect error or disconnect/reset before headers"; + assert.equal(scrubSecretsFromErrorText(text), text); +}); + +test("does not mangle error text mentioning token counts or model names", () => { + const text = "400: max_tokens (128000) exceeds model limit; model=claude-sonnet-4"; + assert.equal(scrubSecretsFromErrorText(text), text); +}); + +// --------------------------------------------------------------------------- +// captureTransportSnapshot +// --------------------------------------------------------------------------- + +test("snapshot keeps header names and routing flags, never values", () => { + const snapshot = captureTransportSnapshot({ + Authorization: "Bearer sk-proj-supersecret", + "x-api-key": "sk-ant-anothersecret", + "x-liveagent-proxy-token": "proxy-token-value", + "x-liveagent-upstream-origin": "https://api.example.com", + "x-liveagent-use-system-proxy": "1", + "x-liveagent-upstream-headers": "eyJzZWNyZXQiOiJ2YWx1ZSJ9", + }); + assert.equal(snapshot.upstreamOrigin, "https://api.example.com"); + assert.equal(snapshot.useSystemProxy, true); + assert.equal(snapshot.fullUrl, false); + assert.deepEqual(snapshot.headerNames, [ + "authorization", + "x-api-key", + "x-liveagent-proxy-token", + "x-liveagent-upstream-headers", + "x-liveagent-upstream-origin", + "x-liveagent-use-system-proxy", + ]); + const serialized = JSON.stringify(snapshot); + assert.ok(!serialized.includes("sk-proj-supersecret")); + assert.ok(!serialized.includes("sk-ant-anothersecret")); + assert.ok(!serialized.includes("proxy-token-value")); + assert.ok(!serialized.includes("eyJzZWNyZXQ"), "base64 override pack value must not leak"); +}); + +test("snapshot flags full-URL mode without recording the URL itself", () => { + const snapshot = captureTransportSnapshot({ + "x-liveagent-upstream-url": "https://api.example.com/v1/chat?key=AIzaSyCsecret012345678901234", + "x-liveagent-upstream-origin": "https://api.example.com", + }); + assert.equal(snapshot.fullUrl, true); + const serialized = JSON.stringify(snapshot); + assert.ok(!serialized.includes("AIzaSy"), "full URL (may contain query keys) must not leak"); + assert.ok(!serialized.includes("/v1/chat")); +}); + +test("null header values (deletion markers) stay out of the snapshot", () => { + const snapshot = captureTransportSnapshot({ + "x-liveagent-use-system-proxy": null, + accept: "text/event-stream", + }); + assert.equal(snapshot.useSystemProxy, false); + assert.deepEqual(snapshot.headerNames, ["accept"]); +}); + +test("a direct-connection candidate snapshot omits the proxy flag entirely", () => { + const withProxy = captureTransportSnapshot({ + "x-liveagent-use-system-proxy": "1", + "x-liveagent-upstream-origin": "https://api.foreign.example", + }); + const direct = captureTransportSnapshot({ + "x-liveagent-upstream-origin": "https://api.domestic.example", + }); + // failover 逐候选独立性:主选走代理、备选直连,两份快照互不泄漏。 + assert.equal(withProxy.useSystemProxy, true); + assert.equal(direct.useSystemProxy, false); + assert.ok(withProxy.headerNames.includes("x-liveagent-use-system-proxy")); + assert.ok(!direct.headerNames.includes("x-liveagent-use-system-proxy")); +}); diff --git a/crates/agent-ui/src/components/trajectory/details/tabs/OptionsTab.tsx b/crates/agent-ui/src/components/trajectory/details/tabs/OptionsTab.tsx index 6f9f7299a..314926133 100644 --- a/crates/agent-ui/src/components/trajectory/details/tabs/OptionsTab.tsx +++ b/crates/agent-ui/src/components/trajectory/details/tabs/OptionsTab.tsx @@ -14,17 +14,65 @@ export function OptionsTab({ record, t }: DetailTabProps) { ["messageIndex", record.messageIndex === undefined ? undefined : String(record.messageIndex)], ]; const visible = entries.filter((entry): entry is [string, string] => entry[1] !== undefined); - if (visible.length === 0 && (record.retries?.length ?? 0) === 0) return ; + if ( + visible.length === 0 && + (record.retries?.length ?? 0) === 0 && + (record.failovers?.length ?? 0) === 0 && + (record.transports?.length ?? 0) === 0 + ) { + return ; + } return (
{visible.map(([label, value]) => ( ))} + {record.transports?.map((transport) => ( + + ))} {record.retries?.map((retry) => ( + ))} + {record.failovers?.map((failover) => ( + diff --git a/crates/agent-ui/src/components/trajectory/details/tabs/OverviewTab.tsx b/crates/agent-ui/src/components/trajectory/details/tabs/OverviewTab.tsx index f9695f6a3..0c8450269 100644 --- a/crates/agent-ui/src/components/trajectory/details/tabs/OverviewTab.tsx +++ b/crates/agent-ui/src/components/trajectory/details/tabs/OverviewTab.tsx @@ -36,6 +36,9 @@ export function OverviewTab({ record, locale, t }: DetailTabProps) { {record.retries !== undefined && record.retries.length > 0 && ( )} + {record.failovers !== undefined && record.failovers.length > 0 && ( + + )} {record.tokensBefore !== undefined && ( ; + if (obj.mode === "off") return { mode: "off" }; + if (obj.mode === "custom") { + const raw = obj.maxRetries; + if (typeof raw !== "number" || !Number.isFinite(raw)) return undefined; + return { + mode: "custom", + maxRetries: clampInt( + raw, + PROVIDER_RETRY_MAX_RETRIES_LIMITS.min, + PROVIDER_RETRY_MAX_RETRIES_LIMITS.max, + PROVIDER_RETRY_MAX_RETRIES_LIMITS.min, + ), + }; + } + return undefined; +} + export function normalizeCustomProvider(input: unknown): CustomProvider { const obj = (input && typeof input === "object" ? input : {}) as Record; const type = normalizeProviderId(obj.type); @@ -1100,6 +1126,10 @@ export function normalizeCustomProvider(input: unknown): CustomProvider { : {}), nativeWebSearchEnabled: obj.nativeWebSearchEnabled !== false, useSystemProxy: obj.useSystemProxy === true, + ...((): { retryPolicy?: ProviderRetryPolicy } => { + const retryPolicy = normalizeProviderRetryPolicy(obj.retryPolicy); + return retryPolicy ? { retryPolicy } : {}; + })(), usageQuery: normalizeUsageQueryConfig(obj.usageQuery), }; } diff --git a/crates/agent-ui/src/lib/settings/types.ts b/crates/agent-ui/src/lib/settings/types.ts index 882f24f98..c27445d83 100644 --- a/crates/agent-ui/src/lib/settings/types.ts +++ b/crates/agent-ui/src/lib/settings/types.ts @@ -499,9 +499,36 @@ export type CustomProvider = { promptCacheRetention?: "short" | "long"; nativeWebSearchEnabled: boolean; useSystemProxy: boolean; + /** 流内重试策略;缺省 = 全局默认行为(等价于 mode:"default")。 */ + retryPolicy?: ProviderRetryPolicy; usageQuery: UsageQueryConfig; }; +/** + * 供应商级流内重试策略。 + * + * - default:沿用全局默认(5 次重试,即 DEFAULT_STREAM_RETRY_MAX_ATTEMPTS-1) + * ——与未配置等价,归一化时直接省略字段,保证旧配置零迁移; + * - off:禁用流内重试(不影响跨供应商 failover); + * - custom:使用 maxRetries——首次失败后的重试次数,不含首次请求(钳位 + * 1..10;0 次重试请直接选 off)。与重试状态提示"正在重试 (n/m)"的 m + * 同一口径。 + */ +export type ProviderRetryPolicy = { mode: "off" } | { mode: "custom"; maxRetries: number }; + +export const PROVIDER_RETRY_MAX_RETRIES_LIMITS = { + min: 1, + max: 10, +} as const; + +/** + * 全局默认流内重试次数(不含首次请求)的 UI 展示镜像。运行时真源是 + * agent-gui streamRetry.ts 的 DEFAULT_STREAM_RETRY_MAX_ATTEMPTS(总尝试 + * 数 = 重试数 + 1;UI 边界禁止反向依赖);两者一致性由 + * provider-retry-policy 单测锁定。 + */ +export const PROVIDER_RETRY_DEFAULT_MAX_RETRIES = 5; + export type EffectiveTheme = "light" | "dark"; export type Theme = EffectiveTheme | "system"; diff --git a/crates/agent-ui/src/lib/trajectory/eventLog.ts b/crates/agent-ui/src/lib/trajectory/eventLog.ts index 72dca39e2..36c09512a 100644 --- a/crates/agent-ui/src/lib/trajectory/eventLog.ts +++ b/crates/agent-ui/src/lib/trajectory/eventLog.ts @@ -10,11 +10,13 @@ import type { LedgerCompaction, + LedgerFailover, LedgerHeader, LedgerInput, LedgerRetry, LedgerStep, LedgerToolCall, + LedgerTransport, LedgerTurn, TrajectoryEvent, TrajectoryLedger, @@ -160,13 +162,15 @@ const EVENT_ORDER: Record = { header: 2, compaction_start: 3, step_start: 4, - first_token: 5, - retry: 6, - tool_start: 7, - tool_end: 8, - step_end: 9, - compaction_end: 10, - turn_end: 11, + transport: 5, + first_token: 6, + retry: 7, + failover: 8, + tool_start: 9, + tool_end: 10, + step_end: 11, + compaction_end: 12, + turn_end: 13, }; /** Normalize once so all later pairing is independent of transport arrival order. */ @@ -229,6 +233,8 @@ type MutableStep = { usage?: LedgerStep["usage"]; headerId?: string; retries: LedgerRetry[]; + failovers: LedgerFailover[]; + transports: LedgerTransport[]; tools: MutableTool[]; /** 当前进程的 live 事件流里出现过该条目的任意事件。 */ sawLive: boolean; @@ -289,6 +295,8 @@ function ensureStep(turnEntry: MutableTurn, step: number): MutableStep { endedAt: null, endStatus: null, retries: [], + failovers: [], + transports: [], tools: [], sawLive: false, }; @@ -465,6 +473,35 @@ export function buildTrajectoryLedger( ...(event.max === undefined ? {} : { maxRetries: event.max }), ...(event.delay === undefined ? {} : { delayMs: event.delay }), ...(event.err === undefined ? {} : { error: event.err }), + ...(event.p === undefined ? {} : { provider: event.p }), + }); + break; + } + case "failover": { + const turnEntry = ensureTurn(turns, event.t); + const step = ensureStep(turnEntry, event.s); + markLive(turnEntry, step); + step.failovers.push({ + attempt: event.n, + at: event.at, + ...(event.from === undefined ? {} : { fromLabel: event.from }), + ...(event.to === undefined ? {} : { toLabel: event.to }), + ...(event.ti === undefined ? {} : { targetIndex: event.ti }), + ...(event.err === undefined ? {} : { error: event.err }), + }); + break; + } + case "transport": { + const turnEntry = ensureTurn(turns, event.t); + const step = ensureStep(turnEntry, event.s); + markLive(turnEntry, step); + step.transports.push({ + at: event.at, + ...(event.p === undefined ? {} : { provider: event.p }), + ...(event.o === undefined ? {} : { upstreamOrigin: event.o }), + ...(event.sp === undefined ? {} : { useSystemProxy: event.sp }), + ...(event.fu === undefined ? {} : { fullUrl: event.fu }), + ...(event.hn === undefined ? {} : { headerNames: [...event.hn] }), }); break; } @@ -644,6 +681,10 @@ export function buildTrajectoryLedger( ...(step.usage === undefined ? {} : { usage: step.usage }), ...(step.headerId === undefined ? {} : { headerId: step.headerId }), retries: [...step.retries].sort((left, right) => left.attempt - right.attempt), + failovers: [...step.failovers].sort( + (left, right) => left.attempt - right.attempt || left.at - right.at, + ), + transports: [...step.transports].sort((left, right) => left.at - right.at), tools, }; }); diff --git a/crates/agent-ui/src/lib/trajectory/fromMessages.ts b/crates/agent-ui/src/lib/trajectory/fromMessages.ts index 1d7e2ae4b..a135bc187 100644 --- a/crates/agent-ui/src/lib/trajectory/fromMessages.ts +++ b/crates/agent-ui/src/lib/trajectory/fromMessages.ts @@ -88,6 +88,8 @@ export function deriveLedgerFromMessages( ...(round.meta?.stopReason === undefined ? {} : { stopReason: round.meta.stopReason }), ...(usage === undefined ? {} : { usage }), retries: [], + failovers: [], + transports: [], tools, }); } diff --git a/crates/agent-ui/src/lib/trajectory/layout.ts b/crates/agent-ui/src/lib/trajectory/layout.ts index 471a86113..f80aa9c37 100644 --- a/crates/agent-ui/src/lib/trajectory/layout.ts +++ b/crates/agent-ui/src/lib/trajectory/layout.ts @@ -399,6 +399,8 @@ function buildAssistantRecord( ...(step.error === undefined ? {} : { error: step.error }), ...(step.headerId === undefined ? {} : { headerId: step.headerId }), ...(step.retries.length === 0 ? {} : { retries: step.retries }), + ...(step.failovers.length === 0 ? {} : { failovers: step.failovers }), + ...(step.transports.length === 0 ? {} : { transports: step.transports }), ...(joined?.text === undefined ? {} : { outputDetail: joined.text }), ...(joined?.thinking === undefined ? {} : { thinkingDetail: joined.thinking }), ...(joined?.blocks === undefined ? {} : { sourceBlocks: joined.blocks }), diff --git a/crates/agent-ui/src/lib/trajectory/types.ts b/crates/agent-ui/src/lib/trajectory/types.ts index ade7149e3..cec134dc5 100644 --- a/crates/agent-ui/src/lib/trajectory/types.ts +++ b/crates/agent-ui/src/lib/trajectory/types.ts @@ -111,7 +111,7 @@ export type TrajectoryEvent = sr?: string; err?: string; } - /** 失败后的重试记录。 */ + /** 失败后的重试记录。`p` 是候选标签("Provider · model"),failover 下区分各候选的重试。 */ | { k: "retry"; t: number; @@ -121,6 +121,34 @@ export type TrajectoryEvent = max?: number; delay?: number; err?: string; + p?: string; + } + /** 跨供应商切换记录。`n` 是本次请求内的切换序号,`ti` 是目标在候选队列里的下标。 */ + | { + k: "failover"; + t: number; + s: number; + at: number; + n: number; + from?: string; + to?: string; + ti?: number; + err?: string; + } + /** + * 一次实际尝试的传输装配快照。脱敏不变量:只含头名(`hn`)与路由标记, + * 绝不含任何头值或密钥;`o` 是上游 origin(scheme+host)。 + */ + | { + k: "transport"; + t: number; + s: number; + at: number; + p?: string; + o?: string; + sp?: boolean; + fu?: boolean; + hn?: readonly string[]; } /** 工具开始执行。`a` 是截断后的参数文本。 */ | { k: "tool_start"; t: number; s: number; at: number; id: string; n: string; a?: string } @@ -163,6 +191,28 @@ export type LedgerRetry = { maxRetries?: number; delayMs?: number; error?: string; + /** 候选标签("Provider · model");failover 下区分各候选自己的重试。 */ + provider?: string; +}; + +export type LedgerFailover = { + attempt: number; + at: number; + fromLabel?: string; + toLabel?: string; + /** 目标在候选队列里的稳定下标(0 = 主选)。 */ + targetIndex?: number; + error?: string; +}; + +/** 一次实际尝试的传输装配快照。只含头名与路由标记,永不含头值。 */ +export type LedgerTransport = { + at: number; + provider?: string; + upstreamOrigin?: string; + useSystemProxy?: boolean; + fullUrl?: boolean; + headerNames?: readonly string[]; }; export type LedgerToolCall = { @@ -192,6 +242,8 @@ export type LedgerStep = { usage?: TrajectoryUsage; headerId?: string; retries: readonly LedgerRetry[]; + failovers: readonly LedgerFailover[]; + transports: readonly LedgerTransport[]; tools: readonly LedgerToolCall[]; }; @@ -338,6 +390,8 @@ export type TrajectoryRecord = { stopReason?: string; error?: string; retries?: readonly LedgerRetry[]; + failovers?: readonly LedgerFailover[]; + transports?: readonly LedgerTransport[]; assistantMetrics?: TrajectoryAssistantMetrics; inputDetail?: string; outputDetail?: string; diff --git a/crates/agent-ui/src/pages/settings/ProviderModal.tsx b/crates/agent-ui/src/pages/settings/ProviderModal.tsx index 170143f00..419c5579c 100644 --- a/crates/agent-ui/src/pages/settings/ProviderModal.tsx +++ b/crates/agent-ui/src/pages/settings/ProviderModal.tsx @@ -3,9 +3,12 @@ import { type CodexRequestFormat, type CustomProvider, getDefaultUsageQueryConfig, + PROVIDER_RETRY_DEFAULT_MAX_RETRIES, + PROVIDER_RETRY_MAX_RETRIES_LIMITS, type PromptCacheHintMode, type ProviderId, type ProviderModelConfig, + type ProviderRetryPolicy, } from "@liveagent/app/lib/settings"; import { useConfirmDialog } from "@liveagent/ui/components/ui/confirm-dialog"; import { useVerticalListReorder } from "@liveagent/ui/components/ui/useVerticalListReorder"; @@ -154,6 +157,17 @@ function useProviderModalController({ providerType, initialData, onSave, onClose initialData?.requestFormat ?? "openai-responses", ); const [useSystemProxy, setUseSystemProxy] = useState(initialData?.useSystemProxy ?? false); + const [streamRetryMode, setStreamRetryMode] = useState<"default" | "off" | "custom">( + initialData?.retryPolicy?.mode ?? "default", + ); + // 数字输入用本地草稿字符串,blur 时 clamp(与 usageTimeoutInput 同范式)。 + const [streamRetryCountInput, setStreamRetryCountInput] = useState(() => + String( + initialData?.retryPolicy?.mode === "custom" + ? initialData.retryPolicy.maxRetries + : PROVIDER_RETRY_DEFAULT_MAX_RETRIES, + ), + ); const [promptCachingEnabled, setPromptCachingEnabled] = useState( initialData?.promptCachingEnabled ?? (providerType !== "gemini" && providerType !== "xai" && providerType !== "deepseek"), @@ -253,6 +267,30 @@ function useProviderModalController({ providerType, initialData, onSave, onClose setUsageQuery((previous) => ({ ...previous, timeoutSecs: next })); } + function clampStreamRetryCount(raw: number): number { + if (!Number.isFinite(raw)) return PROVIDER_RETRY_DEFAULT_MAX_RETRIES; + return Math.min( + PROVIDER_RETRY_MAX_RETRIES_LIMITS.max, + Math.max(PROVIDER_RETRY_MAX_RETRIES_LIMITS.min, Math.round(raw)), + ); + } + + function commitStreamRetryCountInput() { + setStreamRetryCountInput(String(clampStreamRetryCount(Number(streamRetryCountInput.trim())))); + } + + /** default 态不落字段:与 normalizeProviderRetryPolicy 的持久层形态一致。 */ + function serializeStreamRetryPolicy(): ProviderRetryPolicy | undefined { + if (streamRetryMode === "off") return { mode: "off" }; + if (streamRetryMode === "custom") { + return { + mode: "custom", + maxRetries: clampStreamRetryCount(Number(streamRetryCountInput.trim())), + }; + } + return undefined; + } + const doFetch = useCallback( async (url: string, key: string) => { setFetchingModels(true); @@ -696,6 +734,7 @@ function useProviderModalController({ providerType, initialData, onSave, onClose : undefined, nativeWebSearchEnabled: initialData?.nativeWebSearchEnabled ?? true, useSystemProxy, + retryPolicy: serializeStreamRetryPolicy(), usageQuery: serializeUsageQueryDraft(usageQuery, isGatewayWebui), }); requestClose(); @@ -952,11 +991,16 @@ function useProviderModalController({ providerType, initialData, onSave, onClose setRequestFormat, setShowApiKey, setShowUsageVariableApiKey, + setStreamRetryCountInput, + setStreamRetryMode, setUsageQuery, setUsageTimeoutInput, setUseSystemProxy, showApiKey, showUsageVariableApiKey, + streamRetryCountInput, + streamRetryMode, + commitStreamRetryCountInput, t, toggleModel, toggleModelBulkMode, diff --git a/crates/agent-ui/src/pages/settings/ProviderModalView.tsx b/crates/agent-ui/src/pages/settings/ProviderModalView.tsx index ba4e3f060..ca73ea0b5 100644 --- a/crates/agent-ui/src/pages/settings/ProviderModalView.tsx +++ b/crates/agent-ui/src/pages/settings/ProviderModalView.tsx @@ -165,11 +165,16 @@ export function ProviderModalView({ viewModel }: { viewModel: ProviderModalViewM setRequestFormat, setShowApiKey, setShowUsageVariableApiKey, + setStreamRetryCountInput, + setStreamRetryMode, setUsageQuery, setUsageTimeoutInput, setUseSystemProxy, showApiKey, showUsageVariableApiKey, + streamRetryCountInput, + streamRetryMode, + commitStreamRetryCountInput, t, toggleModel, toggleModelBulkMode, @@ -830,6 +835,78 @@ export function ProviderModalView({ viewModel }: { viewModel: ProviderModalViewM />
+
+
+ + + +
+
{t("settings.providerStreamRetry")}
+
+ {t("settings.providerStreamRetryDesc")} +
+
+
+ {( + [ + ["default", "settings.providerStreamRetryDefault"], + ["off", "settings.providerStreamRetryOff"], + ["custom", "settings.providerStreamRetryCustom"], + ] as const + ).map(([value, labelKey]) => ( + + ))} +
+
+ {streamRetryMode === "custom" ? ( +
+ + setStreamRetryCountInput(event.currentTarget.value)} + onBlur={commitStreamRetryCountInput} + /> + + {t("settings.providerStreamRetryMaxRetriesDesc")} + +
+ ) : null} +
+ {providerType !== "gemini" && providerType !== "xai" && providerType !== "deepseek" ? (