Merge remote-tracking branch 'origin/worktree/context-source-cards' into worktree/context-forms-remaining

# Conflicts:
#	apps/web/tests/snapshots/queue-actions/layout.expected.md
#	docs/cordis-catalog/events.md
#	docs/cordis-catalog/services.md
#	docs/core-data-structures/core.i18n.yaml
#	docs/core-data-structures/goal.i18n.yaml
#	docs/core-data-structures/goal.md
#	docs/core-data-structures/goal.zh.md
#	docs/event-producer-consumer.md
#	examples/acp-agent/tests/goal-snapshots/goal-session/session.expected.jsonl
#	examples/acp-agent/tests/goal-snapshots/goal-wrapup/session.expected.jsonl
#	examples/acp-agent/tests/snapshots/advanced-toolchain/session.1.jsonl
#	examples/acp-agent/tests/snapshots/advanced-toolchain/session.2.jsonl
#	examples/acp-agent/tests/snapshots/advanced-toolchain/session.jsonl
#	examples/acp-agent/tests/snapshots/bash-spill/session.jsonl
#	examples/acp-agent/tests/snapshots/bash-tool-turn/session.jsonl
#	examples/acp-agent/tests/snapshots/both-mode-turn/session.jsonl
#	examples/acp-agent/tests/snapshots/cancel-tool-calls/session.jsonl
#	examples/acp-agent/tests/snapshots/cancel/session.jsonl
#	examples/acp-agent/tests/snapshots/code-mode-turn/session.jsonl
#	examples/acp-agent/tests/snapshots/code-mode-workspace-context/session.jsonl
#	examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/session.jsonl
#	examples/acp-agent/tests/snapshots/empty-response-retry/session.jsonl
#	examples/acp-agent/tests/snapshots/error-finish/session.jsonl
#	examples/acp-agent/tests/snapshots/escalation-approved/session.jsonl
#	examples/acp-agent/tests/snapshots/escalation-rejected/session.jsonl
#	examples/acp-agent/tests/snapshots/fs-edit/session.jsonl
#	examples/acp-agent/tests/snapshots/fs-escalation-approved/session.jsonl
#	examples/acp-agent/tests/snapshots/fs-policy-reject/session.jsonl
#	examples/acp-agent/tests/snapshots/fs-read-window/session.jsonl
#	examples/acp-agent/tests/snapshots/fs-read/session.jsonl
#	examples/acp-agent/tests/snapshots/fs-write-overwrite/session.jsonl
#	examples/acp-agent/tests/snapshots/fs-write/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-cc-invalid-matcher/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-cc-posttool-block/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-cc-posttool-context/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-cc-pretool-ask/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-cc-pretool-deny/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-cc-promptsubmit-context/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-cc-stop-continue/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-codex-invalid-matcher/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-codex-posttool-block/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-codex-posttool-context/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-codex-pretool-block/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-codex-promptsubmit-context/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-codex-stop-continue/session.jsonl
#	examples/acp-agent/tests/snapshots/lsp-definition/session.jsonl
#	examples/acp-agent/tests/snapshots/missing-sandbox-runner/session.jsonl
#	examples/acp-agent/tests/snapshots/multi-turn/session.jsonl
#	examples/acp-agent/tests/snapshots/packed-chunks/session.jsonl
#	examples/acp-agent/tests/snapshots/parallel-tool-calls/session.jsonl
#	examples/acp-agent/tests/snapshots/partial-landlock-child-failure/session.jsonl
#	examples/acp-agent/tests/snapshots/pty-tools/session.jsonl
#	examples/acp-agent/tests/snapshots/repeat-tool-guard/session.jsonl
#	examples/acp-agent/tests/snapshots/session-query-spill/session.jsonl
#	examples/acp-agent/tests/snapshots/session-sandbox-root/session.jsonl
#	examples/acp-agent/tests/snapshots/session-title-after-turn/session.jsonl
#	examples/acp-agent/tests/snapshots/skill-load/session.jsonl
#	examples/acp-agent/tests/snapshots/subagent-continuable/session.1.jsonl
#	examples/acp-agent/tests/snapshots/subagent-continuable/session.jsonl
#	examples/acp-agent/tests/snapshots/subagent-depth-two-rejection/session.1.jsonl
#	examples/acp-agent/tests/snapshots/subagent-depth-two-rejection/session.2.jsonl
#	examples/acp-agent/tests/snapshots/subagent-depth-two-rejection/session.jsonl
#	examples/acp-agent/tests/snapshots/subagent-fork/session.1.jsonl
#	examples/acp-agent/tests/snapshots/subagent-fork/session.jsonl
#	examples/acp-agent/tests/snapshots/subagent-list-agents/session.1.jsonl
#	examples/acp-agent/tests/snapshots/subagent-list-agents/session.jsonl
#	examples/acp-agent/tests/snapshots/subagent-mixed/session.1.jsonl
#	examples/acp-agent/tests/snapshots/subagent-mixed/session.2.jsonl
#	examples/acp-agent/tests/snapshots/subagent-mixed/session.jsonl
#	examples/acp-agent/tests/snapshots/subagent-multi/session.1.jsonl
#	examples/acp-agent/tests/snapshots/subagent-multi/session.2.jsonl
#	examples/acp-agent/tests/snapshots/subagent-multi/session.jsonl
#	examples/acp-agent/tests/snapshots/subagent-published-run-failure/session.jsonl
#	examples/acp-agent/tests/snapshots/subagent-report/session.1.jsonl
#	examples/acp-agent/tests/snapshots/subagent-report/session.jsonl
#	examples/acp-agent/tests/snapshots/subagent-spawn/session.1.jsonl
#	examples/acp-agent/tests/snapshots/subagent-spawn/session.jsonl
#	examples/acp-agent/tests/snapshots/text-turn/session.jsonl
#	examples/acp-agent/tests/snapshots/todo-write/session.jsonl
#	examples/acp-agent/tests/snapshots/tool-call-turn/session.jsonl
#	examples/acp-agent/tests/snapshots/web-fetch/session.jsonl
#	examples/acp-agent/tests/snapshots/workflow-run/session.1.jsonl
#	examples/acp-agent/tests/snapshots/workflow-run/session.jsonl
#	examples/acp-agent/tests/snapshots/workspace-context/session.jsonl
#	examples/acp-agent/tests/snapshots/workspace-edit/session.jsonl
#	examples/headless-agent/tests/snapshots/goal-tools/stream-json.expected.jsonl
#	examples/headless-agent/tests/snapshots/pty-tools/session.jsonl
#	examples/headless-agent/tests/snapshots/pty-tools/stream-json.expected.jsonl
#	examples/headless-agent/tests/subagent-inheritance-snapshots/parent-override/child.expected.jsonl
#	examples/headless-agent/tests/subagent-inheritance-snapshots/parent-override/parent.expected.jsonl
#	examples/jsonrpc-agent/tests/snapshots/persistent-tools/notifications.expected.jsonl
#	examples/jsonrpc-agent/tests/snapshots/persistent-tools/session.jsonl
#	packages/bash/tool-bash/tests/integration.spec.ts
#	packages/context/time-context/src/index.ts
#	packages/context/tmux-context/src/index.ts
#	packages/core/agent-loop/src/agent.ts
#	packages/core/system-prompt/src/index.ts
#	packages/goal/goal/src/domain.ts
#	packages/goal/goal/src/index.ts
#	packages/goal/goal/src/render.ts
#	packages/plan/plan-mode/src/index.ts
This commit is contained in:
creatixchu
2026-08-06 11:49:03 +08:00
1357 changed files with 27674 additions and 18803 deletions

View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/llm/README.md
README.md: 66b7beabd73cc3fec7230f209a9da0da48a37c95
README.zh.md: 08f561840d1a560cfc9bced6f8757a0b0fc5770a
README.md: 92d9fbfa2b8c8db4700562009db49229b2189ab3
README.zh.md: 5c6e7aad1db6511bdb660b86e257652128db131f

View File

@@ -6,10 +6,10 @@ The LLM seam and its provider adapters. The interface package (`llm`) owns the a
| Package | Role | ctx key |
|---|---|---|
| `llm/` | Abstract LLM service + content-block vocabulary + chunk assembler | `ctx.llm` |
| `token-meter/` | Replay-aware request and surface token measurement | `ctx.tokenMeter` |
| `llm-retry/` | Exact-provider normal or unbounded request retry policy | (listens to `agent/request-error`) |
| `llm-deepseek/` | DeepSeek API adapter (direct fetch + eventsource-parser SSE) | (registers on `ctx.llm`) |
| `llm-pi-ai/` | Multi-provider adapter via `@earendil-works/pi-ai` | (registers on `ctx.llm`) |
| [`llm/`](llm/README.md) | LLM service and shared streaming vocabulary | `ctx.llm` |
| [`token-meter/`](token-meter/README.md) | Replay-aware token measurement | `ctx.tokenMeter` |
| [`llm-retry/`](llm-retry/README.md) | Provider-scoped retry policy | listens to `agent/request-error` |
| [`llm-deepseek/`](llm-deepseek/README.md) | Direct DeepSeek adapter | registers on `ctx.llm` |
| [`llm-pi-ai/`](llm-pi-ai/README.md) | Multi-provider pi-ai adapter | registers on `ctx.llm` |
The interface lives at `llm/llm/`; adapters, retry policy, and the reusable token meter are flat siblings under the group. Requests route by `provider`, while `model` is passed through to the selected adapter. The route-owning adapter supplies retry policy and resolves available exact-model identity, context capacity, and reasoning metadata; the retry executor and token meter remain provider-agnostic. A new provider adapter registers one or more provider routes on `ctx.llm` without touching the consumers. See [twin LLM adapters](../../.agents/notes/implemented/architecture/2026-06-13-twin-llm-adapters.md) for the two shipping implementations, the [replay token meter Agent Note](../../.agents/notes/implemented/architecture/2026-07-15-replay-token-meter-service.md) for measurement ownership, and the [routed model context Agent Note](../../.agents/notes/implemented/architecture/2026-07-20-routed-model-context-and-compaction-policy.md) for capacity and compaction-policy ownership.
Adapters register provider routes on the seam; retry and token measurement remain separate consumers. The child READMEs own routing, metadata, replay, and provider-wire details; the [LLM architecture decisions](../../.agents/notes/implemented/architecture/2026-06-13-twin-llm-adapters.md) own the rationale.

View File

@@ -1,15 +1,15 @@
# llm/LLM大语言模型能力家族
# llm/ — LLM 能力家族
[English](README.md) | 中文
LLM seam 及其提供方适配器。接口包(`llm`拥有抽象服务、内容块词汇和流分片组装器;适配器是 `ctx.llm` 上注册的具体实现。这些全是**产品**包。
LLM(大语言模型)seam 及其提供方适配器。接口包(`llm`负责抽象服务、内容块词汇和流分片组装器;适配器是注册到 `ctx.llm` 的具体实现。这些全是**产品**包。
| 包 | 职责 | ctx key |
|---|---|---|
| `llm/` | 抽象 LLM 服务 + 内容块词汇 + 分片组装器 | `ctx.llm` |
| `token-meter/` | 感知回放的请求 token 与表层 token 测量 | `ctx.tokenMeter` |
| `llm-retry/` | 确切提供方的常规或无界请求重试策略 | 监听 `agent/request-error` |
| `llm-deepseek/` | DeepSeek API 适配器,直接使用 fetch + eventsource-parser 和 SSEServer-Sent Events | 注册到 `ctx.llm` |
| `llm-pi-ai/` | 通过 `@earendil-works/pi-ai` 实现的多提供方适配器 | 注册到 `ctx.llm` |
| [`llm/`](llm/README.md) | LLM 服务和共享流式词汇 | `ctx.llm` |
| [`token-meter/`](token-meter/README.md) | 感知回放的 token 测量 | `ctx.tokenMeter` |
| [`llm-retry/`](llm-retry/README.md) | 提供方作用域的重试策略 | 监听 `agent/request-error` |
| [`llm-deepseek/`](llm-deepseek/README.md) | 直接 DeepSeek 适配器 | 注册到 `ctx.llm` |
| [`llm-pi-ai/`](llm-pi-ai/README.md) | 多提供方 pi-ai 适配器 | 注册到 `ctx.llm` |
接口位于 `llm/llm/`;适配器、重试策略和可复用的 token 计量器以扁平结构并列在该分组下。请求按 `provider` 路由,而 `model` 会原样传给选中的适配器。负责该路由的适配器提供重试策略,并解析可用的确切模型身份、上下文容量和推理元数据;重试执行器与 token 计量器仍与提供方无关。新的提供方适配器只需在 `ctx.llm` 上注册一个或多个提供方路由,无需改动消费方。两个已交付实现见[双生 LLM 适配器](../../.agents/notes/implemented/architecture/2026-06-13-twin-llm-adapters.md),测量归属见[回放 token 计量器 Agent Note](../../.agents/notes/implemented/architecture/2026-07-15-replay-token-meter-service.md)容量与压缩compaction策略归属见[路由模型上下文 Agent Note](../../.agents/notes/implemented/architecture/2026-07-20-routed-model-context-and-compaction-policy.md)
适配器在 seam 上注册提供方路由;重试与 token 测量仍是独立消费方。子 README 负责路由、元数据、回放和提供方协议细节;[LLM 架构决策](../../.agents/notes/implemented/architecture/2026-06-13-twin-llm-adapters.md)负责设计原理

View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/llm/llm-deepseek/README.md
README.md: 020aa65073495526be3f32912b7cd06667c52a2e
README.zh.md: 0b2c9efd5ec9bc08e21be1966e182a703c5ea405
README.md: 0cd265cadb2b2a619613761062ab2cef209bec83
README.zh.md: 1883b054277adfd6c3d02b2a76ead9b3f8b0138f

View File

@@ -40,7 +40,7 @@ The plugin registers the single provider route `deepseek-official` together with
`contextWindow` is optional per configured model and is not exposed through the advisory catalog. `ctx.llm.resolveModelInfo('deepseek-official', model).context` returns an exact model value first, then `defaultContextWindow` for an entry without capacity or an unlisted pass-through id. The adapter default is 1,000,000; pressure-sensitive plugins therefore get deployment-owned capacity without treating the model selector as authoritative. Registering another adapter for `deepseek-official` throws `LlmError('DUPLICATE_ADAPTER')`.
`maxTokens` is the adapter-configured output cap for conversation requests and defaults to 256,000. Exact-model resolution exposes it as `defaultMaxTokens`; `LlmService` materializes that value into `GenerateOptions.maxTokens` before the agent loop writes `request/header`, so the wire request remains reconstructable. An explicit request or `AgentOptions.maxTokens` value wins and is serialized as `max_tokens`. The adapter does not clamp this request budget against `contextWindow`; deployments with a smaller context or provider output limit must configure a compatible `maxTokens`.
`maxTokens` is the adapter-configured output cap for conversation requests and defaults to 256,000. A catalog entry may carry its own `maxTokens`, which wins for that model; an entry without one, and any unlisted pass-through id, resolve to the profile value, so adding a per-model cap changes one model rather than the route. Exact-model resolution exposes the winner as `defaultMaxTokens`; `LlmService` materializes that value into `GenerateOptions.maxTokens` before the agent loop writes `request/header`, so the wire request remains reconstructable. An explicit request or `AgentOptions.maxTokens` value wins and is serialized as `max_tokens`. The adapter does not clamp this request budget against `contextWindow`; deployments with a smaller context or provider output limit must configure a compatible `maxTokens`.
The same exact-model result exposes ordered `off`, `high`, and `max` efforts under `reasoning` for every pass-through model when deployment policy permits thinking. `reasoningEffort` selects the deployment default and falls back to `high` when omitted. `agent/request` can replace it on each conversation step; the resolved value is logged in `request/header`. `high` and `max` enable thinking and serialize as the official top-level `reasoning_effort`; adapter-owned `off` instead serializes `thinking.type: disabled` and omits `reasoning_effort`. An unsupported value fails with `UNSUPPORTED_REASONING_EFFORT` before network I/O.
@@ -63,7 +63,7 @@ The plugin also declares its route in the configurable-provider directory (`ctx.
Every request carries the shared attribution header from dsh-llm's `attributionHeaders()` - the mandatory `User-Agent` baseline identifying the harness (see [dsh-llm § App attribution](../llm/README.md#app-attribution-attributionts)). Direct DeepSeek requests and OpenAI-compatible gateway requests get no provider-specific app-attribution headers under this adapter contract; OpenRouter app attribution is deferred to a future explicit OpenRouter adapter or mode. A request whose `GenerateOptions.purpose` is `compaction` (dsh-compact-basic's auxiliary summarization call) additionally carries `x-deepseek-harness-compact: 1`, so the host can separate compaction traffic from conversation requests.
## Wire-format notes (verified live + against the official docs)
## Wire-format notes
- Streaming only (`stream_options.include_usage` always on). `usage` may arrive attached to the finish chunk or as a trailing usage-only chunk — the translator defers both to `[DONE]`, so `usage` always precedes `finish` and nothing follows `finish`.
- The adapter-owned `off` effort maps to `thinking: {type: 'disabled'}` and never crosses the wire as `reasoning_effort: 'off'`.
@@ -75,10 +75,6 @@ Every request carries the shared attribution header from dsh-llm's `attributionH
Non-2xx responses throw `LlmError` with stable codes: `AUTH` (401/403), `QUOTA` (a response whose provider details identify exhausted quota, balance, or credits), `RATE_LIMIT` (other 429s), `CONTEXT_WINDOW_EXCEEDED` (a 400 whose provider code, type, or message identifies context overflow), `INVALID_REQUEST` (other 400s), `SERVER` (5xx), `HTTP_<status>` otherwise. Its serializable `failure` retains the HTTP status plus a valid positive `Retry-After` seconds/date delay and `x-request-id` / `x-deepseek-request-id` when present. A pre-response transport failure (DNS, refused connection, TLS, proxy) throws `TRANSPORT` naming the configured endpoint and chaining the original rejection as `cause`; caller aborts throw `ABORTED`, and the loop's cancellation signal remains authoritative. Protocol violations throw `STREAM_CLOSED` (no `[DONE]`) or `MALFORMED_RESPONSE` (bad JSON payload). Unknown wire `finish_reason`s (e.g. `content_filter`, `insufficient_system_resource`) become `finish {kind: 'error', failure}` chunks, and a completed stream whose `stop` (or absent) finish opened no content blocks becomes a `finish {kind: 'error'}` with code `EMPTY_RESPONSE` (retried by default policy).
## Testing
Unit suites run against a local `node:http` mock SSE server (no network), including dynamic `high`/`off`/`max` selection, structured HTTP facts, malformed/truncated streams, caller abort, connection failure, and proof that idle timeout aborts the actual body. `tests/dynamic-config.spec.ts` drives real settings-local and credentials-local providers (next-request base-URL/key pickup, literal precedence, keyless onboarding, last-good snapshots, retry-policy re-registration), and `tests/loader-composition.spec.ts` boots the full chain from a test-only `cordis.yml` through the actual Loader and edits `settings.yaml`/`.env` on disk. Real-API coverage lives in `tests/adapter.e2e.ts` (`pnpm run test:e2e`, key-gated): V4 Flash + V4 Pro across thinking enabled/disabled and both official effort levels, including the thinking+tools round trip with reasoning passback and a request whose key exists only in a credentials-local document.
## Model Experience
### DeepSeek request

View File

@@ -40,7 +40,7 @@ harness LLM大语言模型seam 的 DeepSeek chat-completions 适配器:
`contextWindow` 对每个已配置模型都可选,不会通过建议 catalog 公开。`ctx.llm.resolveModelInfo('deepseek-official', model).context` 先返回精确模型值,再对不含容量的配置项或未列出原样传递 id 返回 `defaultContextWindow`。适配器默认值为 1,000,000因此压力敏感插件可以获得由部署决定的容量不会将模型 selector 视为权威。为 `deepseek-official` 注册另一个适配器会抛出 `LlmError('DUPLICATE_ADAPTER')`
`maxTokens` 是适配器为对话请求配置的输出上限,默认值为 256,000。确切模型解析会将公开为 `defaultMaxTokens``LlmService` 会在 agent loop智能体循环写入 `request/header` 前,将该值填入 `GenerateOptions.maxTokens`,从而仍可根据持久记录重建协议请求。显式的请求值或 `AgentOptions.maxTokens` 值优先,并会序列化为 `max_tokens`。适配器不会根据 `contextWindow` 自动调低该请求预算;上下文或提供方输出上限较小的部署必须配置与其相容的 `maxTokens`
`maxTokens` 是适配器为对话请求配置的输出上限,默认值为 256,000。Catalog 配置项可以自带 `maxTokens`,它对该模型胜出;不含该上限的配置项以及任何未列出原样传递 id 都解析为 profile 值,因此新增按模型的上限只改变一个模型,而非整条路由。确切模型解析会将胜出值公开为 `defaultMaxTokens``LlmService` 会在 agent loop智能体循环写入 `request/header` 前,将该值填入 `GenerateOptions.maxTokens`,从而仍可根据持久记录重建协议请求。显式的请求值或 `AgentOptions.maxTokens` 值优先,并会序列化为 `max_tokens`。适配器不会根据 `contextWindow` 自动调低该请求预算;上下文或提供方输出上限较小的部署必须配置与其相容的 `maxTokens`
同一确切模型结果会在部署策略允许思考时,为每个原样传递模型在 `reasoning` 下公开有序的 `off``high``max` 推理reasoning强度。`reasoningEffort` 选择部署默认值,省略时回退为 `high``agent/request` 可以在每个会话步骤替换它;解析后的值会记录在 `request/header``high``max` 会启用思考,并序列化为官方顶层 `reasoning_effort`;适配器持有的 `off` 则序列化为 `thinking.type: disabled`,且省略 `reasoning_effort`。不支持的值会在网络 I/O 前以 `UNSUPPORTED_REASONING_EFFORT` 失败。
@@ -63,7 +63,7 @@ harness LLM大语言模型seam 的 DeepSeek chat-completions 适配器:
每个请求都携带 dsh-llm `attributionHeaders()` 的共享归因标头,即用于识别 harness 的必需 `User-Agent` 基线(见 [dsh-llm § 应用归因](../llm/README.md#app-attribution-attributionts)。在该适配器契约adapter contract直接 DeepSeek 请求与 OpenAI 兼容 gateway 请求都不会获得提供方特定应用归因标头OpenRouter 应用归因暂缓到未来的显式 OpenRouter 适配器或模式。`GenerateOptions.purpose``compaction` 的请求dsh-compact-basic 的辅助摘要调用)还会携带 `x-deepseek-harness-compact: 1`,让宿主可以将压缩流量与会话请求分开。
## 协议格式说明(已通过实时请求与官方文档验证)
## 协议格式说明
- 只支持流式输出(`stream_options.include_usage` 始终开启)。`usage` 可能附着在 finish 分片上,也可能作为尾随的纯 usage 分片到达;转换器会将两者都延迟到 `[DONE]`,因此 `usage` 始终位于 `finish` 之前,`finish` 之后不会出现任何内容。
- 适配器持有的 `off` 推理强度映射为 `thinking: {type: 'disabled'}`,绝不会以 `reasoning_effort: 'off'` 通过协议发送。
@@ -75,10 +75,6 @@ harness LLM大语言模型seam 的 DeepSeek chat-completions 适配器:
非 2xx 响应会抛出稳定 code 的 `LlmError``AUTH`401/403`QUOTA`(提供方详细信息标识配额、余额或点数耗尽的响应)、`RATE_LIMIT`(其他 429`CONTEXT_WINDOW_EXCEEDED`(提供方 code、type 或 message 标识上下文溢出的 400`INVALID_REQUEST`(其他 400`SERVER`5xx其他情况为 `HTTP_<status>`。其可序列化 `failure` 保留 HTTP 状态,以及有效的正 `Retry-After` 秒数/日期延迟和存在时的 `x-request-id` / `x-deepseek-request-id`。响应前传输失败DNS、连接被拒绝、TLS、proxy会抛出命名已配置端点的 `TRANSPORT`,并将原始拒绝作为 `cause`;调用方 abort 抛出 `ABORTED`,仍以 loop 的取消信号为准。协议违例抛出 `STREAM_CLOSED`(没有 `[DONE]`)或 `MALFORMED_RESPONSE`JSON payload 格式错误)。未知协议 `finish_reason`(例如 `content_filter``insufficient_system_resource`)会变为 `finish {kind: 'error', failure}` 分片;已完成流如果使用 `stop`或缺失finish 但没有开启内容块,就会变为 `finish {kind: 'error'}`code 为 `EMPTY_RESPONSE`(默认策略会重试)。
## 测试
单元套件使用本地 `node:http` mock SSE 服务器(无网络),覆盖动态 `high``off``max` 选择、结构化 HTTP 事实、格式错误/截断流、调用方 abort、连接失败以及 idle 超时确实会 abort 实际 body 的证明。`tests/dynamic-config.spec.ts` 驱动真实的 settings-local 与 credentials-local provider下一请求即生效的 base-URL密钥拾取、字面值优先、无密钥上手、最后可用快照、重试策略重注册`tests/loader-composition.spec.ts` 则从仅测试用的 `cordis.yml` 出发,经真实 Loader 拉起完整链路,并在磁盘上编辑 `settings.yaml`/`.env`。真实 API 覆盖位于 `tests/adapter.e2e.ts``pnpm run test:e2e`,需有 key 才会运行V4 Flash + V4 Pro覆盖思考启用禁用与两种官方 effort 级别,包括思考 + 工具往返与推理回传,以及密钥仅存在于 credentials-local 文档中的请求。
## 模型体验
### DeepSeek 请求

View File

@@ -35,6 +35,8 @@ export interface DeepSeekCatalogModel {
description?: string
/** Known combined request/response context capacity; omitted when deployment metadata is unavailable. */
contextWindow?: number
/** Per-request output cap for this model; omission falls back to the profile's {@link DeepSeekConnectionOptions.maxTokens}. */
maxTokens?: number
}
/**
@@ -181,7 +183,7 @@ export class DeepSeekAdapter extends LlmAdapter {
? { provider, id: model, name: model }
: modelInfo(provider, configured),
context: { contextWindow },
defaultMaxTokens: connection.maxTokens,
defaultMaxTokens: configured?.maxTokens ?? connection.maxTokens,
...connection.defaults.thinking === 'disabled'
? {
reasoning: {

View File

@@ -68,7 +68,7 @@ export interface Config {
thinking?: 'enabled' | 'disabled'
/** Default thinking effort (default `high`); `off` disables thinking per request. */
reasoningEffort?: 'off' | 'high' | 'max'
/** Default per-request output cap (default 256,000); explicit request values win. */
/** Default per-request output cap (default 256,000); a model's own cap and explicit request values win. */
maxTokens?: number
/** Positive context capacity used when the selected model has no exact value (default 1,000,000). */
defaultContextWindow?: number
@@ -85,6 +85,7 @@ const catalogModel: z<DeepSeekCatalogModel> = z.object({
name: z.string(),
description: z.string(),
contextWindow: z.number().step(1).min(1),
maxTokens: z.number().step(1).min(1),
})
export const Config: z<Config> = z.object({
@@ -125,6 +126,12 @@ function resolveModels(models: readonly DeepSeekCatalogModel[] | undefined): Dee
`llm-deepseek: catalog model "${model.id}" contextWindow must be a positive integer`,
)
}
if (model.maxTokens !== undefined
&& (!Number.isInteger(model.maxTokens) || model.maxTokens <= 0)) {
throw new Error(
`llm-deepseek: catalog model "${model.id}" maxTokens must be a positive integer`,
)
}
if (seen.has(model.id)) throw new Error(`llm-deepseek: duplicate catalog model "${model.id}"`)
seen.add(model.id)
return {
@@ -132,6 +139,7 @@ function resolveModels(models: readonly DeepSeekCatalogModel[] | undefined): Dee
...model.name === undefined ? {} : { name: model.name },
...model.description === undefined ? {} : { description: model.description },
...model.contextWindow === undefined ? {} : { contextWindow: model.contextWindow },
...model.maxTokens === undefined ? {} : { maxTokens: model.maxTokens },
}
})
}

View File

@@ -2,8 +2,6 @@ import { afterEach, describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import LlmService, { createUserMessage,
CONTEXT_WINDOW_EXCEEDED_CODE,
errorChain,
LlmError,
ProviderRequestId,
QUOTA_EXCEEDED_CODE,
ReasoningEffortId,
@@ -206,18 +204,22 @@ describe('DeepSeekAdapter against a mock server', () => {
})
})
it('rejects a per-request effort before I/O when thinking is disabled', async () => {
it('reports a per-request effort failure before I/O when thinking is disabled', async () => {
const server = await mockServer([])
const ctx = await harness(server.url, { thinking: 'disabled' })
await expect(assemble(ctx, {
const result = await assemble(ctx, {
model: 'deepseek-v4-flash',
reasoningEffort: ReasoningEffortId('high'),
messages: [createUserMessage({
content: [{ type: 'text', text: 'hi' }],
source: { kind: 'plugin', plugin: 'test' },
})],
})).rejects.toMatchObject({ code: 'UNSUPPORTED_REASONING_EFFORT' })
})
expect(result.finish).toMatchObject({
kind: 'error',
failure: { code: 'UNSUPPORTED_REASONING_EFFORT' },
})
expect(server.requests).toHaveLength(0)
})
@@ -250,23 +252,22 @@ describe('DeepSeekAdapter against a mock server', () => {
[400, 'INVALID_REQUEST'],
[500, 'SERVER'],
[503, 'SERVER'],
])('maps HTTP %d to LlmError code %s with the body message', async (status, code) => {
])('maps HTTP %d to failure code %s with the body message', async (status, code) => {
const behavior: Behavior = {
kind: 'http-error',
status,
body: JSON.stringify({ error: { message: `failed with ${status}`, type: 't', code: 'c' } }),
}
const server = await mockServer([behavior, behavior])
const server = await mockServer([behavior])
const ctx = await harness(server.url)
await expect(assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] }))
.rejects.toThrow(`failed with ${status}`)
await expect(
assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
.catch((error: unknown) => (error as LlmError).code),
).resolves.toBe(code)
const result = await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(result.finish).toEqual({
kind: 'error',
failure: { message: `failed with ${status}`, code, status },
})
})
it('classifies a thrown HTTP context-window rejection with the canonical code', async () => {
it('classifies an HTTP context-window failure with the canonical code', async () => {
const server = await mockServer([{
kind: 'http-error',
status: 400,
@@ -279,9 +280,11 @@ describe('DeepSeekAdapter against a mock server', () => {
}),
}])
const ctx = await harness(server.url)
const code = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
.catch((error: unknown) => (error as LlmError).code)
expect(code).toBe(CONTEXT_WINDOW_EXCEEDED_CODE)
const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(result.finish).toMatchObject({
kind: 'error',
failure: { code: CONTEXT_WINDOW_EXCEEDED_CODE },
})
})
it('retains status, Retry-After seconds, and provider request id as structured facts', async () => {
@@ -292,19 +295,16 @@ describe('DeepSeekAdapter against a mock server', () => {
headers: { 'retry-after': '2', 'x-request-id': 'req-429' },
}])
const ctx = await harness(server.url)
let thrown: unknown
try {
await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
} catch (error: unknown) {
thrown = error
}
expect(thrown).toBeInstanceOf(LlmError)
expect((thrown as LlmError).failure).toEqual({
message: 'slow down',
code: 'RATE_LIMIT',
status: 429,
providerRetryAfterMs: 2_000,
requestId: ProviderRequestId('req-429'),
const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(result.finish).toEqual({
kind: 'error',
failure: {
message: 'slow down',
code: 'RATE_LIMIT',
status: 429,
providerRetryAfterMs: 2_000,
requestId: ProviderRequestId('req-429'),
},
})
})
@@ -322,16 +322,17 @@ describe('DeepSeekAdapter against a mock server', () => {
},
}])
const ctx = await harness(server.url)
await expect(assemble(ctx, { model: 'deepseek-v4-flash', messages: [] }))
.rejects.toMatchObject({
failure: {
message: 'come back later',
code: 'SERVER',
status: 503,
providerRetryAfterMs: 3_000,
requestId: ProviderRequestId('deepseek-503'),
},
})
const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(result.finish).toEqual({
kind: 'error',
failure: {
message: 'come back later',
code: 'SERVER',
status: 503,
providerRetryAfterMs: 3_000,
requestId: ProviderRequestId('deepseek-503'),
},
})
} finally {
dateNow.mockRestore()
}
@@ -352,13 +353,11 @@ describe('DeepSeekAdapter against a mock server', () => {
headers: { 'retry-after': value },
}])
const ctx = await harness(server.url)
let thrown: LlmError | undefined
try {
await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
} catch (error: unknown) {
if (error instanceof LlmError) thrown = error
}
expect(thrown?.failure).toEqual({ message: 'retry later', code: 'RATE_LIMIT', status: 429 })
const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(result.finish).toEqual({
kind: 'error',
failure: { message: 'retry later', code: 'RATE_LIMIT', status: 429 },
})
}
})
@@ -379,53 +378,50 @@ describe('DeepSeekAdapter against a mock server', () => {
it('keeps the status-line message for JSON error bodies without a message', async () => {
const server = await mockServer([{ kind: 'http-error', status: 500, body: '{"error":{"type":"x"}}' }])
const ctx = await harness(server.url)
await expect(assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] }))
.rejects.toThrow(/HTTP 500/)
const result = await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(result.finish.kind).toBe('error')
if (result.finish.kind !== 'error') throw new Error('expected an error finish')
expect(result.finish.failure.code).toBe('SERVER')
expect(result.finish.failure.message).toMatch(/HTTP 500/)
})
it('keeps the status-line message for non-JSON error bodies', async () => {
const server = await mockServer([{ kind: 'http-error', status: 502, body: 'Bad Gateway', contentType: 'text/plain' }])
const ctx = await harness(server.url)
await expect(assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] }))
.rejects.toThrow(/HTTP 502/)
const result = await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(result.finish.kind).toBe('error')
if (result.finish.kind !== 'error') throw new Error('expected an error finish')
expect(result.finish.failure.code).toBe('SERVER')
expect(result.finish.failure.message).toMatch(/HTTP 502/)
})
it('maps unusual statuses to HTTP_<status>', () => {
expect(httpErrorCode(418)).toBe('HTTP_418')
})
it('wraps a transport failure in TRANSPORT with the fetch cause chain in the message', async () => {
// Port 1 is reserved/unbound: fetch rejects with `TypeError: fetch failed`
// whose actionable detail (ECONNREFUSED) lives on `cause`.
it('reports a transport failure with the endpoint in the message', async () => {
// Port 1 is reserved/unbound, so the service normalizes the fetch failure.
const ctx = await harness('http://127.0.0.1:1')
let caught: unknown
try {
await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
} catch (error: unknown) {
caught = error
}
expect(caught).toBeInstanceOf(LlmError)
const llmError = caught as LlmError
expect(llmError.code).toBe('TRANSPORT')
expect(llmError.message).toContain('http://127.0.0.1:1')
expect(llmError.cause).toBeInstanceOf(TypeError)
// The chain renderer reaches the transport diagnosis through the cause.
expect(errorChain(llmError)).toMatch(/ECONNREFUSED|EADDRNOTAVAIL|bad port/)
const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(result.finish).toMatchObject({
kind: 'error',
failure: {
code: 'TRANSPORT',
message: 'DeepSeek API request to http://127.0.0.1:1 failed',
},
})
})
it('classifies an aborted request without losing the transport rejection', async () => {
it('classifies an aborted request as an aborted finish', async () => {
const controller = new AbortController()
controller.abort()
const ctx = await harness('http://127.0.0.1:1')
let caught: unknown
try {
await assemble(ctx, { model: 'deepseek-v4-flash', messages: [], signal: controller.signal })
} catch (error: unknown) {
caught = error
}
expect(caught).toBeInstanceOf(LlmError)
expect(caught).toMatchObject({ code: 'ABORTED' })
expect((caught as LlmError).cause).toMatchObject({ name: 'AbortError' })
const result = await assemble(ctx, {
model: 'deepseek-v4-flash',
messages: [],
signal: controller.signal,
})
expect(result.finish).toMatchObject({ kind: 'aborted', failure: { code: 'ABORTED' } })
})
it('throws EMPTY_RESPONSE when the response has no body', async () => {
@@ -443,20 +439,17 @@ describe('DeepSeekAdapter against a mock server', () => {
}
})
it('classifies an abrupt body close as TRANSPORT and retains its cause', async () => {
it('classifies an abrupt body close as TRANSPORT', async () => {
const server = await mockServer([{
kind: 'close-early',
events: ['{"choices":[{"delta":{"content":"par"}}]}'],
}])
const ctx = await harness(server.url)
let caught: unknown
try {
await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
} catch (error: unknown) {
caught = error
}
expect(caught).toMatchObject({ code: 'TRANSPORT' })
expect(errorChain(caught)).toMatch(/terminated|socket|without \[DONE\]/)
const result = await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(result.finish.kind).toBe('error')
if (result.finish.kind !== 'error') throw new Error('expected an error finish')
expect(result.finish.failure.code).toBe('TRANSPORT')
expect(result.finish.failure.message).toMatch(/^DeepSeek API stream from .* failed$/)
})
it('aborts mid-stream via the request signal', async () => {
@@ -478,7 +471,13 @@ describe('DeepSeekAdapter against a mock server', () => {
})()
setTimeout(() => { controller.abort() }, 30)
await expect(pending).rejects.toMatchObject({ code: 'ABORTED' })
const chunks = await pending
expect(chunks).toHaveLength(1)
expect(chunks[0]?.type).toBe('finish')
if (chunks[0]?.type !== 'finish') throw new Error('expected a finish chunk')
expect(chunks[0].reason.kind).toBe('aborted')
if (chunks[0].reason.kind !== 'aborted') throw new Error('expected an aborted finish')
expect(chunks[0].reason.failure.code).toBe('ABORTED')
})
it('maps connection failures to TRANSPORT without losing the cause', async () => {
@@ -793,6 +792,26 @@ describe('plugin registration and config', () => {
expect(ctx.llm.listProviders()).toEqual([])
})
it.each([0, 1.5])('rejects a per-model output cap of %s', (maxTokens) => {
expect(() => resolveAdapterOptions({ models: [{ id: 'bad-cap', maxTokens }] }))
.toThrow(/maxTokens must be a positive integer/)
})
it('prefers a model\'s own output cap over the profile default', async () => {
// The profile default stays what an unlisted or uncapped model resolves
// to, so adding a per-model cap changes one model rather than the route.
const adapter = adapterOf({ maxTokens: 4096, models: [
{ id: 'capped', maxTokens: 512 },
{ id: 'uncapped' },
] })
await expect(adapter.resolveModel('deepseek-official', 'capped'))
.resolves.toMatchObject({ defaultMaxTokens: 512 })
await expect(adapter.resolveModel('deepseek-official', 'uncapped'))
.resolves.toMatchObject({ defaultMaxTokens: 4096 })
await expect(adapter.resolveModel('deepseek-official', 'not-in-catalog'))
.resolves.toMatchObject({ defaultMaxTokens: 4096 })
})
it('rejects invalid context capacity when apply is called directly', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
@@ -858,12 +877,15 @@ describe('plugin registration and config', () => {
// only the request itself needs a key.
expect(ctx.llm.listProviders()).toEqual([{ id: 'deepseek-official', name: 'DeepSeek' }])
await expect(ctx.llm.listModels('deepseek-official')).resolves.toHaveLength(2)
await expect(assemble(ctx, { model: 'deepseek-v4-flash', messages: [] }))
.rejects.toMatchObject({ code: 'MISSING_CREDENTIAL' })
const first = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(first.finish).toMatchObject({ kind: 'error', failure: { code: 'MISSING_CREDENTIAL' } })
// The guidance leads with the credential store — the path that keeps the
// secret out of configuration files — and mentions a literal key last.
await expect(assemble(ctx, { model: 'deepseek-v4-flash', messages: [] }))
.rejects.toThrow(/store DEEPSEEK_API_KEY through the credentials service.*as a last resort.*"apiKey"/s)
const second = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(second.finish.kind).toBe('error')
if (second.finish.kind !== 'error') throw new Error('expected an error finish')
expect(second.finish.failure.message)
.toMatch(/store DEEPSEEK_API_KEY through the credentials service.*as a last resort.*"apiKey"/s)
})
it('reads the ambient variable when no credentials seam is mounted', async () => {
@@ -883,8 +905,8 @@ describe('plugin registration and config', () => {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmDeepSeek, { baseURL: 'http://127.0.0.1:1' })
await expect(assemble(ctx, { model: 'deepseek-v4-flash', messages: [] }))
.rejects.toMatchObject({ code: 'MISSING_CREDENTIAL' })
const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(result.finish).toMatchObject({ kind: 'error', failure: { code: 'MISSING_CREDENTIAL' } })
})
it('prefers explicit config over env for key and base URL', async () => {

View File

@@ -96,7 +96,8 @@ describe('request-level dynamic configuration', () => {
const server = await mockServer([{ kind: 'sse', events: textEvents }])
const { ctx } = await boot(dir, { baseURL: server.url })
await expect(prompt(ctx)).rejects.toMatchObject({ code: 'MISSING_CREDENTIAL' })
const keyless = await prompt(ctx)
expect(keyless.finish).toMatchObject({ kind: 'error', failure: { code: 'MISSING_CREDENTIAL' } })
await ctx.credentials.set(KEY_REF, 'sk-arrived')
await prompt(ctx)
expect(server.headers[0]?.authorization).toBe('Bearer sk-arrived')

View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/llm/llm-pi-ai/README.md
README.md: e8c2682cbb72ca1ac6a5ad6b26bdf63f0695716b
README.zh.md: 4175b3a751affa65ac68284c7ead47b1f71b5e15
README.md: 75b2136315aed758f18f7fe82afcd4903f4a7b98
README.zh.md: ea67250549f1d23d48455fd185283b00183dd538

View File

@@ -75,10 +75,6 @@ Every request carries the shared attribution header from dsh-llm's `attributionH
pi-ai installs several provider SDKs and lazy-loads the one selected by the catalog model. The dependency weight is isolated to this opt-in adapter package.
## Testing
Unit tests use pi-ai catalog models redirected to local mock servers and cover provider/profile routing, one wire request per adapter call, idle-timeout response termination, caller abort, native API selection, endpoint overrides, attribution, conversion, replay-state validation, and cross-provider/model replay within one adapter instance. `tests/dynamic-config.spec.ts` drives real settings-local and credentials-local providers: a settings-born route registers live and drops when the user layer resets, `apiKeyEnv` credentials rotate between requests, and an unknown-provider snapshot keeps the last good profiles. `tests/loader-composition.spec.ts` boots the dormant posture from a test-only `cordis.yml` through the actual Loader and registers its route from an on-disk `settings.yaml` edit. Real-API coverage remains key-gated under `pnpm run test:e2e`.
## Model Experience
### Provider request through pi-ai

View File

@@ -75,10 +75,6 @@
pi-ai 会安装多个提供方 SDK并延迟加载 catalog 模型所选的 SDK。该可选适配器包将依赖体量隔离在自身范围内。
## 测试
单元测试使用重定向到本地 mock 服务器的 pi-ai catalog 模型覆盖提供方profile 路由、每次适配器调用只发起一个协议请求、idle-timeout 响应终止、调用方 abort、原生 API 选择、端点覆盖、归因、转换、回放状态验证,以及一个适配器实例内的跨提供方/模型回放。`tests/dynamic-config.spec.ts` 驱动真实的 settings-local 与 credentials-local providersettings 里新生的路由实时完成注册,并在用户层重置时随之移除,`apiKeyEnv` 凭据在两次请求之间轮换,点名未知提供方的快照则保留最后可用 profile。`tests/loader-composition.spec.ts` 从仅测试用的 `cordis.yml` 出发,经真实 Loader 拉起休眠姿态,并从磁盘上的一次 `settings.yaml` 编辑注册出它的路由。真实 API 覆盖仍需 key 才会启用,并通过 `pnpm run test:e2e` 运行。
## 模型体验
### 通过 pi-ai 发起的提供方请求

View File

@@ -85,7 +85,7 @@ describe('PiAiAdapter provider routing', () => {
})
})
it('uses a dynamic request effort and rejects unsupported efforts before network I/O', async () => {
it('uses a dynamic request effort and reports unsupported efforts before network I/O', async () => {
const server = await mockServer([{ events: textEvents }, { events: textEvents }])
const ctx = await harness(server.url, { reasoning: 'max' })
@@ -104,11 +104,15 @@ describe('PiAiAdapter provider routing', () => {
expect(server.requests[1]).toMatchObject({ thinking: { type: 'disabled' } })
expect(server.requests[1]).not.toHaveProperty('reasoning_effort')
await expect(assemble(ctx, {
const unsupported = await assemble(ctx, {
model: 'deepseek-v4-flash',
reasoningEffort: ReasoningEffortId('xhigh'),
messages: [],
})).rejects.toMatchObject({ code: 'UNSUPPORTED_REASONING_EFFORT' })
})
expect(unsupported.finish).toMatchObject({
kind: 'error',
failure: { code: 'UNSUPPORTED_REASONING_EFFORT' },
})
expect(server.requests).toHaveLength(2)
})
@@ -125,19 +129,19 @@ describe('PiAiAdapter provider routing', () => {
expect(result.message.content).toEqual([{ type: 'text', text: 'hello' }])
})
it('rejects stop sequences rather than silently ignoring them', async () => {
it('reports unsupported stop sequences rather than silently ignoring them', async () => {
const server = await mockServer([])
const ctx = await harness(server.url)
await expect(assemble(ctx, { model: 'deepseek-v4-flash', messages: [], stop: ['END'] }))
.rejects.toMatchObject({ code: 'UNSUPPORTED_OPTION' })
const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [], stop: ['END'] })
expect(result.finish).toMatchObject({ kind: 'error', failure: { code: 'UNSUPPORTED_OPTION' } })
expect(server.requests).toEqual([])
})
it('rejects unknown catalog models before network I/O', async () => {
it('reports unknown catalog models before network I/O', async () => {
const server = await mockServer([])
const ctx = await harness(server.url)
await expect(assemble(ctx, { model: 'not-in-the-catalog', messages: [] }))
.rejects.toMatchObject({ code: 'UNKNOWN_MODEL' })
const result = await assemble(ctx, { model: 'not-in-the-catalog', messages: [] })
expect(result.finish).toMatchObject({ kind: 'error', failure: { code: 'UNKNOWN_MODEL' } })
expect(server.requests).toEqual([])
})
@@ -237,8 +241,8 @@ describe('PiAiAdapter provider routing', () => {
const server = await mockServer([{ events: textEvents, delayMs: 200 }])
const ctx = await harness(server.url, { streamIdleTimeoutMs: 20 })
await expect(assemble(ctx, { model: 'deepseek-v4-flash', messages: [] }))
.rejects.toMatchObject({ code: 'TIMEOUT' })
const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(result.finish).toMatchObject({ kind: 'error', failure: { code: 'TIMEOUT' } })
await Promise.race([
server.responseClosed,
new Promise<never>((_resolve, reject) => {
@@ -393,10 +397,12 @@ describe('provider profile lifecycle', () => {
vi.stubEnv('DEEPSEEK_API_KEY', 'ambient-key')
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url, { apiKey: undefined, apiKeyEnv: 'PI_CUSTOM_REF_KEY' })
await expect(assemble(ctx, { model: 'deepseek-v4-flash', messages: [] }))
.rejects.toMatchObject({ code: 'MISSING_CREDENTIAL' })
await expect(assemble(ctx, { model: 'deepseek-v4-flash', messages: [] }))
.rejects.toThrow(/provider route "deepseek".*PI_CUSTOM_REF_KEY/s)
const first = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(first.finish).toMatchObject({ kind: 'error', failure: { code: 'MISSING_CREDENTIAL' } })
const second = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(second.finish.kind).toBe('error')
if (second.finish.kind !== 'error') throw new Error('expected an error finish')
expect(second.finish.failure.message).toMatch(/provider route "deepseek".*PI_CUSTOM_REF_KEY/s)
expect(server.requests).toHaveLength(0)
})

View File

@@ -105,8 +105,8 @@ describe('request-level dynamic profiles', () => {
// composition route stays.
await ctx.settings.replace(NS, {})
expect(ctx.llm.listProviders().map(provider => provider.id)).toEqual(['openai'])
await expect(assemble(ctx, { provider: 'deepseek', model: 'deepseek-v4-flash', messages: [] }))
.rejects.toMatchObject({ code: 'NO_ADAPTER' })
const removed = await assemble(ctx, { provider: 'deepseek', model: 'deepseek-v4-flash', messages: [] })
expect(removed.finish).toMatchObject({ kind: 'error', failure: { code: 'NO_ADAPTER' } })
})
it('rotates the per-request credential referenced by apiKeyEnv', async () => {

View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/llm/llm-retry/README.md
README.md: 8de3ea8c9321f04f5af1b0d7ab361f73eaabc822
README.zh.md: 978854e9466e271535a10fcea406d0dcb5607285
README.md: 23b55a30989cc51d4dd9076b61b6595452b0abd0
README.zh.md: 267ef12a87561fd8effef726a781e505225baf03

View File

@@ -48,6 +48,6 @@ The reconstructed request preserves the prior prefix and is eligible for provide
- **Agent turns are the only retry boundary** — direct `ctx.llm.stream()` consumers remain single-attempt because a raw stream cannot separate already-emitted chunks durably.
- **Always mode retries permanent failures** — authentication, quota, invalid-request, protocol, and unrecoverable context errors continue until success, cancellation, or disposal; deployments own provider-specific cost and latency controls.
- **Finite plugin budgets add** — normal mode counts only its configured codes and exact provider policy, while context-overflow compaction owns a separate budget. A future overlapping policy must document and test registration-order behavior.
- **Finite plugin budgets add** — normal mode counts only its configured codes and exact provider policy, while context-overflow compaction owns a separate budget. Any overlapping policy must define registration-order behavior.
- **Recovery policies compose by waterfall order** — always mode accepts a downstream retry before applying its fallback. A later policy that ignores cancellation and never settles also prevents fallback, turn quiescence, and plugin disposal from completing.
- **`llm/retry` records scheduling, not completion** — later step and turn events establish success, exhaustion, or cancellation.

View File

@@ -48,6 +48,6 @@
- **agent 轮次是唯一重试边界**:直接 `ctx.llm.stream()` 消费方仍只尝试一次,因为原始流无法持久地区分各次尝试已经发出的分片。
- **always mode 会重试永久性失败**:身份验证、配额、无效请求、协议和无法恢复的上下文错误都会继续重试,直至成功、取消或 dispose部署负责提供方特定的成本与延迟控制。
- **有限插件预算可叠加**normal mode 只统计已配置 code 和确切提供方策略上下文溢出压缩compaction则拥有独立预算。未来如有重叠策略必须记录并测试注册顺序行为。
- **有限插件预算可叠加**normal mode 只统计已配置 code 和确切提供方策略上下文溢出压缩compaction则拥有独立预算。任何重叠策略必须定义注册顺序行为。
- **恢复策略按 waterfall 顺序组合**always mode 会先接受下游重试,再应用自己的回退。后续策略如果忽略取消且永不结算,也会阻止回退、轮次完全停稳和插件 dispose 完成。
- **`llm/retry` 记录调度,不是完成**:后续步骤与轮次事件用于确立成功、耗尽或取消。

View File

@@ -1,28 +1,29 @@
/** Durable request-route lookup for one closed model step. @module @deepseek-ai/dsh-llm-retry/history */
/** Durable request-route lookup for one open model step. @module @deepseek-ai/dsh-llm-retry/history */
import type { SessionEvent } from '@deepseek-ai/dsh-session'
/**
* Find the provider in force when one step closed, excluding later recovery mutations.
* Find the provider in force for one currently open step.
* Request headers remain effective across turn boundaries until a newer full
* snapshot changes them; every provider change requires a newer full snapshot.
* @param events - session events containing the closed step.
* @param events - session events ending inside the open step.
* @param turn - turn that owns the failed step.
* @param step - failed step whose provider is required.
* @returns the provider from the request header in force at that step boundary.
* @returns the provider from the request header in force for the step.
*/
export function providerForClosedStep(
export function providerForOpenStep(
events: readonly SessionEvent[],
turn: number,
step: number,
): string | undefined {
const stepEndIndex = events.findLastIndex(event =>
event.type === 'step/end'
const stepStartIndex = events.findLastIndex(event =>
event.type === 'step/start'
&& event.data.turn === turn
&& event.data.step === step,
)
if (stepEndIndex < 0) return undefined
for (let index = stepEndIndex; index >= 0; index -= 1) {
if (stepStartIndex < 0 || events.slice(stepStartIndex + 1).some(event =>
event.type === 'step/end' || event.type === 'turn/end')) return undefined
for (let index = events.length - 1; index >= 0; index -= 1) {
// The loop bounds prove this indexed read exists.
// oxlint-disable-next-line typescript/no-non-null-assertion
const event = events[index]!

View File

@@ -1,5 +1,5 @@
/**
* Provider-routed model-request retry policy on the agent loop's closed-step
* Provider-routed model-request retry policy on the agent loop's request
* recovery seam. Each scheduled retry is durable before its cancellable wait.
*
* @module @deepseek-ai/dsh-llm-retry
@@ -7,14 +7,13 @@
import type { Context } from 'cordis'
import z from 'schemastery'
import type { Agent, RequestError, RequestErrorAction } from '@deepseek-ai/dsh-agent'
import type { Agent, RequestErrorAction, RequestFailureContext } from '@deepseek-ai/dsh-agent'
import type { LlmFailure, ResolvedRetryPolicy } from '@deepseek-ai/dsh-llm'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import { providerForClosedStep } from './history.ts'
declare module '@deepseek-ai/dsh-session' {
interface SessionEventMap {
/** Durable, non-surface record of one provider-routed retry scheduled after a closed failed step. */
/** Durable, non-surface record of one provider-routed retry scheduled after a failed request attempt. */
'llm/retry': {
turn: number
step: number
@@ -174,24 +173,12 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna
async function recover(
agent: Agent,
turn: number,
step: number,
_error: RequestError,
failure: LlmFailure,
priorFailures: readonly LlmFailure[],
policy: ResolvedRetryPolicy | undefined,
context: RequestFailureContext,
signal: AbortSignal,
next: () => Promise<RequestErrorAction>,
): Promise<RequestErrorAction> {
const { turn, step, provider, failure, retryPolicy: policy } = context
if (policy === undefined) return next()
// The call-local policy belongs to the registration that served this
// failure. Recover only the durable provider identity from the header;
// downstream recovery may append later state before an always fallback.
const provider = providerForClosedStep(agent.session.events, turn, step)
/* v8 ignore next 3 -- agent-loop closes only steps whose request header was recorded */
if (provider === undefined) {
throw new Error(`llm-retry: no request provider for closed turn ${turn}/step ${step}`)
}
if (policy.mode === 'always') {
if (signal.aborted || lifetime.signal.aborted) return
const fusedSignal = AbortSignal.any([signal, lifetime.signal])
@@ -213,11 +200,10 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna
}
const policyKey = retryPolicyKey(policy)
const firstPriorTurn = turn - priorFailures.length
const priorPolicyRetry = agent.session.events.findLast((event): event is SessionEvent<'llm/retry'> =>
event.type === 'llm/retry'
&& event.data.turn >= firstPriorTurn
&& event.data.turn < turn
&& event.data.turn === turn
&& event.data.step === step
&& event.data.provider === provider
&& event.data.policyKey === policyKey,
)
@@ -243,12 +229,7 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna
const disposeListener = ctx.on('agent/request-error', (
agent: Agent,
turn: number,
step: number,
error: RequestError,
failure: LlmFailure,
priorFailures: readonly LlmFailure[],
policy: ResolvedRetryPolicy | undefined,
context: RequestFailureContext,
signal: AbortSignal,
next: () => Promise<RequestErrorAction>,
) => {
@@ -256,7 +237,7 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna
// removed. Lifetime cancellation must prevent that stale callback from
// entering a downstream policy after disposal.
if (lifetime.signal.aborted) return Promise.resolve<RequestErrorAction>(undefined)
return track(recover(agent, turn, step, error, failure, priorFailures, policy, signal, next))
return track(recover(agent, context, signal, next))
})
ctx.effect(() => async () => {

View File

@@ -5,7 +5,7 @@ import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
import type { LlmFailure } from '@deepseek-ai/dsh-llm'
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants'
import { providerForClosedStep } from './history.ts'
import { providerForOpenStep } from './history.ts'
import type {} from './index.ts'
const PACKAGE_NAME = '@deepseek-ai/dsh-llm-retry'
@@ -41,35 +41,7 @@ function validateFailure(value: unknown, fail: InvariantFailure): asserts value
}
}
/** Find the first turn in the structured-failure retry chain containing `turn`. */
function retryChainStart(history: readonly SessionEvent[], turn: number): number {
let startIndex = history.findLastIndex(
event => event.type === 'turn/start' && event.data.turn === turn,
)
while (startIndex >= 0) {
const start = history[startIndex]
if (start?.type !== 'turn/start' || start.data.trigger.kind !== 'retry') break
let endIndex = startIndex - 1
while (endIndex >= 0 && history[endIndex]?.type !== 'turn/end') endIndex -= 1
const end = history[endIndex]
if (end?.type !== 'turn/end'
|| end.data.reason.kind !== 'error'
|| end.data.reason.failure === undefined) break
const previousStart = history.findLastIndex(
(event, index) =>
index < endIndex
&& event.type === 'turn/start'
&& event.data.turn === end.data.turn,
)
if (previousStart < 0) break
startIndex = previousStart
}
return startIndex
}
/** Validate one retry record against the open turn and most recently closed step. */
/** Validate one retry record against the currently open request step. */
function validateRetry(
history: readonly SessionEvent[],
event: SessionEvent<'llm/retry'>,
@@ -106,49 +78,34 @@ function validateRetry(
fail(`llm/retry delayMs must be a finite number within 0..${MAX_TIMER_DELAY_MS}`)
}
const currentTurnEvents: SessionEvent[] = []
let openTurn: number | undefined
for (const prior of history.slice().reverse()) {
if (prior.type === 'turn/end') fail('llm/retry must be appended inside an open turn')
if (prior.type === 'turn/start') {
openTurn = prior.data.turn
break
}
currentTurnEvents.push(prior)
const turnBoundary = history.findLast(prior =>
prior.type === 'turn/start' || prior.type === 'turn/end')
if (turnBoundary?.type !== 'turn/start') {
fail('llm/retry must be appended inside an open turn')
}
if (openTurn === undefined) fail('llm/retry must be appended inside an open turn')
if (turn !== openTurn) {
fail(`llm/retry names turn ${turn}, but the open turn is ${openTurn}`)
if (turn !== turnBoundary.data.turn) {
fail(`llm/retry names turn ${turn}, but the open turn is ${turnBoundary.data.turn}`)
}
let closedStep: number | undefined
for (const prior of currentTurnEvents) {
if (prior.type === 'step/start') {
fail(`llm/retry must follow step/end, but step ${prior.data.step} is still open`)
}
if (prior.type === 'step/end') {
closedStep = prior.data.step
break
}
const stepBoundary = history.findLast(prior =>
prior.type === 'step/start' || prior.type === 'step/end')
if (stepBoundary?.type !== 'step/start') {
fail('llm/retry must be appended inside an open step')
}
if (closedStep === undefined || step !== closedStep) {
fail(`llm/retry names step ${step}, but the latest closed step is ${String(closedStep)}`)
if (step !== stepBoundary.data.step || turn !== stepBoundary.data.turn) {
fail(`llm/retry names turn ${turn}/step ${step}, but the open step is ${stepBoundary.data.turn}/${stepBoundary.data.step}`)
}
const routedProvider = providerForClosedStep(history, turn, step)
const routedProvider = providerForOpenStep(history, turn, step)
if (routedProvider !== provider) {
fail(`llm/retry provider ${provider} does not match the failed request provider ${String(routedProvider)}`)
}
const chainStart = retryChainStart(history, turn)
const chain = history.slice(Math.max(chainStart, 0))
const lastSuccess = chain.findLastIndex(prior => prior.type === 'assistant/message')
const chainRetries = chain.slice(lastSuccess + 1)
.filter((prior): prior is SessionEvent<'llm/retry'> => prior.type === 'llm/retry')
if (chainRetries.some(prior => prior.data.turn === turn && prior.data.step === step)) {
fail(`llm/retry duplicates the retry record for turn ${turn}/step ${step}`)
}
const priorPolicyRetry = chainRetries.findLast(prior =>
prior.data.provider === provider && prior.data.policyKey === policyKey)
const priorPolicyRetry = history.findLast((prior): prior is SessionEvent<'llm/retry'> =>
prior.type === 'llm/retry'
&& prior.data.turn === turn
&& prior.data.step === step
&& prior.data.provider === provider
&& prior.data.policyKey === policyKey)
const expectedRetry = (priorPolicyRetry?.data.retry ?? 0) + 1
if (retry !== expectedRetry) {
fail(`llm/retry retry ${retry} must equal provider policy retry ${expectedRetry}`)

View File

@@ -1,11 +1,11 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import SessionStore, { SessionId, type Session } from '@deepseek-ai/dsh-session'
import { createUserMessage, ProviderRequestId , createMessage } from '@deepseek-ai/dsh-llm'
import { createUserMessage, ProviderRequestId } from '@deepseek-ai/dsh-llm'
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
import InvariantService from '@deepseek-ai/dsh-invariants'
import * as RetryInvariant from '@deepseek-ai/dsh-llm-retry/invariant'
import { providerForClosedStep } from '../src/history.ts'
import { providerForOpenStep } from '../src/history.ts'
async function setup(): Promise<Context> {
const ctx = new Context()
@@ -15,26 +15,24 @@ async function setup(): Promise<Context> {
return ctx
}
function closeStep(ctx: Context, id: string, turn = 1, step = 1) {
function openStep(ctx: Context, id: string, turn = 1, step = 1) {
const session = ctx.sessions.create(SessionId(id))
session.append('turn/start', { turn, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('turn/start', { turn })
session.append('step/start', { turn, step })
session.append('request/header', {
header: { config: { provider: 'mock', model: 'mock' } },
reason: 'initial',
})
session.append('step/end', { turn, step })
return session
}
function appendRetryTurn(session: Session, turn: number) {
session.append('turn/start', { turn, trigger: { kind: 'retry' } })
session.append('turn/start', { turn })
session.append('step/start', { turn, step: 1 })
session.append('request/header', {
header: { config: { provider: 'mock', model: 'mock' } },
reason: 'initial',
})
session.append('step/end', { turn, step: 1 })
session.append('llm/retry', { turn, step: 1, ...normal })
}
@@ -58,28 +56,24 @@ const always = {
}
describe('llm-retry invariants', () => {
it('has no provider without the requested closed step or a route marker', () => {
expect(providerForClosedStep([], 1, 1)).toBeUndefined()
expect(providerForClosedStep([{
type: 'step/end',
it('has no provider without the requested open step or a route marker', () => {
expect(providerForOpenStep([], 1, 1)).toBeUndefined()
expect(providerForOpenStep([{
type: 'step/start',
data: { turn: 1, step: 1 },
}] as never, 1, 1)).toBeUndefined()
})
it('accepts bounded and unbounded records after successive closed steps', async () => {
it('accepts successive bounded and unbounded records inside their open steps', async () => {
const ctx = await setup()
const session = closeStep(ctx, 'retry-invariant-valid')
const session = openStep(ctx, 'retry-invariant-valid')
expect(() => {
session.append('llm/retry', { turn: 1, step: 1, ...normal })
session.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, failure } })
session.append('turn/start', { turn: 2, trigger: { kind: 'retry' } })
session.append('step/start', { turn: 2, step: 1 })
session.append('step/end', { turn: 2, step: 1 })
session.append('llm/retry', {
turn: 2, step: 1, ...normal, retry: 2, delayMs: 0,
turn: 1, step: 1, ...normal, retry: 2, delayMs: 0,
})
const unbounded = closeStep(ctx, 'retry-invariant-always')
const unbounded = openStep(ctx, 'retry-invariant-always')
unbounded.append('llm/retry', { turn: 1, step: 1, ...always })
}).not.toThrow()
expect(() => { ctx.emit('tools/change') }).not.toThrow()
@@ -87,7 +81,7 @@ describe('llm-retry invariants', () => {
it('validates the complete durable failure payload', async () => {
const ctx = await setup()
const complete = closeStep(ctx, 'retry-invariant-complete-failure')
const complete = openStep(ctx, 'retry-invariant-complete-failure')
expect(() => {
complete.append('llm/retry', {
turn: 1,
@@ -126,7 +120,7 @@ describe('llm-retry invariants', () => {
['request-id-empty', { message: 'failed', code: 'RATE_LIMIT', requestId: '' }, /failure\.requestId/],
]
for (const [name, invalidFailure, message] of invalidFailures) {
const session = closeStep(ctx, `retry-invariant-failure-${name}`)
const session = openStep(ctx, `retry-invariant-failure-${name}`)
expect(() => {
session.append('llm/retry', {
turn: 1, step: 1, ...always, failure: invalidFailure,
@@ -150,95 +144,75 @@ describe('llm-retry invariants', () => {
['delay-type', { ...normal, delayMs: '1' }, /delayMs/],
])('rejects invalid retry data: %s', async (name, data, message) => {
const ctx = await setup()
const session = closeStep(ctx, `retry-invariant-${name}`)
const session = openStep(ctx, `retry-invariant-${name}`)
expect(() => {
session.append('llm/retry', { turn: 1, step: 1, ...data } as never)
}).toThrow(message)
})
it('rejects records outside the latest closed step of an open turn', async () => {
it('rejects records outside the currently open turn and step', async () => {
const ctx = await setup()
const absent = ctx.sessions.create(SessionId('retry-invariant-no-turn'))
expect(() => {
absent.append('llm/retry', { turn: 1, step: 1, ...normal })
}).toThrow(/inside an open turn/)
const wrongTurn = closeStep(ctx, 'retry-invariant-wrong-turn')
const wrongTurn = openStep(ctx, 'retry-invariant-wrong-turn')
expect(() => {
wrongTurn.append('llm/retry', { turn: 2, step: 1, ...normal })
}).toThrow(/open turn is 1/)
const openStep = ctx.sessions.create(SessionId('retry-invariant-open-step'))
openStep.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
openStep.append('step/start', { turn: 1, step: 1 })
const closedStep = openStep(ctx, 'retry-invariant-closed-step')
closedStep.append('step/end', { turn: 1, step: 1 })
expect(() => {
openStep.append('llm/retry', { turn: 1, step: 1, ...normal })
}).toThrow(/step 1 is still open/)
closedStep.append('llm/retry', { turn: 1, step: 1, ...normal })
}).toThrow(/inside an open step/)
const noStep = ctx.sessions.create(SessionId('retry-invariant-no-step'))
noStep.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
noStep.append('turn/start', { turn: 1 })
expect(() => {
noStep.append('llm/retry', { turn: 1, step: 1, ...normal })
}).toThrow(/latest closed step is undefined/)
}).toThrow(/inside an open step/)
const wrongStep = closeStep(ctx, 'retry-invariant-wrong-step')
const wrongStep = openStep(ctx, 'retry-invariant-wrong-step')
expect(() => {
wrongStep.append('llm/retry', { turn: 1, step: 2, ...normal })
}).toThrow(/latest closed step is 1/)
}).toThrow(/open step is 1\/1/)
const closedTurn = closeStep(ctx, 'retry-invariant-closed-turn')
closedTurn.append('turn/end', { turn: 1, reason: { kind: 'aborted' } })
const closedTurn = openStep(ctx, 'retry-invariant-closed-turn')
closedTurn.append('step/end', { turn: 1, step: 1 })
closedTurn.append('turn/end', { turn: 1, reason: { kind: 'aborted', reason: { kind: 'user' } },
})
expect(() => {
closedTurn.append('llm/retry', { turn: 1, step: 1, ...normal })
}).toThrow(/inside an open turn/)
})
it('rejects a second retry record for the same step', async () => {
it('accepts successive retries in one step and rejects skipped numbering', async () => {
const ctx = await setup()
const session = closeStep(ctx, 'retry-invariant-duplicate')
const session = openStep(ctx, 'retry-invariant-number-sequence')
session.append('llm/retry', { turn: 1, step: 1, ...normal })
session.append('llm/retry', { turn: 1, step: 1, ...normal, retry: 2 })
expect(() => {
session.append('llm/retry', { turn: 1, step: 1, ...normal, retry: 2 })
}).toThrow(/duplicates the retry record/)
session.append('llm/retry', { turn: 1, step: 1, ...always, retry: 2 })
}).toThrow(/must equal provider policy retry 1/)
})
it('binds retry numbering to the provider policy and resets it after success', async () => {
it('binds retry numbering to the provider policy and resets it for a new step', async () => {
const ctx = await setup()
const mismatch = closeStep(ctx, 'retry-invariant-numbering')
const mismatch = openStep(ctx, 'retry-invariant-numbering')
mismatch.append('llm/retry', { turn: 1, step: 1, ...normal })
mismatch.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, failure } })
mismatch.append('turn/start', { turn: 2, trigger: { kind: 'retry' } })
mismatch.append('step/start', { turn: 2, step: 1 })
mismatch.append('step/end', { turn: 2, step: 1 })
expect(() => {
mismatch.append('llm/retry', { turn: 2, step: 1, ...normal, retry: 1 })
mismatch.append('llm/retry', { turn: 1, step: 1, ...normal, retry: 1 })
}).toThrow(/must equal provider policy retry 2/)
const reset = closeStep(ctx, 'retry-invariant-reset')
const reset = openStep(ctx, 'retry-invariant-reset')
reset.append('llm/retry', { turn: 1, step: 1, ...normal })
reset.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, failure } })
reset.append('turn/start', { turn: 2, trigger: { kind: 'retry' } })
reset.append('step/start', { turn: 2, step: 1 })
reset.append('assistant/message', {
turn: 2,
step: 1,
message: createMessage({
role: 'assistant',
content: [{ type: 'text', text: 'success' }],
source: {
kind: 'model',
...{ provider: 'mock', model: 'mock' },
},
}),
}, { surfaceOp: 'append' })
reset.append('step/end', { turn: 2, step: 1 })
reset.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
reset.append('turn/start', { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } })
reset.append('step/start', { turn: 3, step: 1 })
reset.append('step/end', { turn: 3, step: 1 })
reset.append('step/end', { turn: 1, step: 1 })
reset.append('step/start', { turn: 1, step: 2 })
expect(() => {
reset.append('llm/retry', { turn: 3, step: 1, ...normal })
reset.append('llm/retry', { turn: 1, step: 2, ...normal })
}).not.toThrow()
})
@@ -262,9 +236,7 @@ describe('llm-retry invariants', () => {
appendRetryTurn(nonFailureEnd, 2)
const missingStart = ctx.sessions.create(SessionId('retry-invariant-missing-start'))
missingStart.append('turn/end', {
turn: 1,
reason: { kind: 'error', step: 1, failure },
missingStart.append('turn/end', { turn: 1, reason: { kind: 'error', error: failure },
})
appendRetryTurn(missingStart, 2)
@@ -274,7 +246,7 @@ describe('llm-retry invariants', () => {
it('rejects a provider that does not match the failed request route', async () => {
const ctx = await setup()
const session = closeStep(ctx, 'retry-invariant-provider')
const session = openStep(ctx, 'retry-invariant-provider')
expect(() => {
session.append('llm/retry', { turn: 1, step: 1, ...always, provider: 'other' })
}).toThrow(/does not match the failed request provider mock/)
@@ -284,7 +256,7 @@ describe('llm-retry invariants', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const session = ctx.sessions.create(SessionId('retry-invariant-late'))
session.append('step/end', { turn: 1, step: 1 })
session.append('step/start', { turn: 1, step: 1 })
session.append('llm/retry', { turn: 1, step: 1, ...normal })
await ctx.plugin(InvariantService)
await expect(ctx.plugin(RetryInvariant)).rejects.toThrow(/inside an open turn/)

View File

@@ -32,13 +32,12 @@ describe.each(['jsonl', 'sqlite'] as const)('%s retry-event persistence', (kind)
const ctx = await backend(kind)
try {
const session = ctx.sessions.create(SessionId(`retry-${kind}`))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('turn/start', { turn: 1 })
session.append('step/start', { turn: 1, step: 1 })
session.append('request/header', {
header: { config: { provider: 'mock', model: 'mock' } },
reason: 'initial',
})
session.append('step/end', { turn: 1, step: 1 })
const event = session.append('llm/retry', {
turn: 1,
step: 1,
@@ -49,13 +48,9 @@ describe.each(['jsonl', 'sqlite'] as const)('%s retry-event persistence', (kind)
delayMs: 750,
failure: { message: 'provider busy', code: 'RATE_LIMIT', status: 429 },
})
session.append('turn/end', {
turn: 1,
reason: {
kind: 'error',
step: 1,
failure: { message: 'provider busy', code: 'RATE_LIMIT', status: 429 },
},
session.append('step/end', { turn: 1, step: 1 })
session.append('turn/end', { turn: 1, reason: { kind: 'error', error: { message: 'provider busy', code: 'RATE_LIMIT', status: 429 },
},
})
expect(session.deriveMessages()).toEqual([])

View File

@@ -149,15 +149,8 @@ function alwaysConfig(backoff: BackoffConfig = {}): AlwaysRetryPolicyConfig {
}
}
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()
}
})
})
function waitForIdle(_ctx: Context, agent: Agent): Promise<void> {
return agent.whenIdle()
}
function waitForRetry(ctx: Context, agent: Agent, retryNumber: number): Promise<Extract<SessionEvent, { type: 'llm/retry' }>> {
@@ -180,7 +173,7 @@ afterEach(async () => {
})
describe('provider-routed retry policy', () => {
it('records the scheduled delay before opening a fresh request attempt', async () => {
it('records the scheduled delay before retrying the request', async () => {
vi.useFakeTimers()
const adapter = new ScriptedAdapter([
new LlmError('busy', 'RATE_LIMIT', { status: 429 }),
@@ -219,7 +212,7 @@ describe('provider-routed retry policy', () => {
expect(adapter.requests).toHaveLength(2)
expect(agent.session.events.filter(item => item.type === 'step/start').map(item => item.data))
.toEqual([{ turn: 1, step: 1 }, { turn: 2, step: 1 }])
.toEqual([{ turn: 1, step: 1 }])
expect(agent.session.deriveMessages().at(-1)).toEqual({
id: expect.any(String) as unknown,
role: 'assistant',
@@ -256,7 +249,7 @@ describe('provider-routed retry policy', () => {
expect(agent.session.events.filter(event => event.type === 'assistant/message').map(event => ({
turn: event.data.turn,
step: event.data.step,
}))).toEqual([{ turn: 2, step: 1 }])
}))).toEqual([{ turn: 1, step: 1 }])
expect(agent.session.deriveMessages().at(-1)).toMatchObject({
role: 'assistant',
content: [{ type: 'text', text: 'recovered' }],
@@ -289,14 +282,21 @@ describe('provider-routed retry policy', () => {
await vi.advanceTimersByTimeAsync(500)
await idle
const retryEvent = agent.session.events.find(event => event.type === 'llm/retry')
const failedChunks = agent.session.events.filter(event =>
event.type === 'assistant/chunk' && event.data.turn === 1 && event.data.step === 1,
event.type === 'assistant/chunk'
&& retryEvent !== undefined
&& event.seq < retryEvent.seq,
)
expect(failedChunks).toHaveLength(6)
expect(agent.session.events.filter(event => event.type === 'assistant/message').map(event => ({
expect(failedChunks).toHaveLength(7)
const assistantMessages = agent.session.events.filter(event => event.type === 'assistant/message')
expect(assistantMessages.map(event => ({
turn: event.data.turn,
step: event.data.step,
}))).toEqual([{ turn: 2, step: 1 }])
}))).toEqual([{ turn: 1, step: 1 }])
expect(failedChunks.every(event =>
!assistantMessages[0]?.sourceEventSeqs?.includes(event.seq),
)).toBe(true)
expect(agent.session.events.some(event => event.type === 'tool/call')).toBe(false)
expect(toolExecutions).toBe(0)
expect(agent.session.deriveMessages().at(-1)).toMatchObject({
@@ -337,7 +337,7 @@ describe('provider-routed retry policy', () => {
expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(2)
expect(agent.session.events.at(-1)).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'error', failure: { message: 'busy three', code: 'SERVER' } } },
data: { reason: { kind: 'error', error: { message: 'busy three', code: 'SERVER' } } },
})
})
@@ -448,10 +448,14 @@ describe('provider-routed retry policy', () => {
expect(adapter.requests).toHaveLength(0)
expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
expect(agent.session.events.at(-1)).toMatchObject({
const end = agent.session.events.at(-1)
expect(end).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'error', failure: { code: 'NO_ADAPTER' } } },
data: { reason: { kind: 'error', error: { code: 'NO_ADAPTER' } } },
})
if (end?.type === 'turn/end' && end.data.reason.kind === 'error') {
expect(end.data.reason.error.message).toContain('no adapter registered for provider')
}
})
it('selects policy by the failed request provider', async () => {
@@ -539,9 +543,9 @@ describe('provider-routed retry policy', () => {
backoff: { initialDelayMs: 1, maxDelayMs: 1 },
}),
}, (ctx) => {
ctx.on('agent/request', async (_agent, turn, _step, _signal, next) => ({
ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => ({
...await next(),
provider: turn === 1 ? 'mock' : 'other',
provider: adapter.requests.length === 0 ? 'mock' : 'other',
}))
}))
const agent = context.agentLoop.create(SessionId('retry-provider-budgets'), {
@@ -913,9 +917,7 @@ describe('provider-routed retry policy', () => {
const captured = Promise.withResolvers<undefined>()
let invokeCaptured: (() => Promise<void>) | undefined
const mounted = await harness(adapter, {}, (ctx) => {
ctx.on('agent/request-error', (
_agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next,
) => {
ctx.on('agent/request-error', (_agent, _context, _signal, next) => {
return new Promise<RequestErrorAction>((resolve) => {
invokeCaptured = async () => { resolve(await next()) }
captured.resolve(undefined)
@@ -924,9 +926,7 @@ describe('provider-routed retry policy', () => {
})
context = mounted.ctx
let downstreamCalls = 0
context.on('agent/request-error', async (
_agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next,
) => {
context.on('agent/request-error', async (_agent, _context, _signal, next) => {
downstreamCalls += 1
return next()
})
@@ -980,9 +980,7 @@ describe('provider-routed retry policy', () => {
textResponse('must not run'),
])
;({ ctx: context } = await harness(adapter, { mock: policy }, (ctx) => {
ctx.on('agent/request-error', async (
agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next,
) => {
ctx.on('agent/request-error', async (agent, _context, _signal, next) => {
agent.cancel({ kind: 'user' })
return next()
})

View File

@@ -55,14 +55,8 @@ async function harness(
return ctx
}
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
if (subject !== agent || status !== 'idle') return
dispose()
resolve()
})
})
function waitForIdle(_ctx: Context, agent: Agent): Promise<void> {
return agent.whenIdle()
}
function sendAndWait(ctx: Context, agent: Agent): Promise<void> {
@@ -109,15 +103,15 @@ describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {
expect(server?.requests).toHaveLength(1)
expect(agent.session.events.filter(event => event.type === 'step/start')
.map(event => [event.data.turn, event.data.step]))
.toEqual([[1, 1], [2, 1]])
.toEqual([[1, 1]])
expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data.failure.code))
.toEqual(['TRANSPORT'])
expect(finalAssistantText(agent)).toBe('connected after retry')
})
it.each([
['stream_disconnect', 0] as const,
['partial_disconnect', 2] as const,
['stream_disconnect', 1] as const,
['partial_disconnect', 3] as const,
])('retries %s without committing failed chunks', async (behavior, failedChunkCount) => {
const server = await start([behavior, 'success'], {
apiKey: 'mock-key',
@@ -136,12 +130,15 @@ describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {
expect(server.requests).toHaveLength(2)
expect(server.requests[0]?.body).toEqual(server.requests[1]?.body)
const retryEvent = agent.session.events.find(event => event.type === 'llm/retry')
expect(agent.session.events.filter(event =>
event.type === 'assistant/chunk' && event.data.turn === 1,
event.type === 'assistant/chunk'
&& retryEvent !== undefined
&& event.seq < retryEvent.seq,
)).toHaveLength(failedChunkCount)
expect(agent.session.events.filter(event => event.type === 'assistant/message')
.map(event => [event.data.turn, event.data.step]))
.toEqual([[2, 1]])
.toEqual([[1, 1]])
expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data.failure.code))
.toEqual(['TRANSPORT'])
expect(finalAssistantText(agent)).toBe('recovered response')
@@ -166,7 +163,7 @@ describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {
.toEqual(['EMPTY_RESPONSE'])
expect(agent.session.events.filter(event => event.type === 'assistant/message')
.map(event => [event.data.turn, event.data.step]))
.toEqual([[2, 1]])
.toEqual([[1, 1]])
expect(agent.session.events.at(-1)).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'completed' } },
@@ -191,12 +188,12 @@ describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {
expect(server.requests).toHaveLength(1)
expect(agent.session.events.filter(event =>
event.type === 'assistant/chunk' && event.data.turn === 1,
)).toHaveLength(2)
)).toHaveLength(3)
expect(agent.session.events.some(event => event.type === 'assistant/message')).toBe(false)
expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
expect(agent.session.events.at(-1)).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'error', failure: { code: 'STREAM_CLOSED' } } },
data: { reason: { kind: 'error', error: { message: 'SSE stream ended without [DONE]', code: 'STREAM_CLOSED' } } },
})
})
@@ -234,11 +231,15 @@ describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {
await sendAndWait(context, agent)
expect(server.requests).toHaveLength(3)
expect(agent.session.events.filter(event => event.type === 'step/start')).toHaveLength(3)
expect(agent.session.events.filter(event => event.type === 'step/start')).toHaveLength(1)
expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(2)
expect(agent.session.events.at(-1)).toMatchObject({
const end = agent.session.events.at(-1)
expect(end).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'error', failure: { code: 'TRANSPORT' } } },
data: { reason: { kind: 'error', error: { code: 'TRANSPORT' } } },
})
if (end?.type === 'turn/end' && end.data.reason.kind === 'error') {
expect(end.data.reason.error.message).toContain('DeepSeek API request to')
}
})
})

View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/llm/llm/README.md
README.md: 21f428fb22c9a59a67d86f446ea866c1629b964a
README.zh.md: 9bd26993bc63b39de3d3a8039fea6b046c875504
README.md: 5d74ed647f4de3c8ed65554dff736eb8aec9eef9
README.zh.md: a362ba8b825238325ce70238c5b2f3725f0d8495

View File

@@ -18,10 +18,10 @@ An adapter registry plus a single streaming call surface, interceptable via a wa
- `ctx.llm.listModels(provider: string): Promise<LlmModelInfo[]>` Discover the models one registered provider currently advertises.
- `ctx.llm.resolveModelInfo(provider: string, model: string, signal?: AbortSignal): Promise<LlmResolvedModelInfo>` Resolve validated exact-model identity plus available context, output-default, and reasoning metadata from the owning adapter, with optional cancellation for asynchronous adapters.
- `ctx.llm.resolveCallConfig(config: LlmCallConfig, signal?: AbortSignal): Promise<LlmCallConfig>` Validate an explicit effort and materialize adapter-configured call defaults without clamping.
- `ctx.llm.prepareCall(config: LlmCallConfig, signal?: AbortSignal): Promise<PreparedLlmCall>` Resolve a config plus detached context metadata and adapter-default provenance in one exact-model lookup, then capture its current adapter registration as one cancellable, one-shot call.
- `ctx.llm.prepareCall(config: LlmCallConfig, signal?: AbortSignal): Promise<PreparedLlmCall>` Resolve a config plus detached context metadata and adapter-default provenance in one exact-model lookup, then capture its current adapter registration and immutable retry policy as one cancellable, one-shot call.
- `ctx.llm.stream(options: GenerateOptions): AsyncIterable<StreamChunk>` Stream one model call as raw chunks (token-level deltas). Consumers assemble the chunks into blocks/messages with `BlockAssembler`.
`LlmService` preserves errors from final adapter selection, synchronous dispatch, iterator construction, and iteration, and binds their provenance to the exact stream handle returned for that model call. `isLlmAdapterFailure(stream, value)` reports only errors from that call's final adapter boundary; `llmFailureOf(stream, value)` returns the adjacent immutable `LlmFailure`; `llmRetryPolicyOf(stream)` returns the immutable policy of the exact registration selected at that boundary, even if the route is later disposed or replaced. A call that never reaches a final adapter has no serving policy. Nested model calls, `llm/stream` middleware, and downstream consumer failures remain unclassified for the outer call. Classification never replaces or mutates the adapter's original coded `Error`.
`LlmService` normalizes failures from final adapter selection, synchronous dispatch, iterator construction, and iteration into the stream protocol's single terminal form: `finish { kind: 'error' | 'aborted', failure }`. A failure after partial deltas may leave content blocks open; consumers discard that incomplete output. Errors from `llm/stream` middleware, nested calls, adapter cleanup, and downstream consumers remain thrown because they are plugin or consumer failures rather than model-request outcomes. A prepared call exposes the immutable retry policy captured with its exact adapter registration; a route handled entirely by middleware has no serving policy.
Provider and model metadata is a discovery surface, not a routing whitelist. `registerAdapter()` still owns provider exclusivity and captures the adapter's retry policy for each route, while an adapter may accept model ids absent from `listModels()`; consumers must not reject a request because its model is unlisted. Returned selector metadata is detached and invalid or duplicate adapter entries fail with `INVALID_ADAPTER` or `INVALID_CATALOG`.
@@ -48,7 +48,7 @@ Exact-model metadata is a separate correctness query, not a catalog decoration o
Message content is an array of typed blocks: `text`, `reasoning`, `tool-call`, `tool-result`. The union is derived from the merge-extensible `ContentBlockMap`, so plugins can add block types via declaration merging. Assistant messages use a model source carrying provider/model provenance and optional adapter-private replay state. Before dispatch, `LlmService` retains that state only when the historical provider route and target provider route are currently owned by the exact same adapter instance; the adapter then decides whether it can restore or convert the state across models/providers. The core block set is limited to blocks every shipping path honors — multimodal content (images, audio, …) has no core block type; a feature that needs one adds it via the map together with the adapter/UI/compaction support that honors it.
Streaming is a raw chunk protocol (`block-start`, `text-delta`, `reasoning-delta`, `tool-call-delta`, `block-end`, `usage`, `finish`). `BlockAssembler` is the single shared implementation that assembles chunks into blocks/messages.
Streaming is a raw chunk protocol (`block-start`, `text-delta`, `reasoning-delta`, `tool-call-delta`, `block-end`, `usage`, `finish`). Every adapter outcome reaches consumers as one terminal `finish`; operational failure uses its `error` or `aborted` reason rather than throwing across the stream API. `BlockAssembler` is the single shared implementation that assembles chunks into blocks/messages.
### Call configuration (`call-config.ts`)
@@ -71,7 +71,7 @@ Every product adapter sends application identity on provider HTTP requests. `att
### Real adapters
Two adapters implement `LlmAdapter` on different internals: [`@deepseek-ai/dsh-llm-deepseek`](../llm-deepseek) uses direct fetch with `eventsource-parser` SSE framing for the `deepseek-official` route, while [`@deepseek-ai/dsh-llm-pi-ai`](../llm-pi-ai) dynamically resolves configured provider/model pairs through `@earendil-works/pi-ai`. Both follow the `StreamChunk` conventions in `types.ts`: usage precedes finish, tool arguments remain raw strings, and errors take one of two sanctioned paths. See [the twin LLM adapters](../../../.agents/notes/implemented/architecture/2026-06-13-twin-llm-adapters.md) for the design rationale.
Two adapters implement `LlmAdapter` on different internals: [`@deepseek-ai/dsh-llm-deepseek`](../llm-deepseek) uses direct fetch with `eventsource-parser` SSE framing for the `deepseek-official` route, while [`@deepseek-ai/dsh-llm-pi-ai`](../llm-pi-ai) dynamically resolves configured provider/model pairs through `@earendil-works/pi-ai`. Both follow the `StreamChunk` conventions in `types.ts`: usage precedes finish and tool arguments remain raw strings. Adapter implementations may throw or emit a failure finish internally; `LlmService` exposes both as a terminal failure finish. See [the twin LLM adapters](../../../.agents/notes/implemented/architecture/2026-06-13-twin-llm-adapters.md) for the adapter rationale and [the terminal-failure decision](../../../.agents/notes/implemented/architecture/2026-07-29-terminal-llm-stream-failures.md) for the service boundary.
## Model Experience

View File

@@ -18,10 +18,10 @@
- `ctx.llm.listModels(provider: string): Promise<LlmModelInfo[]>` 发现某个已注册提供方当前公布的模型。
- `ctx.llm.resolveModelInfo(provider: string, model: string, signal?: AbortSignal): Promise<LlmResolvedModelInfo>` 从拥有精确路由的适配器解析经校验的确切模型身份以及可用上下文、输出默认值和推理reasoning元数据异步适配器可选地支持取消。
- `ctx.llm.resolveCallConfig(config: LlmCallConfig, signal?: AbortSignal): Promise<LlmCallConfig>` 校验显式推理强度,并填入适配器配置的调用默认值,但不自动调整。
- `ctx.llm.prepareCall(config: LlmCallConfig, signal?: AbortSignal): Promise<PreparedLlmCall>` 在一次精确模型查询中解析配置、脱耦的上下文元数据与适配器默认值溯源,再将当前适配器注册捕获为一次可取消、一次性调用。
- `ctx.llm.prepareCall(config: LlmCallConfig, signal?: AbortSignal): Promise<PreparedLlmCall>` 在一次精确模型查询中解析配置、脱耦的上下文元数据与适配器默认值溯源,再将当前适配器注册和不可变重试策略捕获为一次可取消、一次性调用。
- `ctx.llm.stream(options: GenerateOptions): AsyncIterable<StreamChunk>` 将一次模型调用流式输出为原始分片token 级增量)。消费方使用 `BlockAssembler` 将分片组装为块/消息。
`LlmService` 保留来自最终适配器选择、同步 dispatch、iterator 构造与迭代的错误,并将其溯源绑定到该次模型调用返回的精确流句柄。`isLlmAdapterFailure(stream, value)` 只报告该调用最终适配器边界的错误;`llmFailureOf(stream, value)` 返回关联的不可变 `LlmFailure``llmRetryPolicyOf(stream)` 返回在该边界选中的确切注册所对应的不可变策略,即使之后释放或替换路由也不变。未到达最终适配器的调用没有服务策略。嵌套模型调用、`llm/stream` middleware 和下游消费方失败对外层调用仍未分类。分类绝不替换或更改适配器原有的带代码 `Error`
`LlmService` 最终适配器选择、同步 dispatch、iterator 构造与迭代中的失败规范化为流协议唯一的终止形式:`finish { kind: 'error' | 'aborted', failure }`。部分增量输出后发生失败时,内容块可能仍未闭合;消费方会丢弃这些不完整输出。`llm/stream` middleware、嵌套调用、适配器清理和下游消费方的错误仍会抛出因为它们属于插件或消费方失败而非模型请求结果。已准备调用会暴露随其确切适配器注册一同捕获的不可变重试策略完全由 middleware 处理的路由没有服务策略
提供方与模型元数据是发现接口,不是路由白名单。`registerAdapter()` 仍拥有提供方排他性,并为每条路由捕获适配器的重试策略;适配器则可以接受 `listModels()` 中不存在的模型 id消费方禁止因模型未列出而拒绝请求。返回的 selector 元数据与输入脱离,无效或重复适配器配置项会以 `INVALID_ADAPTER``INVALID_CATALOG` 失败。
@@ -48,7 +48,7 @@
消息内容是类型化内容块数组:`text``reasoning``tool-call``tool-result`。联合从可合并扩展的 `ContentBlockMap` 派生,因此插件可以通过 declaration merging 添加块类型。assistant 消息使用模型来源其中携带提供方模型溯源与可选适配器私有回放状态。dispatch 前,`LlmService` 只在历史提供方路由与目标提供方路由当前由完全相同的适配器实例拥有时才保留该状态;随后由适配器判定能否在模型/提供方间恢复或转换该状态。核心块集只包含每条已发布路径都支持的块。多模态内容(图像、音频等)没有核心块类型;需要它的功能会通过 map 添加并一并添加相应的适配器UI压缩compaction支持。
流式输出是原始分片协议(`block-start``text-delta``reasoning-delta``tool-call-delta``block-end``usage``finish`)。`BlockAssembler` 是将分片组装为块/消息的唯一共享实现。
流式输出是原始分片协议(`block-start``text-delta``reasoning-delta``tool-call-delta``block-end``usage``finish`)。每个适配器结果都以一个终止 `finish` 到达消费方;运行故障使用其 `error``aborted` 原因,而不会跨流 API 抛出。`BlockAssembler` 是将分片组装为块/消息的唯一共享实现。
### 调用配置(`call-config.ts`
@@ -71,7 +71,7 @@
### 真实适配器
两个适配器使用不同内部机制实现 `LlmAdapter`[`@deepseek-ai/dsh-llm-deepseek`](../llm-deepseek) 针对 `deepseek-official` 路由使用直接 fetch 加 `eventsource-parser` SSEServer-Sent Events分帧[`@deepseek-ai/dsh-llm-pi-ai`](../llm-pi-ai) 则通过 `@earendil-works/pi-ai` 动态解析已配置提供方/模型对。两者都遵循 `StreamChunk` 约定,定义见 `types.ts`usage 先于 finish工具参数保持原始字符串错误使用两种已批准路径之一。设计理由见 [双 LLM 适配器](../../../.agents/notes/implemented/architecture/2026-06-13-twin-llm-adapters.md)。
两个适配器使用不同内部机制实现 `LlmAdapter`[`@deepseek-ai/dsh-llm-deepseek`](../llm-deepseek) 针对 `deepseek-official` 路由使用直接 fetch 加 `eventsource-parser` SSEServer-Sent Events分帧[`@deepseek-ai/dsh-llm-pi-ai`](../llm-pi-ai) 则通过 `@earendil-works/pi-ai` 动态解析已配置提供方/模型对。两者都遵循 `types.ts` 中的 `StreamChunk` 约定usage 先于 finish工具参数保持原始字符串。适配器实现在内部可以抛出异常或发出失败 finish`LlmService` 会将两者都暴露为终止失败 finish。适配器理由见[双 LLM 适配器](../../../.agents/notes/implemented/architecture/2026-06-13-twin-llm-adapters.md),服务边界见[终止失败决策](../../../.agents/notes/implemented/architecture/2026-07-29-terminal-llm-stream-failures.md)
## 模型体验

View File

@@ -1,67 +1,40 @@
/**
* Private provider-failure tagging shared by `LlmService` and its consumers.
* Normalization for values thrown by a final LLM adapter boundary.
*
* @module @deepseek-ai/dsh-llm/adapter-failure
*/
import { HarnessError } from './error.ts'
import type { LlmFailure, StreamChunk } from './types.ts'
import type { ResolvedRetryPolicy } from './retry-policy.ts'
/** Call-local facts captured when one model call enters its final adapter boundary. */
export interface AdapterFailureScope {
/** Errors and normalized facts proven to originate in this call's final adapter boundary. */
readonly failures: WeakMap<Error, LlmFailure>
/** Immutable policy of the exact adapter registration selected for this call. */
retryPolicy?: ResolvedRetryPolicy
}
/** Call-local failure scopes keyed by the exact stream handle returned to a consumer. */
const adapterFailureScopes = new WeakMap<AsyncIterable<StreamChunk>, AdapterFailureScope>()
import type { LlmFailure } from './types.ts'
/**
* Bind one call's adapter-failure scope to a unique returned stream handle.
* @param stream - the waterfall-selected stream for this call.
* @param failures - errors tagged by this call's final adapter boundary.
* @returns a unique stream handle that delegates iteration to `stream`.
* Detach serializable provider facts from a value thrown by an adapter.
* @param value - arbitrary value thrown during adapter dispatch or iteration.
* @returns immutable provider-neutral facts suitable for a terminal finish chunk.
* @internal
*/
export function bindAdapterFailureScope(
stream: AsyncIterable<StreamChunk>,
failures: AdapterFailureScope,
): AsyncIterable<StreamChunk> {
const call = {
[Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
return stream[Symbol.asyncIterator]()
},
}
adapterFailureScopes.set(call, failures)
return call
}
/**
* Preserve an adapter's Error identity while tagging its provider origin.
* @param failures - the call-local final-adapter failure scope.
* @param value - arbitrary value thrown by adapter dispatch or iteration.
* @returns the original Error, or a coded Error wrapping a non-Error throw.
* @internal
*/
export function markLlmAdapterFailure(
failures: AdapterFailureScope,
value: unknown,
): Error & { code?: string } {
export function normalizeLlmFailure(value: unknown): LlmFailure {
const error = value instanceof Error
? value as Error & { code?: string }
: new HarnessError(String(value), 'UNKNOWN', { cause: value })
? value
: new HarnessError(thrownMessage(value), 'UNKNOWN', { cause: value })
// Cross-package copies preserve own data but not class identity. Trust the
// carried facts only when both own properties agree after validation.
const carried = ownFailureSnapshot(error)
const failure = carried !== undefined && carried.code === ownErrorCode(error) ? carried : Object.freeze({
if (carried !== undefined && carried.code === ownErrorCode(error)) return carried
return Object.freeze({
message: errorMessage(error),
code: harnessErrorCode(error),
})
failures.failures.set(error, failure)
return error
}
/** Render a non-Error throw without letting hostile coercion escape normalization. */
function thrownMessage(value: unknown): string {
try {
const message = String(value)
return message.length > 0 ? message : 'LLM adapter failed'
} catch (_hostileThrownValue) {
return 'LLM adapter failed'
}
}
/** Read a foreign error's own data-backed `code` without invoking accessors. */
@@ -129,46 +102,3 @@ function errorMessage(error: Error): string {
function harnessErrorCode(error: Error): string {
return error instanceof HarnessError ? error.code : 'UNKNOWN'
}
/**
* Whether a failure came from final adapter dispatch, iterator construction,
* or iteration for the call represented by the exact returned stream handle.
* @param stream - the exact stream returned by the model call being classified.
* @param value - arbitrary failure caught by a model-call consumer.
* @returns true only for errors tagged at that call's final adapter boundary.
*/
export function isLlmAdapterFailure(
stream: AsyncIterable<StreamChunk>,
value: unknown,
): value is Error & { code?: string } {
const failures = adapterFailureScopes.get(stream)
return value instanceof Error && failures !== undefined && failures.failures.has(value)
}
/**
* Retrieve normalized provider facts only for an Error tagged by this exact
* model call's final adapter boundary.
* @param stream - the exact stream returned to the consumer.
* @param value - the caught failure.
* @returns the immutable facts for that call, or `undefined` for middleware, nested, or consumer failures.
*/
export function llmFailureOf(
stream: AsyncIterable<StreamChunk>,
value: unknown,
): LlmFailure | undefined {
const failures = adapterFailureScopes.get(stream)
return value instanceof Error ? failures?.failures.get(value) : undefined
}
/**
* Read the retry policy of the exact registration selected at this call's
* final adapter boundary. The policy remains available after that registration
* is disposed or replaced; absence means no final adapter served the call.
* @param stream - the exact stream returned by the model call.
* @returns the immutable serving-registration policy, or `undefined`.
*/
export function llmRetryPolicyOf(
stream: AsyncIterable<StreamChunk>,
): ResolvedRetryPolicy | undefined {
return adapterFailureScopes.get(stream)?.retryPolicy
}

View File

@@ -127,11 +127,15 @@ export class BlockAssembler {
/**
* Assemble all blocks seen so far, in stream order.
* @returns one block per seen index; an open block assembles from its
* accumulated deltas (an unknown block type never closed by `block-end` throws).
* @returns one block per seen index, except that max-token truncation drops
* tool calls that cannot be executed safely; an open block assembles from
* its accumulated deltas (an unknown block type never closed by `block-end` throws).
*/
blocks(): ContentBlock[] {
return this.order.map(index => this.assemble(this.mustGet(index), index))
const blocks = this.order.map(index => this.assemble(this.mustGet(index), index))
return this.finish.kind === 'max-tokens'
? blocks.filter(block => block.type !== 'tool-call')
: blocks
}
/** Usage from the `usage` chunk; undefined until one arrives. */

View File

@@ -93,7 +93,8 @@ export function isQuotaExceededError(detail: string): boolean {
/**
* Render a thrown value with its full `cause` chain and AggregateError
* members, so transport wrappers like undici's `TypeError: fetch failed`
* surface the underlying failure instead of masking it. Diagnostic-surface
* surface the underlying failure instead of masking it. Plain structured
* failures render their own data-backed `message`. Diagnostic-surface
* rendering only (messages, notices, logs) — never parse the result; route on
* {@link HarnessError.code}.
* @param value - the caught value (`unknown` in catch clauses).
@@ -109,7 +110,15 @@ export function errorChain(value: unknown): string {
if (path.has(current)) return '<circular cause>'
path.add(current)
try {
if (!(current instanceof Error)) return String(current)
if (!(current instanceof Error)) {
if (typeof current === 'object' && current !== null) {
const descriptor = Object.getOwnPropertyDescriptor(current, 'message')
if (descriptor !== undefined && 'value' in descriptor && typeof descriptor.value === 'string') {
return descriptor.value
}
}
return String(current)
}
const message = current.message === '' ? current.name : current.message
const members = current instanceof AggregateError && current.errors.length > 0
? ` [${current.errors.map(render).join('; ')}]`

View File

@@ -24,8 +24,7 @@ import type { ProviderRequestId } from './brand.ts'
import { callConfigEquals, deepFreeze } from './call-config.ts'
import type { LlmCallConfig, LlmCallConfigAdapterDefaults } from './call-config.ts'
import { HarnessError } from './error.ts'
import { bindAdapterFailureScope, markLlmAdapterFailure } from './adapter-failure.ts'
import type { AdapterFailureScope } from './adapter-failure.ts'
import { normalizeLlmFailure } from './adapter-failure.ts'
export * from './attribution.ts'
export * from './brand.ts'
@@ -37,7 +36,6 @@ export * from './retry-policy.ts'
export { BlockAssembler } from './assembler.ts'
export { callConfigEquals, deepFreeze, isAgentLoopRequest, markAgentLoopRequest } from './call-config.ts'
export type { LlmCallConfig, LlmCallConfigAdapterDefaults } from './call-config.ts'
export { isLlmAdapterFailure, llmFailureOf, llmRetryPolicyOf } from './adapter-failure.ts'
declare module 'cordis' {
interface Context {
@@ -126,6 +124,8 @@ export class LlmError extends HarnessError {
export interface PreparedLlmCall {
/** Detached, deep-frozen config with any adapter-owned default materialized. */
readonly config: LlmCallConfig
/** Immutable retry policy captured with the adapter registration. */
readonly retryPolicy: ResolvedRetryPolicy
/** Detached context metadata resolved with the registration-bound call. */
readonly context?: LlmModelContext
/** Config fields materialized by the captured adapter rather than proposed by the caller. */
@@ -633,12 +633,19 @@ export class LlmService extends Service {
let dispatched = false
return Object.freeze({
config: resolvedConfig,
retryPolicy: registration.retryPolicy,
adapterDefaults,
...context === undefined ? {} : { context },
stream: (options: GenerateOptions): AsyncIterable<StreamChunk> => {
if (dispatched) {
throw new LlmError('a prepared LLM call can only be dispatched once', 'INVALID_PREPARED_CALL')
}
if (!callConfigEquals(options, resolvedConfig)) {
throw new LlmError(
'prepared LLM call config changed before adapter dispatch',
'INVALID_PREPARED_CALL',
)
}
dispatched = true
return this.streamWithRegistration(options, { registration, config: resolvedConfig })
},
@@ -668,22 +675,17 @@ export class LlmService extends Service {
}
/**
* Final adapter boundary. It tags only failures from adapter selection,
* synchronous dispatch, iterator construction, or iteration while preserving
* the original Error object. Middleware outside this generator remains
* distinguishable as plugin work. An iteration failure skips adapter cleanup
* so it cannot suppress the primary provider error. A downstream close awaits
* adapter cleanup, whose failures remain ordinary untagged work.
* Final adapter boundary. Adapter selection, dispatch, iterator construction,
* and iteration failures become one terminal failure chunk. Middleware and
* downstream consumer failures remain thrown plugin or consumer errors.
*/
private async * adapterStream(
options: GenerateOptions,
failures: AdapterFailureScope,
prepared?: { registration: AdapterRegistration; config: LlmCallConfig },
): AsyncGenerator<StreamChunk> {
let iterator: AsyncIterator<StreamChunk>
try {
const registration = prepared?.registration ?? this.registration(options.provider)
failures.retryPolicy = registration.retryPolicy
const resolvedConfig = prepared === undefined
? (await this.resolveCallFor(registration, options, options.signal)).config
: prepared.config
@@ -702,32 +704,34 @@ export class LlmService extends Service {
const stream = adapter.stream(this.forAdapter(resolvedOptions, adapter))
iterator = stream[Symbol.asyncIterator]()
} catch (error: unknown) {
throw markLlmAdapterFailure(failures, error)
yield adapterFailureChunk(error, options.signal)
return
}
let completed = false
let iterationFailed = false
try {
while (true) {
let value: StreamChunk
let item: { done: true } | { done: false; value: StreamChunk }
try {
const item = await iterator.next()
if (item.done) {
completed = true
return
}
value = item.value
const next = await iterator.next()
item = next.done
? { done: true }
: { done: false, value: next.value }
} catch (error: unknown) {
iterationFailed = true
throw markLlmAdapterFailure(failures, error)
completed = true
yield adapterFailureChunk(error, options.signal)
return
}
if (item.done) {
completed = true
return
}
// End the adapter-owned try before yielding: consumer/middleware
// failures resumed into this generator must remain untagged.
yield value
// failures resumed into this generator must remain thrown.
yield item.value
}
} finally {
// oxlint-disable-next-line typescript/no-unnecessary-condition -- the iteration catch sets its latch before entering finally.
if (!completed && !iterationFailed) {
if (!completed) {
const close = iterator.return?.bind(iterator)
if (close) await close()
}
@@ -735,15 +739,13 @@ export class LlmService extends Service {
}
/**
* Stream one model call as raw chunks (token-level deltas). Throws
* `LlmError` with code `NO_ADAPTER` if no adapter is registered for
* `options.provider`. Replay state is retained only when the same adapter
* instance owns its historical provider and the target provider. Final
* adapter selection remains fixed through asynchronous exact-model resolution
* and dispatch. Selection, dispatch, and iteration failures retain their
* original Error identity and are tagged in a call-local scope for narrow
* agent-loop request recovery; middleware and nested-call failures remain
* untagged for the outer call.
* Stream one model call as raw chunks (token-level deltas). Replay state is
* retained only when the same adapter instance owns its historical provider
* and the target provider. Final adapter selection remains fixed through
* asynchronous exact-model resolution and dispatch. Adapter selection,
* dispatch, and iteration failures become terminal `error` or `aborted`
* finish chunks; middleware, nested-call, cleanup, and consumer failures
* remain thrown.
* @param options - the full request; `options.provider` selects the adapter.
* @returns the chunk stream, possibly wrapped by `llm/stream` listeners.
*/
@@ -755,14 +757,23 @@ export class LlmService extends Service {
options: GenerateOptions,
prepared?: { registration: AdapterRegistration; config: LlmCallConfig },
): AsyncIterable<StreamChunk> {
const failures: AdapterFailureScope = { failures: new WeakMap<Error, LlmFailure>() }
const stream = this.ctx.waterfall(
return this.ctx.waterfall(
this,
'llm/stream',
options,
() => this.adapterStream(options, failures, prepared),
() => this.adapterStream(options, prepared),
)
return bindAdapterFailureScope(stream, failures)
}
}
/** Convert one adapter throw into the stream protocol's terminal outcome. */
function adapterFailureChunk(error: unknown, signal?: AbortSignal): StreamChunk {
const failure = normalizeLlmFailure(error)
return {
type: 'finish',
reason: signal?.aborted || failure.code === 'ABORTED'
? { kind: 'aborted', failure }
: { kind: 'error', failure },
}
}

View File

@@ -72,7 +72,9 @@ async function* validateStream(
usageSeen = true
break
case 'finish':
if (open.size > 0) fail(`LLM stream finished with ${open.size} open block(s)`)
if (open.size > 0 && chunk.reason.kind !== 'error' && chunk.reason.kind !== 'aborted') {
fail(`LLM stream finished with ${open.size} open block(s)`)
}
finished = true
break
}

View File

@@ -192,8 +192,9 @@ export interface LlmResolvedModelInfo extends LlmModelInfo {
* Raw streaming protocol emitted by adapters.
* Block indexes correlate interleaved deltas, and `block-end` carries the
* assembled block. Adapters emit usage before the terminal finish and nothing
* afterward; tool arguments remain raw JSON strings. Failures either throw or
* end with `error`/`aborted`, and consumers must handle both paths.
* afterward; tool arguments remain raw JSON strings. An adapter implementation
* may throw, but `LlmService.stream()` normalizes that failure to a terminal
* `error` or `aborted` finish before exposing it to consumers.
*/
export type StreamChunk =
| { type: 'block-start'; index: number; blockType: ContentBlockType }

View File

@@ -0,0 +1,68 @@
import { describe, expect, it } from 'vitest'
import { normalizeLlmFailure } from '../src/adapter-failure.ts'
describe('adapter failure normalization', () => {
it('contains hostile non-Error coercion', () => {
const thrown = { [Symbol.toPrimitive]: () => { throw new Error('coercion failed') } }
expect(normalizeLlmFailure(thrown)).toEqual({ message: 'LLM adapter failed', code: 'UNKNOWN' })
})
it('normalizes empty primitive throws and data descriptors without values', () => {
expect(normalizeLlmFailure('')).toEqual({ message: 'LLM adapter failed', code: 'UNKNOWN' })
expect(normalizeLlmFailure(null)).toEqual({ message: 'null', code: 'UNKNOWN' })
const error = new Error('provider failed')
Object.defineProperty(error, 'failure', { get: () => ({ message: 'ignored', code: 'IGNORED' }) })
Object.defineProperty(error, 'code', { get: () => 'IGNORED' })
expect(normalizeLlmFailure(error)).toEqual({ message: 'provider failed', code: 'UNKNOWN' })
const accessorCode = Object.assign(new Error('provider failed'), {
failure: { message: 'provider failed', code: 'FOREIGN' },
})
Object.defineProperty(accessorCode, 'code', { get: () => 'FOREIGN' })
expect(normalizeLlmFailure(accessorCode)).toEqual({ message: 'provider failed', code: 'UNKNOWN' })
const primitiveFailure = Object.assign(new Error('provider failed'), {
failure: null,
code: 'FOREIGN',
})
expect(normalizeLlmFailure(primitiveFailure)).toEqual({ message: 'provider failed', code: 'UNKNOWN' })
})
it('contains hostile Error property reflection', () => {
const withFailure = new Error('provider failed') as Error & { failure: unknown; code: string }
withFailure.failure = { message: 'provider failed', code: 'FOREIGN' }
withFailure.code = 'FOREIGN'
const hostileCode = new Proxy(withFailure, {
getOwnPropertyDescriptor(target, property) {
if (property === 'code') throw new Error('code descriptor failed')
return Reflect.getOwnPropertyDescriptor(target, property)
},
})
expect(normalizeLlmFailure(hostileCode)).toEqual({ message: 'provider failed', code: 'UNKNOWN' })
const hostileFailure = new Proxy(new Error('provider failed'), {
getOwnPropertyDescriptor() { throw new Error('failure descriptor failed') },
})
expect(normalizeLlmFailure(hostileFailure)).toEqual({ message: 'provider failed', code: 'UNKNOWN' })
})
it('rejects malformed or accessor-backed failure snapshots', () => {
const malformed = new Error('provider failed') as Error & { failure: unknown; code: string }
malformed.failure = { message: 'provider failed', code: 'FOREIGN', requestId: '' }
malformed.code = 'FOREIGN'
expect(normalizeLlmFailure(malformed)).toEqual({ message: 'provider failed', code: 'UNKNOWN' })
const accessorBacked = new Error('provider failed') as Error & { failure: unknown }
accessorBacked.failure = Object.defineProperty({}, 'message', {
get() { throw new Error('failure getter failed') },
})
expect(normalizeLlmFailure(accessorBacked)).toEqual({ message: 'provider failed', code: 'UNKNOWN' })
})
it('falls back when an Error message accessor throws', () => {
const error = new Error('provider failed')
Object.defineProperty(error, 'message', { get() { throw new Error('message getter failed') } })
expect(normalizeLlmFailure(error)).toEqual({ message: 'LLM adapter failed', code: 'UNKNOWN' })
})
})

View File

@@ -6,11 +6,8 @@ import LlmService, {
HarnessError,
isContextWindowExceededError,
isQuotaExceededError,
isLlmAdapterFailure,
LlmAdapter,
LlmError,
llmFailureOf,
llmRetryPolicyOf,
ProviderRequestId,
ReasoningEffortId,
resolveRetryPolicy,
@@ -95,6 +92,12 @@ const SCRIPT: StreamChunk[] = [
{ type: 'finish', reason: { kind: 'stop' } },
]
async function collect(stream: AsyncIterable<StreamChunk>): Promise<StreamChunk[]> {
const chunks: StreamChunk[] = []
for await (const chunk of stream) chunks.push(chunk)
return chunks
}
describe('LlmService', () => {
it('recognizes structured and model-capacity context-window overflow details', () => {
expect(isContextWindowExceededError('context_length_exceeded maximum context length')).toBe(true)
@@ -142,6 +145,8 @@ describe('LlmService', () => {
it('errorChain survives non-Error values, hostile coercion, and circular causes', () => {
expect(errorChain('plain string')).toBe('plain string')
expect(errorChain({ message: 'structured provider failure', code: 'SERVER' }))
.toBe('structured provider failure')
expect(errorChain({ toString: () => { throw new Error('hostile') } })).toBe('<unrenderable value>')
const circular = new Error('outer')
circular.cause = circular
@@ -219,69 +224,61 @@ describe('LlmService', () => {
)
})
it('keeps the serving registration policy on an in-flight call after route replacement', async () => {
it('keeps a prepared registration and retry policy after route replacement', async () => {
const oldPolicy = resolveRetryPolicy({ mode: 'always' }, 'old retryPolicy')
const newPolicy = resolveRetryPolicy({ mode: 'normal', maxRetries: 0 }, 'new retryPolicy')
const entered = Promise.withResolvers<undefined>()
const release = Promise.withResolvers<undefined>()
const failure = new LlmError('old route failed', 'AUTH')
const oldAdapter = new class extends LlmAdapter {
const oldFailure = new LlmError('old route failed', 'AUTH')
const ctx = new Context()
await ctx.plugin(LlmService)
const disposeOld = ctx.llm.registerAdapter(['route'], new class extends ThrowingAdapter {
override providerRetryPolicy(): typeof oldPolicy {
return oldPolicy
}
}(oldFailure))
const prepared = await ctx.llm.prepareCall({ provider: 'route', model: 'model' })
async * stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
entered.resolve(undefined)
await release.promise
throw failure
}
}()
const newAdapter = new class extends ScriptedAdapter {
disposeOld()
ctx.llm.registerAdapter(['route'], new class extends ScriptedAdapter {
override providerRetryPolicy(): typeof newPolicy {
return newPolicy
}
}(SCRIPT)
const ctx = new Context()
await ctx.plugin(LlmService)
const disposeOld = ctx.llm.registerAdapter(['route'], oldAdapter)
const stream = ctx.llm.stream({ provider: 'route', model: 'model', messages: [] })
const outcome = (async (): Promise<unknown> => {
try {
for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) {
return error
}
return undefined
})()
await entered.promise
}(SCRIPT))
disposeOld()
ctx.llm.registerAdapter(['route'], newAdapter)
release.resolve(undefined)
expect(await outcome).toBe(failure)
expect(llmRetryPolicyOf(stream)).toBe(oldPolicy)
const chunks = await collect(prepared.stream({ ...prepared.config, messages: [] }))
expect(chunks.at(-1)).toEqual({
type: 'finish',
reason: {
kind: 'error',
failure: { message: 'old route failed', code: 'AUTH' },
},
})
expect(prepared.retryPolicy).toBe(oldPolicy)
expect(ctx.llm.providerRetryPolicy('route')).toBe(newPolicy)
})
it('throws NO_ADAPTER for unregistered providers', async () => {
it('normalizes an unregistered provider to a terminal failure', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
const stream = ctx.llm.stream({ provider: 'nope', model: 'any-model', messages: [] })
let caught: unknown
try {
for await (const _ of stream) { /* drain */ }
} catch (error: unknown) {
caught = error
}
expect(caught).toBeInstanceOf(LlmError)
expect((caught as LlmError).code).toBe('NO_ADAPTER')
expect((caught as LlmError).message).toContain('no adapter registered')
expect(isLlmAdapterFailure(stream, caught)).toBe(true)
expect(llmRetryPolicyOf(stream)).toBeUndefined()
const chunks = await collect(ctx.llm.stream({
provider: 'nope',
model: 'any-model',
messages: [],
}))
const finish = chunks.at(-1)
expect(finish).toMatchObject({
type: 'finish',
reason: {
kind: 'error',
failure: { code: 'NO_ADAPTER' },
},
})
if (finish?.type !== 'finish' || finish.reason.kind !== 'error') throw new Error('expected error finish')
expect(finish.reason.failure.message).toContain('no adapter registered')
})
it.each(['done', 'value'] as const)('tags a throwing IteratorResult.%s getter without replacing its Error', async (field) => {
it.each(['done', 'value'] as const)('normalizes a throwing IteratorResult.%s getter', async (field) => {
const original = new LlmError(`${field} getter failed`, 'RESULT_GETTER_FAILED')
const result = field === 'done' ? {} : { done: false }
Object.defineProperty(result, field, { get: () => { throw original } })
@@ -297,31 +294,30 @@ describe('LlmService', () => {
})
const adapter = new class extends LlmAdapter {
stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
return {
[Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
return iterator
},
}
return { [Symbol.asyncIterator]: () => iterator }
}
}()
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], adapter)
ctx.llm.registerAdapter(['test'], adapter)
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
let caught: unknown
try {
for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) {
caught = error
}
const chunks = await collect(ctx.llm.stream({
provider: 'test',
model: 'test',
messages: [],
}))
expect(caught).toBe(original)
expect(isLlmAdapterFailure(stream, caught)).toBe(true)
expect(chunks.at(-1)).toEqual({
type: 'finish',
reason: {
kind: 'error',
failure: { message: `${field} getter failed`, code: 'RESULT_GETTER_FAILED' },
},
})
expect(cleanupLookups).toBe(0)
})
it.each(['dispatch', 'iterator'] as const)('tags synchronous adapter %s failures without replacing their Error', async (boundary) => {
it.each(['dispatch', 'iterator'] as const)('normalizes synchronous adapter %s failures', async (boundary) => {
const original = new LlmError(`${boundary} failed`, 'BOUNDARY_FAILED')
const adapter = new class extends LlmAdapter {
stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
@@ -331,339 +327,63 @@ describe('LlmService', () => {
}()
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], adapter)
ctx.llm.registerAdapter(['test'], adapter)
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
let caught: unknown
try {
for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) {
caught = error
}
const chunks = await collect(ctx.llm.stream({
provider: 'test',
model: 'test',
messages: [],
}))
expect(caught).toBe(original)
expect(isLlmAdapterFailure(stream, caught)).toBe(true)
expect(llmFailureOf(stream, caught)).toEqual({
message: `${boundary} failed`,
code: 'BOUNDARY_FAILED',
expect(chunks.at(-1)).toEqual({
type: 'finish',
reason: {
kind: 'error',
failure: { message: `${boundary} failed`, code: 'BOUNDARY_FAILED' },
},
})
})
it('keeps structured provider facts beside a frozen third-party Error', async () => {
const original = new LlmError('provider busy', 'RATE_LIMIT', {
it('preserves structured LlmError facts in the terminal failure', async () => {
const failure = new LlmError('provider busy', 'RATE_LIMIT', {
status: 429,
providerRetryAfterMs: 1_500,
requestId: ProviderRequestId('req-7'),
})
Object.freeze(original)
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-provider'], new ThrowingAdapter(original))
ctx.llm.registerAdapter(['test'], new ThrowingAdapter(failure))
const stream = ctx.llm.stream({ provider: 'test-provider', model: 'test-model', messages: [] })
let caught: unknown
try {
for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) {
caught = error
}
const chunks = await collect(ctx.llm.stream({
provider: 'test',
model: 'test',
messages: [],
}))
expect(caught).toBe(original)
expect(llmFailureOf(stream, caught)).toEqual({
message: 'provider busy',
code: 'RATE_LIMIT',
status: 429,
providerRetryAfterMs: 1_500,
requestId: ProviderRequestId('req-7'),
})
})
it('does not trust retry facts carried by an unknown third-party Error', async () => {
const carried = { message: 'busy', code: 'SERVER', status: 503 }
const original = Object.assign(new Error('busy'), { failure: carried })
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-provider'], new ThrowingAdapter(original))
const stream = ctx.llm.stream({ provider: 'test-provider', model: 'test-model', messages: [] })
await expect((async () => {
for await (const _chunk of stream) { /* drain */ }
})()).rejects.toBe(original)
const facts = llmFailureOf(stream, original)
carried.status = 500
expect(facts).toEqual({ message: 'busy', code: 'UNKNOWN' })
expect(Object.isFrozen(facts)).toBe(true)
expect(facts).not.toBe(carried)
})
it('keeps validated failure facts across package copies with matching own codes', async () => {
const original = Object.assign(new Error('provider busy'), {
code: 'RATE_LIMIT',
failure: {
message: 'provider busy',
code: 'RATE_LIMIT',
status: 429,
providerRetryAfterMs: 1_500,
requestId: 'req-cross-copy',
expect(chunks.at(-1)).toEqual({
type: 'finish',
reason: {
kind: 'error',
failure: {
message: 'provider busy',
code: 'RATE_LIMIT',
status: 429,
providerRetryAfterMs: 1_500,
requestId: ProviderRequestId('req-7'),
},
},
})
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-provider'], new ThrowingAdapter(original))
const stream = ctx.llm.stream({ provider: 'test-provider', model: 'test-model', messages: [] })
await expect((async () => {
for await (const _chunk of stream) { /* drain */ }
})()).rejects.toBe(original)
expect(llmFailureOf(stream, original)).toEqual({
message: 'provider busy',
code: 'RATE_LIMIT',
status: 429,
providerRetryAfterMs: 1_500,
requestId: 'req-cross-copy',
})
})
it('keeps an unknown SDK Error exact without trusting its private code or accessors', async () => {
const original = Object.assign(new Error('socket closed'), { code: 'ECONNRESET' })
Object.defineProperty(original, 'failure', {
get() { throw new Error('SDK failure accessor must not run') },
})
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-provider'], new ThrowingAdapter(original))
const stream = ctx.llm.stream({ provider: 'test-provider', model: 'test-model', messages: [] })
await expect((async () => {
for await (const _chunk of stream) { /* drain */ }
})()).rejects.toBe(original)
expect(original.code).toBe('ECONNRESET')
expect(llmFailureOf(stream, original)).toEqual({ message: 'socket closed', code: 'UNKNOWN' })
})
it('keeps an SDK Error exact when its message accessor is hostile', async () => {
const original = Object.defineProperty(new Error(), 'message', {
get() { throw new Error('SDK message accessor trap') },
})
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-provider'], new ThrowingAdapter(original))
const stream = ctx.llm.stream({ provider: 'test-provider', model: 'test-model', messages: [] })
await expect((async () => {
for await (const _chunk of stream) { /* drain */ }
})()).rejects.toBe(original)
expect(llmFailureOf(stream, original)).toEqual({ message: 'LLM adapter failed', code: 'UNKNOWN' })
})
it('keeps an SDK Error exact without trusting accessor-backed carried facts', async () => {
const original = Object.assign(new Error('busy'), {
failure: { message: 'busy', code: 'SERVER', status: 503 },
})
Object.defineProperty(original, 'code', {
get() { throw new Error('SDK code accessor must not escape') },
})
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-provider'], new ThrowingAdapter(original))
const stream = ctx.llm.stream({ provider: 'test-provider', model: 'test-model', messages: [] })
await expect((async () => {
for await (const _chunk of stream) { /* drain */ }
})()).rejects.toBe(original)
expect(llmFailureOf(stream, original)).toEqual({ message: 'busy', code: 'UNKNOWN' })
})
it('does not trust carried facts matched only by an inherited code', async () => {
class InheritedCodeError extends Error {
get code(): string { return 'SERVER' }
}
const original = Object.assign(new InheritedCodeError('busy'), {
failure: { message: 'busy', code: 'SERVER', status: 503 },
})
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-provider'], new ThrowingAdapter(original))
const stream = ctx.llm.stream({ provider: 'test-provider', model: 'test-model', messages: [] })
await expect((async () => {
for await (const _chunk of stream) { /* drain */ }
})()).rejects.toBe(original)
expect(llmFailureOf(stream, original)).toEqual({ message: 'busy', code: 'UNKNOWN' })
})
it('keeps an SDK Error exact when code descriptor inspection is trapped', async () => {
const target = Object.assign(new Error('busy'), {
code: 'SERVER',
failure: { message: 'busy', code: 'SERVER', status: 503 },
})
const original = new Proxy(target, {
getOwnPropertyDescriptor(value, property) {
if (property === 'code') throw new Error('SDK code descriptor trap')
return Reflect.getOwnPropertyDescriptor(value, property)
},
})
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-provider'], new ThrowingAdapter(original))
const stream = ctx.llm.stream({ provider: 'test-provider', model: 'test-model', messages: [] })
await expect((async () => {
for await (const _chunk of stream) { /* drain */ }
})()).rejects.toBe(original)
expect(llmFailureOf(stream, original)).toEqual({ message: 'busy', code: 'UNKNOWN' })
})
it('falls back safely when SDK objects trap failure inspection or expose malformed facts', async () => {
const propertyTrap = new Proxy(new HarnessError('descriptor trapped', 'SERVER'), {
getOwnPropertyDescriptor(target, property) {
if (property === 'failure') throw new Error('SDK descriptor trap')
return Reflect.getOwnPropertyDescriptor(target, property)
},
})
const throwingFacts = Object.create(null) as Record<string, unknown>
Object.defineProperty(throwingFacts, 'message', {
get() { throw new Error('SDK fact getter trap') },
})
const carrying = (message: string, failure: unknown): HarnessError => Object.defineProperty(
new HarnessError(message, 'SERVER'),
'failure',
{ value: failure },
)
const factGetter = carrying('fact getter failed', throwingFacts)
const malformed = carrying('malformed facts', { message: 'provider busy', code: 'SERVER', requestId: 1 })
const primitive = carrying('primitive facts', 1)
const nullFacts = carrying('null facts', null)
const mismatched = carrying('mismatched facts', { message: 'busy', code: 'RATE_LIMIT' })
for (const [original, expectedMessage] of [
[propertyTrap, 'descriptor trapped'],
[factGetter, 'fact getter failed'],
[malformed, 'malformed facts'],
[primitive, 'primitive facts'],
[nullFacts, 'null facts'],
[mismatched, 'mismatched facts'],
] as const) {
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-provider'], new ThrowingAdapter(original))
const stream = ctx.llm.stream({ provider: 'test-provider', model: 'test-model', messages: [] })
await expect((async () => {
for await (const _chunk of stream) { /* drain */ }
})()).rejects.toBe(original)
expect(llmFailureOf(stream, original)).toEqual({ message: expectedMessage, code: 'SERVER' })
}
})
it('retains a stable code from a HarnessError without requiring LlmError facts', async () => {
const original = new HarnessError('stable adapter failure', 'ADAPTER_STABLE')
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-provider'], new ThrowingAdapter(original))
const stream = ctx.llm.stream({ provider: 'test-provider', model: 'test-model', messages: [] })
await expect((async () => {
for await (const _chunk of stream) { /* drain */ }
})()).rejects.toBe(original)
expect(llmFailureOf(stream, original)).toEqual({
message: 'stable adapter failure',
code: 'ADAPTER_STABLE',
})
expect(llmFailureOf(stream, 'not an Error')).toBeUndefined()
expect(llmFailureOf({ [Symbol.asyncIterator]: () => stream[Symbol.asyncIterator]() }, original)).toBeUndefined()
})
it('keeps a nested adapter failure scoped to the nested model call', async () => {
const original = new LlmError('nested provider failed', 'NESTED_FAILED')
const outer = new RecordingAdapter(SCRIPT)
const nested = new ThrowingAdapter(original)
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['outer'], outer)
ctx.llm.registerAdapter(['nested'], nested)
let nestedStream: AsyncIterable<StreamChunk> | undefined
ctx.on('llm/stream', (options, next) => {
if (options.provider !== 'outer') return next()
return (async function* () {
nestedStream = ctx.llm.stream({ provider: 'nested', model: 'nested', messages: [] })
yield * nestedStream
})()
})
const outerStream = ctx.llm.stream({ provider: 'outer', model: 'outer', messages: [] })
let caught: unknown
try {
for await (const _chunk of outerStream) { /* drain */ }
} catch (error: unknown) {
caught = error
}
expect(caught).toBe(original)
expect(nestedStream).toBeDefined()
expect(isLlmAdapterFailure(nestedStream!, caught)).toBe(true)
expect(isLlmAdapterFailure(outerStream, caught)).toBe(false)
expect(outer.lastOptions).toBeUndefined()
})
it('keeps call scopes distinct when middleware reuses an iterable', async () => {
const firstFailure = new LlmError('first provider failed', 'FIRST_FAILED')
const secondFailure = new LlmError('second provider failed', 'SECOND_FAILED')
const delegates: AsyncIterable<StreamChunk>[] = []
const shared: AsyncIterable<StreamChunk> = {
[Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
const delegate = delegates.shift()
if (delegate === undefined) throw new Error('shared stream has no call delegate')
return delegate[Symbol.asyncIterator]()
},
}
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['first'], new ThrowingAdapter(firstFailure))
ctx.llm.registerAdapter(['second'], new ThrowingAdapter(secondFailure))
ctx.on('llm/stream', (_options, next) => {
delegates.push(next())
return shared
})
const firstStream = ctx.llm.stream({ provider: 'first', model: 'first', messages: [] })
const secondStream = ctx.llm.stream({ provider: 'second', model: 'second', messages: [] })
const catchFailure = async (stream: AsyncIterable<StreamChunk>): Promise<unknown> => {
try {
for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) {
return error
}
return new Error('expected adapter to fail')
}
expect(firstStream).not.toBe(secondStream)
const firstCaught = await catchFailure(firstStream)
expect(firstCaught).toBe(firstFailure)
expect(isLlmAdapterFailure(firstStream, firstCaught)).toBe(true)
expect(isLlmAdapterFailure(secondStream, firstCaught)).toBe(false)
const secondCaught = await catchFailure(secondStream)
expect(secondCaught).toBe(secondFailure)
expect(isLlmAdapterFailure(secondStream, secondCaught)).toBe(true)
expect(isLlmAdapterFailure(firstStream, secondCaught)).toBe(false)
expect(delegates).toHaveLength(0)
})
it('propagates a rejected next promptly without awaiting a non-settling return', async () => {
const original = new LlmError('provider failed', 'PROVIDER_FAILED')
let cleanupCalls = 0
it('normalizes arbitrary adapter rejections without throwing them downstream', async () => {
const adapter = new class extends LlmAdapter {
stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
return {
[Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
return {
next: () => Promise.reject(original),
return: () => {
cleanupCalls += 1
return new Promise<IteratorResult<StreamChunk>>(() => {})
},
// Third-party adapters can reject with arbitrary values.
// oxlint-disable-next-line typescript/prefer-promise-reject-errors
next: () => Promise.reject('plain provider failure'),
}
},
}
@@ -671,30 +391,73 @@ describe('LlmService', () => {
}()
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], adapter)
ctx.llm.registerAdapter(['test'], adapter)
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
const failure = (async (): Promise<unknown> => {
try {
for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) {
return error
}
return new Error('expected adapter iteration to fail')
})()
let timer: ReturnType<typeof setTimeout> | undefined
const timeout = new Promise<Error>((resolve) => {
timer = setTimeout(() => { resolve(new Error('adapter failure did not settle promptly')) }, 100)
const chunks = await collect(ctx.llm.stream({
provider: 'test',
model: 'test',
messages: [],
}))
expect(chunks.at(-1)).toEqual({
type: 'finish',
reason: {
kind: 'error',
failure: { message: 'plain provider failure', code: 'UNKNOWN' },
},
})
const caught = await Promise.race([failure, timeout])
if (timer !== undefined) clearTimeout(timer)
expect(caught).toBe(original)
expect(isLlmAdapterFailure(stream, caught)).toBe(true)
expect(cleanupCalls).toBe(0)
})
it('awaits one adapter return on downstream close and leaves its rejection unclassified', async () => {
it('maps adapter failure to aborted when the request signal is aborted', async () => {
const controller = new AbortController()
controller.abort('cancelled')
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test'], new ThrowingAdapter(new Error('stopped')))
const chunks = await collect(ctx.llm.stream({
provider: 'test',
model: 'test',
messages: [],
signal: controller.signal,
}))
expect(chunks.at(-1)).toMatchObject({
type: 'finish',
reason: { kind: 'aborted', failure: { message: 'stopped' } },
})
})
it('leaves middleware and consumer failures thrown', async () => {
const middlewareFailure = new Error('middleware failed')
const middlewareCtx = new Context()
await middlewareCtx.plugin(LlmService)
middlewareCtx.llm.registerAdapter(['test'], new ScriptedAdapter(SCRIPT))
middlewareCtx.on('llm/stream', () => (async function* () {
throw middlewareFailure
})())
await expect(collect(middlewareCtx.llm.stream({
provider: 'test',
model: 'test',
messages: [],
}))).rejects.toBe(middlewareFailure)
const consumerFailure = new Error('consumer failed')
const consumerCtx = new Context()
await consumerCtx.plugin(LlmService)
consumerCtx.llm.registerAdapter(['test'], new ScriptedAdapter(SCRIPT))
await expect((async () => {
for await (const _chunk of consumerCtx.llm.stream({
provider: 'test',
model: 'test',
messages: [],
})) {
throw consumerFailure
}
})()).rejects.toBe(consumerFailure)
})
it('awaits adapter cleanup on downstream close and leaves cleanup failure thrown', async () => {
const cleanup = new Error('cleanup failed')
let cleanupCalls = 0
const adapter = new class extends LlmAdapter {
@@ -714,22 +477,19 @@ describe('LlmService', () => {
}()
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], adapter)
ctx.llm.registerAdapter(['test'], adapter)
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
let caught: unknown
try {
for await (const _chunk of stream) break
} catch (error: unknown) {
caught = error
}
expect(caught).toBe(cleanup)
expect(isLlmAdapterFailure(stream, caught)).toBe(false)
await expect((async () => {
for await (const _chunk of ctx.llm.stream({
provider: 'test',
model: 'test',
messages: [],
})) break
})()).rejects.toBe(cleanup)
expect(cleanupCalls).toBe(1)
})
it('allows downstream close when the adapter iterator has no return method', async () => {
it('allows downstream close when an adapter iterator has no return method', async () => {
const adapter = new class extends LlmAdapter {
stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
return {
@@ -741,66 +501,9 @@ describe('LlmService', () => {
}()
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], adapter)
ctx.llm.registerAdapter(['test'], adapter)
let chunks = 0
for await (const _chunk of ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })) {
chunks += 1
break
}
expect(chunks).toBe(1)
})
it('normalizes and tags non-Error adapter failures once', async () => {
const adapter = new class extends LlmAdapter {
stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
return {
[Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
// Third-party adapters can reject with arbitrary values.
// oxlint-disable-next-line typescript/prefer-promise-reject-errors
return { next: () => Promise.reject('plain provider failure') }
},
}
}
}()
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], adapter)
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
let caught: unknown
try {
for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) {
caught = error
}
expect(caught).toBeInstanceOf(HarnessError)
expect(caught).toMatchObject({ code: 'UNKNOWN', cause: 'plain provider failure' })
expect(isLlmAdapterFailure(stream, caught)).toBe(true)
})
it('does not tag a failure thrown downstream while consuming adapter output', async () => {
const downstream = new Error('consumer failed')
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], new ScriptedAdapter(SCRIPT))
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
let caught: unknown
try {
for await (const _chunk of stream) throw downstream
} catch (error: unknown) {
caught = error
}
expect(caught).toBe(downstream)
expect(isLlmAdapterFailure(stream, caught)).toBe(false)
expect(isLlmAdapterFailure(new ScriptedAdapter(SCRIPT).stream({
provider: 'unbound', model: 'unbound', messages: [],
}), caught)).toBe(false)
expect(isLlmAdapterFailure(stream, 'consumer failed')).toBe(false)
for await (const _chunk of ctx.llm.stream({ provider: 'test', model: 'test', messages: [] })) break
})
it('unregisters adapters when the owning fiber is disposed (HMR safety)', async () => {
@@ -1119,19 +822,34 @@ describe('LlmService', () => {
expect(Object.isFrozen(prepared.config)).toBe(true)
expect(Object.isFrozen(prepared.adapterDefaults)).toBe(true)
expect(prepared.adapterDefaults).toEqual({ reasoningEffort: true })
const stream = prepared.stream({
expect(() => prepared.stream({
...prepared.config,
model: 'other',
messages: [],
})
await expect((async () => {
for await (const _chunk of stream) { /* drain */ }
})()).rejects.toMatchObject({ code: 'INVALID_PREPARED_CALL' })
})).toThrow(expect.objectContaining({ code: 'INVALID_PREPARED_CALL' }))
await collect(prepared.stream({
...prepared.config,
messages: [],
}))
expect(() => prepared.stream({
...prepared.config,
messages: [],
})).toThrow(expect.objectContaining({ code: 'INVALID_PREPARED_CALL' }))
const late = await ctx.llm.prepareCall({ provider: 'route', model: 'model' })
const lateOptions = { ...late.config, messages: [] }
const lateStream = late.stream(lateOptions)
lateOptions.model = 'other'
expect(await collect(lateStream)).toContainEqual({
type: 'finish',
reason: {
kind: 'error',
failure: {
message: 'prepared LLM call config changed before adapter dispatch',
code: 'INVALID_PREPARED_CALL',
},
},
})
})
it('reuses one exact-model lookup for prepared config and context metadata', async () => {

View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/llm/token-meter/README.md
README.md: 701893b342f9a93a75bec175634b1054f3d17151
README.zh.md: 0731e05186bec3f21d1f723c5b94ab7945e4139d
README.md: 0935a48a5f5773fbb280bc45e07faaa05c0a4f6e
README.zh.md: 83282ab47e6d406cfca30bbd8af40e0c94050504

View File

@@ -29,13 +29,13 @@ When the composition provides `ctx.sessionProjections`, token-meter registers tw
`contextPressure` carries optional `pressureTokens` — the newest provider-reported prompt size, summing uncached input plus cache reads and writes — and optional `contextWindow` from the newest `request/context` record. Pressure stays absent until a provider reports usage; capacity stays absent for a route whose adapter advertises none. Output is excluded, so the numerator holds still while a turn streams and steps forward when the next request reports its usage.
Both units use the standard projection baseline, live frame, higher-seq-wins store, and JSON checkpoint paths. Unloading token-meter removes both keys. A headless or TUI composition without the projection seam keeps the measurement service's existing behavior.
Both units use the standard projection baseline, live frame, higher-seq-wins store, and JSON checkpoint paths. Unloading token-meter removes both keys. A composition without the projection seam keeps the measurement service's existing behavior.
### Context occupancy is an approximation, by design
`pressureTokens` and `contextWindow` are independent last-wins fields and are **not** one atomic observation of a single request. Switching models pairs the fresh capacity with the previous route's pressure until the next request reports usage, and `pressureTokens` describes the last request rather than the surface as it stands right now.
This is deliberate. An occupancy percentage is a user-facing reference figure, not a billing record or a gating input — nothing in the harness makes decisions from it, and compaction reads `measure()` instead. The TUI status line has always computed occupancy the same way, dividing a `measure()` total by a separately-resolved capacity for the selected model.
This is deliberate. An occupancy percentage is a user-facing reference figure, not a billing record or a gating input — nothing in the harness makes decisions from it, and compaction reads `measure()` instead. A UI computes occupancy by dividing measured pressure by the separately resolved capacity for the selected model.
Making the pair atomic was tried and rejected: it required a transient non-replayable wire frame, which needed lifecycle fencing against cross-stream reordering and left occupancy blank after every reconnect. The [Agent Note](../../../.agents/notes/implemented/architecture/2026-07-29-projected-token-usage-and-request-context.md) records that comparison. Consumers that need an exact same-boundary figure should call `measure()` at their own request boundary rather than read this projection.
@@ -62,4 +62,3 @@ No direct invalidation; the named consumer owns any request-prefix changes.
- **Every measurement clones the current surface** — coherent immutable snapshots make reads O(surface), including below-threshold pressure checks.
- **Provider usage is only reusable for an identical canonical envelope** — prompt, prefix, tools, provider, model, or call-config changes deliberately fall back to full heuristic estimation.
- **Legacy provenance is conservative** — assistant messages without `sourceEventSeqs` cannot distinguish provider output from listener rewrites, so the fold avoids claiming a known empty or exact chunk stream.
- **The TUI and browser fixture retain parallel folds** — `tokenUsage` owns durable session-projection semantics; the TUI keeps its live per-step map because its composition does not mount the generic projection seam, while the browser fixture mirrors the unit for standalone demo data.

View File

@@ -29,13 +29,13 @@ fold 跟踪完整请求标头快照、步骤边界、表层追加与替换、成
`contextPressure` 携带可选的 `pressureTokens`(提供方报告的最新提示词规模,为未缓存输入加缓存读取与写入之和),以及来自最新一条 `request/context` 记录的可选 `contextWindow`。提供方报告用量前压力保持缺失;路由适配器未公布容量时容量也保持缺失。输出不计入其中,因此轮次流式输出期间分子保持不动,等到下一个请求报告用量时才前进。
两个单元都使用标准的投影基线、实时帧、seq 高者胜值仓和 JSON 检查点路径。卸载 token-meter 会移除这两个键。不带投影 seam 的 headless 或 TUI 组合会保留测量服务的既有行为。
两个单元都使用标准的投影基线、实时帧、seq 高者胜值仓和 JSON 检查点路径。卸载 token-meter 会移除这两个键。不带投影 seam 的组合会保留测量服务的既有行为。
### 上下文占用率是刻意为之的近似值
`pressureTokens``contextWindow` 是两个各自后者胜的独立字段,**不是**对单个请求的一次原子观测。切换模型时,新容量会与上一路由的压力配对,直到下一个请求报告用量为止;而 `pressureTokens` 描述的是最后一个请求,不是此刻的表层。
这是刻意的选择。占用率百分比是面向用户的参考数字既不是计费记录也不是门控输入harness 中没有任何环节依据它做决策,压缩改为直接读取 `measure()`TUI 状态行一直以同样的方式计算占用率,即用 `measure()` 总量除以为所选模型单独解析出的容量。
这是刻意的选择。占用率百分比是面向用户的参考数字既不是计费记录也不是门控输入harness 中没有任何环节依据它做决策,压缩改为直接读取 `measure()`。UI 用测得的压力除以为所选模型单独解析出的容量来计算占用率
让这对值保持原子已经尝试过并被否决:它需要一个临时且不可回放的协议帧,进而需要针对跨流重排序的生命周期栅栏,还会让占用率在每次重连后变为空白。[Agent Note](../../../.agents/notes/implemented/architecture/2026-07-29-projected-token-usage-and-request-context.md)记录了这项对比。需要同一边界精确数字的消费方应在自己的请求边界调用 `measure()`,而不是读取该投影。
@@ -62,4 +62,3 @@ fold 跟踪完整请求标头快照、步骤边界、表层追加与替换、成
- **每次测量都会克隆当前表层**:一致且不可变的快照使读取成为 O(surface),包括低于阈值的压力检查。
- **提供方用量只能为完全相同的规范 envelope 复用**:提示词、前缀、工具、提供方、模型或调用配置变更都会有意回退到完整启发式估算。
- **遗留溯源采取保守策略**:没有 `sourceEventSeqs` 的 assistant 消息无法区分提供方输出与 listener 改写,因此 fold 不会声称已知空流或精确分片流。
- **TUI 与浏览器 fixture 仍保留并行 fold**`tokenUsage` 拥有持久会话投影语义TUI 的组合未挂载通用投影 seam因此继续维护实时的逐步骤 map而浏览器 fixture 会为独立 demo 数据镜像该单元。

View File

@@ -93,8 +93,8 @@ export class TokenMeterService extends Service {
super(ctx, 'tokenMeter')
validateConfigKeys(config)
// Projection registration is an optional child: headless and TUI
// compositions without the generic registry keep the meter's old shape.
// Projection registration is an optional child: compositions without the
// generic registry keep the meter's standalone read shape.
ctx.inject(['sessionProjections'], (projectionCtx) => {
projectionCtx.sessionProjections.register(tokenUsageProjectionDefinition)
projectionCtx.sessionProjections.register(contextPressureProjectionDefinition)

View File

@@ -25,8 +25,7 @@ export interface TokenUsageProjection {
* `contextWindow` the newest recorded route capacity. Switching models can
* therefore pair a fresh capacity with the previous route's pressure until the
* next request reports usage. This is an intentional trade — the value is a
* user-facing reference, not a billing or gating input — and it matches how
* the TUI status line has always computed occupancy. See the token-meter
* user-facing reference, not a billing or gating input. See the token-meter
* README for the full rationale.
*/
export interface ContextPressureProjection {

View File

@@ -671,7 +671,7 @@ describe('malformed replay and listener lifecycle', () => {
type: 'turn/start',
seq: 0,
time: 1,
data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
data: { turn: 1 },
}] })
activeMeter.measure(session)
session.append('user/message', createUserMessage({