diff --git a/packages/agent-core-v2/docs/en/llm.md b/packages/agent-core-v2/docs/en/llm.md index 24240c7e858..7d339106c3b 100644 --- a/packages/agent-core-v2/docs/en/llm.md +++ b/packages/agent-core-v2/docs/en/llm.md @@ -8,7 +8,7 @@ llm is a standalone LLM request library inside the human layer (`src/human/llm/` 2. **Streaming-native; events are the contract**. The only outward surface is a single, purely serializable event stream (requester level: `llm.sent / streaming.headers / streaming.part / streaming.usage / streaming.finish / streaming.message_id / failed.syntax / failed.remote / done`; the turn level adds `llm.retrying / llm.recovering`, and `llm.sent` carries the most recent recovery record). Streaming and non-streaming are isomorphic (non-streaming also accumulates over the stream, just without deltas). Events are emitted as they arrive — no caching, no fallback. 3. **format masks inter-protocol differences; traits express provider customizations**. format lives at the protocol layer and handles encoding/decoding of requests, responses, errors, usage, and finish. Each protocol owns a typed trait interface (`OpenAITrait` / `OpenAIResponsesTrait` / `AnthropicTrait` / `GoogleGenAITrait`) exposing only the customization points that protocol actually consumes — a hook a protocol ignores is unrepresentable, never silently dead. format and trait never import each other: both speak only the neutral wire/chunk types in the protocol's `contract.ts`. The requester is the composition root — `generate` runs a fixed per-protocol pipeline (`planOpenAIRequest` and friends) that alternates pure format stages (lower → assemble → encode → stream parser) with trait hooks (cacheKey/thinking → convertMessage → mergeHistory → convertTool → buildParams → extractUsage), so customization is explicit data flow instead of a closure captured inside format. Endpoint/env resolution and default headers form the provider `connection`, error classification is a requester option, and model capability is a provider-variant field — none of them are format business. Each base's public seam is contract + trait + requester; format, lower, and patterns are internal to the requester pipeline — only bases code and tests may import them (lint-enforced). Protocol differences must not leak into the turn or into requester decorators. 4. **Two-layer error model**. Internally, code throws the SDK's native errors; local request validation throws the shared `SyntaxRequestFormatError` (`llm/syntax-errors.ts`), which the requester converts uniformly via `toLlmSyntaxErrorMessage`, with no intermediate layer. Externally there are only `llm.failed.syntax` (local message syntax errors, never retried) and `llm.failed.remote` (remote streaming errors, subdivided into connection / timeout / rate_limit / quota_exhausted / context_overflow / request_structure, etc.), converted by format at the boundary. -5. **Stateless core + turn-driven orchestration**. `generate(config, content, control)` is a stateless function; errors are delivered via onEvent, never thrown. The turn machine invokes the request actor (`createRequestActor`) directly: the actor wraps a single request (messageResolvers, abort scope, event sendBack), and the turn drives retry and recovery through the pure policy functions in retry.ts / recovery.ts: recovery re-sends with replacement messages produced by the pure `propose` function (attempt resets to 1), retry backs off in the `retrying` state (honoring Retry-After), and the turn emits `llm.recovering / llm.retrying` for each. Empty response is judged by `withEmptyResponseGuard` at the requester boundary and raised as `llm.failed.remote`, entering the same retry path. Abort is carried by an AbortController owned by the turn: the controller is passed into the request actor via `LlmInput.signal`, and the turn aborts it directly on `turn.abort`, with the request ending as `llm.failed.remote`; the request actor neither creates its own controller nor touches any signal on teardown, so a finished request can never abort a shared signal. The accumulator is held by the turn and fed by the event stream; on `llm.retrying / llm.recovering` the turn rolls it back and recreates it, so every attempt accumulates from zero while as much interrupted state as possible is preserved (the turn finishes the complete message out of the accumulator at `llm.done`). +5. **Stateless core + turn-driven orchestration**. `generate(config, content, control)` is a stateless function; errors are delivered via onEvent, never thrown. The turn machine invokes the request actor (`createRequestActor`) directly: the actor wraps a single request (messageResolvers, abort scope, event sendBack), and the turn drives retry and recovery through the pure policy functions in retry.ts / recovery.ts: recovery is a strategy chain (credential refresh via `credentialsRecovery`, then replacement-message strategies such as media degradation) whose pure `propose` yields a recovery action — the turn executes the action (invalidating credentials and/or swapping messages) and re-enters `thinking` with attempt reset to 1; retry backs off in the `retrying` state (honoring Retry-After), and the turn emits `llm.recovering / llm.retrying` for each. Empty response is judged by the turn at `llm.done` via the pure `emptyResponseError` and re-raised as `llm.failed.remote`, entering the same failure cascade. Abort is carried by an AbortController owned by the turn: the controller is passed into the request actor via `LlmInput.signal`, and the turn aborts it directly on `turn.abort`, with the request ending as `llm.failed.remote`; the request actor neither creates its own controller nor touches any signal on teardown, so a finished request can never abort a shared signal. The accumulator is held by the turn and fed by the event stream; on `llm.retrying / llm.recovering` the turn rolls it back and recreates it, so every attempt accumulates from zero while as much interrupted state as possible is preserved (the turn finishes the complete message out of the accumulator at `llm.done`). 6. **No silent fallback**. Configuration is taken exactly as given. For beta features, thinking, empty response, and similar scenarios, define explicit error conditions first, fail at request time, and guide the user to fix the configuration — never fall back silently. 7. **Every variable capability is a contribution point**. Providers, media upload/degradation, usage, traceId, and error recovery (compaction / media degradation) all plug in through extension points; the llm core contains none of these concepts. 8. **Data is data**. A model is pure, function-free data (endpoint url + model uniquely identifies a model), serializable and directly usable as generate input. The catalog is a derived `provider -> models` cache; the dependency direction only goes from models-dev into llm internals, never the reverse. @@ -33,11 +33,18 @@ llm/ │ ├── requester/ │ ├── requester.ts LlmRequester.generate(config, content, control); -│ │ ExtraParams typed per protocol {openai?, responses?, anthropic?, googleGenai?} +│ │ ExtraParams typed per protocol {openai?, responses?, anthropic?, googleGenai?}; +│ │ LlmRequestConfig.credentials: credential contribution point +│ │ (resolve/canRecover/invalidate), resolved per attempt by the caller; +│ │ factories and the credentialsRecovery strategy live in human/credentials +│ │ (staticCredentials / oauthCredentials; kimiOAuthCredentialProvider adapts +│ │ Kimi OAuth tokens); the runWithCredentialRecovery / +│ │ streamWithCredentialRecovery executors for direct callers live in +│ │ llm-adapter/model/credential-recovery │ ├── actor.ts request actor: a fromCallback wrapping a single request │ │ (messageResolvers, abort scope, event sendBack); invoked by the turn │ ├── retry.ts / recovery.ts pure retry/recovery policy functions (driven by the turn machine; propose is pure) -│ ├── empty-response.ts withEmptyResponseGuard: judges empty responses at finish and raises llm.failed.remote +│ ├── empty-response.ts emptyResponseError: pure empty-response judgment; the turn raises it as llm.failed.remote at llm.done │ └── bases/ four protocol bases: openai / openai-responses / anthropic / google-genai │ each with contract / format / lower / patterns / capability / extra-params / trait / requester │ (public seam: contract / trait / requester; format / lower / patterns stay internal) @@ -53,7 +60,7 @@ llm/ └── media/ media contribution points: cache / degrade / ref / resolver / store / upload ``` -Request lifecycle: `generate` receives (config, content, control) → the requester's `plan*` function composes pure format stages with trait hooks into protocol requestParams (format lowers the generic Message[] through the Pattern Rewriter; trait adjusts kwargs, converted messages, history, tools, and final params in between) → internalGenerate calls the official SDK → streaming chunks are converted by the stateless parser callbacks into `llm.streaming.part / streaming.usage / streaming.finish / streaming.message_id` events → errors are converted by format into `llm.failed.*`; on success the requester emits `llm.done`, on failure it ends with `llm.failed.syntax / llm.failed.remote` and never emits `llm.done`. At `llm.done` the turn judges empty responses via `emptyResponseError` and re-raises them as `llm.failed.remote`; the turn machine first tries recovery on `llm.failed.remote` (replacement messages from the pure `propose` function, emitting `llm.recovering`), then retries with backoff (honoring Retry-After, emitting `llm.retrying`), and only fails the turn once attempts are exhausted. The turn holds the HistoryAccumulator, fed by the event stream, rolls it back and recreates it on `llm.retrying / llm.recovering`, and finishes the complete message at `llm.done`; usage accounting, tracing, compaction, and media degradation all attach to the event stream as plugins/contribution points. +Request lifecycle: `generate` receives (config, content, control) → the caller resolves `config.credentials` into a fully-credentialed model before each attempt (the request actor on the machine path), so requests always carry fresh credentials and a credential-refresh recovery (recoverable 401 → `credentials.invalidate()`, emitted as `llm.recovering` with strategy `credentials`) naturally re-resolves on the re-send (direct callers outside the state machines — ping, generate, full compaction, media upload — share the same single-retry recovery through `runWithCredentialRecovery` / `streamWithCredentialRecovery`) → the requester's `plan*` function composes pure format stages with trait hooks into protocol requestParams (format lowers the generic Message[] through the Pattern Rewriter; trait adjusts kwargs, converted messages, history, tools, and final params in between) → internalGenerate calls the official SDK → streaming chunks are converted by the stateless parser callbacks into `llm.streaming.part / streaming.usage / streaming.finish / streaming.message_id` events → errors are converted by format into `llm.failed.*`; on success the requester emits `llm.done`, on failure it ends with `llm.failed.syntax / llm.failed.remote` and never emits `llm.done`. At `llm.done` the turn judges empty responses via `emptyResponseError` and re-raises them as `llm.failed.remote`; the turn machine first tries recovery on `llm.failed.remote` (the strategy chain — credential refresh on a recoverable 401, then replacement-message strategies — via the pure `propose`, emitting `llm.recovering`), then retries with backoff (honoring Retry-After, emitting `llm.retrying`), and only fails the turn once attempts are exhausted. The turn holds the HistoryAccumulator, fed by the event stream, rolls it back and recreates it on `llm.retrying / llm.recovering`, and finishes the complete message at `llm.done`; usage accounting, tracing, compaction, and media degradation all attach to the event stream as plugins/contribution points. ## Rejected Schemes (do not reintroduce) diff --git a/packages/agent-core-v2/docs/zh/llm.md b/packages/agent-core-v2/docs/zh/llm.md index 344e3acf71c..0baf49e7aab 100644 --- a/packages/agent-core-v2/docs/zh/llm.md +++ b/packages/agent-core-v2/docs/zh/llm.md @@ -8,7 +8,7 @@ llm 是 human 层内一个独立的 LLM 请求库（`src/human/llm/`），提供 2. **流式原生、事件即契约**。对外只暴露一条纯可序列化的事件流（requester 层：`llm.sent / streaming.headers / streaming.part / streaming.usage / streaming.finish / streaming.message_id / failed.syntax / failed.remote / done`；turn 层补充 `llm.retrying / llm.recovering`，`llm.sent` 携带最近一次 recovery 记录），流式与非流式同构（非流式也走流式累积，只是不发 delta）；事件收到即发，不缓存、不兜底。 3. **format 屏蔽协议间差异，trait 表达 provider 定制**。format 位于 protocol 层，负责请求、响应、错误、usage 和 finish 的编解码。每种协议拥有自己的类型化 trait 接口（`OpenAITrait` / `OpenAIResponsesTrait` / `AnthropicTrait` / `GoogleGenAITrait`），只暴露该协议实际消费的定制点——协议不支持的 hook 在类型上无法表达，而不是配了却静默无效。format 与 trait 互不 import：双方只共享协议 `contract.ts` 里的中立 wire/chunk 类型。requester 是组合根——`generate` 执行每个协议固定的流水线（`planOpenAIRequest` 等），交替调用纯 format 阶段（lower → assemble → encode → stream parser）与 trait hooks（cacheKey/thinking → convertMessage → mergeHistory → convertTool → buildParams → extractUsage），定制逻辑是显式的数据流，而不是捕获在 format 闭包里。endpoint/环境变量解析与默认 headers 属于 provider `connection`，错误归类是 requester 选项，模型能力是 provider variant 字段——都不是 format 的职责。每个 base 的公开接缝是 contract + trait + requester；format、lower、patterns 是 requester 流水线的内部模块——只有 bases 内代码和测试可以 import（lint 强制）。协议差异不允许泄漏到 turn 或 requester 的装饰层。 4. **错误两层模型**。内部 throw SDK 原生错误；本地请求校验抛共享的 `SyntaxRequestFormatError`（`llm/syntax-errors.ts`），由 requester 经 `toLlmSyntaxErrorMessage` 统一转换，不加中间层。对外只有 `llm.failed.syntax`（本地消息语法错误，不重试）与 `llm.failed.remote`（远程流式错误，细分为 connection/timeout/rate_limit/quota_exhausted/context_overflow/request_structure 等），由 format 在边界完成转换。 -5. **无状态内核 + turn 驱动的编排**。`generate(config, content, control)` 是无状态函数，错误走 onEvent 不 throw；turn machine 直接 invoke 请求 actor（`createRequestActor`）：actor 包装单次请求（messageResolvers、abort 作用域、事件 sendBack），turn 借助 retry.ts / recovery.ts 的纯策略函数驱动重试与 recovery：recovery 由纯函数 `propose` 产出替换消息直接重发（attempt 重置为 1），重试走 `retrying` 状态的 backoff（尊重 Retry-After），两者分别由 turn 对外补发 `llm.recovering / llm.retrying` 事件；empty response 由 `withEmptyResponseGuard` 在 requester 边界判定并转为 `llm.failed.remote`，进入同一重试路径；abort 由 turn 持有的 AbortController 承载：controller 经 `LlmInput.signal` 传入 request actor，turn 在 `turn.abort` 时直接 abort 它，请求随即以 `llm.failed.remote` 收尾；request actor 不自建 controller、回收时不触碰任何 signal，正常完成的请求绝不可能误 abort 共享 signal。累积器由 turn 持有并随事件流喂入，在 `llm.retrying / llm.recovering` 时 rollback 并重建，每次 attempt 从零累积，从而尽可能保留中断现场（turn 在 `llm.done` 时从累加器 finish 出完整消息）。 +5. **无状态内核 + turn 驱动的编排**。`generate(config, content, control)` 是无状态函数，错误走 onEvent 不 throw；turn machine 直接 invoke 请求 actor（`createRequestActor`）：actor 包装单次请求（messageResolvers、abort 作用域、事件 sendBack），turn 借助 retry.ts / recovery.ts 的纯策略函数驱动重试与 recovery：recovery 是一条策略链（`credentialsRecovery` 的凭证刷新在前，媒体降级等替换消息策略在后），由纯函数 `propose` 产出恢复动作，turn 负责解释执行（invalidate 凭证和/或替换消息）并重进 `thinking`（attempt 重置为 1），重试走 `retrying` 状态的 backoff（尊重 Retry-After），两者分别由 turn 对外补发 `llm.recovering / llm.retrying` 事件；empty response 由 turn 在 `llm.done` 时经纯函数 `emptyResponseError` 判定并重新转为 `llm.failed.remote`，进入同一失败级联；abort 由 turn 持有的 AbortController 承载：controller 经 `LlmInput.signal` 传入 request actor，turn 在 `turn.abort` 时直接 abort 它，请求随即以 `llm.failed.remote` 收尾；request actor 不自建 controller、回收时不触碰任何 signal，正常完成的请求绝不可能误 abort 共享 signal。累积器由 turn 持有并随事件流喂入，在 `llm.retrying / llm.recovering` 时 rollback 并重建，每次 attempt 从零累积，从而尽可能保留中断现场（turn 在 `llm.done` 时从累加器 finish 出完整消息）。 6. **不兜底**。配置是什么就是什么；beta 特性、thinking、empty response 等场景先定义明确报错条件，在请求阶段报错并引导用户修正，而不是静默兜底。 7. **一切可变能力都是贡献点**。provider、媒体上传/降级、usage、traceId、错误恢复（compaction/媒体降级）都通过扩展点接入，llm 内核不含这些概念。 8. **数据即数据**。model 是无函数的纯数据（endpoint url + model 唯一标识一个模型），可序列化、可直接作为 generate 输入；catalog 是 `provider -> models` 的派生缓存，依赖方向只能从 models-dev 指向 llm 内部，不能反向依赖。 @@ -33,11 +33,18 @@ llm/ │ ├── requester/ │ ├── requester.ts LlmRequester.generate(config, content, control)； -│ │ ExtraParams 按协议带类型 {openai?, responses?, anthropic?, googleGenai?} +│ │ ExtraParams 按协议带类型 {openai?, responses?, anthropic?, googleGenai?}； +│ │ LlmRequestConfig.credentials：凭证贡献点 +│ │ （resolve/canRecover/invalidate），由调用方在每次 attempt 前解析； +│ │ 工厂与 credentialsRecovery 策略位于 human/credentials +│ │ （staticCredentials / oauthCredentials；kimiOAuthCredentialProvider +│ │ 适配 Kimi OAuth token）；供 direct 调用方使用的 +│ │ runWithCredentialRecovery / streamWithCredentialRecovery 执行器 +│ │ 位于 llm-adapter/model/credential-recovery │ ├── actor.ts 请求 actor：包装单次请求的 fromCallback │ │ （messageResolvers、abort 作用域、事件 sendBack），由 turn invoke │ ├── retry.ts / recovery.ts 重试/恢复策略纯函数（由 turn machine 驱动；propose 为纯函数） -│ ├── empty-response.ts withEmptyResponseGuard：finish 时判定空响应并转为 llm.failed.remote +│ ├── empty-response.ts emptyResponseError：空响应判定纯函数，由 turn 在 llm.done 时转为 llm.failed.remote │ └── bases/ 四个协议基座：openai / openai-responses / anthropic / google-genai │ 各自含 contract / format / lower / patterns / capability / extra-params / trait / requester │ （公开接缝：contract / trait / requester；format / lower / patterns 保持内部） @@ -53,7 +60,7 @@ llm/ └── media/ 媒体贡献点：cache / degrade / ref / resolver / store / upload ``` -请求生命周期：`generate` 收到 (config, content, control) → requester 的 `plan*` 函数将纯 format 阶段与 trait hooks 组合为协议 requestParams（format 将通用 Message[] 经 Pattern Rewriter 降低，trait 在其间调整 kwargs、转换消息、合并历史、转换 tools 并收尾 params）→ internalGenerate 调用官方 SDK → 流式 chunk 经无状态 parser 回调转换为 `llm.streaming.part / streaming.usage / streaming.finish / streaming.message_id` 事件 → 错误由 format 转换为 `llm.failed.*`；成功时 requester 发出 `llm.done`，失败时以 `llm.failed.syntax / llm.failed.remote` 收尾、不再发 `llm.done`。turn 在 `llm.done` 时经 `emptyResponseError` 判定空响应并转为 `llm.failed.remote`；turn machine 对 `llm.failed.remote` 先尝试 recovery（纯函数 `propose` 产出替换消息，发 `llm.recovering`），再按策略 backoff 重试（尊重 Retry-After，发 `llm.retrying`），耗尽后才将 turn 置为失败。turn 持有 HistoryAccumulator 随事件流累积，在 `llm.retrying / llm.recovering` 时 rollback 并重建累加器，`llm.done` 时 finish 出完整消息；usage 统计、trace、compaction、媒体降级均以插件/贡献点身份挂接在事件流上。 +请求生命周期：`generate` 收到 (config, content, control) → 调用方在每次 attempt 前把 `config.credentials` 解析成带完整凭证的 model（machine 路径由 request actor 完成），请求因此始终携带新鲜凭证，而凭证刷新恢复（可恢复的 401 → `credentials.invalidate()`，以 `llm.recovering`（strategy 为 `credentials`）发出）在重发时自然重新解析（不经状态机的 direct 调用方——ping、generate、full compaction、媒体上传——通过 `runWithCredentialRecovery` / `streamWithCredentialRecovery` 共享同一套单次重试恢复） → requester 的 `plan*` 函数将纯 format 阶段与 trait hooks 组合为协议 requestParams（format 将通用 Message[] 经 Pattern Rewriter 降低，trait 在其间调整 kwargs、转换消息、合并历史、转换 tools 并收尾 params） → internalGenerate 调用官方 SDK → 流式 chunk 经无状态 parser 回调转换为 `llm.streaming.part / streaming.usage / streaming.finish / streaming.message_id` 事件 → 错误由 format 转换为 `llm.failed.*`；成功时 requester 发出 `llm.done`，失败时以 `llm.failed.syntax / llm.failed.remote` 收尾、不再发 `llm.done`。turn 在 `llm.done` 时经 `emptyResponseError` 判定空响应并重新转为 `llm.failed.remote`；turn machine 对 `llm.failed.remote` 先尝试恢复（策略链——可恢复 401 的凭证刷新在前、替换消息策略在后——经纯函数 `propose` 产出，发 `llm.recovering`），再按策略 backoff 重试（尊重 Retry-After，发 `llm.retrying`），耗尽后才将 turn 置为失败。turn 持有 HistoryAccumulator 随事件流累积，在 `llm.retrying / llm.recovering` 时 rollback 并重建累加器，`llm.done` 时 finish 出完整消息；usage 统计、trace、compaction、媒体降级均以插件/贡献点身份挂接在事件流上。 ## 已被否决的方案（不要再引入） diff --git a/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts b/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts index 196ecbf641c..ce4d3cd862b 100644 --- a/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts +++ b/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts @@ -11,6 +11,7 @@ import { ISessionTokenCountingService } from '#/session/tokenCounting/sessionTok import { IAgentLLMRequesterService, type AgentLLMRequestFinish } from '#/agent/llmRequester/llmRequester'; import type { LLMRequestTrace } from '#/llm-adapter/contract/request-trace'; import { retryBackoffDelays, sleepForRetry } from '#/_base/utils/retry'; +import { runWithCredentialRecovery } from '#/llm-adapter/model/credential-recovery'; import { IAgentLoopService, type LoopErrorContext } from '#/agent/loop/loop'; import { TurnStarted } from '#/agent/loop/turnEvents'; import { TurnEnded } from '#/agent/loop/turnOps'; @@ -652,22 +653,30 @@ export class AgentFullCompactionService extends Service implements IAgentFullCom const estimatedCompactionRequestTokens = this.requestTokens(messages); try { - const request = this.llmRequester.start( - { - messages, - maxOutputSize: compactionMaxOutputSize, - source: { - type: 'operation', - turnId: active.originTurnId, - requestKind: 'full_compaction', - logFields: { droppedCount }, + const runRequest = async () => { + const request = this.llmRequester.start( + { + messages, + maxOutputSize: compactionMaxOutputSize, + source: { + type: 'operation', + turnId: active.originTurnId, + requestKind: 'full_compaction', + logFields: { droppedCount }, + }, }, - }, - undefined, + undefined, + signal, + ); + active.trace = request.trace; + return request.result; + }; + const result = await runWithCredentialRecovery( + this.llmRequester.currentCredentials(), + runRequest, signal, ); - active.trace = request.trace; - attempt = collectSummary(await request.result); + attempt = collectSummary(result); break; } catch (error) { const isContextOverflow = this.shouldRecoverFromContextOverflow( diff --git a/packages/agent-core-v2/src/agent/llmRequester/llmRequester.ts b/packages/agent-core-v2/src/agent/llmRequester/llmRequester.ts index ac414f36975..95a5202294b 100644 --- a/packages/agent-core-v2/src/agent/llmRequester/llmRequester.ts +++ b/packages/agent-core-v2/src/agent/llmRequester/llmRequester.ts @@ -1,5 +1,6 @@ import { createDecorator } from '#/_base/di/instantiation'; import type { FinishReason } from '#human/llm/finish-reason'; +import type { LlmCredentialProvider } from '#human/llm/requester/requester'; import type { ThinkingEffort } from '#human/llm/thinking'; import type { Message } from '#/llm-adapter/contract/message'; import type { StreamedMessagePart, ToolDescription as Tool } from '#human/llm/message'; @@ -59,6 +60,10 @@ export interface IAgentLLMRequesterService { prepareTurnConfig(turnId: number): PreparedTurnRequestConfig | undefined; + currentCredentials(): LlmCredentialProvider | undefined; + + credentialsForTurn(turnId: number): LlmCredentialProvider | undefined; + request( overrides?: AgentLLMRequestOverrides, onPart?: AgentLLMRequestPartHandler, diff --git a/packages/agent-core-v2/src/agent/llmRequester/llmRequesterService.ts b/packages/agent-core-v2/src/agent/llmRequester/llmRequesterService.ts index 95c4567e267..c6a1cbacbc4 100644 --- a/packages/agent-core-v2/src/agent/llmRequester/llmRequesterService.ts +++ b/packages/agent-core-v2/src/agent/llmRequester/llmRequesterService.ts @@ -27,6 +27,7 @@ import { } from '#/llm-adapter/contract/errors'; import type { Message } from '#/llm-adapter/contract/message'; import { type ThinkingEffort } from '#human/llm/thinking'; +import type { LlmCredentialProvider } from '#human/llm/requester/requester'; import { isToolCall, type StreamedMessagePart, type ToolDescription as Tool } from '#human/llm/message'; import { emptyUsage, inputTotal, type TokenUsage } from '#human/llm/usage'; import { ILogService, type LogContext } from '#/_base/log/log'; @@ -210,6 +211,17 @@ export class AgentLLMRequesterService implements IAgentLLMRequesterService { return { thinkingEffort: config.resolved.thinkingLevel }; } + currentCredentials(): LlmCredentialProvider | undefined { + if (!this.profile.hasProvider()) return undefined; + return this.modelCatalog.get(this.profile.resolveModelContext().modelAlias).credentials; + } + + credentialsForTurn(turnId: number): LlmCredentialProvider | undefined { + if (!this.profile.hasProvider()) return undefined; + const resolved = this.turnConfigs.get(turnId)?.resolved ?? this.profile.resolveModelContext(); + return this.modelCatalog.get(resolved.modelAlias).credentials; + } + async request( overrides: AgentLLMRequestOverrides = {}, onPart: AgentLLMRequestPartHandler = noopOnPart, diff --git a/packages/agent-core-v2/src/agent/loop/loopService.ts b/packages/agent-core-v2/src/agent/loop/loopService.ts index d356783ce01..9aaa6c44785 100644 --- a/packages/agent-core-v2/src/agent/loop/loopService.ts +++ b/packages/agent-core-v2/src/agent/loop/loopService.ts @@ -430,6 +430,69 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { this.maybeSettle(); } + private settleUnboundReservation( + pending: { readonly id: number; readonly queueItemId?: string }, + outcome: { readonly outcome: MachineTurnOutcome; readonly error?: unknown }, + ): void { + const active = this.active; + if (active !== undefined) { + active.afterChain = active.afterChain.then(() => { + this.settleUnboundReservation(pending, outcome); + }); + return; + } + if (pending.queueItemId === undefined) { + const seeded = this.nudges.slice(this.nudgeCursor).find( + (nudge) => !nudge.dropped && nudge.contextMessage !== undefined && nudge.contextMessage.content.length > 0, + ); + if (seeded === undefined) { + this.consumeDrainedNudges(); + return; + } + this.beginActiveTurn( + this.createSeededReservation(seeded.contextMessage as ContextMessage), + pending.id, + ); + const seededTurn = this.active; + if (seededTurn === undefined) return; + this.mirrorConsumedNudges(seededTurn); + this.endPreGateTurn(seededTurn, outcome); + return; + } + const index = this.reservations.findIndex( + (entry) => entry.machineQueueId === pending.queueItemId, + ); + if (index < 0) return; + const [reservation] = this.reservations.splice(index, 1); + if (reservation === undefined || reservation.cancelled) return; + this.beginActiveTurn(reservation, pending.id); + reservation.onMaterialize?.(); + this.materializeMessage(reservation.message); + const turn = this.active; + if (turn === undefined) return; + this.endPreGateTurn(turn, outcome); + } + + private endPreGateTurn( + turn: ActiveTurn, + outcome: { readonly outcome: MachineTurnOutcome; readonly error?: unknown }, + ): void { + if (outcome.outcome === 'aborted') { + const reason = turn.controller.signal.aborted + ? turn.controller.signal.reason + : abortError('Turn aborted'); + turn.controller.abort(reason); + turn.afterChain = turn.afterChain.then(() => + this.endTurn(turn, { type: 'cancelled', steps: 0, reason }), + ); + return; + } + const error = outcome.error ?? new Error2(ErrorCodes.INTERNAL, 'Turn ended before first step'); + turn.afterChain = turn.afterChain.then(() => + this.endTurn(turn, { type: 'failed', steps: 0, error }), + ); + } + hasPendingRequests(): boolean { return ( this.reservations.some((reservation) => !reservation.cancelled) || @@ -624,7 +687,14 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { this.machineTurnSuppressed = true; return false; } - const message = seeded.contextMessage as ContextMessage; + this.beginActiveTurn( + this.createSeededReservation(seeded.contextMessage as ContextMessage), + pending.id, + ); + return true; + } + + private createSeededReservation(message: ContextMessage): TurnReservation { const controller = new AbortController(); const ready = createControlledPromise(); const result = createControlledPromise(); @@ -650,8 +720,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { result, turn, }; - this.beginActiveTurn(reservation, pending.id); - return true; + return reservation; } private beginActiveTurn(reservation: TurnReservation, id: number): void { @@ -732,7 +801,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { this.context.append(message); } - private mirrorConsumedNudges(turn: ActiveTurn): { readonly live: number; readonly bypass: boolean } { + private consumeDrainedNudges(): { readonly live: number; readonly bypass: boolean } { const engine = this.engine; if (engine === undefined) return { live: 0, bypass: false }; const notificationCount = engine.snapshot().notificationCount; @@ -752,10 +821,15 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { } nudge.onConsume?.(); } - turn.nudgeCursor = this.nudgeCursor; return { live, bypass }; } + private mirrorConsumedNudges(turn: ActiveTurn): { readonly live: number; readonly bypass: boolean } { + const consumed = this.consumeDrainedNudges(); + turn.nudgeCursor = this.nudgeCursor; + return consumed; + } + private projectMachineEvent(event: MachineEngineEvent): void { switch (event.type) { case 'turnStarted': { @@ -772,8 +846,10 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { return; } if (this.pendingMachineTurn !== undefined) { + const pending = this.pendingMachineTurn; this.pendingMachineTurn = undefined; this.machineTurnSuppressed = false; + this.settleUnboundReservation(pending, outcome); this.maybeSettle(); return; } @@ -948,6 +1024,16 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { this.machineEngine().abort(); return; } + case 'recovering': { + const turn = this.active; + const step = turn?.current; + if (turn === undefined) return; + if (step !== undefined) { + this.closeFailedMachineStep(turn, step, 'error'); + } + turn.current = undefined; + return; + } case 'retrying': { const turn = this.active; const step = turn?.current; diff --git a/packages/agent-core-v2/src/agent/loop/machine/engine.ts b/packages/agent-core-v2/src/agent/loop/machine/engine.ts index a4f021392ec..d0df250cc06 100644 --- a/packages/agent-core-v2/src/agent/loop/machine/engine.ts +++ b/packages/agent-core-v2/src/agent/loop/machine/engine.ts @@ -15,6 +15,7 @@ import type { FinishInfo } from '#human/llm/finish-reason'; import type { StreamedMessagePart, UserMessage } from '#human/llm/message'; import type { LlmModel } from '#human/llm/model'; import type { LlmRecovery, LlmRecoveryRecord } from '#human/llm/requester/recovery'; +import type { LlmCredentialProvider } from '#human/llm/requester/requester'; import { resolveMaxAttempts } from '#human/llm/requester/retry'; import type { ToolResult as MachineToolResult, ToolUpdate } from '#human/tool/executor'; import type { TokenUsage } from '#human/llm/usage'; @@ -269,6 +270,17 @@ export function createMachineEngine(options: CreateMachineEngineOptions): Machin publish({ type: 'toolBatchFailed', error }); }, }); + const current = (): LlmCredentialProvider | undefined => { + const source = options.source?.(); + return source?.type === 'turn' + ? options.llmRequester.credentialsForTurn(source.turnId) + : options.llmRequester.currentCredentials(); + }; + const credentials: LlmCredentialProvider = { + resolve: () => current()?.resolve(), + canRecover: (error) => current()?.canRecover?.(error) === true, + invalidate: () => current()?.invalidate?.(), + }; const journal = memoryJournal(); const initialTurnId = options.initialTurnId ?? 0; if (initialTurnId > 0) { @@ -288,7 +300,12 @@ export function createMachineEngine(options: CreateMachineEngineOptions): Machin }), abortTimeoutMs: options.abortTimeoutMs, }), - { input: { request: { model: options.model, systemPrompt: options.systemPrompt }, store } }, + { + input: { + request: { model: options.model, systemPrompt: options.systemPrompt, credentials }, + store, + }, + }, ); const subscriptions: Subscription[] = [ actor.on('turn.started', (event) => { diff --git a/packages/agent-core-v2/src/agent/media/mediaResolverService.ts b/packages/agent-core-v2/src/agent/media/mediaResolverService.ts index b1781099778..c3f3ac767ea 100644 --- a/packages/agent-core-v2/src/agent/media/mediaResolverService.ts +++ b/packages/agent-core-v2/src/agent/media/mediaResolverService.ts @@ -9,6 +9,7 @@ import { ITelemetryService } from '#/app/telemetry/telemetry'; import type { Message } from '#/llm-adapter/contract/message'; import type { ContentPart } from '#human/llm/message'; import type { ModelRequester } from '#/llm-adapter/model/model-requester'; +import { runWithCredentialRecovery } from '#/llm-adapter/model/credential-recovery'; import { IBlobStore } from '#/persistence/interface/blobStore'; import { detectFileType, MEDIA_SNIFF_BYTES } from './file-type'; @@ -271,7 +272,11 @@ export class AgentMediaResolverService implements IAgentMediaResolverService { } try { - const uploaded = await uploader({ data: bytes, mimeType, filename }, { signal }); + const uploaded = await runWithCredentialRecovery( + requester.model.credentials, + () => uploader({ data: bytes, mimeType, filename }, { signal }), + signal, + ); const llmFileId = uploaded.videoUrl.id ?? msFileIdFromUrl(uploaded.videoUrl.url); if (llmFileId !== undefined) await this.writeCachedUpload(cacheKey, llmFileId); return { part: uploaded, memoize: true }; diff --git a/packages/agent-core-v2/src/agent/media/mediaToolsRegistrar.ts b/packages/agent-core-v2/src/agent/media/mediaToolsRegistrar.ts index b076a36bb07..7d13829d016 100644 --- a/packages/agent-core-v2/src/agent/media/mediaToolsRegistrar.ts +++ b/packages/agent-core-v2/src/agent/media/mediaToolsRegistrar.ts @@ -9,6 +9,7 @@ import { AgentStatusUpdated } from '#/agent/usage/usageEvents'; import { ITelemetryService } from '#/app/telemetry/telemetry'; import { IModelCatalog, type Model } from '#/llm-adapter/model/catalog'; import { type ModelRequester } from '#/llm-adapter/model/model-requester'; +import { runWithCredentialRecovery } from '#/llm-adapter/model/credential-recovery'; import { IAgentRuntimeService } from '#/agent/runtimeBinding/agentRuntime'; import { ISessionSkillCatalog } from '#/features/skill/session/skillCatalog'; import { ISessionWorkspaceContext } from '#/session/workspaceContext/workspaceContext'; @@ -117,6 +118,14 @@ export class AgentMediaToolsRegistrar extends Service implements IAgentMediaTool requester = undefined; } } + const uploader = createVideoUploader(requester, { + client: this.telemetry, + props: { + model: modelAlias, + provider_type: model?.providerType ?? model?.protocol, + protocol: model?.protocol, + }, + }); this.registration = registerMediaTools(this.toolRegistry, { attachmentStore: this.attachmentStore, runtime, @@ -133,14 +142,15 @@ export class AgentMediaToolsRegistrar extends Service implements IAgentMediaTool }, }, capabilities, - videoUploader: createVideoUploader(requester, { - client: this.telemetry, - props: { - model: modelAlias, - provider_type: model?.providerType ?? model?.protocol, - protocol: model?.protocol, - }, - }), + videoUploader: + uploader === undefined || requester === undefined + ? undefined + : (input, options) => + runWithCredentialRecovery( + requester.model.credentials, + () => uploader(input, options), + options?.signal, + ), inlineVideoSupported: model?.protocol !== 'openai' && model?.protocol !== 'openai_responses', providerType: model?.providerType, telemetry: this.telemetry, diff --git a/packages/agent-core-v2/src/human/agent/turn.ts b/packages/agent-core-v2/src/human/agent/turn.ts index 8b8ed8088e4..8566dd046d5 100644 --- a/packages/agent-core-v2/src/human/agent/turn.ts +++ b/packages/agent-core-v2/src/human/agent/turn.ts @@ -1,8 +1,9 @@ import { assign, raise, setup } from '#/xstate2'; import { emptyResponseError } from '#/llm/empty-response'; -import type { LlmErrorMessage, LlmRemoteErrorMessage } from '#/llm/errors'; +import type { LlmErrorMessage } from '#/llm/errors'; import { NO_FINISH, type FinishInfo } from '#/llm/finish-reason'; +import { credentialsRecovery } from '#/credentials/credentials'; import { createMessageAccumulator, createToolMessage, @@ -187,7 +188,12 @@ export type TurnEvent = | LlmEvent | TurnToolEvent | { type: 'turn.notify'; messages: HistoryMessage[] } - | { type: 'turn.abort' }; + | { type: 'turn.abort' } + | { + type: 'turn.failure.triaged'; + cause: Extract; + proposal?: LlmRecoveryProposal & LlmRecoveryRecord; + }; export type TurnLlmEvent = | Exclude @@ -217,7 +223,7 @@ export interface TurnMachineContext { attempt: number; delayMs: number; appliedRecoveries: LlmRecoveryRecord[]; - lastError?: LlmRemoteErrorMessage; + recoveryMessages?: readonly Message[]; outcome?: 'done' | 'failed' | 'aborted'; error?: unknown; } @@ -280,30 +286,23 @@ function baseMessages(context: TurnMachineContext): readonly Message[] { return toInputMessages([...context.input.history, ...context.produced]); } -function attemptMessages( - context: TurnMachineContext, - recovery: LlmRecovery | undefined, -): readonly Message[] { - const base = baseMessages(context); - const lastError = context.lastError; - if (lastError === undefined || context.appliedRecoveries.length === 0) return base; - return ( - proposeRecovery(recovery, { - error: lastError, - messages: base, - applied: context.appliedRecoveries.slice(0, -1), - })?.messages ?? base - ); +function attemptMessages(context: TurnMachineContext): readonly Message[] { + return context.recoveryMessages ?? baseMessages(context); } function proposeRecovery( recovery: LlmRecovery | undefined, ctx: LlmRecoveryContext, ): (LlmRecoveryProposal & LlmRecoveryRecord) | undefined { + const credentialProposal = credentialsRecovery.propose(ctx); + if (credentialProposal !== undefined) { + return { strategy: credentialsRecovery.id, ...credentialProposal }; + } if (recovery === undefined) return undefined; const proposal = recovery.propose(ctx); - if (proposal === undefined || proposal.messages === ctx.messages) return undefined; - return { strategy: recovery.id, action: proposal.action, messages: proposal.messages }; + if (proposal === undefined) return undefined; + if (proposal.messages !== undefined && proposal.messages === ctx.messages) return undefined; + return { strategy: recovery.id, ...proposal }; } function llmRetryingEvent( @@ -448,7 +447,7 @@ export function createTurnMachine( config: context.input.request, signal: context.llmScope.signal, content: { - messages: attemptMessages(context, recovery), + messages: attemptMessages(context), usedContextTokens: estimateUsedContextTokens(entries, { systemPrompt: context.input.request.systemPrompt, tools: context.input.request.tools, @@ -583,33 +582,38 @@ export function createTurnMachine( }), ], }, - 'llm.failed.remote': [ + 'llm.failed.remote': { + actions: raise(({ context, event }) => ({ + type: 'turn.failure.triaged' as const, + cause: event, + proposal: proposeRecovery(recovery, { + error: event.error, + messages: baseMessages(context), + applied: context.appliedRecoveries, + credentials: context.input.request.credentials, + }), + })), + }, + 'turn.failure.triaged': [ { - guard: ({ context, event }) => - proposeRecovery(recovery, { - error: event.error, - messages: baseMessages(context), - applied: context.appliedRecoveries, - }) !== undefined, + guard: ({ event }) => event.proposal !== undefined, target: 'thinking', reenter: true, actions: [ - ({ context }) => { + ({ context, event }) => { context.accumulator.rollback(); + if (event.proposal?.refreshCredentials === true) { + context.input.request.credentials?.invalidate?.(); + } }, assign(({ context, event }) => { - const proposal = proposeRecovery(recovery, { - error: event.error, - messages: baseMessages(context), - applied: context.appliedRecoveries, - }); - if (proposal === undefined) return {}; + const proposal = event.proposal as LlmRecoveryProposal & LlmRecoveryRecord; return { - lastError: event.error, appliedRecoveries: [ ...context.appliedRecoveries, { strategy: proposal.strategy, action: proposal.action }, ], + recoveryMessages: proposal.messages ?? context.recoveryMessages, attempt: 1, }; }), @@ -618,14 +622,14 @@ export function createTurnMachine( params: ({ context, event }) => llmRecoveringEvent( context.appliedRecoveries.at(-1) as LlmRecoveryRecord, - event.error, + event.cause.error, ), }, ], }, { guard: ({ context, event }) => - shouldRetry(retry, context.attempt, event.error), + shouldRetry(retry, context.attempt, event.cause.error), target: 'retrying', actions: [ ({ context }) => { @@ -633,22 +637,23 @@ export function createTurnMachine( }, assign({ delayMs: ({ context, event }) => - readRetryAfterMs(event.error) ?? retryBackoffDelay(context.attempt - 1), + readRetryAfterMs(event.cause.error) ?? + retryBackoffDelay(context.attempt - 1), }), { type: 'sendToParent', params: ({ context, event }) => - llmRetryingEvent(retry, context.attempt, context.delayMs, event.error), + llmRetryingEvent(retry, context.attempt, context.delayMs, event.cause.error), }, ], }, { target: 'failed', actions: [ - 'forwardToParent', + { type: 'sendToParent', params: ({ event }) => event.cause }, assign({ outcome: 'failed' as const, - error: ({ event }) => event.error, + error: ({ event }) => event.cause.rawError ?? event.cause.error, }), ], }, @@ -796,7 +801,7 @@ export function createTurnMachine( steps: event.messages.length > 0 ? 1 : context.steps + 1, attempt: 1, appliedRecoveries: [], - lastError: undefined, + recoveryMessages: undefined, })), 'signalRemindersConsumed', ], diff --git a/packages/agent-core-v2/src/human/credentials/credentials.ts b/packages/agent-core-v2/src/human/credentials/credentials.ts new file mode 100644 index 00000000000..e0e0666a987 --- /dev/null +++ b/packages/agent-core-v2/src/human/credentials/credentials.ts @@ -0,0 +1,66 @@ +import { errorStatusCode } from '#/llm/errors'; +import type { LlmModel } from '#/llm/model'; +import type { LlmRecovery } from '#/llm/requester/recovery'; +import { + mergeRequestHeaders, + type LlmCredential, + type LlmCredentialProvider, +} from '#/llm/requester/requester'; + +export interface CredentialTokenSource { + (options?: { readonly force?: boolean }): Promise; +} + +export function staticCredentials(apiKey?: string): LlmCredentialProvider { + return { + resolve: () => + apiKey === undefined || apiKey.trim().length === 0 ? undefined : { apiKey }, + }; +} + +export function oauthCredentials(getToken: CredentialTokenSource): LlmCredentialProvider { + let refreshed: Promise | undefined; + return { + resolve: async () => { + const pending = refreshed; + refreshed = undefined; + const apiKey = pending === undefined ? await getToken() : await pending; + return apiKey === undefined ? undefined : { apiKey }; + }, + canRecover: (error) => errorStatusCode(error) === 401, + invalidate: () => { + refreshed ??= getToken({ force: true }); + refreshed.catch(() => {}); + }, + }; +} + +export function applyCredential( + model: LlmModel, + credential: LlmCredential | undefined, +): LlmModel { + if (credential === undefined) { + return model; + } + return { + ...model, + apiKey: credential.apiKey ?? model.apiKey, + defaultHeaders: mergeRequestHeaders(model.defaultHeaders, credential.headers), + }; +} + +export async function resolveModelCredentials( + model: LlmModel, + credentials: LlmCredentialProvider | undefined, +): Promise { + return applyCredential(model, await credentials?.resolve()); +} + +export const credentialsRecovery: LlmRecovery = { + id: 'credentials', + propose: ({ error, applied, credentials }) => + credentials?.canRecover?.(error) === true && + !applied.some((record) => record.strategy === 'credentials') + ? { action: 'refresh', refreshCredentials: true } + : undefined, +}; diff --git a/packages/agent-core-v2/src/human/credentials/index.ts b/packages/agent-core-v2/src/human/credentials/index.ts new file mode 100644 index 00000000000..374911ccabd --- /dev/null +++ b/packages/agent-core-v2/src/human/credentials/index.ts @@ -0,0 +1,2 @@ +export * from './credentials'; +export * from './kimi-oauth'; diff --git a/packages/agent-core-v2/src/human/credentials/kimi-oauth.ts b/packages/agent-core-v2/src/human/credentials/kimi-oauth.ts new file mode 100644 index 00000000000..091e5b804d5 --- /dev/null +++ b/packages/agent-core-v2/src/human/credentials/kimi-oauth.ts @@ -0,0 +1,8 @@ +import type { BearerTokenProvider } from '@moonshot-ai/kimi-code-oauth'; + +import { oauthCredentials } from '#/credentials/credentials'; +import type { LlmCredentialProvider } from '#/llm/requester/requester'; + +export function kimiOAuthCredentialProvider(tokens: BearerTokenProvider): LlmCredentialProvider { + return oauthCredentials((options) => tokens.getAccessToken(options)); +} diff --git a/packages/agent-core-v2/src/human/index.ts b/packages/agent-core-v2/src/human/index.ts index 39acdd4a2fb..508b2bc2d01 100644 --- a/packages/agent-core-v2/src/human/index.ts +++ b/packages/agent-core-v2/src/human/index.ts @@ -53,7 +53,7 @@ export * from './llm-kimi/media'; export * from './llm-kimi/schema'; export * from './llm-kimi/trait'; export * from './llm/provider/providers/standard'; -export * from './kimi-oauth'; +export * from './credentials'; export * from './tool/executor'; export * from './tool/machine'; export * from './tool/wait-for'; diff --git a/packages/agent-core-v2/src/human/kimi-oauth/credential-source.ts b/packages/agent-core-v2/src/human/kimi-oauth/credential-source.ts deleted file mode 100644 index 28a148c9b1c..00000000000 --- a/packages/agent-core-v2/src/human/kimi-oauth/credential-source.ts +++ /dev/null @@ -1,79 +0,0 @@ -import type { MediaVideoUploader } from '#/llm/media/upload'; -import type { LlmRemoteErrorMessage } from '#/llm/errors'; -import type { LlmModel } from '#/llm/model'; -import type { - LlmRequestConfig, - LlmRequestContent, - LlmRequestControl, - LlmRequester, -} from '#/llm/requester/requester'; - -export interface CredentialResolveOptions { - readonly force?: boolean; -} - -export interface CredentialSource { - resolve(model: LlmModel, options?: CredentialResolveOptions): Promise | LlmModel; - canRecover?(model: LlmModel, error: unknown): boolean; -} - -async function runWithCredentials( - source: CredentialSource, - model: LlmModel, - aborted: () => boolean, - run: (model: LlmModel) => Promise, -): Promise { - const resolved = await source.resolve(model); - try { - return await run(resolved); - } catch (error) { - if (aborted() || source.canRecover?.(resolved, error) !== true) { - throw error; - } - } - const refreshed = await source.resolve(model, { force: true }); - return run(refreshed); -} - -export function withAuth(inner: LlmRequester, source: CredentialSource): LlmRequester { - return { - async generate( - config: LlmRequestConfig, - content: LlmRequestContent, - control: LlmRequestControl, - ): Promise { - const resolved = await source.resolve(config.model); - let failed: LlmRemoteErrorMessage | undefined; - await inner.generate({ ...config, model: resolved }, content, { - ...control, - onEvent: (event) => { - if (event.type === 'llm.failed.remote') { - failed = event.error; - return; - } - control.onEvent?.(event); - }, - }); - if (failed === undefined) { - return; - } - const failure: LlmRemoteErrorMessage = failed; - if (control.signal.aborted || source.canRecover?.(resolved, failure) !== true) { - control.onEvent?.({ type: 'llm.failed.remote', error: failure }); - return; - } - const refreshed = await source.resolve(config.model, { force: true }); - await inner.generate({ ...config, model: refreshed }, content, control); - }, - }; -} - -export function withAuthUpload( - inner: MediaVideoUploader, - source: CredentialSource, -): MediaVideoUploader { - return (video, options) => - runWithCredentials(source, options.model, () => options.signal?.aborted === true, (model) => - inner(video, { ...options, model }), - ); -} diff --git a/packages/agent-core-v2/src/human/kimi-oauth/index.ts b/packages/agent-core-v2/src/human/kimi-oauth/index.ts deleted file mode 100644 index e22369501ad..00000000000 --- a/packages/agent-core-v2/src/human/kimi-oauth/index.ts +++ /dev/null @@ -1,2 +0,0 @@ -export * from './credential-source'; -export * from './kimi-oauth'; diff --git a/packages/agent-core-v2/src/human/kimi-oauth/kimi-oauth.ts b/packages/agent-core-v2/src/human/kimi-oauth/kimi-oauth.ts deleted file mode 100644 index 4690301462b..00000000000 --- a/packages/agent-core-v2/src/human/kimi-oauth/kimi-oauth.ts +++ /dev/null @@ -1,22 +0,0 @@ -import type { BearerTokenProvider } from '@moonshot-ai/kimi-code-oauth'; - -import type { CredentialSource } from './credential-source'; - -function statusOf(error: unknown): number | undefined { - if (typeof error !== 'object' || error === null) { - return undefined; - } - const record = error as Record; - const status = record['status'] ?? record['statusCode']; - return typeof status === 'number' ? status : undefined; -} - -export function kimiOAuthCredentialSource(tokens: BearerTokenProvider): CredentialSource { - return { - resolve: async (model, options) => ({ - ...model, - apiKey: await tokens.getAccessToken({ force: options?.force === true }), - }), - canRecover: (_model, error) => statusOf(error) === 401, - }; -} diff --git a/packages/agent-core-v2/src/human/llm/errors.ts b/packages/agent-core-v2/src/human/llm/errors.ts index 9f99b67c0b1..ebb25e9f772 100644 --- a/packages/agent-core-v2/src/human/llm/errors.ts +++ b/packages/agent-core-v2/src/human/llm/errors.ts @@ -17,6 +17,15 @@ export function isAbortError(error: unknown): boolean { ); } +export function errorStatusCode(error: unknown): number | undefined { + if (typeof error !== 'object' || error === null) { + return undefined; + } + const record = error as Record; + const status = record['status'] ?? record['statusCode']; + return typeof status === 'number' ? status : undefined; +} + export type LlmErrorKind = | 'syntax' | 'abort' diff --git a/packages/agent-core-v2/src/human/llm/requester/actor.ts b/packages/agent-core-v2/src/human/llm/requester/actor.ts index d33c7d57652..0ca257f497d 100644 --- a/packages/agent-core-v2/src/human/llm/requester/actor.ts +++ b/packages/agent-core-v2/src/human/llm/requester/actor.ts @@ -1,5 +1,7 @@ import { fromCallback } from '#/xstate2'; +import { applyCredential } from '#/credentials/credentials'; +import { isAbortError, toLlmErrorMessage } from '#/llm/errors'; import type { Message } from '#/llm/message'; import type { LlmModel } from '#/llm/model'; @@ -58,21 +60,36 @@ export function createRequestActor( ) { return fromCallback(({ input, sendBack }) => { void (async () => { - let messages = input.content.messages; - for (const resolver of messageResolvers) { - messages = await resolver.resolve(messages, { - model: input.config.model, - signal: input.signal, - }); + try { + const credential = input.config.credentials?.resolve(); + const config = + credential === undefined + ? input.config + : credential instanceof Promise + ? { + ...input.config, + model: applyCredential(input.config.model, await credential), + } + : { ...input.config, model: applyCredential(input.config.model, credential) }; + let messages = input.content.messages; + for (const resolver of messageResolvers) { + messages = await resolver.resolve(messages, { + model: config.model, + signal: input.signal, + }); + } + await requester.generate( + config, + { ...input.content, messages }, + { + signal: input.signal, + onEvent: sendBack, + }, + ); + } catch (error) { + if (isAbortError(error) || input.signal.aborted) return; + sendBack({ type: 'llm.failed.remote', error: toLlmErrorMessage(error), rawError: error }); } - await requester.generate( - input.config, - { ...input.content, messages }, - { - signal: input.signal, - onEvent: sendBack, - }, - ); })(); }); } diff --git a/packages/agent-core-v2/src/human/llm/requester/recovery.ts b/packages/agent-core-v2/src/human/llm/requester/recovery.ts index e65a19dfbd2..0d930a5b2d3 100644 --- a/packages/agent-core-v2/src/human/llm/requester/recovery.ts +++ b/packages/agent-core-v2/src/human/llm/requester/recovery.ts @@ -1,5 +1,6 @@ import type { LlmRemoteErrorMessage } from '#/llm/errors'; import type { Message } from '#/llm/message'; +import type { LlmCredentialProvider } from '#/llm/requester/requester'; export interface LlmRecoveryRecord { readonly strategy: string; @@ -10,11 +11,13 @@ export interface LlmRecoveryContext { readonly error: LlmRemoteErrorMessage; readonly messages: readonly Message[]; readonly applied: readonly LlmRecoveryRecord[]; + readonly credentials?: LlmCredentialProvider; } export interface LlmRecoveryProposal { readonly action: string; - readonly messages: readonly Message[]; + readonly messages?: readonly Message[]; + readonly refreshCredentials?: boolean; } export interface LlmRecovery { diff --git a/packages/agent-core-v2/src/human/llm/requester/requester.ts b/packages/agent-core-v2/src/human/llm/requester/requester.ts index fe65a3bcba2..185e4abf9b2 100644 --- a/packages/agent-core-v2/src/human/llm/requester/requester.ts +++ b/packages/agent-core-v2/src/human/llm/requester/requester.ts @@ -30,7 +30,7 @@ export type LlmRequestEvent = | { type: 'llm.streaming.finish'; finish: FinishInfo } | { type: 'llm.streaming.message_id'; messageId: string } | { type: 'llm.failed.syntax'; error: LlmErrorMessage<'syntax'> } - | { type: 'llm.failed.remote'; error: LlmRemoteErrorMessage } + | { type: 'llm.failed.remote'; error: LlmRemoteErrorMessage; rawError?: unknown } | { type: 'llm.done' }; export interface ExtraParams { @@ -42,8 +42,20 @@ export interface ExtraParams { export type ToolMessageConversion = 'extract_text' | 'keep_parts'; +export interface LlmCredential { + readonly apiKey?: string; + readonly headers?: Record; +} + +export interface LlmCredentialProvider { + resolve(): Promise | LlmCredential | undefined; + canRecover?(error: unknown): boolean; + invalidate?(): void; +} + export interface LlmRequestConfig { readonly model: LlmModel; + readonly credentials?: LlmCredentialProvider; readonly systemPrompt?: string; readonly tools?: readonly ToolDescription[]; readonly cacheKey?: string; diff --git a/packages/agent-core-v2/src/human/test/agent/turn.test.ts b/packages/agent-core-v2/src/human/test/agent/turn.test.ts index 7ada3304dd7..acae1bc8440 100644 --- a/packages/agent-core-v2/src/human/test/agent/turn.test.ts +++ b/packages/agent-core-v2/src/human/test/agent/turn.test.ts @@ -6,9 +6,9 @@ import type { LlmErrorMessage } from '#/llm/errors'; import type { ContentPart, Message, UserMessage } from '#/llm/message'; import { createMediaDegradeRecovery } from '#/llm/media/degrade'; import type { LlmModel } from '#/llm/model'; -import type { LlmEvent } from '#/llm/requester/actor'; +import { createRequestActor, type LlmEvent } from '#/llm/requester/actor'; import type { LlmRecovery } from '#/llm/requester/recovery'; -import type { LlmRequester } from '#/llm/requester/requester'; +import type { LlmCredentialProvider, LlmRequester } from '#/llm/requester/requester'; import type { LlmRetryOptions } from '#/llm/requester/retry'; import { createTurnMachine, @@ -132,6 +132,44 @@ function startTurnActor( return { actor, retrying, recovering, sent, failed }; } +function startRequestActor(requester: LlmRequester, signal: AbortSignal) { + const harness = setup({ + types: { + input: {} as { signal: AbortSignal }, + context: {} as { signal: AbortSignal }, + events: {} as LlmEvent, + emitted: {} as LlmEvent, + }, + actors: { request: createRequestActor(requester) }, + }).createMachine({ + id: 'request-harness', + initial: 'running', + context: ({ input }) => input, + states: { + running: { + invoke: { + src: 'request', + input: ({ context }) => ({ + config: { model }, + content: { messages: [] }, + signal: context.signal, + }), + }, + on: { + '*': { + actions: emit(({ event }) => event), + }, + }, + }, + }, + }); + const failed: unknown[] = []; + const actor = createActor(harness, { input: { signal } }); + actor.on('llm.failed.remote', (event) => failed.push(event)); + actor.start(); + return { failed }; +} + async function flush(): Promise { await vi.advanceTimersByTimeAsync(0); } @@ -534,3 +572,177 @@ describe('turn machine media recovery', () => { expect(recovering).toHaveLength(0); }); }); + +describe('turn machine credential recovery', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + function createCredentials(onInvalidate: () => void): { + provider: LlmCredentialProvider; + tokens: readonly string[]; + } { + const tokens = ['tok-1', 'tok-2']; + let resolutions = 0; + return { + tokens, + provider: { + resolve: () => { + const apiKey = tokens[Math.min(resolutions, tokens.length - 1)] as string; + resolutions += 1; + return { apiKey }; + }, + canRecover: (error) => + typeof error === 'object' && + error !== null && + (error as { statusCode?: number }).statusCode === 401, + invalidate: onInvalidate, + }, + }; + } + + it('refreshes credentials once on a recoverable 401 and retries', async () => { + let invalidations = 0; + const { provider } = createCredentials(() => (invalidations += 1)); + const apiKeys: (string | undefined)[] = []; + const requester: LlmRequester = { + generate: (config, _content, control) => { + apiKeys.push(config.model.apiKey); + control.onEvent?.({ type: 'llm.sent' }); + if (apiKeys.length === 1) { + control.onEvent?.({ type: 'llm.failed.remote', error: statusError(401, 'unauthorized') }); + return Promise.resolve(); + } + control.onEvent?.({ type: 'llm.streaming.part', part: { type: 'text', text: 'done' } }); + control.onEvent?.({ type: 'llm.done' }); + return Promise.resolve(); + }, + }; + const { actor, recovering, sent, failed } = startTurnActor(requester, undefined, { + request: { model, credentials: provider }, + }); + + await drain(); + + expect(apiKeys).toEqual(['tok-1', 'tok-2']); + expect(invalidations).toBe(1); + expect(recovering).toHaveLength(1); + expect(recovering[0]).toMatchObject({ + strategy: 'credentials', + action: 'refresh', + statusCode: 401, + }); + expect(sent.map((event) => event.recovery?.action)).toEqual([undefined, 'refresh']); + expect(actor.getSnapshot().context.turnOutput).toMatchObject({ type: 'done' }); + expect(failed).toHaveLength(0); + }); + + it('keeps recovered messages when a credential refresh follows a message recovery', async () => { + let invalidations = 0; + const { provider } = createCredentials(() => (invalidations += 1)); + const { requester, calls, seen } = createCapturingRequester([ + tooLargeError(), + statusError(401, 'unauthorized'), + 'ok', + ]); + const { actor, recovering } = startTurnActor( + requester, + { recovery: createMediaDegradeRecovery() }, + { + ...mediaHistory([mediaMessage('a', 2), mediaMessage('b', 1), mediaMessage('c', 1)]), + request: { model, credentials: provider }, + }, + ); + + await drain(); + + expect(calls()).toBe(3); + expect(invalidations).toBe(1); + expect(recovering.map((event) => `${event.strategy}:${event.action}`)).toEqual([ + 'media-degrade:degraded', + 'credentials:refresh', + ]); + expect(countImageParts(seen[1] ?? [])).toBe(2); + expect(countImageParts(seen[2] ?? [])).toBe(2); + expect(actor.getSnapshot().context.turnOutput).toMatchObject({ type: 'done' }); + }); + + it('fails when the attempt after a credential refresh also fails', async () => { + let invalidations = 0; + const { provider } = createCredentials(() => (invalidations += 1)); + const { requester, calls } = createStubRequester([ + statusError(401, 'unauthorized'), + statusError(401, 'still unauthorized'), + ]); + const { actor, recovering, failed } = startTurnActor(requester, undefined, { + request: { model, credentials: provider }, + }); + + await drain(); + + expect(calls()).toBe(2); + expect(invalidations).toBe(1); + expect(recovering).toHaveLength(1); + expect(actor.getSnapshot().context.turnOutput).toMatchObject({ type: 'failed' }); + expect(failed).toHaveLength(1); + }); + + it('does not refresh when the request carries no recoverable credentials', async () => { + const { requester, calls } = createStubRequester([statusError(401, 'unauthorized')]); + const { actor, recovering, failed } = startTurnActor(requester); + + await drain(); + + expect(calls()).toBe(1); + expect(recovering).toHaveLength(0); + expect(actor.getSnapshot().context.turnOutput).toMatchObject({ type: 'failed' }); + expect(failed).toHaveLength(1); + }); + + it('fails the turn instead of hanging when credential resolution rejects', async () => { + const { requester, calls } = createStubRequester(['ok']); + const provider: LlmCredentialProvider = { + resolve: () => Promise.reject(new Error('login required')), + }; + const { actor, failed } = startTurnActor(requester, undefined, { + request: { model, credentials: provider }, + }); + + await drain(); + + expect(calls()).toBe(0); + expect(actor.getSnapshot().context.turnOutput).toMatchObject({ type: 'failed' }); + expect(failed).toHaveLength(1); + expect((failed[0] as { message?: string }).message).toContain('login required'); + }); + + it('does not report llm.failed.remote when the request aborts', async () => { + const requester: LlmRequester = { + generate: () => Promise.reject(new DOMException('The operation was aborted.', 'AbortError')), + }; + const { failed } = startRequestActor(requester, new AbortController().signal); + + await drain(); + + expect(failed).toHaveLength(0); + }); + + it('does not report llm.failed.remote when the signal is already aborted', async () => { + const controller = new AbortController(); + const requester: LlmRequester = { + generate: () => { + controller.abort(); + return Promise.reject(new Error('boom')); + }, + }; + const { failed } = startRequestActor(requester, controller.signal); + + await drain(); + + expect(failed).toHaveLength(0); + }); +}); diff --git a/packages/agent-core-v2/src/human/test/credentials/credentials.test.ts b/packages/agent-core-v2/src/human/test/credentials/credentials.test.ts new file mode 100644 index 00000000000..aa83192b6aa --- /dev/null +++ b/packages/agent-core-v2/src/human/test/credentials/credentials.test.ts @@ -0,0 +1,164 @@ +import { describe, expect, it } from 'vitest'; + +import { + applyCredential, + credentialsRecovery, + oauthCredentials, + resolveModelCredentials, + staticCredentials, +} from '#/credentials/credentials'; +import type { LlmModel } from '#/llm/model'; +import type { LlmRecoveryContext, LlmRecoveryRecord } from '#/llm/requester/recovery'; +import type { LlmCredentialProvider } from '#/llm/requester/requester'; + +const MODEL: LlmModel = { + provider: 'fake', + model: 'fake-model', + apiKey: 'base-key', + defaultHeaders: { 'x-base': '1' }, +}; + +describe('staticCredentials', () => { + it('resolves the static api key and never recovers', async () => { + const provider = staticCredentials('sk-1'); + expect(await provider.resolve()).toEqual({ apiKey: 'sk-1' }); + expect(provider.canRecover).toBeUndefined(); + expect(provider.invalidate).toBeUndefined(); + }); + + it('resolves undefined for missing or blank keys', async () => { + expect(await staticCredentials(undefined).resolve()).toBeUndefined(); + expect(await staticCredentials(' ').resolve()).toBeUndefined(); + }); +}); + +describe('oauthCredentials', () => { + it('refreshes with force on invalidate and consumes the refresh on the next resolve', async () => { + const calls: (boolean | undefined)[] = []; + const provider = oauthCredentials((options) => { + calls.push(options?.force); + return Promise.resolve('tok'); + }); + + await provider.resolve(); + await provider.resolve(); + provider.invalidate?.(); + await provider.resolve(); + await provider.resolve(); + + expect(calls).toEqual([undefined, undefined, true, undefined]); + }); + + it('starts the forced refresh eagerly on invalidate, before the next resolve', async () => { + const calls: (boolean | undefined)[] = []; + const provider = oauthCredentials((options) => { + calls.push(options?.force); + return Promise.resolve('tok'); + }); + + provider.invalidate?.(); + + expect(calls).toEqual([true]); + + await provider.resolve(); + + expect(calls).toEqual([true]); + }); + + it('coalesces repeated invalidates into a single refresh', async () => { + const calls: (boolean | undefined)[] = []; + const provider = oauthCredentials((options) => { + calls.push(options?.force); + return Promise.resolve('tok'); + }); + + provider.invalidate?.(); + provider.invalidate?.(); + await provider.resolve(); + + expect(calls).toEqual([true]); + }); + + it('propagates a failed refresh to the consuming resolve and recovers afterwards', async () => { + let calls = 0; + const provider = oauthCredentials(() => { + calls += 1; + return calls === 1 ? Promise.reject(new Error('login required')) : Promise.resolve('tok'); + }); + + provider.invalidate?.(); + + await expect(provider.resolve()).rejects.toThrow('login required'); + await expect(provider.resolve()).resolves.toEqual({ apiKey: 'tok' }); + }); + + it('recovers only from 401 errors', () => { + const provider = oauthCredentials(() => Promise.resolve('tok')); + expect(provider.canRecover?.(Object.assign(new Error('x'), { status: 401 }))).toBe(true); + expect(provider.canRecover?.(Object.assign(new Error('x'), { statusCode: 401 }))).toBe(true); + expect(provider.canRecover?.(Object.assign(new Error('x'), { statusCode: 403 }))).toBe(false); + expect(provider.canRecover?.(new Error('boom'))).toBe(false); + }); + + it('resolves undefined when the token source has no token', async () => { + const provider = oauthCredentials(() => Promise.resolve(undefined)); + await expect(provider.resolve()).resolves.toBeUndefined(); + }); +}); + +describe('applyCredential / resolveModelCredentials', () => { + it('returns the model unchanged when the credential is undefined', async () => { + expect(applyCredential(MODEL, undefined)).toBe(MODEL); + await expect(resolveModelCredentials(MODEL, undefined)).resolves.toBe(MODEL); + }); + + it('overrides the api key and merges headers', () => { + const applied = applyCredential(MODEL, { apiKey: 'fresh', headers: { 'x-auth': 't' } }); + expect(applied.apiKey).toBe('fresh'); + expect(applied.defaultHeaders).toEqual({ 'x-base': '1', 'x-auth': 't' }); + }); + + it('keeps the model api key when the credential carries none', () => { + const applied = applyCredential(MODEL, { headers: { 'x-auth': 't' } }); + expect(applied.apiKey).toBe('base-key'); + }); +}); + +function recoveryContext( + error: unknown, + applied: readonly LlmRecoveryRecord[] = [], + credentials?: LlmCredentialProvider, +): LlmRecoveryContext { + return { error: error as LlmRecoveryContext['error'], messages: [], applied, credentials }; +} + +const unauthorized = Object.assign(new Error('unauthorized'), { status: 401 }); +const forbidden = Object.assign(new Error('forbidden'), { status: 403 }); + +describe('credentialsRecovery', () => { + it('proposes a credentials refresh on a recoverable error', () => { + const provider = oauthCredentials(() => Promise.resolve('tok')); + expect(credentialsRecovery.propose(recoveryContext(unauthorized, [], provider))).toEqual({ + action: 'refresh', + refreshCredentials: true, + }); + }); + + it('does not propose when the strategy was already applied', () => { + const provider = oauthCredentials(() => Promise.resolve('tok')); + const applied: LlmRecoveryRecord[] = [{ strategy: 'credentials', action: 'refresh' }]; + expect(credentialsRecovery.propose(recoveryContext(unauthorized, applied, provider))).toBeUndefined(); + }); + + it('does not propose without recoverable credentials', () => { + expect(credentialsRecovery.propose(recoveryContext(unauthorized))).toBeUndefined(); + expect( + credentialsRecovery.propose(recoveryContext(unauthorized, [], staticCredentials('sk-1'))), + ).toBeUndefined(); + expect( + credentialsRecovery.propose( + recoveryContext(forbidden, [], oauthCredentials(() => Promise.resolve('tok'))), + ), + ).toBeUndefined(); + }); +}); diff --git a/packages/agent-core-v2/src/human/test/credentials/kimi-oauth.test.ts b/packages/agent-core-v2/src/human/test/credentials/kimi-oauth.test.ts new file mode 100644 index 00000000000..25bc2b0069c --- /dev/null +++ b/packages/agent-core-v2/src/human/test/credentials/kimi-oauth.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from 'vitest'; + +import { kimiOAuthCredentialProvider } from '#/credentials/kimi-oauth'; + +describe('kimiOAuthCredentialProvider', () => { + function createTokens() { + const calls: (boolean | undefined)[] = []; + return { + calls, + tokens: { + getAccessToken: (options?: { readonly force?: boolean }) => { + calls.push(options?.force); + return Promise.resolve('access-token'); + }, + }, + }; + } + + it('resolves the access token from the token provider', async () => { + const { calls, tokens } = createTokens(); + const provider = kimiOAuthCredentialProvider(tokens); + + await expect(provider.resolve()).resolves.toEqual({ apiKey: 'access-token' }); + expect(calls).toEqual([undefined]); + }); +}); diff --git a/packages/agent-core-v2/src/human/test/kimi-oauth/credential-source.test.ts b/packages/agent-core-v2/src/human/test/kimi-oauth/credential-source.test.ts deleted file mode 100644 index 91d03121ee8..00000000000 --- a/packages/agent-core-v2/src/human/test/kimi-oauth/credential-source.test.ts +++ /dev/null @@ -1,296 +0,0 @@ -import { describe, expect, it } from 'vitest'; - -import { UNKNOWN_CAPABILITY } from '#/llm/capability'; -import type { StreamedMessagePart, VideoURLPart } from '#/llm/message'; -import type { MediaVideoUploader } from '#/llm/media/upload'; -import type { LlmErrorMessage } from '#/llm/errors'; -import type { LlmModel } from '#/llm/model'; -import type { LlmRequestControl, LlmRequester } from '#/llm/requester/requester'; -import { - kimiOAuthCredentialSource, - withAuth, - withAuthUpload, - type CredentialSource, -} from '#/kimi-oauth/index'; - -const model: LlmModel = { provider: 'test', model: 'test-model', capability: UNKNOWN_CAPABILITY }; - -type GenerateArgs = Parameters; - -function generateArgs(control: Partial = {}): GenerateArgs { - return [{ model }, { messages: [] }, { signal: new AbortController().signal, ...control }]; -} - -function statusError(status: number): LlmErrorMessage { - return { - kind: 'status', - statusCode: status, - message: `status ${status}`, - requestId: null, - retryAfterMs: null, - headers: null, - }; -} - -interface InnerCall { - readonly model: LlmModel; -} - -function createInner(plan: readonly (LlmErrorMessage | 'ok')[]) { - const calls: InnerCall[] = []; - const requester: LlmRequester = { - generate: (config, _content, { onEvent }) => { - calls.push({ model: config.model }); - const step = plan[Math.min(calls.length - 1, plan.length - 1)]; - if (step === 'ok') { - onEvent?.({ - type: 'llm.streaming.part', - part: { type: 'text', text: `call-${calls.length}` }, - }); - onEvent?.({ type: 'llm.done' }); - return Promise.resolve(); - } - onEvent?.({ type: 'llm.failed.remote', error: step }); - return Promise.resolve(); - }, - }; - return { requester, calls }; -} - -async function generateFailures( - requester: LlmRequester, - control: Partial = {}, -): Promise { - const failures: LlmErrorMessage[] = []; - await requester.generate( - ...generateArgs({ - ...control, - onEvent: (event) => { - if (event.type === 'llm.failed.remote' || event.type === 'llm.failed.syntax') { - failures.push(event.error); - } - }, - }), - ); - return failures; -} - -describe('withAuth', () => { - it('resolves credentials before each generate and forwards the resolved model', async () => { - const { requester, calls } = createInner(['ok']); - const source: CredentialSource = { - resolve: (m) => ({ ...m, apiKey: 'token-1' }), - }; - const wrapped = withAuth(requester, source); - - const parts: StreamedMessagePart[] = []; - await wrapped.generate( - ...generateArgs({ - onEvent: (event) => { - if (event.type === 'llm.streaming.part') { - parts.push(event.part); - } - }, - }), - ); - - expect(calls).toHaveLength(1); - expect(calls[0]?.model).toEqual({ ...model, apiKey: 'token-1' }); - expect(parts).toEqual([{ type: 'text', text: 'call-1' }]); - }); - - it('retries once with forced credentials when canRecover accepts the error', async () => { - const { requester, calls } = createInner([statusError(401), 'ok']); - const resolveOptions: unknown[] = []; - const source: CredentialSource = { - resolve: (m, options) => { - resolveOptions.push(options); - return { ...m, apiKey: options?.force === true ? 'token-2' : 'token-1' }; - }, - canRecover: (_m, error) => statusErrorStatus(error) === 401, - }; - const wrapped = withAuth(requester, source); - - const parts: StreamedMessagePart[] = []; - await wrapped.generate( - ...generateArgs({ - onEvent: (event) => { - if (event.type === 'llm.streaming.part') { - parts.push(event.part); - } - }, - }), - ); - - expect(calls).toHaveLength(2); - expect(calls[0]?.model.apiKey).toBe('token-1'); - expect(calls[1]?.model.apiKey).toBe('token-2'); - expect(resolveOptions).toEqual([undefined, { force: true }]); - expect(parts).toEqual([{ type: 'text', text: 'call-2' }]); - }); - - it('emits the failure when the retry also fails', async () => { - const { requester, calls } = createInner([statusError(401), statusError(401)]); - const source: CredentialSource = { - resolve: (m) => m, - canRecover: () => true, - }; - const wrapped = withAuth(requester, source); - - const failures = await generateFailures(wrapped); - - expect(calls).toHaveLength(2); - expect(failures).toHaveLength(1); - expect(failures[0]).toMatchObject({ kind: 'status', statusCode: 401 }); - }); - - it('does not retry when canRecover rejects the error', async () => { - const { requester, calls } = createInner([statusError(401)]); - const source: CredentialSource = { - resolve: (m) => m, - canRecover: () => false, - }; - const wrapped = withAuth(requester, source); - - const failures = await generateFailures(wrapped); - - expect(calls).toHaveLength(1); - expect(failures).toHaveLength(1); - expect(failures[0]).toMatchObject({ kind: 'status', statusCode: 401 }); - }); - - it('does not retry when the source has no canRecover', async () => { - const { requester, calls } = createInner([statusError(401)]); - const wrapped = withAuth(requester, { resolve: (m) => m }); - - const failures = await generateFailures(wrapped); - - expect(calls).toHaveLength(1); - expect(failures).toHaveLength(1); - expect(failures[0]).toMatchObject({ kind: 'status', statusCode: 401 }); - }); - - it('does not retry when the signal is aborted', async () => { - const { requester, calls } = createInner([statusError(401)]); - const controller = new AbortController(); - controller.abort(); - const source: CredentialSource = { - resolve: (m) => m, - canRecover: () => true, - }; - const wrapped = withAuth(requester, source); - - const failures = await generateFailures(wrapped, { signal: controller.signal }); - - expect(calls).toHaveLength(1); - expect(failures).toHaveLength(1); - expect(failures[0]).toMatchObject({ kind: 'status', statusCode: 401 }); - }); - - it('wraps uploadVideo with the same credential flow', async () => { - const part: VideoURLPart = { type: 'video_url', videoUrl: { url: 'ms://file-1', id: 'file-1' } }; - const seen: (string | undefined)[] = []; - let attempts = 0; - const inner: MediaVideoUploader = (_video, options) => { - attempts += 1; - seen.push(options.model.apiKey); - if (attempts === 1) { - return Promise.reject(statusError(401)); - } - return Promise.resolve(part); - }; - const source: CredentialSource = { - resolve: (m, options) => ({ ...m, apiKey: options?.force === true ? 'fresh' : 'stale' }), - canRecover: () => true, - }; - const wrapped = withAuthUpload(inner, source); - - const result = await wrapped({ data: new Uint8Array([1]), mimeType: 'video/mp4' }, { model }); - - expect(result).toBe(part); - expect(seen).toEqual(['stale', 'fresh']); - }); - - it('does not retry the upload when the signal is aborted', async () => { - const failure = statusError(401); - let attempts = 0; - const inner: MediaVideoUploader = () => { - attempts += 1; - return Promise.reject(failure); - }; - const controller = new AbortController(); - controller.abort(); - const wrapped = withAuthUpload(inner, { - resolve: (m) => m, - canRecover: () => true, - }); - - await expect( - wrapped( - { data: new Uint8Array([1]), mimeType: 'video/mp4' }, - { model, signal: controller.signal }, - ), - ).rejects.toBe(failure); - expect(attempts).toBe(1); - }); -}); - -describe('kimiOAuthCredentialSource', () => { - function createTokens() { - const calls: (boolean | undefined)[] = []; - return { - calls, - tokens: { - getAccessToken: (options?: { readonly force?: boolean }) => { - calls.push(options?.force); - return Promise.resolve('access-token'); - }, - }, - }; - } - - it('resolves the model apiKey from the token provider', async () => { - const { calls, tokens } = createTokens(); - const source = kimiOAuthCredentialSource(tokens); - - const resolved = await source.resolve({ ...model, baseUrl: 'https://example.com/v1' }); - - expect(resolved).toEqual({ - ...model, - baseUrl: 'https://example.com/v1', - apiKey: 'access-token', - }); - expect(calls).toEqual([false]); - }); - - it('passes force through to the token provider', async () => { - const { calls, tokens } = createTokens(); - const source = kimiOAuthCredentialSource(tokens); - - await source.resolve(model, { force: true }); - - expect(calls).toEqual([true]); - }); - - it('recovers only from 401 errors', () => { - const { tokens } = createTokens(); - const source = kimiOAuthCredentialSource(tokens); - - expect(source.canRecover?.(model, statusError(401))).toBe(true); - expect(source.canRecover?.(model, Object.assign(new Error('x'), { statusCode: 401 }))).toBe( - true, - ); - expect(source.canRecover?.(model, statusError(403))).toBe(false); - expect(source.canRecover?.(model, new Error('boom'))).toBe(false); - expect(source.canRecover?.(model, 'nope')).toBe(false); - }); -}); - -function statusErrorStatus(error: unknown): number | undefined { - if (typeof error !== 'object' || error === null) { - return undefined; - } - const record = error as Record; - const status = record['status'] ?? record['statusCode']; - return typeof status === 'number' ? status : undefined; -} diff --git a/packages/agent-core-v2/src/index.ts b/packages/agent-core-v2/src/index.ts index a1e2b74ffeb..6d7e5dcc5a9 100644 --- a/packages/agent-core-v2/src/index.ts +++ b/packages/agent-core-v2/src/index.ts @@ -159,7 +159,6 @@ export type { export type { ThinkingEffort, ThinkingRequestOptions } from '#human/llm/thinking'; export type { VideoUploadInput } from '#human/llm/media/upload'; export type { ToolCallIdPolicy } from '#human/llm/requester/requester'; -export type { ProviderRequestAuth } from '#/llm-adapter/model/catalog'; export type { SamplingOptions } from '#/llm-adapter/model/model-requester'; export * from '#/llm-adapter/contract/request-trace'; export type { KimiThinkingConfig } from '#human/llm-kimi/trait'; diff --git a/packages/agent-core-v2/src/llm-adapter/model/catalog-service.ts b/packages/agent-core-v2/src/llm-adapter/model/catalog-service.ts index 45ccc9d08dd..9b3e724cd10 100644 --- a/packages/agent-core-v2/src/llm-adapter/model/catalog-service.ts +++ b/packages/agent-core-v2/src/llm-adapter/model/catalog-service.ts @@ -6,6 +6,8 @@ import { ScopeActivation, registerScopedService } from '#/_base/di/scope'; import { Error2 } from '#/_base/errors/errors'; import type { CatalogModel, CatalogProviderInfo } from '#human/llm/provider-catalog'; +import { oauthCredentials, staticCredentials } from '#human/credentials/credentials'; +import type { LlmCredentialProvider } from '#human/llm/requester/requester'; import type { ModelCapability } from '../contract/capability'; import { CONFIG_INVALID_ERROR_CODE } from '../contract/errors'; import type { TokenUsage } from '#human/llm/usage'; @@ -21,21 +23,22 @@ import { } from '../provider/provider-definition'; import { - type AuthProvider, IModelCatalog, type Model, type ModelCatalogItem, type ModelPingResult, type ProviderCatalogItem, type ProviderCredentialState, - type ProviderRequestAuth, type SetDefaultModelResponse, - StaticAuthProvider, toProtocolModel, toProtocolModelFallback, toProtocolProvider, } from './catalog'; import { IProviderCatalogRuntime, rawRecordOf } from './catalog-runtime'; +import { + runWithCredentialRecovery, + streamWithCredentialRecovery, +} from './credential-recovery'; import { ModelCatalogErrors } from './errors'; import { IHostRequestHeaders } from './host-request-headers'; import { IModelService, type ModelRecord } from './model'; @@ -49,7 +52,12 @@ import { } from './model-auth'; import { IModelOAuthTokens } from './model-oauth'; import type { ResolvedModelAuthMaterial } from './model.types'; -import type { ModelRequester } from './model-requester'; +import type { + ModelRequestEvent, + ModelRequestInput, + ModelRequestParams, + ModelRequester, +} from './model-requester'; import { ModelRequesterImpl } from './model-requester-impl'; type MutableProtocolProviderOptions = { @@ -131,31 +139,55 @@ export class ModelCatalog extends Disposable implements IModelCatalog { return entry; } + async *generate( + id: string, + input: ModelRequestInput, + signal?: AbortSignal, + params?: ModelRequestParams, + ): AsyncIterable { + const { requester } = this.entry(id); + yield* streamWithCredentialRecovery( + requester.model.credentials, + () => requester.request(input, signal, params), + signal, + ); + } + async ping(id: string): Promise { const { requester } = this.entry(id); const startedAt = Date.now(); try { - let text = ''; - let usage: TokenUsage | undefined; - let finishReason: string | undefined; - for await (const event of requester.request( - { - systemPrompt: 'You are a connectivity probe. Answer with the single word "pong".', - tools: [], - messages: [{ role: 'user', content: [{ type: 'text', text: 'ping' }], toolCalls: [] }], - }, - undefined, - { maxCompletionTokens: 512 }, - )) { - if (event.type === 'part' && event.part.type === 'text') { - text += event.part.text; - } else if (event.type === 'usage') { - usage = event.usage; - } else if (event.type === 'finish') { - finishReason = event.providerFinishReason ?? event.rawFinishReason; + const consume = async () => { + let text = ''; + let usage: TokenUsage | undefined; + let finishReason: string | undefined; + for await (const event of requester.request( + { + systemPrompt: 'You are a connectivity probe. Answer with the single word "pong".', + tools: [], + messages: [{ role: 'user', content: [{ type: 'text', text: 'ping' }], toolCalls: [] }], + }, + undefined, + { maxCompletionTokens: 512 }, + )) { + if (event.type === 'part' && event.part.type === 'text') { + text += event.part.text; + } else if (event.type === 'usage') { + usage = event.usage; + } else if (event.type === 'finish') { + finishReason = event.providerFinishReason ?? event.rawFinishReason; + } } - } - return { ok: true, durationMs: Date.now() - startedAt, text: text.trim(), finishReason, usage }; + return { text: text.trim(), usage, finishReason }; + }; + const result = await runWithCredentialRecovery(requester.model.credentials, consume); + return { + ok: true, + durationMs: Date.now() - startedAt, + text: result.text, + finishReason: result.finishReason, + usage: result.usage, + }; } catch (error) { return { ok: false, @@ -293,7 +325,7 @@ export class ModelCatalog extends Disposable implements IModelCatalog { provider: providerConfig, providerName, }); - const authProvider = this.buildAuthProvider(providerName, auth); + const credentials = this.buildCredentials(providerName, auth); const providerType = providerConfig?.type ?? protocol; const resolvedBaseUrl = @@ -354,7 +386,7 @@ export class ModelCatalog extends Disposable implements IModelCatalog { alwaysThinking: declared.has('always_thinking'), providerType, providerName, - authProvider, + credentials, providerOptions, }; } @@ -413,25 +445,22 @@ export class ModelCatalog extends Disposable implements IModelCatalog { return protocol; } - private buildAuthProvider(providerName: string, auth: ResolvedModelAuthMaterial): AuthProvider { + private buildCredentials( + providerName: string, + auth: ResolvedModelAuthMaterial, + ): LlmCredentialProvider { if (auth.apiKey !== undefined) { - return new StaticAuthProvider(auth.apiKey); + return staticCredentials(auth.apiKey); } if (auth.oauth !== undefined) { const oauthRef = auth.oauth; const providerKey = auth.oauthProviderKey ?? providerName; const tokens = this.oauth; - return { - canRefresh: true, - async getAuth(options): Promise { - const apiKey = await tokens.getAccessToken(providerKey, oauthRef, { - force: options?.force === true, - }); - return { apiKey }; - }, - }; + return oauthCredentials((options) => + tokens.getAccessToken(providerKey, oauthRef, { force: options?.force === true }), + ); } - return new StaticAuthProvider(undefined); + return staticCredentials(undefined); } } diff --git a/packages/agent-core-v2/src/llm-adapter/model/catalog.ts b/packages/agent-core-v2/src/llm-adapter/model/catalog.ts index fc3d5614b66..8fb634be28e 100644 --- a/packages/agent-core-v2/src/llm-adapter/model/catalog.ts +++ b/packages/agent-core-v2/src/llm-adapter/model/catalog.ts @@ -3,33 +3,19 @@ import { z } from 'zod'; import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; import type { ModelCapability } from '../contract/capability'; +import type { LlmCredentialProvider } from '#human/llm/requester/requester'; import type { TokenUsage } from '#human/llm/usage'; import type { Protocol, ProtocolProviderOptions } from '../protocol/protocol'; import type { ProviderConfig } from '../provider/provider'; import type { ModelRecord } from './model'; import { effectiveModelConfig } from './model-auth'; -import type { ModelRequester } from './model-requester'; -export interface ProviderRequestAuth { - apiKey?: string; - headers?: Record; -} - -export interface AuthProvider { - readonly canRefresh?: boolean; - - getAuth(options?: { readonly force?: boolean }): Promise; -} - -export class StaticAuthProvider implements AuthProvider { - readonly canRefresh = false; - - constructor(private readonly apiKey: string | undefined) {} - async getAuth(): Promise { - if (this.apiKey === undefined || this.apiKey.trim().length === 0) return undefined; - return { apiKey: this.apiKey }; - } -} +import type { + ModelRequestEvent, + ModelRequestInput, + ModelRequestParams, + ModelRequester, +} from './model-requester'; export interface Model { readonly id: string; @@ -51,7 +37,7 @@ export interface Model { readonly providerType?: string; readonly providerName: string; - readonly authProvider: AuthProvider; + readonly credentials?: LlmCredentialProvider; readonly providerOptions?: ProtocolProviderOptions; } @@ -182,6 +168,12 @@ export interface IModelCatalog { get(id: string): Model; getRequester(id: string): ModelRequester; + generate( + id: string, + input: ModelRequestInput, + signal?: AbortSignal, + params?: ModelRequestParams, + ): AsyncIterable; ping(id: string): Promise; findByName(name: string): readonly string[]; diff --git a/packages/agent-core-v2/src/llm-adapter/model/credential-recovery.ts b/packages/agent-core-v2/src/llm-adapter/model/credential-recovery.ts new file mode 100644 index 00000000000..19424961f3c --- /dev/null +++ b/packages/agent-core-v2/src/llm-adapter/model/credential-recovery.ts @@ -0,0 +1,35 @@ +import type { LlmCredentialProvider } from '#human/llm/requester/requester'; + +export async function runWithCredentialRecovery( + credentials: LlmCredentialProvider | undefined, + run: () => Promise, + signal?: AbortSignal, +): Promise { + try { + return await run(); + } catch (error) { + if (signal?.aborted === true || credentials?.canRecover?.(error) !== true) throw error; + credentials?.invalidate?.(); + return run(); + } +} + +export async function* streamWithCredentialRecovery( + credentials: LlmCredentialProvider | undefined, + stream: () => AsyncIterable, + signal?: AbortSignal, +): AsyncIterable { + let recovered = false; + for (;;) { + try { + yield* stream(); + return; + } catch (error) { + if (recovered || signal?.aborted === true || credentials?.canRecover?.(error) !== true) { + throw error; + } + recovered = true; + credentials?.invalidate?.(); + } + } +} diff --git a/packages/agent-core-v2/src/llm-adapter/model/model-requester-impl.ts b/packages/agent-core-v2/src/llm-adapter/model/model-requester-impl.ts index d347a8e9cbd..2cbb94a2405 100644 --- a/packages/agent-core-v2/src/llm-adapter/model/model-requester-impl.ts +++ b/packages/agent-core-v2/src/llm-adapter/model/model-requester-impl.ts @@ -10,8 +10,8 @@ import type { ProviderMediaContribution, VideoUploadInput } from '#human/llm/med import { createMessageAccumulator, type VideoURLPart } from '#human/llm/message'; import type { LlmModel } from '#human/llm/model'; import type { ProtocolName } from '#human/llm/protocol/base'; +import { applyCredential, resolveModelCredentials } from '#human/credentials/credentials'; import { - mergeRequestHeaders, type ExtraParams, type LlmRequestConfig, type LlmRequestContent, @@ -19,17 +19,11 @@ import { type LlmRequester, } from '#human/llm/requester/requester'; import type { TokenUsage } from '#human/llm/usage'; -import { - withAuth, - withAuthUpload, - type CredentialSource, -} from '#human/kimi-oauth/credential-source'; import { ChatProviderError, errorFromLlmMessage, isAbortError, - isUnauthorizedLlmError, llmMessageFromError, traceIdFromHeadersRecord, VideoUploadUnsupportedError, @@ -37,7 +31,7 @@ import { import { fromLlmAssistantMessage, toLlmMessage, type Tool } from '../contract/message'; import { mergeUsagePatch } from '#human/llm/usage'; -import type { Model, ProviderRequestAuth } from './catalog'; +import type { Model } from './catalog'; import type { ModelRequestEvent, ModelRequestInput, @@ -83,20 +77,11 @@ export class ModelRequesterImpl implements ModelRequester { private requesterFor(resolved: ResolvedLlmModel): LlmRequester { if (this.cachedRequester === undefined) { - this.cachedRequester = withAuth(throwToEvent(resolved.requester), this.credentialSource); + this.cachedRequester = throwToEvent(resolved.requester); } return this.cachedRequester; } - private readonly credentialSource: CredentialSource = { - resolve: async (model, options) => { - const auth = await this.model.authProvider.getAuth({ force: options?.force }); - return applyAuth(model, auth); - }, - canRecover: (_model, error) => - this.model.authProvider.canRefresh === true && isUnauthorizedLlmError(error), - }; - request( input: ModelRequestInput, signal?: AbortSignal, @@ -122,8 +107,8 @@ export class ModelRequesterImpl implements ModelRequester { ); } const video = typeof input === 'string' ? readVideoFile(input) : input; - const wrapped = withAuthUpload(uploader, this.credentialSource); - return wrapped(video, { model: resolved.model, signal: options?.signal }); + const model = await resolveModelCredentials(resolved.model, this.model.credentials); + return uploader(video, { model, signal: options?.signal }); } private async runRequest( @@ -172,71 +157,76 @@ export class ModelRequesterImpl implements ModelRequester { usedContextTokens: params?.usedContextTokens, }; - await requester.generate(config, content, { - signal: signal ?? new AbortController().signal, - onEvent: (event: LlmRequestEvent) => { - switch (event.type) { - case 'llm.sent': { - const now = Date.now(); - if (requestSentAt !== undefined) { - requestStartedAt = now; - accumulator = createMessageAccumulator(); - usage = undefined; - finish = undefined; - messageId = undefined; + const credential = await this.model.credentials?.resolve(); + await requester.generate( + { ...config, model: applyCredential(resolved.model, credential) }, + content, + { + signal: signal ?? new AbortController().signal, + onEvent: (event: LlmRequestEvent) => { + switch (event.type) { + case 'llm.sent': { + const now = Date.now(); + if (requestSentAt !== undefined) { + requestStartedAt = now; + accumulator = createMessageAccumulator(); + usage = undefined; + finish = undefined; + messageId = undefined; + } + requestSentAt = now; + return; } - requestSentAt = now; - return; - } - case 'llm.streaming.headers': { - traceId = traceIdFromHeadersRecord(event.headers); - params?.onTraceId?.(traceId); - return; - } - case 'llm.streaming.part': { - const arrivedAt = Date.now(); - if (firstChunkAt === undefined) { - firstChunkAt = arrivedAt; - decodeEluStart = performance.eventLoopUtilization(); - } else { - serverDecodeMs += arrivedAt - lastResumeAt; + case 'llm.streaming.headers': { + traceId = traceIdFromHeadersRecord(event.headers); + params?.onTraceId?.(traceId); + return; } - accumulator.push(event.part); - queue.push({ type: 'part', part: event.part }); - lastResumeAt = Date.now(); - clientConsumeMs += lastResumeAt - arrivedAt; - return; - } - case 'llm.streaming.usage': { - usage = mergeUsagePatch(usage, event.usage); - return; - } - case 'llm.streaming.finish': { - finish = event.finish; - return; - } - case 'llm.streaming.message_id': { - messageId = event.messageId; - return; - } - case 'llm.failed.syntax': - case 'llm.failed.remote': { - failed = event.error; - return; - } - case 'llm.done': { - streamEndedAt = Date.now(); - if (firstChunkAt !== undefined) { - serverDecodeMs += streamEndedAt - lastResumeAt; - if (decodeEluStart !== undefined) { - decodeEluEnd = performance.eventLoopUtilization(decodeEluStart); + case 'llm.streaming.part': { + const arrivedAt = Date.now(); + if (firstChunkAt === undefined) { + firstChunkAt = arrivedAt; + decodeEluStart = performance.eventLoopUtilization(); + } else { + serverDecodeMs += arrivedAt - lastResumeAt; } + accumulator.push(event.part); + queue.push({ type: 'part', part: event.part }); + lastResumeAt = Date.now(); + clientConsumeMs += lastResumeAt - arrivedAt; + return; + } + case 'llm.streaming.usage': { + usage = mergeUsagePatch(usage, event.usage); + return; + } + case 'llm.streaming.finish': { + finish = event.finish; + return; + } + case 'llm.streaming.message_id': { + messageId = event.messageId; + return; + } + case 'llm.failed.syntax': + case 'llm.failed.remote': { + failed = event.error; + return; + } + case 'llm.done': { + streamEndedAt = Date.now(); + if (firstChunkAt !== undefined) { + serverDecodeMs += streamEndedAt - lastResumeAt; + if (decodeEluStart !== undefined) { + decodeEluEnd = performance.eventLoopUtilization(decodeEluStart); + } + } + return; } - return; } - } + }, }, - }); + ); if (failed !== undefined) { throw errorFromLlmMessage(failed); @@ -293,15 +283,6 @@ function finalizeDecodeStats( }; } -function applyAuth(model: LlmModel, auth: ProviderRequestAuth | undefined): LlmModel { - if (auth === undefined) return model; - return { - ...model, - apiKey: auth.apiKey ?? model.apiKey, - defaultHeaders: mergeRequestHeaders(model.defaultHeaders, auth.headers), - }; -} - function throwToEvent(inner: LlmRequester): LlmRequester { return { async generate(config, content, control) { diff --git a/packages/agent-core-v2/test/agent/llmRequester/llmRequesterService.test.ts b/packages/agent-core-v2/test/agent/llmRequester/llmRequesterService.test.ts index b3cc79804fe..daef4fa7e27 100644 --- a/packages/agent-core-v2/test/agent/llmRequester/llmRequesterService.test.ts +++ b/packages/agent-core-v2/test/agent/llmRequester/llmRequesterService.test.ts @@ -117,7 +117,6 @@ function createRequester( maxContextSize: 1000, alwaysThinking: false, providerName: 'p', - authProvider: { getAuth: async () => undefined }, }; return { model, @@ -168,6 +167,7 @@ function createService( ix.stub(IBootstrapService, stubBootstrap('/tmp/kimi-code-llm-requester-test', options.env ?? {})); const thinkingLevel = options.thinkingLevel ?? 'off'; const profile: Partial = { + hasProvider: () => true, resolveModelContext: () => ({ modelAlias: 'm', modelCapabilities: capabilities, @@ -754,7 +754,6 @@ describe('AgentLLMRequesterService trace id', () => { maxContextSize: 1000, alwaysThinking: false, providerName: 'p', - authProvider: { getAuth: async () => undefined }, }; return { model, diff --git a/packages/agent-core-v2/test/agent/loop/loop.test.ts b/packages/agent-core-v2/test/agent/loop/loop.test.ts index 3a10f5c824e..d158ee0324a 100644 --- a/packages/agent-core-v2/test/agent/loop/loop.test.ts +++ b/packages/agent-core-v2/test/agent/loop/loop.test.ts @@ -1918,6 +1918,33 @@ describe('aborted step tool execution', () => { await ctx.dispose(); } }); + + it('settles a message-less notification when credential resolution rejects before the first request', async () => { + const rejectingCredentials = () => ({ + resolve: () => Promise.reject(new Error('OAuth login required')), + }); + const requester: IAgentLLMRequesterService = { + _serviceBrand: undefined, + prepareTurnConfig: () => ({ thinkingEffort: 'off' }), + currentCredentials: rejectingCredentials, + credentialsForTurn: rejectingCredentials, + async request() { + throw new Error('request must not run'); + }, + start() { + throw new Error('request must not run'); + }, + }; + const ctx = createTestAgent(agentService(IAgentLLMRequesterService, requester)); + try { + const loopService = ctx.get(IAgentLoopService); + const handle = loopService.notify(); + await loopService.settled(); + expect(handle.dropped).toBe(false); + } finally { + await ctx.dispose(); + } + }); }); function submitTurn(loop: IAgentLoopService, text: string): { readonly turn: Turn } { @@ -1945,6 +1972,8 @@ function createTimingRequester(): IAgentLLMRequesterService { const requester: IAgentLLMRequesterService = { _serviceBrand: undefined, prepareTurnConfig: () => ({ thinkingEffort: 'off' }), + currentCredentials: () => undefined, + credentialsForTurn: () => undefined, async request(_overrides, onPart = () => {}) { await onPart({ type: 'text', text: 'answer' }); return { diff --git a/packages/agent-core-v2/test/agent/media/mediaResolver.test.ts b/packages/agent-core-v2/test/agent/media/mediaResolver.test.ts index 30332a495a3..3d551f0aae1 100644 --- a/packages/agent-core-v2/test/agent/media/mediaResolver.test.ts +++ b/packages/agent-core-v2/test/agent/media/mediaResolver.test.ts @@ -24,6 +24,7 @@ import { ITelemetryService } from '#/app/telemetry/telemetry'; import type { ModelCapability } from '#/llm-adapter/contract/capability'; import type { Message } from '#/llm-adapter/contract/message'; import type { ContentPart, VideoURLPart } from '#human/llm/message'; +import type { LlmCredentialProvider } from '#human/llm/requester/requester'; import type { ModelRequester } from '#/llm-adapter/model/model-requester'; import type { Protocol } from '#/llm-adapter/protocol/protocol'; import { IBlobStore } from '#/persistence/interface/blobStore'; @@ -161,6 +162,7 @@ function requester(opts: { protocol?: Protocol; providerType?: string; uploadVideo?: ModelRequester['uploadVideo']; + credentials?: LlmCredentialProvider; }): ModelRequester { return { model: { @@ -177,7 +179,7 @@ function requester(opts: { alwaysThinking: false, providerName: 'p', providerType: opts.providerType ?? 'kimi', - authProvider: {} as never, + credentials: opts.credentials, }, request: () => { throw new Error('unused'); @@ -339,6 +341,29 @@ describe('AgentMediaResolverService video strategy', () => { ).rejects.toThrow('unauthorized'); }); + it('invalidates recoverable credentials and retries the upload once on a 401', async () => { + let invalidations = 0; + const credentials: LlmCredentialProvider = { + resolve: () => ({ apiKey: 'tok' }), + canRecover: (error) => (error as { statusCode?: number }).statusCode === 401, + invalidate: () => { + invalidations += 1; + }, + }; + const upload = vi.fn(async (): Promise => msPart('prov-9')); + upload.mockRejectedValueOnce(Object.assign(new Error('unauthorized'), { statusCode: 401 })); + const res = resolver(new Map([[FILE_ID, { name: 'clip.mp4', bytes: VIDEO_BYTES }]])); + + const out = await res.resolve( + [videoMessage(buildKimiFileUrl(FILE_ID))], + requester({ uploadVideo: upload, credentials }), + ); + + expect(firstPart(out)).toEqual(msPart('prov-9')); + expect(upload).toHaveBeenCalledTimes(2); + expect(invalidations).toBe(1); + }); + it('rethrows a cancelled upload without memoizing the fallback', async () => { const controller = new AbortController(); const interrupted = vi.fn(async () => { diff --git a/packages/agent-core-v2/test/agent/profile/profileOps.test.ts b/packages/agent-core-v2/test/agent/profile/profileOps.test.ts index e10a786bf75..3a3a3331e4d 100644 --- a/packages/agent-core-v2/test/agent/profile/profileOps.test.ts +++ b/packages/agent-core-v2/test/agent/profile/profileOps.test.ts @@ -92,7 +92,6 @@ function createTestModel( alwaysThinking: false, providerType, providerName: 'kimi', - authProvider: { getAuth: async () => undefined }, }; } @@ -107,6 +106,9 @@ function createModelCatalogStub(models: Readonly> = {}): I getRequester: () => { throw new Error('not exercised'); }, + generate: () => { + throw new Error('not exercised'); + }, ping: () => { throw new Error('not exercised'); }, diff --git a/packages/agent-core-v2/test/llm-adapter/model/catalog.test.ts b/packages/agent-core-v2/test/llm-adapter/model/catalog.test.ts index 01f4fc7b17a..2fcfc325dc6 100644 --- a/packages/agent-core-v2/test/llm-adapter/model/catalog.test.ts +++ b/packages/agent-core-v2/test/llm-adapter/model/catalog.test.ts @@ -286,7 +286,7 @@ describe('Model assembly (pure data)', () => { } }); - it('resolves provider env-bag credentials and endpoints through the registry', () => { + it('resolves provider env-bag credentials and endpoints through the registry', async () => { const { host, catalog } = createHost({ providers: { kimi: { type: 'kimi', env: { KIMI_API_KEY: 'env-token', KIMI_BASE_URL: 'https://kimi-env.example.test/v1' } }, @@ -300,7 +300,7 @@ describe('Model assembly (pure data)', () => { try { const kimi = catalog.get('k1'); expect(kimi.baseUrl).toBe('https://kimi-env.example.test/v1'); - return expect(kimi.authProvider.getAuth()).resolves.toEqual({ apiKey: 'env-token' }); + return expect(await kimi.credentials?.resolve()).toEqual({ apiKey: 'env-token' }); } finally { host.dispose(); } @@ -473,7 +473,7 @@ describe('Model assembly (pure data)', () => { } }); - it('builds a refreshable OAuth auth provider for oauth-backed models', async () => { + it('builds recoverable OAuth credentials for oauth-backed models', async () => { const tokenProvider = stubTokenProvider(['tok-1']); const { host, catalog } = createHost( { @@ -486,8 +486,10 @@ describe('Model assembly (pure data)', () => { ); try { const model = catalog.get('k1'); - expect(model.authProvider.canRefresh).toBe(true); - await expect(model.authProvider.getAuth()).resolves.toEqual({ apiKey: 'tok-1' }); + expect(model.credentials?.canRecover?.(Object.assign(new Error('x'), { status: 401 }))).toBe( + true, + ); + await expect(model.credentials?.resolve()).resolves.toEqual({ apiKey: 'tok-1' }); } finally { host.dispose(); } diff --git a/packages/agent-core-v2/test/llm-adapter/model/credential-recovery.test.ts b/packages/agent-core-v2/test/llm-adapter/model/credential-recovery.test.ts new file mode 100644 index 00000000000..aef8dfc6fcf --- /dev/null +++ b/packages/agent-core-v2/test/llm-adapter/model/credential-recovery.test.ts @@ -0,0 +1,127 @@ +import { describe, expect, it } from 'vitest'; + +import { + runWithCredentialRecovery, + streamWithCredentialRecovery, +} from '#/llm-adapter/model/credential-recovery'; +import type { LlmCredentialProvider } from '#human/llm/requester/requester'; + +const unauthorized = Object.assign(new Error('unauthorized'), { status: 401 }); +const forbidden = Object.assign(new Error('forbidden'), { status: 403 }); + +describe('runWithCredentialRecovery', () => { + it('returns the result without touching credentials on success', async () => { + let invalidations = 0; + const provider: LlmCredentialProvider = { + resolve: () => undefined, + canRecover: () => true, + invalidate: () => (invalidations += 1), + }; + const result = await runWithCredentialRecovery(provider, () => Promise.resolve('ok')); + expect(result).toBe('ok'); + expect(invalidations).toBe(0); + }); + + it('invalidates and retries once on a recoverable error', async () => { + let invalidations = 0; + let runs = 0; + const provider: LlmCredentialProvider = { + resolve: () => undefined, + canRecover: (error) => error === unauthorized, + invalidate: () => (invalidations += 1), + }; + const result = await runWithCredentialRecovery(provider, () => { + runs += 1; + return runs === 1 ? Promise.reject(unauthorized) : Promise.resolve('ok'); + }); + expect(result).toBe('ok'); + expect(runs).toBe(2); + expect(invalidations).toBe(1); + }); + + it('rethrows when the error is not recoverable or the signal is aborted', async () => { + const provider: LlmCredentialProvider = { + resolve: () => undefined, + canRecover: (error) => error === unauthorized, + invalidate: () => {}, + }; + await expect( + runWithCredentialRecovery(provider, () => Promise.reject(forbidden)), + ).rejects.toBe(forbidden); + + const controller = new AbortController(); + controller.abort(); + await expect( + runWithCredentialRecovery(provider, () => Promise.reject(unauthorized), controller.signal), + ).rejects.toBe(unauthorized); + }); + + it('propagates the failure of the retry attempt', async () => { + const provider: LlmCredentialProvider = { + resolve: () => undefined, + canRecover: () => true, + invalidate: () => {}, + }; + let runs = 0; + await expect( + runWithCredentialRecovery(provider, () => { + runs += 1; + return Promise.reject(unauthorized); + }), + ).rejects.toBe(unauthorized); + expect(runs).toBe(2); + }); +}); + +describe('streamWithCredentialRecovery', () => { + async function collect(stream: AsyncIterable): Promise { + const items: T[] = []; + for await (const item of stream) items.push(item); + return items; + } + + it('re-creates the stream once after a recoverable error', async () => { + let invalidations = 0; + let factories = 0; + const provider: LlmCredentialProvider = { + resolve: () => undefined, + canRecover: (error) => error === unauthorized, + invalidate: () => (invalidations += 1), + }; + const items = await collect( + streamWithCredentialRecovery(provider, async function* () { + factories += 1; + yield 'a'; + if (factories === 1) throw unauthorized; + yield 'b'; + }), + ); + expect(items).toEqual(['a', 'a', 'b']); + expect(factories).toBe(2); + expect(invalidations).toBe(1); + }); + + it('rethrows a second recoverable error and non-recoverable errors', async () => { + const provider: LlmCredentialProvider = { + resolve: () => undefined, + canRecover: (error) => error === unauthorized, + invalidate: () => {}, + }; + await expect( + collect( + streamWithCredentialRecovery(provider, async function* () { + yield 'x'; + throw unauthorized; + }), + ), + ).rejects.toBe(unauthorized); + await expect( + collect( + streamWithCredentialRecovery(provider, async function* () { + yield 'x'; + throw forbidden; + }), + ), + ).rejects.toBe(forbidden); + }); +}); diff --git a/packages/agent-core-v2/test/llm-adapter/model/modelRequester.test.ts b/packages/agent-core-v2/test/llm-adapter/model/modelRequester.test.ts index 3928ff1513f..311601f6d0b 100644 --- a/packages/agent-core-v2/test/llm-adapter/model/modelRequester.test.ts +++ b/packages/agent-core-v2/test/llm-adapter/model/modelRequester.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'vitest'; import { isError2 } from '#/_base/errors/errors'; +import { oauthCredentials, staticCredentials } from '#human/credentials/credentials'; import type { ProviderMediaContribution } from '#human/llm/media/upload'; import type { LlmModel } from '#human/llm/model'; import type { @@ -83,7 +84,7 @@ function textStream(emit: (event: LlmRequestEvent) => void, text = 'hello'): voi ]); } -function modelWith(authProvider: Model['authProvider']): Model { +function modelWith(credentials: Model['credentials']): Model { return { id: 'm1', name: 'fake-model', @@ -102,15 +103,10 @@ function modelWith(authProvider: Model['authProvider']): Model { alwaysThinking: false, providerType: 'fake', providerName: 'fake', - authProvider, + credentials, }; } -const staticAuth = (apiKey?: string): Model['authProvider'] => ({ - canRefresh: false, - getAuth: () => Promise.resolve(apiKey === undefined ? undefined : { apiKey }), -}); - async function collect(stream: AsyncIterable): Promise { const events: ModelRequestEvent[] = []; for await (const event of stream) events.push(event); @@ -123,7 +119,7 @@ describe('ModelRequesterImpl request execution', () => { it('maps ModelRequestParams onto LlmRequestConfig, content and control', async () => { const requester = new FakeLlmRequester(); requester.handler = (_i, emit) => textStream(emit); - const impl = new ModelRequesterImpl(modelWith(staticAuth('sk-1')), gatewayReturning(requester)); + const impl = new ModelRequesterImpl(modelWith(staticCredentials('sk-1')), gatewayReturning(requester)); const signal = AbortSignal.timeout(1000); const messages: Message[] = [ { @@ -198,7 +194,7 @@ describe('ModelRequesterImpl request execution', () => { it('omits the thinking intent when no effort is requested', async () => { const requester = new FakeLlmRequester(); requester.handler = (_i, emit) => textStream(emit); - const impl = new ModelRequesterImpl(modelWith(staticAuth()), gatewayReturning(requester)); + const impl = new ModelRequesterImpl(modelWith(staticCredentials()), gatewayReturning(requester)); await collect(impl.request(INPUT)); expect(requester.calls[0]?.config.thinking).toBeUndefined(); expect(requester.calls[0]?.config.extraParams).toBeUndefined(); @@ -225,7 +221,7 @@ describe('ModelRequesterImpl request execution', () => { { type: 'llm.done' }, ]); const traceIds: Array = []; - const impl = new ModelRequesterImpl(modelWith(staticAuth()), gatewayReturning(requester)); + const impl = new ModelRequesterImpl(modelWith(staticCredentials()), gatewayReturning(requester)); const events = await collect( impl.request(INPUT, undefined, { onTraceId: (id) => traceIds.push(id) }), ); @@ -261,46 +257,7 @@ describe('ModelRequesterImpl request execution', () => { expect(traceIds).toEqual(['trace-1']); }); - it('replays once after a forced token refresh on 401', async () => { - const requester = new FakeLlmRequester(); - requester.handler = (callIndex, emit) => { - if (callIndex === 0) { - emit({ - type: 'llm.failed.remote', - error: { - kind: 'status', - statusCode: 401, - message: 'unauthorized', - requestId: null, - retryAfterMs: null, - headers: null, - }, - }); - return; - } - textStream(emit, 'ok'); - }; - const authCalls: Array<{ force?: boolean }> = []; - const impl = new ModelRequesterImpl( - modelWith({ - canRefresh: true, - getAuth: (options) => { - authCalls.push(options ?? {}); - return Promise.resolve({ apiKey: authCalls.length === 1 ? 'tok-1' : 'tok-2' }); - }, - }), - gatewayReturning(requester), - ); - - const events = await collect(impl.request(INPUT)); - expect(events.some((e) => e.type === 'finish')).toBe(true); - expect(requester.calls).toHaveLength(2); - expect(requester.calls[0]?.config.model.apiKey).toBe('tok-1'); - expect(requester.calls[1]?.config.model.apiKey).toBe('tok-2'); - expect(authCalls).toEqual([{ force: undefined }, { force: true }]); - }); - - it('surfaces a replay-surviving 401 as provider.auth_error', async () => { + it('surfaces a 401 as provider.auth_error without replaying', async () => { const requester = new FakeLlmRequester(); requester.handler = (_i, emit) => emit({ @@ -315,10 +272,7 @@ describe('ModelRequesterImpl request execution', () => { }, }); const impl = new ModelRequesterImpl( - modelWith({ - canRefresh: true, - getAuth: () => Promise.resolve({ apiKey: 'tok' }), - }), + modelWith(oauthCredentials(() => Promise.resolve('tok'))), gatewayReturning(requester), ); @@ -326,10 +280,10 @@ describe('ModelRequesterImpl request execution', () => { expect(isError2(failure)).toBe(true); expect((failure as { code: string }).code).toBe(PROVIDER_AUTH_ERROR_CODE); expect((failure as Error).message).toContain('account rejected'); - expect(requester.calls).toHaveLength(2); + expect(requester.calls).toHaveLength(1); }); - it('does not replay 401s against a non-refreshable auth provider', async () => { + it('does not replay 401s against static credentials', async () => { const requester = new FakeLlmRequester(); requester.handler = (_i, emit) => emit({ @@ -344,7 +298,7 @@ describe('ModelRequesterImpl request execution', () => { }, }); const impl = new ModelRequesterImpl( - modelWith(staticAuth('sk-bad')), + modelWith(staticCredentials('sk-bad')), gatewayReturning(requester), ); @@ -367,7 +321,7 @@ describe('ModelRequesterImpl request execution', () => { headers: null, }, }); - const impl = new ModelRequesterImpl(modelWith(staticAuth()), gatewayReturning(requester)); + const impl = new ModelRequesterImpl(modelWith(staticCredentials()), gatewayReturning(requester)); const failure = await collect(impl.request(INPUT)).catch((error: unknown) => error); expect((failure as { code: string }).code).toBe(PROVIDER_API_ERROR_CODE); @@ -394,7 +348,7 @@ describe('ModelRequesterImpl request execution', () => { it('uploadVideo presence is the capability declaration', async () => { const requester = new FakeLlmRequester(); const impl = new ModelRequesterImpl( - modelWith(staticAuth('sk-1')), + modelWith(staticCredentials('sk-1')), gatewayReturning(requester), ); await expect(impl.uploadVideo('file-id')).rejects.toThrow(/does not support video upload/); @@ -410,7 +364,7 @@ describe('ModelRequesterImpl request execution', () => { }, }; const withMedia = new ModelRequesterImpl( - modelWith(staticAuth('sk-1')), + modelWith(staticCredentials('sk-1')), gatewayReturning(requester, media), ); const part = await withMedia.uploadVideo({ data: new Uint8Array([1]), mimeType: 'video/mp4' }); @@ -429,7 +383,7 @@ describe('ModelRequesterImpl request execution', () => { emit({ type: 'llm.streaming.finish', finish: { finishReason: 'completed', rawFinishReason: 'stop' } }); emit({ type: 'llm.done' }); }; - const impl = new ModelRequesterImpl(modelWith(staticAuth()), gatewayReturning(requester)); + const impl = new ModelRequesterImpl(modelWith(staticCredentials()), gatewayReturning(requester)); const events = await collect(impl.request(INPUT)); const timing = events.find((event) => event.type === 'timing'); expect(timing).toBeDefined(); diff --git a/packages/agent-core-v2/test/llm-adapter/protocol/protocolAdapterRegistry.test.ts b/packages/agent-core-v2/test/llm-adapter/protocol/protocolAdapterRegistry.test.ts index ddbf72fb74b..3d9e31dd57a 100644 --- a/packages/agent-core-v2/test/llm-adapter/protocol/protocolAdapterRegistry.test.ts +++ b/packages/agent-core-v2/test/llm-adapter/protocol/protocolAdapterRegistry.test.ts @@ -86,7 +86,6 @@ function modelWith(spec: { providerName: spec.providerType ?? spec.protocol, reasoningKey: spec.reasoningKey, supportEfforts: spec.supportEfforts, - authProvider: { canRefresh: false, getAuth: () => Promise.resolve(undefined) }, providerOptions: spec.providerOptions, }; } diff --git a/packages/kap-server/test/helpers/fakeModelCatalog.ts b/packages/kap-server/test/helpers/fakeModelCatalog.ts index 52ee1a9ef17..1752cfaca8d 100644 --- a/packages/kap-server/test/helpers/fakeModelCatalog.ts +++ b/packages/kap-server/test/helpers/fakeModelCatalog.ts @@ -9,6 +9,9 @@ export function fakeModelCatalog(): IModelCatalog { getRequester: () => { throw new Error('modelCatalog.getRequester not exercised in this test'); }, + generate: () => { + throw new Error('modelCatalog.generate not exercised in this test'); + }, ping: () => { throw new Error('modelCatalog.ping not exercised in this test'); }, diff --git a/packages/kap-server/test/history.test.ts b/packages/kap-server/test/history.test.ts index a40fed954c0..2896fe0ac20 100644 --- a/packages/kap-server/test/history.test.ts +++ b/packages/kap-server/test/history.test.ts @@ -122,6 +122,9 @@ describe('server /api/v1/sessions/{sid}/history', () => { getRequester: () => { throw new Error('modelCatalog.getRequester not exercised in this test'); }, + generate: () => { + throw new Error('modelCatalog.generate not exercised in this test'); + }, ping: () => { throw new Error('modelCatalog.ping not exercised in this test'); }, diff --git a/packages/kap-server/test/messages.test.ts b/packages/kap-server/test/messages.test.ts index 72a1dd5c556..83a79424336 100644 --- a/packages/kap-server/test/messages.test.ts +++ b/packages/kap-server/test/messages.test.ts @@ -57,6 +57,9 @@ describe('server-v2 /api/v1/sessions/{sid}/messages', () => { getRequester: () => { throw new Error('modelCatalog.getRequester not exercised in this test'); }, + generate: () => { + throw new Error('modelCatalog.generate not exercised in this test'); + }, ping: () => { throw new Error('modelCatalog.ping not exercised in this test'); }, diff --git a/packages/kap-server/test/modelCatalog.test.ts b/packages/kap-server/test/modelCatalog.test.ts index 8e3c204d2d5..f647175f09a 100644 --- a/packages/kap-server/test/modelCatalog.test.ts +++ b/packages/kap-server/test/modelCatalog.test.ts @@ -294,6 +294,9 @@ describe('server-v2 /api/v1 model/provider catalog', () => { getRequester: () => { throw new Error('unused'); }, + generate: () => { + throw new Error('unused'); + }, ping: async () => { throw new Error('unused'); }, diff --git a/packages/kap-server/test/tasks.test.ts b/packages/kap-server/test/tasks.test.ts index b6cb13f82b3..59662569f77 100644 --- a/packages/kap-server/test/tasks.test.ts +++ b/packages/kap-server/test/tasks.test.ts @@ -60,6 +60,9 @@ describe('server-v2 /api/v1/sessions/{sid}/tasks', () => { getRequester: () => { throw new Error('modelCatalog.getRequester not exercised in this test'); }, + generate: () => { + throw new Error('modelCatalog.generate not exercised in this test'); + }, ping: () => { throw new Error('modelCatalog.ping not exercised in this test'); }, @@ -381,7 +384,7 @@ describe('server-v2 /api/v1/sessions/{sid}/tasks', () => { ]); expect(first.body.code).toBe(0); expect(second.body.code).toBe(0); - const detachedFlags = [first.body.data.detached, second.body.data.detached].sort(); + const detachedFlags = [first.body.data.detached, second.body.data.detached].toSorted(); expect(detachedFlags).toEqual([false, true]); expect(first.body.data.status).toBe('running'); expect(second.body.data.status).toBe('running'); diff --git a/packages/kap-server/test/tools.test.ts b/packages/kap-server/test/tools.test.ts index 626cd15eda1..b0091a38faf 100644 --- a/packages/kap-server/test/tools.test.ts +++ b/packages/kap-server/test/tools.test.ts @@ -51,6 +51,9 @@ describe('server-v2 /api/v1 tools + mcp', () => { getRequester: () => { throw new Error('modelCatalog.getRequester not exercised in this test'); }, + generate: () => { + throw new Error('modelCatalog.generate not exercised in this test'); + }, ping: () => { throw new Error('modelCatalog.ping not exercised in this test'); }, diff --git a/packages/kap-server/test/transcript.test.ts b/packages/kap-server/test/transcript.test.ts index 4cb3ca25a02..df8fe96055e 100644 --- a/packages/kap-server/test/transcript.test.ts +++ b/packages/kap-server/test/transcript.test.ts @@ -132,6 +132,9 @@ describe('server-v2 /api/v1/sessions/{sid}/transcript', () => { getRequester: () => { throw new Error('modelCatalog.getRequester not exercised in this test'); }, + generate: () => { + throw new Error('modelCatalog.generate not exercised in this test'); + }, ping: () => { throw new Error('modelCatalog.ping not exercised in this test'); }, diff --git a/packages/klient/examples/kimi-select-tools.ts b/packages/klient/examples/kimi-select-tools.ts index a6b72b3ff98..a791d780e97 100644 --- a/packages/klient/examples/kimi-select-tools.ts +++ b/packages/klient/examples/kimi-select-tools.ts @@ -63,7 +63,8 @@ import { renderLoadableToolsAnnouncement } from '@moonshot-ai/agent-core-v2/agen import { UNKNOWN_CAPABILITY } from '@moonshot-ai/agent-core-v2/llm-adapter/contract/capability'; import type { Message } from '@moonshot-ai/agent-core-v2/llm-adapter/contract/message'; import type { ToolDescription as Tool } from '@moonshot-ai/agent-core-v2/human/llm/message'; -import type { AuthProvider, Model } from '@moonshot-ai/agent-core-v2/llm-adapter/model/catalog'; +import { staticCredentials } from '@moonshot-ai/agent-core-v2/human/credentials/credentials'; +import type { Model } from '@moonshot-ai/agent-core-v2/llm-adapter/model/catalog'; import { IModelCatalog } from '@moonshot-ai/agent-core-v2/llm-adapter/model/catalog'; import type { ModelRequestInput, @@ -281,10 +282,6 @@ async function probeWireEncoding(): Promise { const port = (server.address() as AddressInfo).port; const registry = new ProtocolAdapterRegistry(); - const staticKey: AuthProvider = { - canRefresh: false, - getAuth: () => Promise.resolve({ apiKey: 'sk-probe' }), - }; const makeRequester = (providerType?: string): ModelRequester => { const model: Model = { id: 'probe', @@ -298,7 +295,7 @@ async function probeWireEncoding(): Promise { alwaysThinking: false, providerType, providerName: providerType ?? 'probe', - authProvider: staticKey, + credentials: staticCredentials('sk-probe'), }; return new ModelRequesterImpl(model, registry); }; diff --git a/packages/klient/examples/model-requester-boundary.ts b/packages/klient/examples/model-requester-boundary.ts index 7d30459b364..46f24bb806b 100644 --- a/packages/klient/examples/model-requester-boundary.ts +++ b/packages/klient/examples/model-requester-boundary.ts @@ -17,9 +17,12 @@ * APIConnectionError / APIEmptyResponseError / ...), so the requester could * translate it into a coded `Error2` (`provider.*` / `context.overflow`). * - owned by ModelRequester — behavior the ChatProvider layer CANNOT - * provide: per-request auth injection and the OAuth 401 → force-refresh → - * single replay, plus the final `translateProviderError` safety net that + * provide: per-request credential injection (credentials.resolve per + * attempt), plus the final `translateProviderError` safety net that * turns even unwrapped raw errors into `Error2` (`internal`). + * - owned by the caller — the OAuth 401 → credentials.invalidate → + * single replay: the requester surfaces the 401, and the call site + * re-resolves and re-sends (mirroring IModelCatalog.generate / ping). * - owned by neither — user cancellation: the standard AbortError * DOMException passes through BOTH layers untranslated, by design. * @@ -56,11 +59,16 @@ import { isAbortError, isToolExchangeAdjacencyError, } from '@moonshot-ai/agent-core-v2/llm-adapter/contract/errors'; +import { + oauthCredentials, + staticCredentials, +} from '@moonshot-ai/agent-core-v2/human/credentials/credentials'; import type { ToolCall, ToolDescription as Tool, } from '@moonshot-ai/agent-core-v2/human/llm/message'; -import type { AuthProvider, Model } from '@moonshot-ai/agent-core-v2/llm-adapter/model/catalog'; +import type { LlmCredentialProvider } from '@moonshot-ai/agent-core-v2/human/llm/requester/requester'; +import type { Model } from '@moonshot-ai/agent-core-v2/llm-adapter/model/catalog'; import { IModelCatalog } from '@moonshot-ai/agent-core-v2/llm-adapter/model/catalog'; import type { ModelRequestInput, @@ -207,6 +215,17 @@ async function collect( return { events, text, toolCalls, usage, finishReason }; } +async function collectWithRecovery(requester: ModelRequester): Promise { + const credentials = requester.model.credentials; + try { + return await collect(requester); + } catch (error) { + if (credentials?.canRecover?.(error) !== true) throw error; + credentials?.invalidate?.(); + return collect(requester); + } +} + // --- stub server ----------------------------------------------------------- type StubHandler = (req: IncomingMessage, res: ServerResponse) => void; @@ -345,7 +364,7 @@ async function probeBoundaries(): Promise { const baseUrl = `http://127.0.0.1:${String(port)}`; const registry = new ProtocolAdapterRegistry(); - const makeRequester = (authProvider: AuthProvider, url = baseUrl): ModelRequester => { + const makeRequester = (credentials: LlmCredentialProvider, url = baseUrl): ModelRequester => { const model: Model = { id: 'probe', name: 'probe-model', @@ -357,14 +376,10 @@ async function probeBoundaries(): Promise { maxContextSize: 8192, alwaysThinking: false, providerName: 'probe', - authProvider, + credentials, }; return new ModelRequesterImpl(model, registry); }; - const staticKey = (apiKey: string): AuthProvider => ({ - canRefresh: false, - getAuth: () => Promise.resolve({ apiKey }), - }); const resetCounts = (): void => { requestCount = 0; lastAuth = undefined; @@ -374,7 +389,7 @@ async function probeBoundaries(): Promise { // 1) happy path — the requester's event envelope on top of the raw stream. resetCounts(); handler = (_req, res) => writePong(res); - const ok = await collect(makeRequester(staticKey('sk-probe'))); + const ok = await collect(makeRequester(staticCredentials('sk-probe'))); assert(ok.text === 'pong', 'happy path assembles streamed text'); assert(ok.events.includes('usage'), 'happy path emits a usage event'); assert(ok.events.includes('finish'), 'happy path emits a finish event'); @@ -383,11 +398,11 @@ async function probeBoundaries(): Promise { report('happy-path', `events=${ok.events.join('>')} text=${JSON.stringify(ok.text)}`, '—'); // 2) 401 with a static key: ChatProvider wraps to APIStatusError(401), the - // requester translates to provider.auth_error. No replay (canRefresh=false). + // requester translates to provider.auth_error. No replay (no canRecover). resetCounts(); handler = (_req, res) => writeJsonError(res, 401, 'invalid api key'); try { - await collect(makeRequester(staticKey('sk-bad'))); + await collect(makeRequester(staticCredentials('sk-bad'))); throw new Error('expected a failure'); } catch (error) { const { outcome, wrappedBy } = describeCaught(error); @@ -397,33 +412,30 @@ async function probeBoundaries(): Promise { report('auth-401-static-key', outcome, wrappedBy); } - // 3) 401 with a refreshable auth provider: the requester force-refreshes - // and replays ONCE — behavior the ChatProvider layer cannot own. + // 3) 401 with refreshable credentials: the call site invalidates and + // replays ONCE — recovery the requester deliberately does not own. resetCounts(); handler = (req, res) => { if (req.headers.authorization === 'Bearer sk-good') writePong(res); else writeJsonError(res, 401, 'token expired'); }; - let getAuthCalls = 0; - const refreshable: AuthProvider = { - canRefresh: true, - getAuth: (options) => { - getAuthCalls += 1; - return Promise.resolve({ apiKey: options?.force === true ? 'sk-good' : 'sk-stale' }); - }, - }; - const replayed = await collect(makeRequester(refreshable)); + let resolveCalls = 0; + const refreshable = oauthCredentials((options) => { + resolveCalls += 1; + return Promise.resolve(options?.force === true ? 'sk-good' : 'sk-stale'); + }); + const replayed = await collectWithRecovery(makeRequester(refreshable)); assert(replayed.text === 'pong', 'refresh+replay succeeds'); - assert(getAuthCalls === 2, 'getAuth called twice (normal + forced)'); + assert(resolveCalls === 2, 'resolve called twice (normal + forced)'); assert(requests() === 2, 'exactly one replay after the 401'); - report('auth-401-refresh-replay', `success after ${String(requestCount)} attempts`, 'ModelRequester ONLY (ChatProvider just throws the 401)'); + report('auth-401-refresh-replay', `success after ${String(requestCount)} attempts`, 'caller-owned recovery (ModelRequester surfaces the 401)'); // 4) 401 that survives a forced refresh: the provider rejected the account // — surfaced as provider.auth_error, not a re-login prompt. resetCounts(); handler = (_req, res) => writeJsonError(res, 401, 'account disabled'); try { - await collect(makeRequester(refreshable)); + await collectWithRecovery(makeRequester(refreshable)); throw new Error('expected a failure'); } catch (error) { const { outcome, wrappedBy } = describeCaught(error); @@ -436,7 +448,7 @@ async function probeBoundaries(): Promise { resetCounts(); handler = (_req, res) => writeJsonError(res, 429, 'too many requests', { 'retry-after': '2' }); try { - await collect(makeRequester(staticKey('sk-probe'))); + await collect(makeRequester(staticCredentials('sk-probe'))); throw new Error('expected a failure'); } catch (error) { const { outcome, wrappedBy } = describeCaught(error); @@ -451,7 +463,7 @@ async function probeBoundaries(): Promise { handler = (_req, res) => writeJsonError(res, 400, 'This model\'s maximum context length is 8192 tokens.'); try { - await collect(makeRequester(staticKey('sk-probe'))); + await collect(makeRequester(staticCredentials('sk-probe'))); throw new Error('expected a failure'); } catch (error) { const { outcome, wrappedBy } = describeCaught(error); @@ -467,7 +479,7 @@ async function probeBoundaries(): Promise { res.end('500 Internal Server Erroroops'); }; try { - await collect(makeRequester(staticKey('sk-probe'))); + await collect(makeRequester(staticCredentials('sk-probe'))); throw new Error('expected a failure'); } catch (error) { const { outcome, wrappedBy } = describeCaught(error); @@ -488,7 +500,7 @@ async function probeBoundaries(): Promise { }); handler = (_req, res) => writePong(res); // unused — nothing listens there try { - await collect(makeRequester(staticKey('sk-probe'), `http://127.0.0.1:${String(deadPort)}`)); + await collect(makeRequester(staticCredentials('sk-probe'), `http://127.0.0.1:${String(deadPort)}`)); throw new Error('expected a failure'); } catch (error) { const { outcome, wrappedBy } = describeCaught(error); @@ -500,7 +512,7 @@ async function probeBoundaries(): Promise { resetCounts(); handler = (_req, res) => writeSse(res, []); try { - await collect(makeRequester(staticKey('sk-probe'))); + await collect(makeRequester(staticCredentials('sk-probe'))); throw new Error('expected a failure'); } catch (error) { const { outcome, wrappedBy } = describeCaught(error); @@ -516,7 +528,7 @@ async function probeBoundaries(): Promise { res.end('data: {this is not json}\n\ndata: [DONE]\n\n'); }; try { - await collect(makeRequester(staticKey('sk-probe'))); + await collect(makeRequester(staticCredentials('sk-probe'))); throw new Error('expected a failure'); } catch (error) { const { outcome, wrappedBy } = describeCaught(error); @@ -533,7 +545,7 @@ async function probeBoundaries(): Promise { }); }; try { - await collect(makeRequester(staticKey('sk-probe'))); + await collect(makeRequester(staticCredentials('sk-probe'))); throw new Error('expected a failure'); } catch (error) { const { outcome, wrappedBy } = describeCaught(error); @@ -566,7 +578,7 @@ async function probeBoundaries(): Promise { sseToolDelta([], 'tool_calls'), SSE_USAGE, ]); - const toolOk = await collect(makeRequester(staticKey('sk-probe')), undefined, TOOL_INPUT); + const toolOk = await collect(makeRequester(staticCredentials('sk-probe')), undefined, TOOL_INPUT); const wireTools = (lastRequestBody as { tools?: { function?: { name?: string } }[] }).tools; assert( wireTools?.some((t) => t.function?.name === 'get_weather') === true, @@ -605,7 +617,7 @@ async function probeBoundaries(): Promise { sseToolDelta([], 'tool_calls'), SSE_USAGE, ]); - const parallel = await collect(makeRequester(staticKey('sk-probe')), undefined, TOOL_INPUT); + const parallel = await collect(makeRequester(staticCredentials('sk-probe')), undefined, TOOL_INPUT); assert(parallel.toolCalls.length === 2, 'two parallel tool calls assembled'); assert( parallel.toolCalls[0]?.name === 'tool_a' && parallel.toolCalls[0]?.arguments === '{"a":1}', @@ -639,7 +651,7 @@ async function probeBoundaries(): Promise { sseToolDelta([], 'tool_calls'), SSE_USAGE, ]); - const malformedArgs = await collect(makeRequester(staticKey('sk-probe')), undefined, TOOL_INPUT); + const malformedArgs = await collect(makeRequester(staticCredentials('sk-probe')), undefined, TOOL_INPUT); assert( malformedArgs.toolCalls[0]?.arguments === '{not json', 'malformed arguments pass through untouched', @@ -664,7 +676,7 @@ async function probeBoundaries(): Promise { sseToolDelta([], 'tool_calls'), SSE_USAGE, ]); - const indexless = await collect(makeRequester(staticKey('sk-probe')), undefined, TOOL_INPUT); + const indexless = await collect(makeRequester(staticCredentials('sk-probe')), undefined, TOOL_INPUT); assert( indexless.toolCalls[0]?.arguments === '{"location":"HZ"}', 'index-less fragments merge into the pending call', @@ -682,7 +694,7 @@ async function probeBoundaries(): Promise { handler = (_req, res) => writeJsonError(res, 400, 'tool_call_id "call_1" is not found'); try { - await collect(makeRequester(staticKey('sk-probe')), undefined, TOOL_HISTORY_INPUT); + await collect(makeRequester(staticCredentials('sk-probe')), undefined, TOOL_HISTORY_INPUT); throw new Error('expected a failure'); } catch (error) { const { outcome, wrappedBy } = describeCaught(error); @@ -698,7 +710,7 @@ async function probeBoundaries(): Promise { // the tool result must hit the wire in the provider's shape. resetCounts(); handler = (_req, res) => writePong(res); - await collect(makeRequester(staticKey('sk-probe')), undefined, TOOL_HISTORY_INPUT); + await collect(makeRequester(staticCredentials('sk-probe')), undefined, TOOL_HISTORY_INPUT); const wireMessages = (lastRequestBody as { messages?: Record[] }).messages; assert( wireMessages?.some( @@ -728,7 +740,7 @@ async function probeBoundaries(): Promise { }; const ac = new AbortController(); try { - for await (const event of makeRequester(staticKey('sk-probe')).request(PING_INPUT, ac.signal)) { + for await (const event of makeRequester(staticCredentials('sk-probe')).request(PING_INPUT, ac.signal)) { if (event.type === 'part') ac.abort(); } throw new Error('expected an abort'); diff --git a/packages/klient/src/transports/memory/dispatcher.ts b/packages/klient/src/transports/memory/dispatcher.ts index abe7ff32a79..a16efd84a82 100644 --- a/packages/klient/src/transports/memory/dispatcher.ts +++ b/packages/klient/src/transports/memory/dispatcher.ts @@ -473,9 +473,9 @@ export function createMemoryDispatcher(root: ScopeLike): MemoryDispatcher { }, stream(scope, service, method, args): AsyncIterable { - // Special case: modelResolver.generate routes to - // getRequester(modelId).request(input, signal, params) because the - // catalog has no `generate` method — the facade synthesises the call. + // Special case: modelResolver.generate routes to IModelCatalog.generate + // (which owns credential recovery); the dispatcher only supplies the + // abort signal so client cancellation still reaches the request. if (service === 'modelResolver' && method === 'generate') { return { [Symbol.asyncIterator]() { @@ -488,9 +488,17 @@ export function createMemoryDispatcher(root: ScopeLike): MemoryDispatcher { const resolved = await resolveScope(scope); const catalog = resolveService(resolved, 'modelResolver'); const [modelId, input, params] = args; - const requester = (catalog as { getRequester(id: string): { request(...a: unknown[]): AsyncIterable } }) - .getRequester(modelId as string); - const iterable = requester.request( + const iterable = ( + catalog as { + generate( + id: string, + input: unknown, + signal: AbortSignal, + params: unknown, + ): AsyncIterable; + } + ).generate( + modelId as string, wireClone(input), controller.signal, wireClone(params),