From 015ba14bae3bcd817b88636d375add27ee31d6cf Mon Sep 17 00:00:00 2001 From: Turtle Date: Sat, 25 Jul 2026 13:30:34 +0800 Subject: [PATCH] fix(llm): preserve serving retry policy --- ...26-07-24-provider-retry-policies.i18n.yaml | 4 +- .../2026-07-24-provider-retry-policies.md | 8 +- .../2026-07-24-provider-retry-policies.zh.md | 8 +- docs/config-catalog.md | 4 +- docs/cordis-catalog/events.md | 13 ++- .../llm-streaming.i18n.yaml | 4 +- docs/core-data-structures/llm-streaming.md | 4 +- docs/core-data-structures/llm-streaming.zh.md | 4 +- docs/event-producer-consumer.md | 8 +- packages/compact/compact-basic/src/index.ts | 1 + .../compact-basic/tests/compact-basic.spec.ts | 2 +- .../cordis/tool-cordis/src/api-catalog.ts | 4 +- packages/core/agent-loop/README.md | 2 +- packages/core/agent-loop/src/loop.ts | 23 ++-- .../agent-loop/tests/request-recovery.spec.ts | 44 ++++++-- packages/core/agent/src/types.ts | 6 +- packages/core/scope/tests/invariant.spec.ts | 2 +- packages/llm/llm-retry/README.md | 2 +- packages/llm/llm-retry/src/index.ts | 15 +-- packages/llm/llm-retry/tests/retry.spec.ts | 106 ++++++++++++++++-- packages/llm/llm/README.md | 2 +- packages/llm/llm/src/adapter-failure.ts | 29 ++++- packages/llm/llm/src/index.ts | 8 +- packages/llm/llm/tests/service.spec.ts | 47 ++++++++ packages/plan/plan-mode/src/index.ts | 1 + .../plan/plan-mode/tests/integration.spec.ts | 4 +- .../plan/plan-mode/tests/plan-mode.spec.ts | 5 +- 27 files changed, 278 insertions(+), 82 deletions(-) diff --git a/.agents/notes/implemented/feature/2026-07-24-provider-retry-policies.i18n.yaml b/.agents/notes/implemented/feature/2026-07-24-provider-retry-policies.i18n.yaml index ee221716ce..000b11d7f0 100644 --- a/.agents/notes/implemented/feature/2026-07-24-provider-retry-policies.i18n.yaml +++ b/.agents/notes/implemented/feature/2026-07-24-provider-retry-policies.i18n.yaml @@ -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 -2026-07-24-provider-retry-policies.md: 0ff304715aa5e1bfaf296e3f53e1da28d8b4042d -2026-07-24-provider-retry-policies.zh.md: 3b1d8f12a5fb37424e8b964f7bbb2be278263056 +2026-07-24-provider-retry-policies.md: 0b9456eb1563bfcbfa06d93403fd68124ed1f707 +2026-07-24-provider-retry-policies.zh.md: d3ef90ec739eec7300b7d0bb526cb5de958ed93f diff --git a/.agents/notes/implemented/feature/2026-07-24-provider-retry-policies.md b/.agents/notes/implemented/feature/2026-07-24-provider-retry-policies.md index 0ff304715a..0b9456eb15 100644 --- a/.agents/notes/implemented/feature/2026-07-24-provider-retry-policies.md +++ b/.agents/notes/implemented/feature/2026-07-24-provider-retry-policies.md @@ -12,7 +12,7 @@ Provider policy must follow the request that actually failed, including a route ## Decision -Each concrete adapter accepts an optional `retryPolicy` inside its provider configuration. The adapter validates and resolves the policy, and `ctx.llm` captures it when that exact provider route registers. `@deepseek-ai/dsh-llm-retry` reads the registered policy for the provider whose step failed. A provider without `retryPolicy` uses the normal defaults. +Each concrete adapter accepts an optional `retryPolicy` inside its provider configuration. The adapter validates and resolves the policy, and `ctx.llm` captures it when that exact provider route registers. When a call enters its final adapter boundary, `ctx.llm` binds the serving registration's immutable policy to that call; the agent loop passes it to closed-step recovery even if the route is disposed or replaced while the request is in flight. `@deepseek-ai/dsh-llm-retry` combines that call-local policy with the failed step's durable provider identity. A call that never reaches a final adapter has no serving policy and delegates. A provider without `retryPolicy` uses the normal defaults. ```yaml providers: @@ -34,7 +34,7 @@ providers: jitterRatio: 0.2 ``` -The listener selects the policy from the durable `request/header` in force when the failed step closed, excluding later recovery mutations. Normal mode retains the bounded transient behavior: it retries configured codes up to `maxRetries`, counts retries scheduled by the same provider policy in the current consecutive failure sequence, and otherwise delegates. +The listener reads the provider from the durable `request/header` in force when the failed step closed, excluding later recovery mutations, but never re-resolves policy from the mutable provider registry. Normal mode retains the bounded transient behavior: it retries configured codes up to `maxRetries`, counts retries scheduled by the same provider policy in the current consecutive failure sequence, and otherwise delegates. Always mode asks downstream recovery first so a specialized policy such as context-overflow compaction can make progress. A downstream retry wins. A downstream failure decision or thrown recovery error falls back to an unbounded retry of the same provider request; the thrown error is logged. Success, turn cancellation, and plugin disposal are the only termination paths. @@ -56,10 +56,10 @@ Each scheduled retry appends a non-surface `llm/retry` event with the failed pro ## Verification -Adapter tests validate nested policies at provider load and prove registration captures configured and default policies. Unit and real-Loader composition tests select policies from the failed request's provider, exercise always mode beyond the normal budget, pin jitter and delay caps, prove downstream recovery ordering, prove cancellation interrupts stalled downstream recovery, and prove cancellation and disposal stop active backoff waits. Request-level coverage compares the complete messages of failed and retried attempts and rejects both provider error text and discarded partial output. JSONL and SQLite tests round-trip an always event without `Infinity`; invariant tests bind its provider to the request header and its retry number to the active provider policy; ACP and TUI tests render finite and infinite limits. +Adapter tests validate nested policies at provider load, prove registration captures configured and default policies, and retain the serving policy across in-flight route replacement. Unit and real-Loader composition tests select policies from the failed request's serving registration, exercise always mode beyond the normal budget, pin jitter and delay caps, prove downstream recovery ordering, prove cancellation interrupts stalled downstream recovery, and prove cancellation and disposal stop active backoff waits. Request-level coverage compares the complete messages of failed and retried attempts and rejects both provider error text and discarded partial output. JSONL and SQLite tests round-trip an always event without `Infinity`; invariant tests bind its provider to the request header and its retry number to the active provider policy; ACP and TUI tests render finite and infinite limits. ## Consequences -Normal mode remains a finite default, while an explicit always policy can spend unbounded requests and time on permanent authentication, quota, invalid-request, protocol, or context failures. Operators must pair always mode with a cancellable caller and provider-specific cost controls. Retry state stays observable and durable without becoming model-visible, and exact-provider selection keeps one provider's exceptional policy from changing another provider's recovery behavior. +Normal mode remains a finite default, while an explicit always policy can spend unbounded requests and time on permanent authentication, quota, invalid-request, protocol, or context failures. Operators must pair always mode with a cancellable caller and provider-specific cost controls. Retry state stays observable and durable without becoming model-visible, and serving-registration capture prevents adapter lifecycle changes from retroactively changing an in-flight request's recovery contract. This decision extends the closed-step recovery, single visible adapter attempt, structured failure, and durable status design in [bounded recovery for transient LLM request failures](../architecture/2026-06-21-bounded-llm-request-recovery.md). diff --git a/.agents/notes/implemented/feature/2026-07-24-provider-retry-policies.zh.md b/.agents/notes/implemented/feature/2026-07-24-provider-retry-policies.zh.md index 3b1d8f12a5..d3ef90ec73 100644 --- a/.agents/notes/implemented/feature/2026-07-24-provider-retry-policies.zh.md +++ b/.agents/notes/implemented/feature/2026-07-24-provider-retry-policies.zh.md @@ -12,7 +12,7 @@ Status: implemented ## 决策 -每个具体适配器都在其提供方配置中接受可选的 `retryPolicy`。适配器负责校验并解析策略，`ctx.llm` 则在该精确提供方路由注册时捕获策略。`@deepseek-ai/dsh-llm-retry` 读取失败步骤对应提供方的已注册策略。未配置 `retryPolicy` 的提供方使用 normal 默认值。 +每个具体适配器都在其提供方配置中接受可选的 `retryPolicy`。适配器负责校验并解析策略，`ctx.llm` 则在该精确提供方路由注册时捕获策略。当调用进入最终适配器边界时，`ctx.llm` 会把实际提供服务的注册项所持不可变策略绑定到该调用；即使路由在请求进行期间被 dispose 或替换，agent loop 仍会把该策略传给已关闭步骤恢复。`@deepseek-ai/dsh-llm-retry` 会把绑定到该调用的策略与失败步骤的持久提供方标识结合起来。未到达最终适配器的调用没有实际提供服务的策略，因而会委托后续处理。未配置 `retryPolicy` 的提供方使用 normal 默认值。 ```yaml providers: @@ -34,7 +34,7 @@ providers: jitterRatio: 0.2 ``` -监听器根据失败步骤关闭时生效的持久 `request/header` 选择策略，后续恢复产生的改动不参与选择。normal 模式保留有界瞬态错误处理行为：它重试配置的错误代码，次数不超过 `maxRetries`；在当前连续失败序列中，同一提供方策略安排的重试都计入次数；其他情况委托后续处理。 +监听器从失败步骤关闭时生效的持久 `request/header` 读取提供方，后续恢复产生的改动不参与选择，但绝不会从可变的提供方注册表重新解析策略。normal 模式保留有界瞬态错误处理行为：它重试配置的错误代码，次数不超过 `maxRetries`；在当前连续失败序列中，同一提供方策略安排的重试都计入次数；其他情况委托后续处理。 always 模式先请求下游恢复，使上下文溢出压缩（compaction）之类的专用策略有机会取得进展。下游若决定重试，则以该决定为准。下游若决定失败或恢复过程抛出错误，则回退为无界重试同一提供方请求；抛出的错误会写入日志。成功、轮次取消和插件 dispose（资源释放）是仅有的终止路径。 @@ -56,10 +56,10 @@ always 模式先请求下游恢复，使上下文溢出压缩（compaction）之 ## 验证 -适配器测试会在提供方加载时校验嵌套策略，并证明注册流程会捕获已配置策略和默认策略。单元测试与真实 Loader 组合测试根据失败请求的提供方选择策略、验证 always 模式可越过 normal 预算、固定抖动和延迟上限、证明下游恢复顺序、证明取消会中断停滞的下游恢复，并证明取消与 dispose 会停止正在进行的退避等待。请求级覆盖会比较失败尝试与重试尝试的完整消息，并排除提供方错误文本和丢弃的部分输出。JSONL 与 SQLite 测试会往返读写不含 `Infinity` 的 always 事件；不变式测试会将事件中的提供方绑定到请求头，并将重试编号绑定到活跃的提供方策略；ACP 与 TUI 测试会分别渲染有限和无限上限。 +适配器测试会在提供方加载时校验嵌套策略，证明注册流程会捕获已配置策略和默认策略，并证明请求进行期间替换路由后仍会保留实际提供服务的策略。单元测试与真实 Loader 组合测试根据失败请求实际使用的注册项选择策略、验证 always 模式可越过 normal 预算、固定抖动和延迟上限、证明下游恢复顺序、证明取消会中断停滞的下游恢复，并证明取消与 dispose 会停止正在进行的退避等待。请求级覆盖会比较失败尝试与重试尝试的完整消息，并排除提供方错误文本和丢弃的部分输出。JSONL 与 SQLite 测试会往返读写不含 `Infinity` 的 always 事件；不变式测试会将事件中的提供方绑定到请求头，并将重试编号绑定到活跃的提供方策略；ACP 与 TUI 测试会分别渲染有限和无限上限。 ## 后果 -normal 模式仍是有限的默认策略；显式的 always 策略可能在永久性的身份验证、配额、无效请求、协议或上下文错误上耗费无限次请求和无限时间。运维方必须为 always 模式配备可取消的调用方和针对提供方的成本控制。重试状态保持可观察且持久，但不会对模型可见；精确提供方选择也能避免某个提供方的例外策略改变其他提供方的恢复行为。 +normal 模式仍是有限的默认策略；显式的 always 策略可能在永久性的身份验证、配额、无效请求、协议或上下文错误上耗费无限次请求和无限时间。运维方必须为 always 模式配备可取消的调用方和针对提供方的成本控制。重试状态保持可观察且持久，但不会对模型可见；捕获实际提供服务的注册项，也能防止适配器生命周期变化反过来改变进行中请求的恢复契约。 本决策扩展了[瞬态 LLM（大语言模型）请求失败的有界恢复](../architecture/2026-06-21-bounded-llm-request-recovery.md)中确定的已关闭 step 恢复、单次可见适配器尝试、结构化失败与持久状态设计。 diff --git a/docs/config-catalog.md b/docs/config-catalog.md index 2036a34307..9219ee0f10 100644 --- a/docs/config-catalog.md +++ b/docs/config-catalog.md @@ -646,11 +646,11 @@ Source: [`packages/support/llm-replay/src/index.ts:387`](../packages/support/llm ## `@deepseek-ai/dsh-llm-retry` -Requires: `agents` · `llm` +Requires: `agents` ```ts config-catalog /** This policy executor has no config; providers own `retryPolicy`. */ -export type Config = Readonly> +export type Config = Readonly> ``` Source: [`packages/llm/llm-retry/src/index.ts:43`](../packages/llm/llm-retry/src/index.ts) diff --git a/docs/cordis-catalog/events.md b/docs/cordis-catalog/events.md index b6a15c19b6..941c92eacc 100644 --- a/docs/cordis-catalog/events.md +++ b/docs/cordis-catalog/events.md @@ -96,7 +96,7 @@ A step or turn errored. The loop reports a failure here (plus the logger) even w Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:498`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:499`](../../packages/core/agent/src/types.ts) ### `agent/inbox/dequeue` — emit @@ -281,16 +281,17 @@ Recover a model-request failure after its failed step has closed. `retry` opens * @param error - the original model-request failure. * @param failure - serializable facts normalized at the final adapter boundary. * @param priorFailures - immutable failures that already authorized another request in this consecutive sequence. + * @param retryPolicy - immutable policy of the adapter registration that served the failed request, or `undefined` if no final adapter served it. * @param signal - the turn abort signal. * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. * @mode waterfall */ -'agent/request-error'(this: Scoped, agent: Agent, turn: number, step: number, error: RequestError, failure: LlmFailure, priorFailures: readonly LlmFailure[], signal: AbortSignal, next: () => Promise): Promise +'agent/request-error'(this: Scoped, agent: Agent, turn: number, step: number, error: RequestError, failure: LlmFailure, priorFailures: readonly LlmFailure[], retryPolicy: ResolvedRetryPolicy | undefined, signal: AbortSignal, next: () => Promise): Promise ``` -Types: [Agent](../core-data-structures/core.md) · [LlmFailure](../core-data-structures/llm-streaming.md) · [RequestError](../core-data-structures/core.md) · [RequestErrorDecision](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) +Types: [Agent](../core-data-structures/core.md) · [LlmFailure](../core-data-structures/llm-streaming.md) · [RequestError](../core-data-structures/core.md) · [RequestErrorDecision](../core-data-structures/core.md) · [ResolvedRetryPolicy](../core-data-structures/llm-streaming.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:463`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:464`](../../packages/core/agent/src/types.ts) ### `agent/session-prefix` — waterfall @@ -403,7 +404,7 @@ Override whether the turn continues. The default continues after tool calls or s Types: [Agent](../core-data-structures/core.md) · [ContinuationDecision](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:474`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:475`](../../packages/core/agent/src/types.ts) ### `agent/turn-stop` — serial @@ -425,7 +426,7 @@ Monotonic terminal-stop checkpoint after continuation and steering are folded; a Types: [Agent](../core-data-structures/core.md) · [ContinuationStop](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:485`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:486`](../../packages/core/agent/src/types.ts) ## `agent-loop/*` diff --git a/docs/core-data-structures/llm-streaming.i18n.yaml b/docs/core-data-structures/llm-streaming.i18n.yaml index 0d7edf9dfe..d7bab0aa3e 100644 --- a/docs/core-data-structures/llm-streaming.i18n.yaml +++ b/docs/core-data-structures/llm-streaming.i18n.yaml @@ -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 -llm-streaming.md: 2917d9796b32956f11ed5703f1f34c7510d0526c -llm-streaming.zh.md: 1bf7b3c226b5878deff3252396d96c2e97bf2982 +llm-streaming.md: 2d185e694f272114d4efa4f6be9020ef7c9a950f +llm-streaming.zh.md: 9734c643724e19fc39aff2ee94bd8de61935e16c diff --git a/docs/core-data-structures/llm-streaming.md b/docs/core-data-structures/llm-streaming.md index 2917d9796b..2d185e694f 100644 --- a/docs/core-data-structures/llm-streaming.md +++ b/docs/core-data-structures/llm-streaming.md @@ -59,7 +59,7 @@ Every adapter MUST obey these, and every consumer may rely on them: - **`usage` before `finish`, nothing after `finish`.** Defer both to the provider's end-of-stream marker so a trailing usage-only chunk can't violate the ordering. - **Tool-call `arguments` stay raw JSON strings end-to-end.** Partial fragments stream via `argumentsDelta`; a provider that hands back parsed objects re-stringifies at `block-end`. -- **Two sanctioned error paths, one fact shape.** A failure may either THROW from `stream()` (transport/protocol errors) **or** end the stream with `finish {kind:'error'|'aborted', failure}` (provider in-band errors, for adapters that can't throw mid-stream). `LlmError.failure` carries the same `LlmFailure`. The final adapter boundary preserves the exact thrown `Error` object and associates immutable facts with that call; the agent loop closes the failed step and offers the error, facts, and immutable prior-retried facts to `agent/request-error`. Absent recovery the structured failure becomes the turn error, and no normal assistant message or tool side effect is committed for that attempt. +- **Two sanctioned error paths, one fact shape.** A failure may either THROW from `stream()` (transport/protocol errors) **or** end the stream with `finish {kind:'error'|'aborted', failure}` (provider in-band errors, for adapters that can't throw mid-stream). `LlmError.failure` carries the same `LlmFailure`. The final adapter boundary preserves the exact thrown `Error` object and associates immutable facts plus the serving registration's immutable retry policy with that call; the agent loop closes the failed step and offers the error, facts, immutable prior-retried facts, and serving policy to `agent/request-error`. Absent recovery the structured failure becomes the turn error, and no normal assistant message or tool side effect is committed for that attempt. - **One adapter call is one provider attempt.** Adapters disable library retries. Agent-level recovery opens another durable numbered step; direct `ctx.llm.stream()` callers remain single-attempt. - **Provider stalls are bounded at the transport.** Both shipping remote adapters expose positive finite `streamIdleTimeoutMs` with a five-minute default. The watchdog arms only while iterator `next()` is outstanding, uses one stable signal for the whole request, maps its own expiry to `TIMEOUT`, and keeps an earlier caller abort as `ABORTED`. - **Context overflow has one canonical code.** Both DeepSeek adapters classify explicit provider detail through `isContextWindowExceededError()` and surface `CONTEXT_WINDOW_EXCEEDED`, whether the failure arrives as a thrown HTTP `LlmError` or an in-band finish error. Consumers route on the code, never provider text. @@ -70,7 +70,7 @@ This contract is pinned down by two deliberately independent implementations: `d ## `ResolvedRetryPolicy` -Provider configuration resolves before route registration into an immutable discriminated union. Normal mode carries `mode: 'normal'`, finite `maxRetries`, `retryableCodes`, and required `initialDelayMs`, `maxDelayMs`, and `jitterRatio`; always mode carries `mode: 'always'` and the same required backoff fields without a finite maximum. `LlmService.providerRetryPolicy(provider)` returns the captured value and supplies normal defaults when the adapter omits one. The [generated config catalog](../config-catalog.md) owns the optional input shapes. +Provider configuration resolves before route registration into an immutable discriminated union. Normal mode carries `mode: 'normal'`, finite `maxRetries`, `retryableCodes`, and required `initialDelayMs`, `maxDelayMs`, and `jitterRatio`; always mode carries `mode: 'always'` and the same required backoff fields without a finite maximum. `LlmService.providerRetryPolicy(provider)` returns the currently registered value and supplies normal defaults when the adapter omits one; `llmRetryPolicyOf(stream)` returns the exact serving registration's captured value after that call enters its final adapter boundary, so later route disposal or replacement cannot change an in-flight failure's recovery policy. The [generated config catalog](../config-catalog.md) owns the optional input shapes. ## `AppIdentity` — app attribution diff --git a/docs/core-data-structures/llm-streaming.zh.md b/docs/core-data-structures/llm-streaming.zh.md index 1bf7b3c226..9734c64372 100644 --- a/docs/core-data-structures/llm-streaming.zh.md +++ b/docs/core-data-structures/llm-streaming.zh.md @@ -59,7 +59,7 @@ interface LlmFailure { - **`usage` 在 `finish` 之前，`finish` 之后不再有任何分片。** 将两者都推迟到提供方的流结束标记，这样尾部的 usage-only 分片就不会违反顺序。 - **工具调用的 `arguments` 全程保持原始 JSON 字符串。** 部分片段通过 `argumentsDelta` 流式传输；如果提供方返回的是已解析的对象，适配器在 `block-end` 时重新序列化为字符串。 -- **两条受支持的错误路径，一种事实形状。** 失败可以从 `stream()` 抛出（传输/协议错误），**或者**以 `finish {kind:'error'|'aborted', failure}` 结束流（无法在流中途抛异常的适配器用它表示提供方带内错误）。`LlmError.failure` 携带同一个 `LlmFailure`。最终适配器边界保留被抛出的确切 `Error` 对象，并将不可变事实关联到该调用；agent loop（智能体循环）关闭失败的步骤，再把错误、事实与不可变的先前已重试事实提供给 `agent/request-error`。若未恢复，结构化失败会成为轮次错误，并且该次尝试不会提交正常 assistant 消息或工具副作用。 +- **两条受支持的错误路径，一种事实形状。** 失败可以从 `stream()` 抛出（传输/协议错误），**或者**以 `finish {kind:'error'|'aborted', failure}` 结束流（无法在流中途抛异常的适配器用它表示提供方带内错误）。`LlmError.failure` 携带同一个 `LlmFailure`。最终适配器边界保留被抛出的确切 `Error` 对象，并将不可变事实以及实际提供服务的注册项所持不可变重试策略关联到该调用；agent loop（智能体循环）关闭失败的步骤，再把错误、事实、不可变的先前已重试事实与实际提供服务的策略提供给 `agent/request-error`。若未恢复，结构化失败会成为轮次错误，并且该次尝试不会提交正常 assistant 消息或工具副作用。 - **一次适配器调用就是一次提供方尝试。** 适配器禁用库重试。agent 层恢复会打开另一个持久、带编号的步骤；直接调用 `ctx.llm.stream()` 的调用方仍然只尝试一次。 - **提供方停顿在传输层受到时限约束。** 两个已交付的远程适配器都暴露正数且有限的 `streamIdleTimeoutMs`，默认五分钟。watchdog 只在 iterator `next()` 尚未完成时启动，整个请求使用同一个稳定 signal，把自身到期映射为 `TIMEOUT`，并把更早发生的调用方中止保留为 `ABORTED`。 - **上下文溢出只有一个规范 code。** 两个 DeepSeek 适配器都通过 `isContextWindowExceededError()` 对提供方的显式细节分类并暴露 `CONTEXT_WINDOW_EXCEEDED`，无论失败以抛出的 HTTP `LlmError` 还是带内 finish error 到达。消费方按 code 路由，绝不依赖提供方文本。 @@ -70,7 +70,7 @@ interface LlmFailure { ## `ResolvedRetryPolicy` -提供方配置会在路由注册前解析为不可变的可辨识联合类型。normal 模式包含 `mode: 'normal'`、有限的 `maxRetries`、`retryableCodes`，以及必填的 `initialDelayMs`、`maxDelayMs` 和 `jitterRatio`；always 模式包含 `mode: 'always'` 和相同的必填退避字段，但不含有限上限。`LlmService.providerRetryPolicy(provider)` 返回捕获的值；适配器未提供策略时，该方法会补上 normal 默认值。可选输入形状由[生成的配置目录](../config-catalog.md)定义。 +提供方配置会在路由注册前解析为不可变的可辨识联合类型。normal 模式包含 `mode: 'normal'`、有限的 `maxRetries`、`retryableCodes`，以及必填的 `initialDelayMs`、`maxDelayMs` 和 `jitterRatio`；always 模式包含 `mode: 'always'` 和相同的必填退避字段，但不含有限上限。`LlmService.providerRetryPolicy(provider)` 返回当前已注册的值；适配器未提供策略时，该方法会补上 normal 默认值。调用进入最终适配器边界后，`llmRetryPolicyOf(stream)` 返回实际提供服务的确切注册项所捕获的值，因此后续路由 dispose 或替换无法改变请求进行期间发生的失败所用恢复策略。可选输入形状由[生成的配置目录](../config-catalog.md)定义。 ## `AppIdentity`：应用归属 diff --git a/docs/event-producer-consumer.md b/docs/event-producer-consumer.md index 08dcc19d54..3a2802588b 100644 --- a/docs/event-producer-consumer.md +++ b/docs/event-producer-consumer.md @@ -11,7 +11,7 @@ This matrix shows which packages dispatch each harness-owned event and which pac | `agent/cancel-requested` | `emit` | [`packages/core/agent/src/types.ts:350`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`goal-session`](../packages/goal/goal-session) | | `agent/created` | `emit` | [`packages/core/agent/src/types.ts:285`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`goal-session`](../packages/goal/goal-session), [`tui`](../packages/ui/tui) | | `agent/disposed` | `emit` | [`packages/core/agent/src/types.ts:294`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`agent-loop`](../packages/core/agent-loop), [`goal-session`](../packages/goal/goal-session), [`tui`](../packages/ui/tui) | -| `agent/error` | `emit` | [`packages/core/agent/src/types.ts:498`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`goal-session`](../packages/goal/goal-session), `runtime`, [`tui`](../packages/ui/tui) | +| `agent/error` | `emit` | [`packages/core/agent/src/types.ts:499`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`goal-session`](../packages/goal/goal-session), `runtime`, [`tui`](../packages/ui/tui) | | `agent/inbox/dequeue` | `emit` | [`packages/core/agent/src/types.ts:326`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`agent`](../packages/core/agent) | | `agent/inbox/discard` | `emit` | [`packages/core/agent/src/types.ts:340`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`agent`](../packages/core/agent) | | `agent/inbox/enqueue` | `emit` | [`packages/core/agent/src/types.ts:316`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`agent`](../packages/core/agent), [`goal-session`](../packages/goal/goal-session), [`tui`](../packages/ui/tui) | @@ -19,13 +19,13 @@ This matrix shows which packages dispatch each harness-owned event and which pac | `agent/pre-step` | `serial` | [`packages/core/agent/src/types.ts:379`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`time-context`](../packages/context/time-context), [`user-approval`](../packages/ui/user-approval) | | `agent/prompt-submit` | `waterfall` | [`packages/core/agent/src/types.ts:395`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`acp`](../packages/ui/acp), [`goal-session`](../packages/goal/goal-session), [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`plan-mode`](../packages/plan/plan-mode), [`repeat-tool-guard`](../packages/guard/repeat-tool-guard) | | `agent/request` | `waterfall` | [`packages/core/agent/src/types.ts:409`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`agent`](../packages/core/agent) | -| `agent/request-error` | `waterfall` | [`packages/core/agent/src/types.ts:463`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`compact-basic`](../packages/compact/compact-basic), [`llm-retry`](../packages/llm/llm-retry), [`plan-mode`](../packages/plan/plan-mode) | +| `agent/request-error` | `waterfall` | [`packages/core/agent/src/types.ts:464`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`compact-basic`](../packages/compact/compact-basic), [`llm-retry`](../packages/llm/llm-retry), [`plan-mode`](../packages/plan/plan-mode) | | `agent/session-prefix` | `waterfall` | [`packages/core/agent/src/types.ts:424`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`tool-skill`](../packages/skill/tool-skill), [`workspace-context`](../packages/context/workspace-context) | | `agent/session-start` | `emit` | [`packages/core/agent/src/types.ts:363`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`goal`](../packages/goal/goal), [`goal-session`](../packages/goal/goal-session), [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex) | | `agent/status` | `emit` | [`packages/core/agent/src/types.ts:303`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`agent`](../packages/core/agent), [`goal-session`](../packages/goal/goal-session), `runtime`, [`tui`](../packages/ui/tui) | | `agent/step-result` | `waterfall` | [`packages/core/agent/src/types.ts:436`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | - | -| `agent/turn-continuation` | `waterfall` | [`packages/core/agent/src/types.ts:474`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`plan-mode`](../packages/plan/plan-mode) | -| `agent/turn-stop` | `serial` | [`packages/core/agent/src/types.ts:485`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`subagent-inprocess`](../packages/subagent/subagent-inprocess), [`tool-goal`](../packages/goal/tool-goal) | +| `agent/turn-continuation` | `waterfall` | [`packages/core/agent/src/types.ts:475`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`plan-mode`](../packages/plan/plan-mode) | +| `agent/turn-stop` | `serial` | [`packages/core/agent/src/types.ts:486`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`subagent-inprocess`](../packages/subagent/subagent-inprocess), [`tool-goal`](../packages/goal/tool-goal) | | `approval/request` | `waterfall` | [`packages/ui/user-approval/src/index.ts:30`](../packages/ui/user-approval/src/index.ts) | [`user-approval`](../packages/ui/user-approval) (`waterfall`) | [`acp`](../packages/ui/acp) | | `commands/change` | `emit` | [`packages/ui/commands/src/index.ts:103`](../packages/ui/commands/src/index.ts) | [`commands`](../packages/ui/commands) (`events.dispatch`) | [`acp`](../packages/ui/acp), [`tui`](../packages/ui/tui) | | `fs/edit-intent` | `waterfall` | [`packages/fs/fs/src/index.ts:62`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`waterfall`) | [`fs-policy`](../packages/fs/fs-policy) | diff --git a/packages/compact/compact-basic/src/index.ts b/packages/compact/compact-basic/src/index.ts index 10c64d10ca..0ae0e02f7b 100644 --- a/packages/compact/compact-basic/src/index.ts +++ b/packages/compact/compact-basic/src/index.ts @@ -160,6 +160,7 @@ export class BasicCompactService extends CompactService { _error, failure, priorFailures, + _retryPolicy, signal, next, ) => { diff --git a/packages/compact/compact-basic/tests/compact-basic.spec.ts b/packages/compact/compact-basic/tests/compact-basic.spec.ts index 1db86d38e4..711dc3ee95 100644 --- a/packages/compact/compact-basic/tests/compact-basic.spec.ts +++ b/packages/compact/compact-basic/tests/compact-basic.spec.ts @@ -1277,7 +1277,7 @@ describe('automatic listener and loader composition', () => { const failure: LlmFailure = { message: error.message, code: error.code ?? 'UNKNOWN' } const priorFailures = Object.freeze(Array.from({ length: retryAttempt }, () => failure)) return agentEvents(ctx, owner).waterfall( - 'agent/request-error', 1, 1, error, failure, priorFailures, signal, next, + 'agent/request-error', 1, 1, error, failure, priorFailures, undefined, signal, next, ) } diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index 8284aa022c..acbf9c74ef 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -941,8 +941,8 @@ export const EVENT_API: readonly EventApiEntry[] = [ { name: 'agent/request-error', mode: 'waterfall', - signature: '\'agent/request-error\'(this: Scoped, agent: Agent, turn: number, step: number, error: RequestError, failure: LlmFailure, priorFailures: readonly LlmFailure[], signal: AbortSignal, next: () => Promise): Promise', - jsDoc: '/**\n * Recover a model-request failure after its failed step has closed. `retry`\n * opens a new numbered step; `fail` preserves the original request error.\n * Call `next()` to delegate to the next recovery listener or the default.\n * @param agent - the agent whose request failed.\n * @param turn - the open turn number.\n * @param step - the failed step number.\n * @param error - the original model-request failure.\n * @param failure - serializable facts normalized at the final adapter boundary.\n * @param priorFailures - immutable failures that already authorized another request in this consecutive sequence.\n * @param signal - the turn abort signal.\n * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.\n * @mode waterfall\n */', + signature: '\'agent/request-error\'(this: Scoped, agent: Agent, turn: number, step: number, error: RequestError, failure: LlmFailure, priorFailures: readonly LlmFailure[], retryPolicy: ResolvedRetryPolicy | undefined, signal: AbortSignal, next: () => Promise): Promise', + jsDoc: '/**\n * Recover a model-request failure after its failed step has closed. `retry`\n * opens a new numbered step; `fail` preserves the original request error.\n * Call `next()` to delegate to the next recovery listener or the default.\n * @param agent - the agent whose request failed.\n * @param turn - the open turn number.\n * @param step - the failed step number.\n * @param error - the original model-request failure.\n * @param failure - serializable facts normalized at the final adapter boundary.\n * @param priorFailures - immutable failures that already authorized another request in this consecutive sequence.\n * @param retryPolicy - immutable policy of the adapter registration that served the failed request, or `undefined` if no final adapter served it.\n * @param signal - the turn abort signal.\n * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.\n * @mode waterfall\n */', summary: 'Recover a model-request failure after its failed step has closed.', }, { diff --git a/packages/core/agent-loop/README.md b/packages/core/agent-loop/README.md index f937067588..2a9a2468f3 100644 --- a/packages/core/agent-loop/README.md +++ b/packages/core/agent-loop/README.md @@ -60,7 +60,7 @@ The driver owns one agent for its lifetime and runs inside `ctx.agents.withIniti Every provider call that reaches a successful finish appends exactly one `assistant/message` completion anchor, including content-less calls and `max-tokens` finishes. A successful `agent/step-result` stores its transformed content; a rejected result records empty content before the original failure continues. The anchor retains exact chunk provenance (`[]` for a stream with no chunks) and usage when available, while empty content stays out of derived message history. -Plugin failure ends the current turn, not the loop. Only final adapter dispatch/iteration failures and terminal in-band error or aborted finishes enter `agent/request-error`; middleware, result processing, tools, and `agent/post-step` remain ordinary turn failures. Recovery receives the exact live error, immutable provider facts, and immutable prior failures after the failed step closes. A retry rebuilds from the durable log in a new numbered step, success clears the consecutive history, and exhaustion records the structured failure once on `turn/end`. AgentLoop privately owns one cancellation holder whose explicit signal spans prompt policy, assembly, every step, model and tool work, recovery, continuation, and terminal stop; it retires the holder immediately before publishing `turn/end`, while the driver may remain `running` through the durability flush. An effective `cancel()` emits the typed runtime-only `user | parent` cause before clearing pending work and cooperatively aborting the holder; notification failures cannot veto cancellation, work queued by a notification observer is cleared, work queued by a later abort observer belongs to the next turn, and idle cancellation emits nothing. Durable `turn/end` remains coarse `aborted`; undispatched model tool calls receive synthetic `tool/call` and `ABORTED_BEFORE_DISPATCH` result pairs. Disposal wins terminal classification, and work that ignores the signal must settle before quiescence. The [explicit-cancellation decision](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md) owns the lifecycle and race contract. Terminal continuation stops remain authoritative through turn close and durability flush. +Plugin failure ends the current turn, not the loop. Only final adapter dispatch/iteration failures and terminal in-band error or aborted finishes enter `agent/request-error`; middleware, result processing, tools, and `agent/post-step` remain ordinary turn failures. Recovery receives the exact live error, immutable provider facts, immutable prior failures, and the immutable retry policy of the adapter registration that served the request after the failed step closes; the policy is absent if no final adapter served it. A retry rebuilds from the durable log in a new numbered step, success clears the consecutive history, and exhaustion records the structured failure once on `turn/end`. AgentLoop privately owns one cancellation holder whose explicit signal spans prompt policy, assembly, every step, model and tool work, recovery, continuation, and terminal stop; it retires the holder immediately before publishing `turn/end`, while the driver may remain `running` through the durability flush. An effective `cancel()` emits the typed runtime-only `user | parent` cause before clearing pending work and cooperatively aborting the holder; notification failures cannot veto cancellation, work queued by a notification observer is cleared, work queued by a later abort observer belongs to the next turn, and idle cancellation emits nothing. Durable `turn/end` remains coarse `aborted`; undispatched model tool calls receive synthetic `tool/call` and `ABORTED_BEFORE_DISPATCH` result pairs. Disposal wins terminal classification, and work that ignores the signal must settle before quiescence. The [explicit-cancellation decision](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md) owns the lifecycle and race contract. Terminal continuation stops remain authoritative through turn close and durability flush. Within a step, exclusive calls form barriers; parallel-safe calls use a bounded rolling pool and are reclassified before start. Only dispatch/body overlaps. Policy, durable results, and result context remain model-ordered. Abort stops new calls, drains started results, then drains accepted batch context before the turn closes through the normal abort path. diff --git a/packages/core/agent-loop/src/loop.ts b/packages/core/agent-loop/src/loop.ts index 4324deca8d..fcf4dfa846 100644 --- a/packages/core/agent-loop/src/loop.ts +++ b/packages/core/agent-loop/src/loop.ts @@ -7,9 +7,9 @@ import { randomUUID } from 'node:crypto' import type { Context } from 'cordis' -import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, LlmFailure, Message } from '@deepseek-ai/dsh-llm' +import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, LlmFailure, Message, ResolvedRetryPolicy } from '@deepseek-ai/dsh-llm' import { isDeepStrictEqual } from 'node:util' -import { BlockAssembler, HarnessError, LlmError, assertNever, deepFreeze, errorChain, llmFailureOf, markAgentLoopRequest } from '@deepseek-ai/dsh-llm' +import { BlockAssembler, HarnessError, LlmError, assertNever, deepFreeze, errorChain, llmFailureOf, llmRetryPolicyOf, markAgentLoopRequest } from '@deepseek-ai/dsh-llm' import { agentEvents, agentInterruptReasonOf, assembleContextFor, AgentMessageId } from '@deepseek-ai/dsh-agent' import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision, RequestError, RequestErrorDecision } from '@deepseek-ai/dsh-agent' import { canonicalHeader } from '@deepseek-ai/dsh-session' @@ -33,6 +33,7 @@ class TerminalModelRequestFailure extends Error { constructor( readonly requestError: RequestError, readonly failure: LlmFailure, + readonly retryPolicy: ResolvedRetryPolicy | undefined, ) { super(failure.message, { cause: requestError }) this.name = 'TerminalModelRequestFailure' @@ -442,14 +443,18 @@ async function runTurn( let stepOutcome: | { hadToolCalls: boolean; finish: FinishReason } - | { requestError: RequestError; failure: LlmFailure } + | { requestError: RequestError; failure: LlmFailure; retryPolicy: ResolvedRetryPolicy | undefined } | { error: RequestError } try { stepOutcome = await runStep( ctx, events, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, signal) } catch (error: unknown) { if (error instanceof TerminalModelRequestFailure) { - stepOutcome = { requestError: error.requestError, failure: error.failure } + stepOutcome = { + requestError: error.requestError, + failure: error.failure, + retryPolicy: error.retryPolicy, + } } else { stepOutcome = { error: toError(error) } } @@ -470,7 +475,7 @@ async function runTurn( try { recoveryDecision = await events.waterfall( 'agent/request-error', turn, step, stepOutcome.requestError, - stepOutcome.failure, requestFailureHistory, signal, + stepOutcome.failure, requestFailureHistory, stepOutcome.retryPolicy, signal, () => Promise.resolve(defaultDecision), ) } catch (recoveryError: unknown) { @@ -714,14 +719,18 @@ async function runStep( } } catch (error: unknown) { const failure = llmFailureOf(stream, error) - if (failure !== undefined && error instanceof Error) throw new TerminalModelRequestFailure(error, failure) + if (failure !== undefined && error instanceof Error) { + throw new TerminalModelRequestFailure(error, failure, llmRetryPolicyOf(stream)) + } throw error } interruptionCheckpoint(signal) // Normalize failure finish chunks into the same path as thrown stream errors. const stepError = finishError(assembler.finish) - if (stepError) throw new TerminalModelRequestFailure(stepError.error, stepError.failure) + if (stepError) { + throw new TerminalModelRequestFailure(stepError.error, stepError.failure, llmRetryPolicyOf(stream)) + } const recordAssistantMessage = ( assembledContent: ContentBlock[], diff --git a/packages/core/agent-loop/tests/request-recovery.spec.ts b/packages/core/agent-loop/tests/request-recovery.spec.ts index 1fc44e3431..494c2f95a9 100644 --- a/packages/core/agent-loop/tests/request-recovery.spec.ts +++ b/packages/core/agent-loop/tests/request-recovery.spec.ts @@ -269,10 +269,11 @@ describe('agent post-step and request-error lifecycle', () => { const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId(`recover-${_style}`), { provider: 'mock', model: 'mock' }) const attempts: number[] = [] - ctx.on('agent/request-error', async (subject, turn, step, error, facts, history) => { + ctx.on('agent/request-error', async (subject, turn, step, error, facts, history, retryPolicy) => { expect(subject).toBe(agent) expect({ turn, step, code: error.code }).toEqual({ turn: 1, step: 1, code: CONTEXT_WINDOW_EXCEEDED_CODE }) expect(facts.code).toBe(CONTEXT_WINDOW_EXCEEDED_CODE) + expect(retryPolicy).toMatchObject({ mode: 'normal', maxRetries: 2 }) attempts.push(history.length) subject.session.append('user/message', { content: [{ type: 'text', text: 'RECOVERY SURFACE MUTATION' }], @@ -301,7 +302,9 @@ describe('agent post-step and request-error lifecycle', () => { const agent = ctx.agentLoop.create(SessionId(`stream-plugin-${_name.replaceAll(' ', '-')}`), { provider: 'mock', model: 'mock' }) let recoveries = 0 install(ctx) - ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, _signal, next) => { + ctx.on('agent/request-error', async ( + _agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next, + ) => { recoveries += 1 return next() }) @@ -332,7 +335,9 @@ describe('agent post-step and request-error lifecycle', () => { }) const agent = ctx.agentLoop.create(SessionId('nested-stream-not-recoverable'), { provider: 'mock', model: 'mock' }) let recoveries = 0 - ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, _signal, next) => { + ctx.on('agent/request-error', async ( + _agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next, + ) => { recoveries += 1 return next() }) @@ -365,7 +370,9 @@ describe('agent post-step and request-error lifecycle', () => { } const agent = ctx.agentLoop.create(SessionId(`${boundary}-not-recoverable`), { provider: 'mock', model: 'mock' }) let recoveries = 0 - ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, _signal, next) => { + ctx.on('agent/request-error', async ( + _agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next, + ) => { recoveries += 1 return next() }) @@ -393,7 +400,9 @@ describe('agent post-step and request-error lifecycle', () => { } const agent = ctx.agentLoop.create(SessionId(`${failure}-not-recoverable`), { provider: 'mock', model: 'mock' }) let recoveries = 0 - ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, _signal, next) => { + ctx.on('agent/request-error', async ( + _agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next, + ) => { recoveries += 1 return next() }) @@ -412,7 +421,9 @@ describe('agent post-step and request-error lifecycle', () => { const ctx = await harness(makeAdapter(original)) const agent = ctx.agentLoop.create(SessionId(`identity-${_name.replaceAll(' ', '-')}`), { provider: 'mock', model: 'mock' }) let seen: Error | undefined - ctx.on('agent/request-error', async (_agent, _turn, _step, error, _failure, _history, _signal, next) => { + ctx.on('agent/request-error', async ( + _agent, _turn, _step, error, _failure, _history, _retryPolicy, _signal, next, + ) => { seen = error return next() }) @@ -431,7 +442,9 @@ describe('agent post-step and request-error lifecycle', () => { const agent = ctx.agentLoop.create(SessionId('hostile-message-recovery'), { provider: 'mock', model: 'mock' }) let seenError: Error | undefined let seenFailure: LlmFailure | undefined - ctx.on('agent/request-error', async (_agent, _turn, _step, error, failure, _history, _signal, next) => { + ctx.on('agent/request-error', async ( + _agent, _turn, _step, error, failure, _history, _retryPolicy, _signal, next, + ) => { seenError = error seenFailure = failure return next() @@ -462,7 +475,7 @@ describe('agent post-step and request-error lifecycle', () => { let seenFailure: LlmFailure | undefined let seenHistory: readonly LlmFailure[] | undefined ctx.on('agent/request-error', async ( - _agent, _turn, _step, error, failure, history, _signal, next, + _agent, _turn, _step, error, failure, history, _retryPolicy, _signal, next, ) => { seenError = error seenFailure = failure @@ -506,13 +519,18 @@ describe('agent post-step and request-error lifecycle', () => { const ctx = scenario === 'iterator' ? await harness(new IteratorConstructionFailureAdapter()) : await harness() const agent = ctx.agentLoop.create(SessionId(`request-boundary-${scenario}`), { provider: 'mock', model: 'mock' }) let seen = '' - ctx.on('agent/request-error', async (_agent, _turn, _step, error, _failure, _history, _signal, next) => { + let sawServingPolicy = false + ctx.on('agent/request-error', async ( + _agent, _turn, _step, error, _failure, _history, retryPolicy, _signal, next, + ) => { seen = error.code ?? '' + sawServingPolicy = retryPolicy !== undefined return next() }) send(agent) await waitForIdle(ctx, agent) expect(seen).toBe(scenario === 'iterator' ? 'ITERATOR_CONSTRUCTION' : 'NO_ADAPTER') + expect(sawServingPolicy).toBe(scenario === 'iterator') } }) @@ -522,7 +540,7 @@ describe('agent post-step and request-error lifecycle', () => { const cappedAgent = cappedCtx.agentLoop.create(SessionId('retry-cap'), { provider: 'mock', model: 'mock' }) const cappedHistories: string[][] = [] cappedCtx.on('agent/request-error', async ( - _agent, _turn, _step, _error, _failure, history, _signal, next, + _agent, _turn, _step, _error, _failure, history, _retryPolicy, _signal, next, ) => { const codes = history.map(entry => entry.code) cappedHistories.push(codes) @@ -547,7 +565,7 @@ describe('agent post-step and request-error lifecycle', () => { const resetAgent = resetCtx.agentLoop.create(SessionId('retry-reset'), { provider: 'mock', model: 'mock' }) const resetHistories: { step: number; codes: string[] }[] = [] resetCtx.on('agent/request-error', async ( - _agent, _turn, step, _error, _failure, history, _signal, next, + _agent, _turn, step, _error, _failure, history, _retryPolicy, _signal, next, ) => { resetHistories.push({ step, codes: history.map(entry => entry.code) }) return resetHistories.length === 1 ? { action: 'retry' } : next() @@ -578,7 +596,9 @@ describe('agent post-step and request-error lifecycle', () => { const agent = ctx.agentLoop.create(SessionId(`${action}-recovery`), { provider: 'mock', model: 'mock' }) let entered!: () => void const recoveryEntered = new Promise((resolve) => { entered = resolve }) - ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, signal) => { + ctx.on('agent/request-error', async ( + _agent, _turn, _step, _error, _failure, _history, _retryPolicy, signal, + ) => { entered() await new Promise((resolve) => { signal.addEventListener('abort', () => { resolve() }, { once: true }) diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index 0f442e3f88..0efe140b22 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -8,7 +8,7 @@ import type { Context } from 'cordis' import type { Branded } from '@deepseek-ai/dsh-brand' import type { Scoped } from '@deepseek-ai/dsh-scope' -import type { ContentBlock, LlmCallConfig, LlmFailure, Message, MessageSource } from '@deepseek-ai/dsh-llm' +import type { ContentBlock, LlmCallConfig, LlmFailure, Message, MessageSource, ResolvedRetryPolicy } from '@deepseek-ai/dsh-llm' import type { JsonValue, Session, SessionId } from '@deepseek-ai/dsh-session' import type {} from '@deepseek-ai/dsh-system-prompt' declare module '@deepseek-ai/dsh-system-prompt' { @@ -456,11 +456,13 @@ declare module 'cordis' { * @param error - the original model-request failure. * @param failure - serializable facts normalized at the final adapter boundary. * @param priorFailures - immutable failures that already authorized another request in this consecutive sequence. + * @param retryPolicy - immutable policy of the adapter registration that served + * the failed request, or `undefined` if no final adapter served it. * @param signal - the turn abort signal. * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. * @mode waterfall */ - 'agent/request-error'(this: Scoped, agent: Agent, turn: number, step: number, error: RequestError, failure: LlmFailure, priorFailures: readonly LlmFailure[], signal: AbortSignal, next: () => Promise): Promise + 'agent/request-error'(this: Scoped, agent: Agent, turn: number, step: number, error: RequestError, failure: LlmFailure, priorFailures: readonly LlmFailure[], retryPolicy: ResolvedRetryPolicy | undefined, signal: AbortSignal, next: () => Promise): Promise /** * Override whether the turn continues. The default continues after tool * calls or steering and stops otherwise; a continue reason becomes steering. diff --git a/packages/core/scope/tests/invariant.spec.ts b/packages/core/scope/tests/invariant.spec.ts index ca0841165b..c54753c301 100644 --- a/packages/core/scope/tests/invariant.spec.ts +++ b/packages/core/scope/tests/invariant.spec.ts @@ -51,7 +51,7 @@ describe('scoped-dispatch invariants', () => { 'agent/post-step': [agent, 1, 1, signal], 'agent/prompt-submit': [agent, [], { kind: 'user' }, signal, () => Promise.resolve({ kind: 'allow' })], 'agent/request': [agent, 1, 1, config, signal, () => Promise.resolve(config)], - 'agent/request-error': [agent, 1, 1, new Error('request failed'), { message: 'request failed', code: 'UNKNOWN' }, [], signal, () => Promise.resolve({ action: 'fail' })], + 'agent/request-error': [agent, 1, 1, new Error('request failed'), { message: 'request failed', code: 'UNKNOWN' }, [], undefined, signal, () => Promise.resolve({ action: 'fail' })], 'agent/session-prefix': [agent, [], signal, () => Promise.resolve([])], 'agent/step-result': [agent, 1, 1, message, signal, () => Promise.resolve(message)], 'agent/turn-continuation': [agent, 1, { action: 'stop' }, signal, () => Promise.resolve({ action: 'stop' })], diff --git a/packages/llm/llm-retry/README.md b/packages/llm/llm-retry/README.md index 711666b8a0..72d7ac6831 100644 --- a/packages/llm/llm-retry/README.md +++ b/packages/llm/llm-retry/README.md @@ -2,7 +2,7 @@ Function plugin that applies exact-provider retry policy on the agent loop's closed-step recovery seam. It does not wrap `ctx.llm.stream()`: every adapter call remains one provider attempt, and every retry opens a fresh numbered step. -Each provider adapter owns an optional nested `retryPolicy`, captured when its route registers on `ctx.llm`. Omission uses normal mode: two retries for `RATE_LIMIT`, `SERVER`, `TIMEOUT`, and `TRANSPORT`. A normal policy can change its finite budget, eligible codes, and backoff. Always mode asks downstream recovery first, then retries every model-request failure without an attempt limit; success, cancellation, or plugin disposal stops it. +Each provider adapter owns an optional nested `retryPolicy`, captured when its route registers on `ctx.llm` and carried with each call that reaches that registration's final adapter boundary. An in-flight failure retains that serving policy if the route is later disposed or replaced; a failure before any final adapter is selected has no provider policy and delegates. Omission uses normal mode: two retries for `RATE_LIMIT`, `SERVER`, `TIMEOUT`, and `TRANSPORT`. A normal policy can change its finite budget, eligible codes, and backoff. Always mode asks downstream recovery first, then retries every model-request failure without an attempt limit; success, cancellation, or plugin disposal stops it. Both modes use bounded exponential backoff with symmetric jitter. A valid `providerRetryAfterMs` at or below `maxDelayMs` replaces local backoff without jitter. An over-cap provider delay makes normal mode delegate, while always mode uses its configured local backoff so it cannot terminate on that instruction. diff --git a/packages/llm/llm-retry/src/index.ts b/packages/llm/llm-retry/src/index.ts index 04f1b0996a..7e9c2d56e6 100644 --- a/packages/llm/llm-retry/src/index.ts +++ b/packages/llm/llm-retry/src/index.ts @@ -37,13 +37,13 @@ declare module '@deepseek-ai/dsh-session' { } export const name = 'llm-retry' -export const inject = ['agents', 'llm'] +export const inject = ['agents'] /** This policy executor has no config; providers own `retryPolicy`. */ -export type Config = Readonly> +export type Config = Readonly> /** Runtime schema for {@link Config}. */ -export const Config: z = z.object({}) +export const Config = z.object({}) as unknown as z function validateConfig(config: Config): void { const [key] = Object.keys(config) @@ -170,6 +170,7 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna _error: RequestError, failure: LlmFailure, priorFailures: readonly LlmFailure[], + policy: ResolvedRetryPolicy | undefined, signal: AbortSignal, next: () => Promise, ) => { @@ -177,15 +178,15 @@ 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({ action: 'fail' }) - // Bind policy to the header in force when this step closed. Downstream - // recovery may append later state before an always fallback runs. + 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}`) } - const policy = ctx.llm.providerRetryPolicy(provider) - if (policy.mode === 'always') { const downstream = await downstreamUntilAbort( next, diff --git a/packages/llm/llm-retry/tests/retry.spec.ts b/packages/llm/llm-retry/tests/retry.spec.ts index 56f785f38c..232f4b4459 100644 --- a/packages/llm/llm-retry/tests/retry.spec.ts +++ b/packages/llm/llm-retry/tests/retry.spec.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, it, vi } from 'vitest' +import { afterEach, describe, expect, expectTypeOf, it, vi } from 'vitest' import { Context } from 'cordis' import type { Fiber } from 'cordis' import LlmService, { CallId, LlmAdapter, LlmError, resolveRetryPolicy } from '@deepseek-ai/dsh-llm' @@ -79,7 +79,7 @@ async function harness( policies: Readonly> = { mock: normalConfig() }, beforeRetry?: (ctx: Context) => void, internals: retry.RetryInternals = {}, -): Promise<{ ctx: Context; retryFiber: Fiber }> { +): Promise<{ ctx: Context; retryFiber: Fiber; disposeAdapter: () => void }> { const ctx = new Context() await ctx.plugin(LlmService) await ctx.plugin(SessionStore) @@ -92,8 +92,8 @@ async function harness( retry.apply(inner, {}, internals) }, { inject: retry.inject })) await ctx.plugin(AgentLoop, { agents: [] }) - ctx.llm.registerAdapter(['mock', 'other'], adapter) - return { ctx, retryFiber } + const disposeAdapter = ctx.llm.registerAdapter(['mock', 'other'], adapter) + return { ctx, retryFiber, disposeAdapter } } function normalConfig( @@ -413,6 +413,28 @@ describe('provider-routed retry policy', () => { expect(vi.getTimerCount()).toBe(0) }) + it('delegates when no final adapter served the failed request', async () => { + const adapter = new ScriptedAdapter([textResponse('must not run')]) + const mounted = await harness(adapter, { mock: alwaysConfig() }) + context = mounted.ctx + mounted.disposeAdapter() + const agent = context.agentLoop.create(SessionId('retry-no-serving-policy'), { + provider: 'mock', + model: 'mock', + }) + const idle = waitForIdle(context, agent) + + agent.followup([{ type: 'text', text: 'missing route' }]) + await idle + + expect(adapter.requests).toHaveLength(0) + 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: 'NO_ADAPTER' } } }, + }) + }) + it('selects policy by the failed request provider', async () => { vi.useFakeTimers() const adapter = new ScriptedAdapter([ @@ -481,6 +503,64 @@ describe('provider-routed retry policy', () => { expect(adapter.requests.map(request => request.provider)).toEqual(['other', 'other']) }) + it.each(['thrown', 'in-band'] as const)( + 'uses the serving registration policy when an in-flight route is replaced after a %s failure', + async (failureKind) => { + vi.useFakeTimers() + const entered = Promise.withResolvers() + const release = Promise.withResolvers() + const oldAdapter = new ScriptedAdapter([(async function * (): AsyncGenerator { + entered.resolve(undefined) + await release.promise + if (failureKind === 'thrown') { + throw new LlmError('old route auth failed', 'AUTH') + } + yield { + type: 'finish', + reason: { + kind: 'error', + failure: { message: 'old route auth failed', code: 'AUTH' }, + }, + } + })()]) + const mounted = await harness(oldAdapter, { mock: alwaysConfig({ + initialDelayMs: 1, + maxDelayMs: 1, + }) }) + context = mounted.ctx + const agent = context.agentLoop.create(SessionId('retry-serving-registration'), { + provider: 'mock', + model: 'mock', + }) + const scheduled = waitForRetry(context, agent, 1) + agent.followup([{ type: 'text', text: 'replace while in flight' }]) + await entered.promise + + mounted.disposeAdapter() + const replacement = new ScriptedAdapter([textResponse('replacement recovered')]) + replacement.configureRetryPolicies({ mock: normalConfig({ maxRetries: 0 }) }) + context.llm.registerAdapter(['mock'], replacement) + release.resolve(undefined) + + expect((await scheduled).data).toMatchObject({ + provider: 'mock', + mode: 'always', + retry: 1, + delayMs: 1, + }) + const idle = waitForIdle(context, agent) + await vi.advanceTimersByTimeAsync(1) + await idle + + expect(oldAdapter.requests).toHaveLength(1) + expect(replacement.requests).toHaveLength(1) + expect(agent.session.deriveMessages().at(-1)).toMatchObject({ + role: 'assistant', + content: [{ type: 'text', text: 'replacement recovered' }], + }) + }, + ) + it('keeps always mode unbounded while preserving cancellable jittered backoff', async () => { vi.useFakeTimers() const adapter = new ScriptedAdapter([ @@ -719,7 +799,9 @@ describe('provider-routed retry policy', () => { const captured = Promise.withResolvers() let invokeCaptured: (() => Promise) | undefined const mounted = await harness(adapter, {}, (ctx) => { - ctx.on('agent/request-error', (_agent, _turn, _step, _error, _failure, _history, _signal, next) => { + ctx.on('agent/request-error', ( + _agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next, + ) => { return new Promise((resolve) => { invokeCaptured = async () => { resolve(await next()) } captured.resolve(undefined) @@ -728,7 +810,9 @@ describe('provider-routed retry policy', () => { }) context = mounted.ctx let downstreamCalls = 0 - context.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, _signal, next) => { + context.on('agent/request-error', async ( + _agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next, + ) => { downstreamCalls += 1 return next() }) @@ -782,7 +866,9 @@ 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, _signal, next) => { + ctx.on('agent/request-error', async ( + agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next, + ) => { agent.cancel({ kind: 'user' }) return next() }) @@ -823,14 +909,16 @@ describe('provider-routed retry policy', () => { }) it('rejects retry policy configured on the executor instead of a provider', () => { + expectTypeOf<{}>().toExtend() + expectTypeOf<{ retryPolicy: { mode: 'always' } }>().not.toExtend() expect(() => { - retry.apply(new Context(), { retryPolicy: { mode: 'always' } }) + retry.apply(new Context(), { retryPolicy: { mode: 'always' } } as unknown as retry.Config) }).toThrow(/retryPolicy belongs under each provider/) }) it('rejects unknown executor config', () => { expect(() => { - retry.apply(new Context(), { retryPolciy: {} }) + retry.apply(new Context(), { retryPolciy: {} } as unknown as retry.Config) }).toThrow(/unknown key "retryPolciy"/) }) }) diff --git a/packages/llm/llm/README.md b/packages/llm/llm/README.md index 74dafa3e48..6a855f6004 100644 --- a/packages/llm/llm/README.md +++ b/packages/llm/llm/README.md @@ -15,7 +15,7 @@ An adapter registry plus a single streaming call surface, interceptable via a wa - `ctx.llm.resolveModelContext(provider: string, model: string): Promise` Resolve authoritative context capacity for one exact route from its owning adapter. - `ctx.llm.stream(options: GenerateOptions): AsyncIterable` 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`. 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` 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`. 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`. diff --git a/packages/llm/llm/src/adapter-failure.ts b/packages/llm/llm/src/adapter-failure.ts index 390282327d..87e783dcea 100644 --- a/packages/llm/llm/src/adapter-failure.ts +++ b/packages/llm/llm/src/adapter-failure.ts @@ -6,9 +6,15 @@ import { HarnessError } from './error.ts' import type { LlmFailure, StreamChunk } from './types.ts' +import type { ResolvedRetryPolicy } from './retry-policy.ts' -/** Errors and normalized facts proven to originate in one model call's final adapter boundary. */ -export type AdapterFailureScope = WeakMap +/** 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 + /** 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, AdapterFailureScope>() @@ -52,7 +58,7 @@ export function markLlmAdapterFailure( message: errorMessage(error), code: harnessErrorCode(error), }) - failures.set(error, failure) + failures.failures.set(error, failure) return error } @@ -124,7 +130,7 @@ export function isLlmAdapterFailure( value: unknown, ): value is Error & { code?: string } { const failures = adapterFailureScopes.get(stream) - return value instanceof Error && failures !== undefined && failures.has(value) + return value instanceof Error && failures !== undefined && failures.failures.has(value) } /** @@ -139,5 +145,18 @@ export function llmFailureOf( value: unknown, ): LlmFailure | undefined { const failures = adapterFailureScopes.get(stream) - return value instanceof Error ? failures?.get(value) : undefined + 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, +): ResolvedRetryPolicy | undefined { + return adapterFailureScopes.get(stream)?.retryPolicy } diff --git a/packages/llm/llm/src/index.ts b/packages/llm/llm/src/index.ts index 773b9a5ca1..6b714a7074 100644 --- a/packages/llm/llm/src/index.ts +++ b/packages/llm/llm/src/index.ts @@ -33,7 +33,7 @@ export * from './retry-policy.ts' export { BlockAssembler } from './assembler.ts' export { callConfigEquals, deepFreeze, isAgentLoopRequest, markAgentLoopRequest } from './call-config.ts' export type { LlmCallConfig } from './call-config.ts' -export { isLlmAdapterFailure, llmFailureOf } from './adapter-failure.ts' +export { isLlmAdapterFailure, llmFailureOf, llmRetryPolicyOf } from './adapter-failure.ts' declare module 'cordis' { interface Context { @@ -337,7 +337,9 @@ export class LlmService extends Service { ): AsyncGenerator { let iterator: AsyncIterator try { - const adapter = this.registration(options.provider).adapter + const registration = this.registration(options.provider) + failures.retryPolicy = registration.retryPolicy + const adapter = registration.adapter const stream = adapter.stream(this.forAdapter(options, adapter)) iterator = stream[Symbol.asyncIterator]() } catch (error: unknown) { @@ -386,7 +388,7 @@ export class LlmService extends Service { * @returns the chunk stream, possibly wrapped by `llm/stream` listeners. */ stream(options: GenerateOptions): AsyncIterable { - const failures: AdapterFailureScope = new WeakMap() + const failures: AdapterFailureScope = { failures: new WeakMap() } const stream = this.ctx.waterfall(this, 'llm/stream', options, () => this.adapterStream(options, failures)) return bindAdapterFailureScope(stream, failures) } diff --git a/packages/llm/llm/tests/service.spec.ts b/packages/llm/llm/tests/service.spec.ts index 52c51852ba..5631191b22 100644 --- a/packages/llm/llm/tests/service.spec.ts +++ b/packages/llm/llm/tests/service.spec.ts @@ -10,6 +10,7 @@ import LlmService, { LlmAdapter, LlmError, llmFailureOf, + llmRetryPolicyOf, ProviderRequestId, resolveRetryPolicy, StreamChunk, @@ -181,6 +182,51 @@ describe('LlmService', () => { ) }) + it('keeps the serving registration policy on an in-flight call after route replacement', async () => { + const oldPolicy = resolveRetryPolicy({ mode: 'always' }, 'old retryPolicy') + const newPolicy = resolveRetryPolicy({ mode: 'normal', maxRetries: 0 }, 'new retryPolicy') + const entered = Promise.withResolvers() + const release = Promise.withResolvers() + const failure = new LlmError('old route failed', 'AUTH') + const oldAdapter = new class extends LlmAdapter { + override providerRetryPolicy(): typeof oldPolicy { + return oldPolicy + } + + async * stream(_options: GenerateOptions): AsyncIterable { + entered.resolve(undefined) + await release.promise + throw failure + } + }() + const newAdapter = 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 => { + try { + for await (const _chunk of stream) { /* drain */ } + } catch (error: unknown) { + return error + } + return undefined + })() + await entered.promise + + disposeOld() + ctx.llm.registerAdapter(['route'], newAdapter) + release.resolve(undefined) + + expect(await outcome).toBe(failure) + expect(llmRetryPolicyOf(stream)).toBe(oldPolicy) + expect(ctx.llm.providerRetryPolicy('route')).toBe(newPolicy) + }) + it('throws NO_ADAPTER for unregistered providers', async () => { const ctx = new Context() await ctx.plugin(LlmService) @@ -195,6 +241,7 @@ describe('LlmService', () => { 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() }) it.each(['done', 'value'] as const)('tags a throwing IteratorResult.%s getter without replacing its Error', async (field) => { diff --git a/packages/plan/plan-mode/src/index.ts b/packages/plan/plan-mode/src/index.ts index 26ad904d30..6ba7a719d7 100644 --- a/packages/plan/plan-mode/src/index.ts +++ b/packages/plan/plan-mode/src/index.ts @@ -184,6 +184,7 @@ export class PlanModeService extends Service { _error, _failure, _priorFailures, + _retryPolicy, _signal, next, ) => { diff --git a/packages/plan/plan-mode/tests/integration.spec.ts b/packages/plan/plan-mode/tests/integration.spec.ts index e195ed54b2..1500554ad7 100644 --- a/packages/plan/plan-mode/tests/integration.spec.ts +++ b/packages/plan/plan-mode/tests/integration.spec.ts @@ -138,7 +138,9 @@ describe('plan mode through the agent loop', () => { const agent = ctx.agentLoop.create(SessionId('it-plan-retry-flip'), { provider: 'mock', model: 'mock' }) const recoveryEntered = Promise.withResolvers() const releaseRecovery = Promise.withResolvers() - ctx.on('agent/request-error', async (subject, _turn, _step, _error, _failure, _history, _signal, next) => { + ctx.on('agent/request-error', async ( + subject, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next, + ) => { if (subject !== agent) return next() recoveryEntered.resolve(true) await releaseRecovery.promise diff --git a/packages/plan/plan-mode/tests/plan-mode.spec.ts b/packages/plan/plan-mode/tests/plan-mode.spec.ts index a7f7743497..786d1425c2 100644 --- a/packages/plan/plan-mode/tests/plan-mode.spec.ts +++ b/packages/plan/plan-mode/tests/plan-mode.spec.ts @@ -83,6 +83,7 @@ function recoveryBoundary( new Error('request failed'), { message: 'request failed', code: 'SERVER' }, [], + undefined, new AbortController().signal, () => Promise.resolve(decision), ) @@ -945,7 +946,9 @@ describe('HMR disposal', () => { const agent = await agentWithSession(ctx, 'disposed-in-flight-recovery') const recoveryEntered = Promise.withResolvers() const releaseRecovery = Promise.withResolvers() - ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, _signal, _next) => { + ctx.on('agent/request-error', async ( + _agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, _next, + ) => { recoveryEntered.resolve(true) await releaseRecovery.promise return { action: 'retry' }