Merge remote-tracking branch 'origin/master' into worktree/web-model-request-retry
# Conflicts: # .agents/notes/implemented/architecture/2026-06-21-bounded-llm-request-recovery.i18n.yaml # .agents/notes/implemented/architecture/2026-06-21-bounded-llm-request-recovery.md # .agents/notes/implemented/architecture/2026-06-21-bounded-llm-request-recovery.zh.md # apps/web/tests/session-title.snapshot.ts # docs/config-catalog.md # packages/client/runtime/README.i18n.yaml # packages/client/runtime/README.md # packages/client/runtime/README.zh.md # packages/client/ui-conversation/README.i18n.yaml # packages/client/ui-conversation/src/client/chat/ChatView.tsx # packages/client/ui-conversation/tests/chat-view.spec.tsx # packages/llm/llm-retry/README.i18n.yaml # packages/llm/llm-retry/README.md # packages/llm/llm-retry/README.zh.md
This commit is contained in:
Yichen Jiang
@@ -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: a742f14a137dcad151c6dfe1e179081a184d6d23
|
||||
README.zh.md: 789caac3b67c35c2ab9d071c59a7f78edae1e142
|
||||
README.md: 8de3ea8c9321f04f5af1b0d7ab361f73eaabc822
|
||||
README.zh.md: 227dfc31b08622b197debca9855f35201c5becf5
|
||||
|
||||
@@ -2,42 +2,52 @@
|
||||
|
||||
English | [中文](README.zh.md)
|
||||
|
||||
Function plugin that retries selected transient model-request failures 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.
|
||||
Function plugin that applies exact-provider retry policy through the agent loop's closed-step `agent/request-error` waterfall. It does not wrap `ctx.llm.stream()`: every adapter call remains one provider attempt, and every retry opens a fresh numbered turn.
|
||||
|
||||
The default policy permits two retries for `EMPTY_RESPONSE`, `RATE_LIMIT`, `SERVER`, `TIMEOUT`, and `TRANSPORT`, using bounded exponential backoff from 500 ms to 10 seconds with 10 percent jitter. `EMPTY_RESPONSE` is the adapters' classification of a degenerate provider completion (a terminal stop with zero content blocks); the attempt produced nothing durable, so repeating it is safe. Delay bounds must fit Node's supported timer range. A valid `providerRetryAfterMs` replaces local backoff when it is within the configured cap; an over-cap instruction delegates to the next recovery policy instead.
|
||||
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 `EMPTY_RESPONSE`, `RATE_LIMIT`, `SERVER`, `TIMEOUT`, and `TRANSPORT`, with bounded exponential backoff from 500 ms to 10 seconds and 10 percent jitter. `EMPTY_RESPONSE` is the adapters' classification of a degenerate provider completion that produced no durable content, so repeating it is safe. 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 after active delegated recovery reaches quiescence.
|
||||
|
||||
Before waiting, the plugin appends a non-surface `llm/retry` event with the failure and scheduled delay. Its payload is available from the browser-safe `@deepseek-ai/dsh-llm-retry/types` subpath, so remote renderers can consume the durable status without loading the policy runtime. Cancellation and plugin disposal abort the wait; disposal drains the plugin's active backoffs, and a callback captured before disposal fails closed if invoked afterward.
|
||||
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.
|
||||
|
||||
The separately published `./invariant` companion checks that every retry record names the current open turn and its latest closed step, has a unique step record and increasing retry number, and carries a positive bounded retry budget and non-negative bounded timer delay. Full jitter may schedule zero milliseconds at its lower boundary.
|
||||
Before waiting, the plugin appends a non-surface `llm/retry` event with the provider, mode, canonical resolved-policy key, failure, and scheduled delay. Its payload is available from the browser-safe `@deepseek-ai/dsh-llm-retry/types` subpath, so remote renderers can consume the durable status without loading the policy runtime. The key includes every behavior-affecting field and sorts normal-mode codes because eligibility uses set membership. Retry numbers continue only across events with the same provider and complete policy key, so a route replacement with different limits, code membership, or backoff starts its own history. Normal events include the finite maximum; always events omit it, and UIs render `∞`. After the wait, the listener returns `{ kind: 'retry' }`, and the loop closes the failed turn and opens a retry turn over the same durable history. Cancellation and plugin disposal abort active backoff, drain active delegated recovery before applying the abort, and make a callback captured before disposal fail closed.
|
||||
|
||||
The separately published `./invariant` companion checks that every retry record names the current open turn and latest closed step, matches the failed request's durable provider, carries non-empty provider and policy identities, has mode-specific bounds, a unique step record, the correct provider-policy retry number, and a bounded timer delay. Full jitter may schedule zero milliseconds at its lower boundary.
|
||||
|
||||
```yaml
|
||||
- name: '@deepseek-ai/dsh-llm-retry'
|
||||
- name: '@deepseek-ai/dsh-llm-deepseek'
|
||||
config:
|
||||
maxTransientRetries: 2
|
||||
initialDelayMs: 500
|
||||
maxDelayMs: 10000
|
||||
jitterRatio: 0.1
|
||||
retryableCodes: [EMPTY_RESPONSE, RATE_LIMIT, SERVER, TIMEOUT, TRANSPORT]
|
||||
apiKey: !!js process.env.DEEPSEEK_API_KEY
|
||||
retryPolicy:
|
||||
mode: always
|
||||
backoff:
|
||||
initialDelayMs: 1000
|
||||
maxDelayMs: 30000
|
||||
jitterRatio: 0.2
|
||||
|
||||
- name: '@deepseek-ai/dsh-llm-retry'
|
||||
```
|
||||
|
||||
The executor has no policy config. Multi-provider adapters such as `dsh-llm-pi-ai` place `retryPolicy` inside each provider profile, avoiding a second provider-name list.
|
||||
|
||||
## Model Experience
|
||||
|
||||
### Transient request recovery
|
||||
### Model-request recovery
|
||||
|
||||
#### What the model sees
|
||||
|
||||
No retry event, delay, or failure prose is model-visible. After a retry, the next numbered step reconstructs the same explicit provider/model request from durable session history; failed chunks never enter derived messages.
|
||||
No retry event, delay, provider error, or failed partial output is model-visible. The retry turn reconstructs the same explicit provider/model request from durable surface history unless a downstream recovery policy deliberately changes that surface; failed chunks never enter derived messages.
|
||||
|
||||
#### Token effect
|
||||
|
||||
Each retry is a new provider request and may repeat input-token billing. The finite budget caps attempts; `llm/retry` itself contributes no tokens.
|
||||
Each retry is a new provider request and may repeat input-token billing. Normal mode has a finite budget; always mode can consume unbounded requests until success or cancellation. `llm/retry` itself contributes no tokens.
|
||||
|
||||
#### KV Cache effect
|
||||
|
||||
The reconstructed request preserves the prior prefix and is eligible for provider cache reuse under that provider's rules. The non-surface status event does not change cache identity.
|
||||
The reconstructed request preserves the prior prefix and is eligible for provider cache reuse under that provider's rules. The non-surface retry event does not change cache identity.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **Agent steps are the only retry boundary** — direct `ctx.llm.stream()` consumers remain single-attempt because a raw stream cannot separate already-emitted chunks durably.
|
||||
- **Finite plugin budgets add** — this policy counts only configured transient codes; context-overflow compaction counts only its own code. A future policy with overlapping codes must document and test registration-order behavior.
|
||||
- **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.
|
||||
- **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.
|
||||
|
||||
@@ -2,42 +2,52 @@
|
||||
|
||||
[English](README.md) | 中文
|
||||
|
||||
一个函数插件，在 agent loop 的已关闭步骤恢复 seam 上重试特定的短暂模型请求失败。它不包装 `ctx.llm.stream()`：每次适配器调用仍是一次提供方尝试，每次重试都会开启新的编号步骤。
|
||||
一个函数插件，通过 agent loop（智能体循环）在已关闭步骤上触发的 `agent/request-error` waterfall（瀑布式事件）应用确切提供方重试策略。它不包装 `ctx.llm.stream()`：每次适配器调用仍是一次提供方尝试，每次重试都会开启新的编号轮次。
|
||||
|
||||
默认策略允许为 `EMPTY_RESPONSE`、`RATE_LIMIT`、`SERVER`、`TIMEOUT` 和 `TRANSPORT` 重试两次，使用从 500 ms 到 10 秒的有界指数退避与 10% jitter。`EMPTY_RESPONSE` 是适配器对退化提供方完成的分类（携带零个内容块的终止 stop）；该尝试未产生持久内容，因此可安全重复。延迟边界必须适合 Node 支持的定时器范围。有效 `providerRetryAfterMs` 在已配置上限内时替换本地退避；超出上限的指令会委托给下一项恢复策略。
|
||||
每个提供方适配器都拥有可选的嵌套 `retryPolicy`；路由在 `ctx.llm` 上注册时会捕获该策略，任何到达该注册最终适配器边界的调用都会携带它。如果之后释放或替换路由，进行中的失败仍会保留为其提供服务的策略；在选中任何最终适配器前发生的失败没有提供方策略，会继续委托。省略策略时使用 normal mode：为 `EMPTY_RESPONSE`、`RATE_LIMIT`、`SERVER`、`TIMEOUT` 和 `TRANSPORT` 重试两次，并采用从 500 ms 到 10 秒的有界指数退避与 10% jitter。`EMPTY_RESPONSE` 是适配器对未产生任何持久内容的退化提供方完成所作的分类，因此可安全重复。normal 策略可以更改其有限预算、合格 code 和退避配置。always mode 会先请求下游恢复，再无次数上限地重试每个模型请求失败；成功、取消或插件 dispose（资源释放）会在活跃的委托恢复完全停稳后终止它。
|
||||
|
||||
等待之前，插件会追加一个非表层 `llm/retry` 事件，携带失败与计划延迟。该载荷由可安全用于浏览器的 `@deepseek-ai/dsh-llm-retry/types` 子路径导出，因此远程渲染器无需加载策略运行时即可使用该持久状态。取消与插件 dispose 会中止等待；dispose 会排空插件的活跃退避，dispose 前捕获的 callback 如果在之后调用，将快速失败。
|
||||
两种 mode 都使用带对称 jitter 的有界指数退避。有效 `providerRetryAfterMs` 不超过 `maxDelayMs` 时会替换本地退避，并且不加 jitter。超出上限的提供方延迟会使 normal mode 继续委托；always mode 则改用已配置的本地退避，避免该指令终止重试。
|
||||
|
||||
单独发布的 `./invariant` 配套模块会检查每个重试记录是否指向当前开启轮次及其最新已关闭步骤，是否拥有唯一步骤记录与递增重试编号，以及是否携带正数有界重试预算和非负有界定时器延迟。完整 jitter 可以在下界调度为零毫秒。
|
||||
等待前，插件会追加一条不进入表层的 `llm/retry` 事件，其中包含提供方、mode、规范的解析策略 key、失败和计划延迟。该载荷由可安全用于浏览器的 `@deepseek-ai/dsh-llm-retry/types` 子路径导出，因此远程渲染器无需加载策略运行时即可使用该持久状态。该 key 包含所有影响行为的字段，并对 normal mode 的 code 排序，因为合格性采用集合成员关系判断。只有提供方与完整策略 key 都相同的事件才会延续重试编号；因此，用限制、code 成员关系或退避不同的路由替换后，会开始自己的历史。normal 事件包含有限上限；always 事件省略该上限，UI 会渲染 `∞`。等待结束后，监听器返回 `{ kind: 'retry' }`，循环关闭失败轮次，并在同一持久历史上开启重试轮次。取消与插件 dispose 会中止活跃退避，在应用中止前排空活跃的委托恢复，并使 dispose 前捕获的 callback 只能以失败结束。
|
||||
|
||||
单独发布的 `./invariant` 配套模块会检查每个重试记录是否指向当前开启轮次及其最新已关闭步骤，是否与失败请求的持久提供方匹配，是否携带非空的提供方与策略身份，是否满足 mode 特定边界，是否拥有唯一步骤记录和正确的提供方策略重试编号，以及是否携带有界定时器延迟。完整 jitter 可以在下界调度为零毫秒。
|
||||
|
||||
```yaml
|
||||
- name: '@deepseek-ai/dsh-llm-retry'
|
||||
- name: '@deepseek-ai/dsh-llm-deepseek'
|
||||
config:
|
||||
maxTransientRetries: 2
|
||||
initialDelayMs: 500
|
||||
maxDelayMs: 10000
|
||||
jitterRatio: 0.1
|
||||
retryableCodes: [EMPTY_RESPONSE, RATE_LIMIT, SERVER, TIMEOUT, TRANSPORT]
|
||||
apiKey: !!js process.env.DEEPSEEK_API_KEY
|
||||
retryPolicy:
|
||||
mode: always
|
||||
backoff:
|
||||
initialDelayMs: 1000
|
||||
maxDelayMs: 30000
|
||||
jitterRatio: 0.2
|
||||
|
||||
- name: '@deepseek-ai/dsh-llm-retry'
|
||||
```
|
||||
|
||||
执行器没有策略配置。`dsh-llm-pi-ai` 等多提供方适配器会把 `retryPolicy` 放在每个提供方 profile 内，避免维护第二份提供方名称列表。
|
||||
|
||||
## 模型体验
|
||||
|
||||
### 短暂请求恢复
|
||||
### 模型请求恢复
|
||||
|
||||
#### 模型看到的内容
|
||||
|
||||
模型不会看到重试事件、延迟或失败文本。重试后，下一个编号步骤会从持久会话历史中重建相同的显式提供方／模型请求；失败 chunk 绝不会进入派生消息。
|
||||
模型不会看到重试事件、延迟、提供方错误或失败的部分输出。重试轮次会从持久表层历史中重建相同的显式提供方／模型请求，除非下游恢复策略有意更改该表层；失败分片绝不会进入派生消息。
|
||||
|
||||
#### Token 影响
|
||||
|
||||
每次重试都是新的提供方请求，可能重复计费输入 token。有限预算会限制尝试次数；`llm/retry` 自身不产生 token。
|
||||
每次重试都是新的提供方请求，可能重复计费输入 token。normal mode 具有有限预算；always mode 可以在成功或取消前消耗无界数量的请求。`llm/retry` 自身不产生 token。
|
||||
|
||||
#### KV Cache 影响
|
||||
|
||||
重建请求保留之前的前缀，并可根据该提供方的规则复用 cache。非表层状态事件不会改变 cache 身份。
|
||||
重建请求保留之前的前缀，并可根据该提供方的规则复用 cache。非表层重试事件不会改变 cache 身份。
|
||||
|
||||
## 已知限制与暂缓事项
|
||||
|
||||
- **Agent 步骤是唯一重试边界**：直接 `ctx.llm.stream()` 消费方仍只尝试一次，因为原始流无法将已发出 chunk 持久分隔为不同尝试。
|
||||
- **有限插件预算可叠加**：该策略只统计已配置短暂 code；上下文溢出压缩只统计自身 code。未来如有 code 重叠的策略，必须记录并测试注册顺序行为。
|
||||
- **Agent 轮次是唯一重试边界**：直接 `ctx.llm.stream()` 消费方仍只尝试一次，因为原始流无法将已发出 chunk 持久分隔为不同尝试。
|
||||
- **always mode 会重试永久性失败**：身份验证、配额、无效请求、协议和无法恢复的上下文错误都会继续重试，直至成功、取消或 dispose；部署负责提供方特定的成本与延迟控制。
|
||||
- **有限插件预算可叠加**：normal mode 只统计已配置 code 和确切提供方策略，上下文溢出压缩则拥有独立预算。未来如有重叠策略，必须记录并测试注册顺序行为。
|
||||
- **恢复策略按 waterfall 顺序组合**：always mode 会先接受下游重试，再应用自己的回退。后续策略如果忽略取消且永不结算，也会阻止回退、轮次完全停稳和插件 dispose 完成。
|
||||
- **`llm/retry` 记录调度，不是完成**：后续步骤与轮次事件用于确立成功、耗尽或取消。
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@deepseek-ai/dsh-llm-retry",
|
||||
"description": "Bounded transient LLM request retry policy for the DeepSeek Harness",
|
||||
"description": "Provider-routed LLM request retry policy for the DeepSeek Harness",
|
||||
"version": "0.0.1",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
/** Durable request-route lookup for one closed 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.
|
||||
* 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 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.
|
||||
*/
|
||||
export function providerForClosedStep(
|
||||
events: readonly SessionEvent[],
|
||||
turn: number,
|
||||
step: number,
|
||||
): string | undefined {
|
||||
const stepEndIndex = events.findLastIndex(event =>
|
||||
event.type === 'step/end'
|
||||
&& event.data.turn === turn
|
||||
&& event.data.step === step,
|
||||
)
|
||||
if (stepEndIndex < 0) return undefined
|
||||
for (let index = stepEndIndex; index >= 0; index -= 1) {
|
||||
// The loop bounds prove this indexed read exists.
|
||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
|
||||
const event = events[index]!
|
||||
if (event.type === 'request/header') return event.data.header.config.provider
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
+166
-114
@@ -1,5 +1,5 @@
|
||||
/**
|
||||
* Bounded transient model-request retry policy on the agent loop's closed-step
|
||||
* Provider-routed model-request retry policy on the agent loop's closed-step
|
||||
* recovery seam. Each scheduled retry is durable before its cancellable wait.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-llm-retry
|
||||
@@ -7,21 +7,33 @@
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import z from 'schemastery'
|
||||
import type { Agent, RequestError, RequestErrorDecision } from '@deepseek-ai/dsh-agent'
|
||||
import type { LlmFailure } from '@deepseek-ai/dsh-llm'
|
||||
import type {} from '@deepseek-ai/dsh-session'
|
||||
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
|
||||
import type { Agent, RequestError, RequestErrorAction } 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 transient retry scheduled after a closed failed step. */
|
||||
/** Durable, non-surface record of one provider-routed retry scheduled after a closed failed step. */
|
||||
'llm/retry': {
|
||||
turn: number
|
||||
step: number
|
||||
provider: string
|
||||
mode: 'normal'
|
||||
policyKey: string
|
||||
retry: number
|
||||
maxRetries: number
|
||||
delayMs: number
|
||||
failure: LlmFailure
|
||||
} | {
|
||||
turn: number
|
||||
step: number
|
||||
provider: string
|
||||
mode: 'always'
|
||||
policyKey: string
|
||||
retry: number
|
||||
delayMs: number
|
||||
failure: LlmFailure
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -31,82 +43,19 @@ export type { LlmRetryEventData } from './types.ts'
|
||||
export const name = 'llm-retry'
|
||||
export const inject = ['agents']
|
||||
|
||||
const DEFAULT_MAX_TRANSIENT_RETRIES = 2
|
||||
const DEFAULT_INITIAL_DELAY_MS = 500
|
||||
const DEFAULT_MAX_DELAY_MS = 10_000
|
||||
const DEFAULT_JITTER_RATIO = 0.1
|
||||
const DEFAULT_RETRYABLE_CODES = Object.freeze(['EMPTY_RESPONSE', 'RATE_LIMIT', 'SERVER', 'TIMEOUT', 'TRANSPORT'])
|
||||
|
||||
/** Deployment-owned limits and classification for transient request recovery. */
|
||||
export interface Config {
|
||||
/** Maximum transient retries after the first request (default 2). */
|
||||
maxTransientRetries?: number
|
||||
/** Initial local exponential-backoff delay in milliseconds (default 500). */
|
||||
initialDelayMs?: number
|
||||
/** Maximum accepted or locally scheduled delay in milliseconds (default 10000). */
|
||||
maxDelayMs?: number
|
||||
/** Symmetric random multiplier range around one (default 0.1). */
|
||||
jitterRatio?: number
|
||||
/** Stable failure codes eligible for this policy. */
|
||||
retryableCodes?: string[]
|
||||
}
|
||||
/** This policy executor has no config; providers own `retryPolicy`. */
|
||||
export type Config = Readonly<Record<string, never>>
|
||||
|
||||
/** Runtime schema for {@link Config}. */
|
||||
export const Config: z<Config> = z.object({
|
||||
maxTransientRetries: z.number().step(1).min(0).default(DEFAULT_MAX_TRANSIENT_RETRIES),
|
||||
initialDelayMs: z.number().max(MAX_TIMER_DELAY_MS).default(DEFAULT_INITIAL_DELAY_MS),
|
||||
maxDelayMs: z.number().max(MAX_TIMER_DELAY_MS).default(DEFAULT_MAX_DELAY_MS),
|
||||
jitterRatio: z.number().min(0).max(1).default(DEFAULT_JITTER_RATIO),
|
||||
retryableCodes: z.array(z.string()).default([...DEFAULT_RETRYABLE_CODES]),
|
||||
})
|
||||
export const Config = z.object({}) as unknown as z<Config>
|
||||
|
||||
interface ResolvedConfig {
|
||||
readonly maxTransientRetries: number
|
||||
readonly initialDelayMs: number
|
||||
readonly maxDelayMs: number
|
||||
readonly jitterRatio: number
|
||||
readonly retryableCodes: ReadonlySet<string>
|
||||
}
|
||||
|
||||
function resolveConfig(config: Config): ResolvedConfig {
|
||||
const maxTransientRetries = config.maxTransientRetries ?? DEFAULT_MAX_TRANSIENT_RETRIES
|
||||
const initialDelayMs = config.initialDelayMs ?? DEFAULT_INITIAL_DELAY_MS
|
||||
const maxDelayMs = config.maxDelayMs ?? DEFAULT_MAX_DELAY_MS
|
||||
const jitterRatio = config.jitterRatio ?? DEFAULT_JITTER_RATIO
|
||||
const codes = config.retryableCodes ?? [...DEFAULT_RETRYABLE_CODES]
|
||||
|
||||
if (!Number.isInteger(maxTransientRetries) || maxTransientRetries < 0) {
|
||||
throw new Error('llm-retry: maxTransientRetries must be a non-negative integer')
|
||||
function validateConfig(config: Config): void {
|
||||
const [key] = Object.keys(config)
|
||||
if (key === undefined) return
|
||||
if (key === 'retryPolicy') {
|
||||
throw new Error('llm-retry: retryPolicy belongs under each provider configuration')
|
||||
}
|
||||
if (!Number.isFinite(initialDelayMs) || initialDelayMs <= 0 || initialDelayMs > MAX_TIMER_DELAY_MS) {
|
||||
throw new Error(`llm-retry: initialDelayMs must be a positive finite number no greater than ${MAX_TIMER_DELAY_MS}`)
|
||||
}
|
||||
if (!Number.isFinite(maxDelayMs) || maxDelayMs <= 0 || maxDelayMs > MAX_TIMER_DELAY_MS) {
|
||||
throw new Error(`llm-retry: maxDelayMs must be a positive finite number no greater than ${MAX_TIMER_DELAY_MS}`)
|
||||
}
|
||||
if (initialDelayMs > maxDelayMs) {
|
||||
throw new Error('llm-retry: initialDelayMs must be less than or equal to maxDelayMs')
|
||||
}
|
||||
if (!Number.isFinite(jitterRatio) || jitterRatio < 0 || jitterRatio > 1) {
|
||||
throw new Error('llm-retry: jitterRatio must be between 0 and 1')
|
||||
}
|
||||
if (codes.length === 0) {
|
||||
throw new Error('llm-retry: retryableCodes must not be empty')
|
||||
}
|
||||
if (codes.some(code => code.length === 0)) {
|
||||
throw new Error('llm-retry: retryableCodes must contain only non-empty strings')
|
||||
}
|
||||
if (new Set(codes).size !== codes.length) {
|
||||
throw new Error('llm-retry: retryableCodes must not contain duplicates')
|
||||
}
|
||||
|
||||
return Object.freeze({
|
||||
maxTransientRetries,
|
||||
initialDelayMs,
|
||||
maxDelayMs,
|
||||
jitterRatio,
|
||||
retryableCodes: new Set(codes),
|
||||
})
|
||||
throw new Error(`llm-retry: unknown key "${key}"`)
|
||||
}
|
||||
|
||||
/** Non-serializable seams used to make timing policy deterministic in tests. */
|
||||
@@ -115,13 +64,40 @@ export interface RetryInternals {
|
||||
random?: () => number
|
||||
}
|
||||
|
||||
function localDelay(config: ResolvedConfig, retry: number, random: () => number): number {
|
||||
type DownstreamOutcome =
|
||||
| { readonly type: 'decision'; readonly decision: RequestErrorAction }
|
||||
| { readonly type: 'error'; readonly error: unknown }
|
||||
|
||||
async function settleDownstream(
|
||||
next: () => Promise<RequestErrorAction>,
|
||||
): Promise<DownstreamOutcome> {
|
||||
try {
|
||||
return { type: 'decision', decision: await next() }
|
||||
} catch (error: unknown) {
|
||||
return { type: 'error', error }
|
||||
}
|
||||
}
|
||||
|
||||
function localDelay(config: ResolvedRetryPolicy, retry: number, random: () => number): number {
|
||||
const exponent = Math.min(retry - 1, 1024)
|
||||
const exponential = Math.min(config.initialDelayMs * 2 ** exponent, config.maxDelayMs)
|
||||
const jitter = 1 - config.jitterRatio + 2 * config.jitterRatio * random()
|
||||
return Math.min(exponential * jitter, config.maxDelayMs)
|
||||
}
|
||||
|
||||
function retryPolicyKey(policy: ResolvedRetryPolicy): string {
|
||||
return policy.mode === 'always'
|
||||
? JSON.stringify([policy.mode, policy.initialDelayMs, policy.maxDelayMs, policy.jitterRatio])
|
||||
: JSON.stringify([
|
||||
policy.mode,
|
||||
policy.maxRetries,
|
||||
[...policy.retryableCodes].sort(),
|
||||
policy.initialDelayMs,
|
||||
policy.maxDelayMs,
|
||||
policy.jitterRatio,
|
||||
])
|
||||
}
|
||||
|
||||
function cancellableDelay(delayMs: number, signal: AbortSignal): Promise<boolean> {
|
||||
if (signal.aborted) return Promise.resolve(false)
|
||||
return new Promise((resolve) => {
|
||||
@@ -138,78 +114,154 @@ function cancellableDelay(delayMs: number, signal: AbortSignal): Promise<boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Install bounded transient request recovery.
|
||||
* Install provider-routed normal or unbounded request recovery.
|
||||
* @param ctx - plugin context that owns the listener and active waits.
|
||||
* @param config - retry budget, delay bounds, jitter, and eligible codes.
|
||||
* @param config - empty executor config; provider registrations own policy.
|
||||
* @param internals - non-serializable deterministic seams for tests.
|
||||
*/
|
||||
export function apply(ctx: Context, config: Config = {}, internals: RetryInternals = {}): void {
|
||||
const resolved = resolveConfig(config)
|
||||
validateConfig(config)
|
||||
const random = internals.random ?? Math.random
|
||||
const lifetime = new AbortController()
|
||||
const active = new Set<Promise<RequestErrorDecision>>()
|
||||
const active = new Set<Promise<RequestErrorAction>>()
|
||||
|
||||
function track(operation: Promise<RequestErrorAction>): Promise<RequestErrorAction> {
|
||||
const tracked = operation.finally(() => active.delete(tracked))
|
||||
active.add(tracked)
|
||||
return tracked
|
||||
}
|
||||
|
||||
async function backoff(
|
||||
agent: Agent,
|
||||
turn: number,
|
||||
step: number,
|
||||
failure: LlmFailure,
|
||||
provider: string,
|
||||
policy: ResolvedRetryPolicy,
|
||||
policyKey: string,
|
||||
retry: number,
|
||||
delayMs: number,
|
||||
signal: AbortSignal,
|
||||
): Promise<RequestErrorDecision> {
|
||||
): Promise<RequestErrorAction> {
|
||||
const fusedSignal = AbortSignal.any([signal, lifetime.signal])
|
||||
if (fusedSignal.aborted) return { action: 'fail' }
|
||||
agent.session.append('llm/retry', {
|
||||
turn,
|
||||
step,
|
||||
retry,
|
||||
maxRetries: resolved.maxTransientRetries,
|
||||
delayMs,
|
||||
failure,
|
||||
})
|
||||
if (!await cancellableDelay(delayMs, fusedSignal)) return { action: 'fail' }
|
||||
return { action: 'retry' }
|
||||
if (fusedSignal.aborted) return
|
||||
const eventData = policy.mode === 'normal'
|
||||
? {
|
||||
turn,
|
||||
step,
|
||||
provider,
|
||||
mode: policy.mode,
|
||||
policyKey,
|
||||
retry,
|
||||
maxRetries: policy.maxRetries,
|
||||
delayMs,
|
||||
failure,
|
||||
}
|
||||
: {
|
||||
turn,
|
||||
step,
|
||||
provider,
|
||||
mode: policy.mode,
|
||||
policyKey,
|
||||
retry,
|
||||
delayMs,
|
||||
failure,
|
||||
}
|
||||
agent.session.append('llm/retry', eventData)
|
||||
if (!await cancellableDelay(delayMs, fusedSignal)) return
|
||||
return { kind: 'retry' }
|
||||
}
|
||||
|
||||
const disposeListener = ctx.on('agent/request-error', (
|
||||
async function recover(
|
||||
agent: Agent,
|
||||
turn: number,
|
||||
step: number,
|
||||
_error: RequestError,
|
||||
failure: LlmFailure,
|
||||
priorFailures: readonly LlmFailure[],
|
||||
policy: ResolvedRetryPolicy | undefined,
|
||||
signal: AbortSignal,
|
||||
next: () => Promise<RequestErrorDecision>,
|
||||
) => {
|
||||
// A waterfall may have captured this callback before its registration was
|
||||
// removed. Lifetime cancellation must prevent that stale callback from
|
||||
// entering a downstream policy after disposal.
|
||||
if (lifetime.signal.aborted) return Promise.resolve<RequestErrorDecision>({ action: 'fail' })
|
||||
if (!resolved.retryableCodes.has(failure.code)) return next()
|
||||
const priorTransientFailures = priorFailures.filter(item => resolved.retryableCodes.has(item.code)).length
|
||||
if (priorTransientFailures >= resolved.maxTransientRetries) return next()
|
||||
next: () => Promise<RequestErrorAction>,
|
||||
): Promise<RequestErrorAction> {
|
||||
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])
|
||||
// The loop and plugin lifetime stay open until delegated recovery settles.
|
||||
// An abort then wins before the decision or fallback can mutate later state.
|
||||
const downstream = await settleDownstream(next)
|
||||
if (fusedSignal.aborted) return
|
||||
if (downstream.type === 'error') {
|
||||
ctx.logger.warn(
|
||||
`llm-retry: provider "${provider}" always policy ignored a downstream recovery failure: %o`,
|
||||
downstream.error,
|
||||
)
|
||||
}
|
||||
if (downstream.type === 'decision' && downstream.decision?.kind === 'retry') {
|
||||
return downstream.decision
|
||||
}
|
||||
} else if (!policy.retryableCodes.includes(failure.code)) {
|
||||
return next()
|
||||
}
|
||||
|
||||
const retry = priorTransientFailures + 1
|
||||
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.provider === provider
|
||||
&& event.data.policyKey === policyKey,
|
||||
)
|
||||
const previousRetry = priorPolicyRetry?.data.retry ?? 0
|
||||
if (policy.mode === 'normal' && previousRetry >= policy.maxRetries) return next()
|
||||
const retry = previousRetry + 1
|
||||
let delayMs: number
|
||||
if (failure.providerRetryAfterMs !== undefined
|
||||
&& Number.isFinite(failure.providerRetryAfterMs)
|
||||
&& failure.providerRetryAfterMs > 0) {
|
||||
if (failure.providerRetryAfterMs > resolved.maxDelayMs) return next()
|
||||
delayMs = failure.providerRetryAfterMs
|
||||
if (failure.providerRetryAfterMs > policy.maxDelayMs) {
|
||||
if (policy.mode === 'normal') return next()
|
||||
delayMs = localDelay(policy, retry, random)
|
||||
} else {
|
||||
delayMs = failure.providerRetryAfterMs
|
||||
}
|
||||
} else {
|
||||
delayMs = localDelay(resolved, retry, random)
|
||||
delayMs = localDelay(policy, retry, random)
|
||||
}
|
||||
|
||||
const tracked = backoff(agent, turn, step, failure, retry, delayMs, signal)
|
||||
.finally(() => active.delete(tracked))
|
||||
active.add(tracked)
|
||||
return tracked
|
||||
return backoff(agent, turn, step, failure, provider, policy, policyKey, retry, delayMs, signal)
|
||||
}
|
||||
|
||||
const disposeListener = ctx.on('agent/request-error', (
|
||||
agent: Agent,
|
||||
turn: number,
|
||||
step: number,
|
||||
error: RequestError,
|
||||
failure: LlmFailure,
|
||||
priorFailures: readonly LlmFailure[],
|
||||
policy: ResolvedRetryPolicy | undefined,
|
||||
signal: AbortSignal,
|
||||
next: () => Promise<RequestErrorAction>,
|
||||
) => {
|
||||
// A waterfall may have captured this callback before its registration was
|
||||
// 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))
|
||||
})
|
||||
|
||||
ctx.effect(() => async () => {
|
||||
disposeListener()
|
||||
lifetime.abort(new Error('llm-retry plugin disposed'))
|
||||
await Promise.allSettled([...active])
|
||||
}, 'llm-retry: abort and drain backoffs')
|
||||
}, 'llm-retry: abort and drain active recovery')
|
||||
}
|
||||
|
||||
@@ -2,8 +2,10 @@
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
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 type {} from './index.ts'
|
||||
|
||||
const PACKAGE_NAME = '@deepseek-ai/dsh-llm-retry'
|
||||
@@ -13,21 +15,95 @@ export const name = 'llm-retry-invariant'
|
||||
/** Service required before the companion can reserve package ownership. */
|
||||
export const inject = ['invariants']
|
||||
|
||||
/** Validate the complete provider-neutral failure payload at the durable boundary. */
|
||||
function validateFailure(value: unknown, fail: InvariantFailure): asserts value is LlmFailure {
|
||||
if (typeof value !== 'object' || value === null) {
|
||||
fail('llm/retry failure must be an object')
|
||||
}
|
||||
const failure = value as Partial<LlmFailure>
|
||||
if (typeof failure.message !== 'string' || failure.message.length === 0) {
|
||||
fail('llm/retry failure.message must be a non-empty string')
|
||||
}
|
||||
if (typeof failure.code !== 'string' || failure.code.length === 0) {
|
||||
fail('llm/retry failure.code must be a non-empty string')
|
||||
}
|
||||
if (failure.status !== undefined
|
||||
&& (!Number.isInteger(failure.status) || failure.status < 100 || failure.status > 599)) {
|
||||
fail('llm/retry failure.status must be an integer from 100 through 599 when present')
|
||||
}
|
||||
if (failure.providerRetryAfterMs !== undefined
|
||||
&& (!Number.isFinite(failure.providerRetryAfterMs) || failure.providerRetryAfterMs <= 0)) {
|
||||
fail('llm/retry failure.providerRetryAfterMs must be a positive finite number when present')
|
||||
}
|
||||
if (failure.requestId !== undefined
|
||||
&& (typeof failure.requestId !== 'string' || failure.requestId.length === 0)) {
|
||||
fail('llm/retry failure.requestId must be a non-empty string when present')
|
||||
}
|
||||
}
|
||||
|
||||
/** 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. */
|
||||
function validateRetry(
|
||||
history: readonly SessionEvent[],
|
||||
event: SessionEvent<'llm/retry'>,
|
||||
fail: InvariantFailure,
|
||||
): void {
|
||||
const { turn, step, retry, maxRetries, delayMs } = event.data
|
||||
const { turn, step, provider, mode, policyKey, retry, delayMs } = event.data
|
||||
const failure: unknown = event.data.failure
|
||||
validateFailure(failure, fail)
|
||||
if (!Number.isSafeInteger(retry) || retry < 1) {
|
||||
fail('llm/retry retry must be a positive safe integer')
|
||||
}
|
||||
if (!Number.isSafeInteger(maxRetries) || maxRetries < 1 || retry > maxRetries) {
|
||||
fail(`llm/retry retry ${retry} must not exceed a positive safe maxRetries ${maxRetries}`)
|
||||
if (typeof provider !== 'string' || provider.length === 0) {
|
||||
fail('llm/retry provider must be a non-empty string')
|
||||
}
|
||||
if (!(delayMs >= 0 && delayMs <= MAX_TIMER_DELAY_MS)) {
|
||||
fail(`llm/retry delayMs must be within 0..${MAX_TIMER_DELAY_MS}`)
|
||||
if (typeof policyKey !== 'string' || policyKey.length === 0) {
|
||||
fail('llm/retry policyKey must be a non-empty string')
|
||||
}
|
||||
switch (mode) {
|
||||
case 'normal': {
|
||||
const { maxRetries } = event.data
|
||||
if (!Number.isSafeInteger(maxRetries) || maxRetries < 1 || retry > maxRetries) {
|
||||
fail(`llm/retry retry ${retry} must not exceed a positive safe maxRetries ${maxRetries}`)
|
||||
}
|
||||
break
|
||||
}
|
||||
case 'always':
|
||||
if ('maxRetries' in event.data) fail('llm/retry always mode must omit maxRetries')
|
||||
break
|
||||
default:
|
||||
fail(`llm/retry mode must be normal or always, got ${String(mode)}`)
|
||||
}
|
||||
if (typeof delayMs !== 'number' || !Number.isFinite(delayMs)
|
||||
|| delayMs < 0 || delayMs > MAX_TIMER_DELAY_MS) {
|
||||
fail(`llm/retry delayMs must be a finite number within 0..${MAX_TIMER_DELAY_MS}`)
|
||||
}
|
||||
|
||||
const currentTurnEvents: SessionEvent[] = []
|
||||
@@ -58,15 +134,24 @@ function validateRetry(
|
||||
if (closedStep === undefined || step !== closedStep) {
|
||||
fail(`llm/retry names step ${step}, but the latest closed step is ${String(closedStep)}`)
|
||||
}
|
||||
const routedProvider = providerForClosedStep(history, turn, step)
|
||||
if (routedProvider !== provider) {
|
||||
fail(`llm/retry provider ${provider} does not match the failed request provider ${String(routedProvider)}`)
|
||||
}
|
||||
|
||||
const priorRetries = currentTurnEvents
|
||||
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 (priorRetries.some(prior => prior.data.step === step)) {
|
||||
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 priorRetry = priorRetries[0]
|
||||
if (priorRetry !== undefined && retry <= priorRetry.data.retry) {
|
||||
fail(`llm/retry retry ${retry} must increase after retry ${priorRetry.data.retry}`)
|
||||
const priorPolicyRetry = chainRetries.findLast(prior =>
|
||||
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}`)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,11 +1,25 @@
|
||||
import type { LlmFailure } from '@deepseek-ai/dsh-llm/types'
|
||||
|
||||
/** Durable payload recorded before one transient model-request retry wait. */
|
||||
export interface LlmRetryEventData {
|
||||
turn: number
|
||||
step: number
|
||||
retry: number
|
||||
maxRetries: number
|
||||
delayMs: number
|
||||
failure: LlmFailure
|
||||
}
|
||||
/** Durable payload recorded before one provider-routed model-request retry wait. */
|
||||
export type LlmRetryEventData =
|
||||
| {
|
||||
turn: number
|
||||
step: number
|
||||
provider: string
|
||||
mode: 'normal'
|
||||
policyKey: string
|
||||
retry: number
|
||||
maxRetries: number
|
||||
delayMs: number
|
||||
failure: LlmFailure
|
||||
}
|
||||
| {
|
||||
turn: number
|
||||
step: number
|
||||
provider: string
|
||||
mode: 'always'
|
||||
policyKey: string
|
||||
retry: number
|
||||
delayMs: number
|
||||
failure: LlmFailure
|
||||
}
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import SessionStore, { SessionId, type Session } from '@deepseek-ai/dsh-session'
|
||||
import { 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'
|
||||
|
||||
async function setup(): Promise<Context> {
|
||||
const ctx = new Context()
|
||||
@@ -17,121 +19,259 @@ function closeStep(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('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('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 })
|
||||
}
|
||||
|
||||
const failure = { message: 'provider busy', code: 'RATE_LIMIT', status: 429 }
|
||||
const normal = {
|
||||
provider: 'mock',
|
||||
mode: 'normal' as const,
|
||||
policyKey: 'normal-policy',
|
||||
retry: 1,
|
||||
maxRetries: 2,
|
||||
delayMs: 1,
|
||||
failure,
|
||||
}
|
||||
const always = {
|
||||
provider: 'mock',
|
||||
mode: 'always' as const,
|
||||
policyKey: 'always-policy',
|
||||
retry: 1,
|
||||
delayMs: 1,
|
||||
failure,
|
||||
}
|
||||
|
||||
describe('llm-retry invariants', () => {
|
||||
it('accepts increasing retry records for successive closed steps and ignores unrelated events', async () => {
|
||||
it('has no provider without the requested closed step or a route marker', () => {
|
||||
expect(providerForClosedStep([], 1, 1)).toBeUndefined()
|
||||
expect(providerForClosedStep([{
|
||||
type: 'step/end',
|
||||
data: { turn: 1, step: 1 },
|
||||
}] as never, 1, 1)).toBeUndefined()
|
||||
})
|
||||
|
||||
it('accepts bounded and unbounded records after successive closed steps', async () => {
|
||||
const ctx = await setup()
|
||||
const session = closeStep(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: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 500, failure,
|
||||
})
|
||||
session.append('step/start', { turn: 1, step: 2 })
|
||||
session.append('step/end', { turn: 1, step: 2 })
|
||||
session.append('llm/retry', {
|
||||
turn: 1, step: 2, retry: 2, maxRetries: 2, delayMs: 1_000, failure,
|
||||
})
|
||||
const zeroDelay = closeStep(ctx, 'retry-invariant-zero-delay')
|
||||
zeroDelay.append('llm/retry', {
|
||||
turn: 1, step: 1, retry: 1, maxRetries: 1, delayMs: 0, failure,
|
||||
turn: 2, step: 1, ...normal, retry: 2, delayMs: 0,
|
||||
})
|
||||
const unbounded = closeStep(ctx, 'retry-invariant-always')
|
||||
unbounded.append('llm/retry', { turn: 1, step: 1, ...always })
|
||||
}).not.toThrow()
|
||||
expect(() => { ctx.emit('tools/change') }).not.toThrow()
|
||||
})
|
||||
|
||||
it.each([
|
||||
[{ retry: 0, maxRetries: 2, delayMs: 1 }, /positive safe integer/],
|
||||
[{ retry: 1.5, maxRetries: 2, delayMs: 1 }, /positive safe integer/],
|
||||
[{ retry: 1, maxRetries: 0, delayMs: 1 }, /positive safe maxRetries/],
|
||||
[{ retry: 1, maxRetries: 1.5, delayMs: 1 }, /positive safe maxRetries/],
|
||||
[{ retry: 3, maxRetries: 2, delayMs: 1 }, /must not exceed/],
|
||||
[{ retry: 1, maxRetries: 2, delayMs: -1 }, /delayMs/],
|
||||
[{ retry: 1, maxRetries: 2, delayMs: MAX_TIMER_DELAY_MS + 1 }, /delayMs/],
|
||||
])('rejects invalid retry bounds %#', async (data, message) => {
|
||||
it('validates the complete durable failure payload', async () => {
|
||||
const ctx = await setup()
|
||||
const session = closeStep(ctx, `retry-invariant-bounds-${data.retry}-${data.maxRetries}-${data.delayMs}`)
|
||||
const complete = closeStep(ctx, 'retry-invariant-complete-failure')
|
||||
expect(() => {
|
||||
session.append('llm/retry', { turn: 1, step: 1, ...data, failure })
|
||||
complete.append('llm/retry', {
|
||||
turn: 1,
|
||||
step: 1,
|
||||
...always,
|
||||
failure: {
|
||||
message: 'provider busy',
|
||||
code: 'RATE_LIMIT',
|
||||
status: 429,
|
||||
providerRetryAfterMs: 25,
|
||||
requestId: ProviderRequestId('request-1'),
|
||||
},
|
||||
})
|
||||
}).not.toThrow()
|
||||
|
||||
const invalidFailures: readonly [string, unknown, RegExp][] = [
|
||||
['null', null, /failure must be an object/],
|
||||
['message-type', { message: 1, code: 'RATE_LIMIT' }, /failure\.message/],
|
||||
['message-empty', { message: '', code: 'RATE_LIMIT' }, /failure\.message/],
|
||||
['code-type', { message: 'failed', code: 1 }, /failure\.code/],
|
||||
['code-empty', { message: 'failed', code: '' }, /failure\.code/],
|
||||
['status-type', { message: 'failed', code: 'RATE_LIMIT', status: 429.5 }, /failure\.status/],
|
||||
['status-low', { message: 'failed', code: 'RATE_LIMIT', status: 99 }, /failure\.status/],
|
||||
['status-high', { message: 'failed', code: 'RATE_LIMIT', status: 600 }, /failure\.status/],
|
||||
[
|
||||
'retry-after-type',
|
||||
{ message: 'failed', code: 'RATE_LIMIT', providerRetryAfterMs: '25' },
|
||||
/failure\.providerRetryAfterMs/,
|
||||
],
|
||||
[
|
||||
'retry-after-zero',
|
||||
{ message: 'failed', code: 'RATE_LIMIT', providerRetryAfterMs: 0 },
|
||||
/failure\.providerRetryAfterMs/,
|
||||
],
|
||||
['request-id-type', { message: 'failed', code: 'RATE_LIMIT', requestId: 1 }, /failure\.requestId/],
|
||||
['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}`)
|
||||
expect(() => {
|
||||
session.append('llm/retry', {
|
||||
turn: 1, step: 1, ...always, failure: invalidFailure,
|
||||
} as never)
|
||||
}).toThrow(message)
|
||||
}
|
||||
})
|
||||
|
||||
it.each([
|
||||
['retry-zero', { ...normal, retry: 0 }, /positive safe integer/],
|
||||
['retry-fraction', { ...normal, retry: 1.5 }, /positive safe integer/],
|
||||
['max-zero', { ...normal, maxRetries: 0 }, /positive safe maxRetries/],
|
||||
['max-fraction', { ...normal, maxRetries: 1.5 }, /positive safe maxRetries/],
|
||||
['over-budget', { ...normal, retry: 3 }, /must not exceed/],
|
||||
['always-maximum', { ...always, maxRetries: 2 }, /always mode must omit maxRetries/],
|
||||
['unknown-mode', { ...always, mode: 'sometimes' }, /mode must be normal or always/],
|
||||
['empty-provider', { ...always, provider: '' }, /provider must be a non-empty string/],
|
||||
['empty-policy-key', { ...always, policyKey: '' }, /policyKey must be a non-empty string/],
|
||||
['delay-negative', { ...normal, delayMs: -1 }, /delayMs/],
|
||||
['delay-overflow', { ...normal, delayMs: MAX_TIMER_DELAY_MS + 1 }, /delayMs/],
|
||||
['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}`)
|
||||
expect(() => {
|
||||
session.append('llm/retry', { turn: 1, step: 1, ...data } as never)
|
||||
}).toThrow(message)
|
||||
})
|
||||
|
||||
it('rejects retry records outside the matching closed-step boundary', async () => {
|
||||
it('rejects records outside the latest closed step of an open turn', async () => {
|
||||
const ctx = await setup()
|
||||
const absent = ctx.sessions.create(SessionId('retry-invariant-no-turn'))
|
||||
expect(() => {
|
||||
absent.append('llm/retry', {
|
||||
turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
|
||||
})
|
||||
absent.append('llm/retry', { turn: 1, step: 1, ...normal })
|
||||
}).toThrow(/inside an open turn/)
|
||||
|
||||
const wrongTurn = closeStep(ctx, 'retry-invariant-wrong-turn')
|
||||
expect(() => {
|
||||
wrongTurn.append('llm/retry', {
|
||||
turn: 2, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
|
||||
})
|
||||
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 })
|
||||
expect(() => {
|
||||
openStep.append('llm/retry', {
|
||||
turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
|
||||
})
|
||||
openStep.append('llm/retry', { turn: 1, step: 1, ...normal })
|
||||
}).toThrow(/step 1 is still open/)
|
||||
|
||||
const noStep = ctx.sessions.create(SessionId('retry-invariant-no-step'))
|
||||
noStep.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
expect(() => {
|
||||
noStep.append('llm/retry', {
|
||||
turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
|
||||
})
|
||||
noStep.append('llm/retry', { turn: 1, step: 1, ...normal })
|
||||
}).toThrow(/latest closed step is undefined/)
|
||||
|
||||
const wrongStep = closeStep(ctx, 'retry-invariant-wrong-step')
|
||||
expect(() => {
|
||||
wrongStep.append('llm/retry', {
|
||||
turn: 1, step: 2, retry: 1, maxRetries: 2, delayMs: 1, failure,
|
||||
})
|
||||
wrongStep.append('llm/retry', { turn: 1, step: 2, ...normal })
|
||||
}).toThrow(/latest closed step is 1/)
|
||||
|
||||
const closedTurn = closeStep(ctx, 'retry-invariant-closed-turn')
|
||||
closedTurn.append('turn/end', { turn: 1, reason: { kind: 'aborted' } })
|
||||
expect(() => {
|
||||
closedTurn.append('llm/retry', {
|
||||
turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
|
||||
})
|
||||
closedTurn.append('llm/retry', { turn: 1, step: 1, ...normal })
|
||||
}).toThrow(/inside an open turn/)
|
||||
})
|
||||
|
||||
it('rejects duplicate and non-increasing retry records', async () => {
|
||||
it('rejects a second retry record for the same step', async () => {
|
||||
const ctx = await setup()
|
||||
const duplicate = closeStep(ctx, 'retry-invariant-duplicate')
|
||||
duplicate.append('llm/retry', {
|
||||
turn: 1, step: 1, retry: 1, maxRetries: 3, delayMs: 1, failure,
|
||||
})
|
||||
expect(() => {
|
||||
duplicate.append('llm/retry', {
|
||||
turn: 1, step: 1, retry: 2, maxRetries: 3, delayMs: 1, failure,
|
||||
})
|
||||
}).toThrow(/duplicates the retry record/)
|
||||
const session = closeStep(ctx, 'retry-invariant-duplicate')
|
||||
session.append('llm/retry', { turn: 1, step: 1, ...normal })
|
||||
|
||||
const nonIncreasing = closeStep(ctx, 'retry-invariant-non-increasing')
|
||||
nonIncreasing.append('llm/retry', {
|
||||
turn: 1, step: 1, retry: 1, maxRetries: 3, delayMs: 1, failure,
|
||||
})
|
||||
nonIncreasing.append('step/start', { turn: 1, step: 2 })
|
||||
nonIncreasing.append('step/end', { turn: 1, step: 2 })
|
||||
expect(() => {
|
||||
nonIncreasing.append('llm/retry', {
|
||||
turn: 1, step: 2, retry: 1, maxRetries: 3, delayMs: 1, failure,
|
||||
})
|
||||
}).toThrow(/must increase/)
|
||||
session.append('llm/retry', { turn: 1, step: 1, ...normal, retry: 2 })
|
||||
}).toThrow(/duplicates the retry record/)
|
||||
})
|
||||
|
||||
it('binds retry numbering to the provider policy and resets it after success', async () => {
|
||||
const ctx = await setup()
|
||||
const mismatch = closeStep(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 })
|
||||
}).toThrow(/must equal provider policy retry 2/)
|
||||
|
||||
const reset = closeStep(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,
|
||||
content: [{ type: 'text', text: 'success' }],
|
||||
provenance: { 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 })
|
||||
expect(() => {
|
||||
reset.append('llm/retry', { turn: 3, step: 1, ...normal })
|
||||
}).not.toThrow()
|
||||
})
|
||||
|
||||
it('starts a fresh retry chain after incomplete predecessor boundaries', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
|
||||
const missingEnd = ctx.sessions.create(SessionId('retry-invariant-missing-end'))
|
||||
missingEnd.append('user/message', {
|
||||
content: [{ type: 'text', text: 'idle context' }],
|
||||
source: { kind: 'user' },
|
||||
}, { surfaceOp: 'append' })
|
||||
appendRetryTurn(missingEnd, 2)
|
||||
|
||||
const nonFailureEnd = ctx.sessions.create(SessionId('retry-invariant-non-failure-end'))
|
||||
nonFailureEnd.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
nonFailureEnd.append('user/message', {
|
||||
content: [{ type: 'text', text: 'idle context' }],
|
||||
source: { kind: 'user' },
|
||||
}, { surfaceOp: 'append' })
|
||||
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 },
|
||||
})
|
||||
appendRetryTurn(missingStart, 2)
|
||||
|
||||
await ctx.plugin(InvariantService)
|
||||
await expect(ctx.plugin(RetryInvariant)).resolves.toBeDefined()
|
||||
})
|
||||
|
||||
it('rejects a provider that does not match the failed request route', async () => {
|
||||
const ctx = await setup()
|
||||
const session = closeStep(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/)
|
||||
})
|
||||
|
||||
it('validates existing histories on late registration', async () => {
|
||||
@@ -139,9 +279,7 @@ describe('llm-retry invariants', () => {
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('retry-invariant-late'))
|
||||
session.append('step/end', { turn: 1, step: 1 })
|
||||
session.append('llm/retry', {
|
||||
turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
|
||||
})
|
||||
session.append('llm/retry', { turn: 1, step: 1, ...normal })
|
||||
await ctx.plugin(InvariantService)
|
||||
await expect(ctx.plugin(RetryInvariant)).rejects.toThrow(/inside an open turn/)
|
||||
})
|
||||
|
||||
@@ -7,10 +7,9 @@ import { Context } from 'cordis'
|
||||
import Loader from '@cordisjs/plugin-loader'
|
||||
import Include from '@cordisjs/plugin-include'
|
||||
import AgentRegistry from '@deepseek-ai/dsh-agent'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
||||
import LlmService, { LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
|
||||
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import LlmService, { LlmAdapter, LlmError, resolveRetryPolicy } from '@deepseek-ai/dsh-llm'
|
||||
import type { GenerateOptions, ResolvedRetryPolicy, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry from '@deepseek-ai/dsh-tools'
|
||||
@@ -21,6 +20,16 @@ let context: Context | undefined
|
||||
|
||||
class TransientOnceAdapter extends LlmAdapter {
|
||||
requests = 0
|
||||
private readonly retryPolicy = resolveRetryPolicy({
|
||||
mode: 'normal',
|
||||
maxRetries: 1,
|
||||
retryableCodes: ['RATE_LIMIT', 'SERVER'],
|
||||
backoff: { initialDelayMs: 1, maxDelayMs: 1, jitterRatio: 0 },
|
||||
}, 'loader test provider retryPolicy')
|
||||
|
||||
override providerRetryPolicy(_provider: string): ResolvedRetryPolicy {
|
||||
return this.retryPolicy
|
||||
}
|
||||
|
||||
async * stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
this.requests += 1
|
||||
@@ -32,17 +41,6 @@ class TransientOnceAdapter extends LlmAdapter {
|
||||
}
|
||||
}
|
||||
|
||||
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()
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
await context?.fiber.dispose()
|
||||
context = undefined
|
||||
@@ -87,7 +85,7 @@ describe('real Loader composition', () => {
|
||||
// Real-Loader composition resolves workspace packages through tsx at test
|
||||
// time; first resolution after the host/client program split is slow enough
|
||||
// to trip the default 5s budget on cold caches.
|
||||
it('loads the flat policy and records recovery through the shipping loop', { timeout: 60_000 }, async () => {
|
||||
it('loads provider-supplied policy and records recovery through the shipping loop', { timeout: 60_000 }, async () => {
|
||||
const loaded = await loadYaml([
|
||||
"- name: '@deepseek-ai/dsh-llm'",
|
||||
"- name: '@deepseek-ai/dsh-session'",
|
||||
@@ -95,12 +93,6 @@ describe('real Loader composition', () => {
|
||||
"- name: '@deepseek-ai/dsh-tools'",
|
||||
"- name: '@deepseek-ai/dsh-agent'",
|
||||
"- name: '@deepseek-ai/dsh-llm-retry'",
|
||||
' config:',
|
||||
' maxTransientRetries: 1',
|
||||
' initialDelayMs: 1',
|
||||
' maxDelayMs: 1',
|
||||
' jitterRatio: 0',
|
||||
' retryableCodes: [RATE_LIMIT, SERVER]',
|
||||
"- name: '@deepseek-ai/dsh-agent-loop'",
|
||||
])
|
||||
|
||||
@@ -113,9 +105,8 @@ describe('real Loader composition', () => {
|
||||
const adapter = new TransientOnceAdapter()
|
||||
loaded.llm.registerAdapter(['mock'], adapter)
|
||||
const agent = loaded.agentLoop.create(SessionId('loader-retry'), { provider: 'mock', model: 'mock' })
|
||||
const idle = waitForIdle(loaded, agent)
|
||||
agent.followup([{ type: 'text', text: 'recover' }])
|
||||
await idle
|
||||
agent.followup({ content: [{ type: 'text', text: 'recover' }], source: { kind: 'user' } })
|
||||
await agent.whenIdle()
|
||||
|
||||
expect(adapter.requests).toBe(2)
|
||||
expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(1)
|
||||
|
||||
@@ -34,16 +34,29 @@ describe.each(['jsonl', 'sqlite'] as const)('%s retry-event persistence', (kind)
|
||||
const session = ctx.sessions.create(SessionId(`retry-${kind}`))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
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,
|
||||
provider: 'mock',
|
||||
mode: 'always',
|
||||
policyKey: '["always",500,10000,0.1]',
|
||||
retry: 1,
|
||||
maxRetries: 2,
|
||||
delayMs: 750,
|
||||
failure: { message: 'provider busy', code: 'RATE_LIMIT', status: 429 },
|
||||
})
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'aborted' } })
|
||||
session.append('turn/end', {
|
||||
turn: 1,
|
||||
reason: {
|
||||
kind: 'error',
|
||||
step: 1,
|
||||
failure: { message: 'provider busy', code: 'RATE_LIMIT', status: 429 },
|
||||
},
|
||||
})
|
||||
|
||||
expect(session.deriveMessages()).toEqual([])
|
||||
await ctx.sessions.flush(session)
|
||||
|
||||
@@ -1,17 +1,24 @@
|
||||
import { afterEach, describe, expect, expectTypeOf, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import type { Fiber } from 'cordis'
|
||||
import LlmService, { CallId, EMPTY_RESPONSE_CODE, LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
|
||||
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import LlmService, { CallId, EMPTY_RESPONSE_CODE, LlmAdapter, LlmError, resolveRetryPolicy } from '@deepseek-ai/dsh-llm'
|
||||
import type {
|
||||
AlwaysRetryPolicyConfig,
|
||||
BackoffConfig,
|
||||
GenerateOptions,
|
||||
NormalRetryPolicyConfig,
|
||||
ResolvedRetryPolicy,
|
||||
RetryPolicyConfig,
|
||||
StreamChunk,
|
||||
} from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionEvent, SessionEventMap } from '@deepseek-ai/dsh-session'
|
||||
import type { LlmRetryEventData } from '@deepseek-ai/dsh-llm-retry/types'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry, { defineContentToolFixture } from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry from '@deepseek-ai/dsh-agent'
|
||||
import type { Agent, RequestErrorDecision } from '@deepseek-ai/dsh-agent'
|
||||
import type { Agent, RequestErrorAction } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
|
||||
import * as retry from '../src/index.ts'
|
||||
|
||||
type ScriptEntry = Error | Iterable<StreamChunk> | AsyncIterable<StreamChunk>
|
||||
@@ -22,6 +29,7 @@ it('keeps the browser-safe retry payload identical to the session event', () =>
|
||||
|
||||
class ScriptedAdapter extends LlmAdapter {
|
||||
readonly requests: GenerateOptions[] = []
|
||||
private retryPolicies: Readonly<Record<string, ResolvedRetryPolicy | undefined>> = {}
|
||||
|
||||
constructor(private readonly entries: ScriptEntry[]) {
|
||||
super()
|
||||
@@ -34,6 +42,21 @@ class ScriptedAdapter extends LlmAdapter {
|
||||
if (entry instanceof Error) throw entry
|
||||
yield* entry
|
||||
}
|
||||
|
||||
configureRetryPolicies(
|
||||
policies: Readonly<Record<string, RetryPolicyConfig | undefined>>,
|
||||
): void {
|
||||
this.retryPolicies = Object.fromEntries(Object.entries(policies).map(([provider, policy]) => [
|
||||
provider,
|
||||
policy === undefined
|
||||
? undefined
|
||||
: resolveRetryPolicy(policy, `retry test provider "${provider}" retryPolicy`),
|
||||
]))
|
||||
}
|
||||
|
||||
override providerRetryPolicy(provider: string): ResolvedRetryPolicy | undefined {
|
||||
return this.retryPolicies[provider]
|
||||
}
|
||||
}
|
||||
|
||||
async function* partialToolFailure(error: Error): AsyncGenerator<StreamChunk> {
|
||||
@@ -76,11 +99,11 @@ function emptyCompletion(): StreamChunk[] {
|
||||
}
|
||||
|
||||
async function harness(
|
||||
adapter: LlmAdapter,
|
||||
config: retry.Config = {},
|
||||
adapter: ScriptedAdapter,
|
||||
policies: Readonly<Record<string, RetryPolicyConfig | undefined>> = { 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)
|
||||
@@ -88,18 +111,42 @@ async function harness(
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
beforeRetry?.(ctx)
|
||||
const resolvedConfig = Object.assign({
|
||||
maxTransientRetries: 2,
|
||||
initialDelayMs: 500,
|
||||
maxDelayMs: 10_000,
|
||||
jitterRatio: 0,
|
||||
}, config)
|
||||
adapter.configureRetryPolicies(policies)
|
||||
const retryFiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
retry.apply(inner, resolvedConfig, internals)
|
||||
retry.apply(inner, {}, internals)
|
||||
}, { inject: retry.inject }))
|
||||
await ctx.plugin(AgentLoop, { agents: [] })
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
return { ctx, retryFiber }
|
||||
const disposeAdapter = ctx.llm.registerAdapter(['mock', 'other'], adapter)
|
||||
return { ctx, retryFiber, disposeAdapter }
|
||||
}
|
||||
|
||||
function normalConfig(
|
||||
overrides: Partial<Omit<NormalRetryPolicyConfig, 'mode'>> = {},
|
||||
): NormalRetryPolicyConfig {
|
||||
const { backoff, ...policy } = overrides
|
||||
return {
|
||||
mode: 'normal',
|
||||
maxRetries: 2,
|
||||
...policy,
|
||||
backoff: {
|
||||
initialDelayMs: 500,
|
||||
maxDelayMs: 10_000,
|
||||
jitterRatio: 0,
|
||||
...backoff,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function alwaysConfig(backoff: BackoffConfig = {}): AlwaysRetryPolicyConfig {
|
||||
return {
|
||||
mode: 'always',
|
||||
backoff: {
|
||||
initialDelayMs: 500,
|
||||
maxDelayMs: 10_000,
|
||||
jitterRatio: 0,
|
||||
...backoff,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
|
||||
@@ -132,33 +179,31 @@ afterEach(async () => {
|
||||
context = undefined
|
||||
})
|
||||
|
||||
describe('bounded transient retry policy', () => {
|
||||
describe('provider-routed retry policy', () => {
|
||||
it('records the scheduled delay before opening a fresh request attempt', async () => {
|
||||
vi.useFakeTimers()
|
||||
const adapter = new ScriptedAdapter([
|
||||
new LlmError('busy', 'RATE_LIMIT', { status: 429 }),
|
||||
textResponse('done'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter))
|
||||
;({ ctx: context } = await harness(adapter, {
|
||||
mock: normalConfig({ retryableCodes: ['SERVER', 'RATE_LIMIT'] }),
|
||||
}, undefined, { random: () => 0.5 }))
|
||||
const agent = context.agentLoop.create(SessionId('retry-success'), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
const scheduled = new Promise<Extract<(typeof agent.session.events)[number], { type: 'llm/retry' }>>((resolve) => {
|
||||
const dispose = context?.on('session/event', (session, event) => {
|
||||
if (session === agent.session && event.type === 'llm/retry') {
|
||||
dispose?.()
|
||||
resolve(event)
|
||||
}
|
||||
})
|
||||
})
|
||||
const scheduled = waitForRetry(context, agent, 1)
|
||||
|
||||
agent.followup([{ type: 'text', text: 'go' }])
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
const event = await scheduled
|
||||
|
||||
expect(event.data).toEqual({
|
||||
turn: 1,
|
||||
step: 1,
|
||||
provider: 'mock',
|
||||
mode: 'normal',
|
||||
policyKey: '["normal",2,["RATE_LIMIT","SERVER"],500,10000,0]',
|
||||
retry: 1,
|
||||
maxRetries: 2,
|
||||
delayMs: 500,
|
||||
@@ -173,8 +218,8 @@ describe('bounded transient retry policy', () => {
|
||||
await idle
|
||||
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
expect(agent.session.events.filter(item => item.type === 'step/start').map(item => item.data.step))
|
||||
.toEqual([1, 2])
|
||||
expect(agent.session.events.filter(item => item.type === 'step/start').map(item => item.data))
|
||||
.toEqual([{ turn: 1, step: 1 }, { turn: 2, step: 1 }])
|
||||
expect(agent.session.deriveMessages().at(-1)).toEqual({
|
||||
role: 'assistant',
|
||||
content: [{ type: 'text', text: 'done' }],
|
||||
@@ -195,7 +240,7 @@ describe('bounded transient retry policy', () => {
|
||||
const agent = context.agentLoop.create(SessionId('retry-empty-response'), { provider: 'mock', model: 'mock' })
|
||||
const scheduled = waitForRetry(context, agent, 1)
|
||||
|
||||
agent.followup([{ type: 'text', text: 'go' }])
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
const event = await scheduled
|
||||
expect(event.data.failure).toEqual({
|
||||
message: 'model returned a completed response with no content',
|
||||
@@ -207,8 +252,10 @@ describe('bounded transient retry policy', () => {
|
||||
await idle
|
||||
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
expect(agent.session.events.filter(event => event.type === 'assistant/message').map(event => event.data.step))
|
||||
.toEqual([2])
|
||||
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 }])
|
||||
expect(agent.session.deriveMessages().at(-1)).toMatchObject({
|
||||
role: 'assistant',
|
||||
content: [{ type: 'text', text: 'recovered' }],
|
||||
@@ -235,18 +282,20 @@ describe('bounded transient retry policy', () => {
|
||||
const agent = context.agentLoop.create(SessionId('retry-partial'), { provider: 'mock', model: 'mock' })
|
||||
const scheduled = waitForRetry(context, agent, 1)
|
||||
|
||||
agent.followup([{ type: 'text', text: 'go' }])
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
await scheduled
|
||||
const idle = waitForIdle(context, agent)
|
||||
await vi.advanceTimersByTimeAsync(500)
|
||||
await idle
|
||||
|
||||
const failedChunks = agent.session.events.filter(event =>
|
||||
event.type === 'assistant/chunk' && event.data.step === 1,
|
||||
event.type === 'assistant/chunk' && event.data.turn === 1 && event.data.step === 1,
|
||||
)
|
||||
expect(failedChunks).toHaveLength(6)
|
||||
expect(agent.session.events.filter(event => event.type === 'assistant/message').map(event => event.data.step))
|
||||
.toEqual([2])
|
||||
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 }])
|
||||
expect(agent.session.events.some(event => event.type === 'tool/call')).toBe(false)
|
||||
expect(toolExecutions).toBe(0)
|
||||
expect(agent.session.deriveMessages().at(-1)).toMatchObject({
|
||||
@@ -264,13 +313,15 @@ describe('bounded transient retry policy', () => {
|
||||
new LlmError('busy two', 'SERVER'),
|
||||
new LlmError('busy three', 'SERVER'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter, { jitterRatio: 0.1 }, undefined, {
|
||||
;({ ctx: context } = await harness(adapter, { mock: normalConfig({
|
||||
backoff: { jitterRatio: 0.1 },
|
||||
}) }, undefined, {
|
||||
random: () => samples.shift() ?? 0.5,
|
||||
}))
|
||||
const agent = context.agentLoop.create(SessionId('retry-exhausted'), { provider: 'mock', model: 'mock' })
|
||||
const first = waitForRetry(context, agent, 1)
|
||||
|
||||
agent.followup([{ type: 'text', text: 'go' }])
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
expect((await first).data.delayMs).toBe(450)
|
||||
|
||||
const second = waitForRetry(context, agent, 2)
|
||||
@@ -295,15 +346,13 @@ describe('bounded transient retry policy', () => {
|
||||
new LlmError('busy', 'SERVER'),
|
||||
textResponse('done'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter, {
|
||||
initialDelayMs: 1,
|
||||
maxDelayMs: 1,
|
||||
jitterRatio: 1,
|
||||
}, undefined, { random: () => 0 }))
|
||||
;({ ctx: context } = await harness(adapter, { mock: normalConfig({
|
||||
backoff: { initialDelayMs: 1, maxDelayMs: 1, jitterRatio: 1 },
|
||||
}) }, undefined, { random: () => 0 }))
|
||||
const agent = context.agentLoop.create(SessionId('retry-zero-delay'), { provider: 'mock', model: 'mock' })
|
||||
const scheduled = waitForRetry(context, agent, 1)
|
||||
|
||||
agent.followup([{ type: 'text', text: 'go' }])
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
expect((await scheduled).data.delayMs).toBe(0)
|
||||
|
||||
const idle = waitForIdle(context, agent)
|
||||
@@ -318,10 +367,12 @@ describe('bounded transient retry policy', () => {
|
||||
new LlmError('wait', 'RATE_LIMIT', { providerRetryAfterMs: 2_000 }),
|
||||
textResponse('done'),
|
||||
])
|
||||
;({ ctx: context } = await harness(accepted, { jitterRatio: 1 }))
|
||||
;({ ctx: context } = await harness(accepted, { mock: normalConfig({
|
||||
backoff: { jitterRatio: 1 },
|
||||
}) }))
|
||||
const acceptedAgent = context.agentLoop.create(SessionId('retry-after-accepted'), { provider: 'mock', model: 'mock' })
|
||||
const scheduled = waitForRetry(context, acceptedAgent, 1)
|
||||
acceptedAgent.followup([{ type: 'text', text: 'go' }])
|
||||
acceptedAgent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
expect((await scheduled).data.delayMs).toBe(2_000)
|
||||
const acceptedIdle = waitForIdle(context, acceptedAgent)
|
||||
await vi.advanceTimersByTimeAsync(2_000)
|
||||
@@ -335,36 +386,397 @@ describe('bounded transient retry policy', () => {
|
||||
;({ ctx: context } = await harness(rejected))
|
||||
const rejectedAgent = context.agentLoop.create(SessionId('retry-after-rejected'), { provider: 'mock', model: 'mock' })
|
||||
const rejectedIdle = waitForIdle(context, rejectedAgent)
|
||||
rejectedAgent.followup([{ type: 'text', text: 'go' }])
|
||||
rejectedAgent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
await rejectedIdle
|
||||
expect(rejected.requests).toHaveLength(1)
|
||||
expect(rejectedAgent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
|
||||
})
|
||||
|
||||
it('uses local jittered backoff when always mode receives an over-cap Retry-After', async () => {
|
||||
vi.useFakeTimers()
|
||||
const adapter = new ScriptedAdapter([
|
||||
new LlmError('wait too long', 'AUTH', { providerRetryAfterMs: 10 }),
|
||||
textResponse('done'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter, { mock: alwaysConfig({
|
||||
initialDelayMs: 2,
|
||||
maxDelayMs: 4,
|
||||
jitterRatio: 0.5,
|
||||
}) }, undefined, { random: () => 1 }))
|
||||
const agent = context.agentLoop.create(SessionId('retry-always-over-cap'), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
const scheduled = waitForRetry(context, agent, 1)
|
||||
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
expect((await scheduled).data.delayMs).toBe(3)
|
||||
const idle = waitForIdle(context, agent)
|
||||
await vi.advanceTimersByTimeAsync(3)
|
||||
await idle
|
||||
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('delegates non-transient failures without scheduling a timer', async () => {
|
||||
vi.useFakeTimers()
|
||||
const adapter = new ScriptedAdapter([new LlmError('bad key', 'AUTH')])
|
||||
;({ ctx: context } = await harness(adapter))
|
||||
const agent = context.agentLoop.create(SessionId('retry-auth'), { provider: 'mock', model: 'mock' })
|
||||
const idle = waitForIdle(context, agent)
|
||||
agent.followup([{ type: 'text', text: 'go' }])
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
await idle
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
|
||||
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({ content: [{ type: 'text', text: 'missing route' }], source: { kind: 'user' } })
|
||||
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([
|
||||
new LlmError('mock auth failed', 'AUTH'),
|
||||
new LlmError('other auth failed', 'AUTH'),
|
||||
textResponse('other recovered'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter, {
|
||||
other: alwaysConfig({ initialDelayMs: 1, maxDelayMs: 1, jitterRatio: 0 }),
|
||||
}))
|
||||
|
||||
const normalAgent = context.agentLoop.create(SessionId('retry-provider-normal'), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
const normalIdle = waitForIdle(context, normalAgent)
|
||||
normalAgent.followup({ content: [{ type: 'text', text: 'normal' }], source: { kind: 'user' } })
|
||||
await normalIdle
|
||||
expect(normalAgent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
|
||||
|
||||
const alwaysAgent = context.agentLoop.create(SessionId('retry-provider-always'), {
|
||||
provider: 'other',
|
||||
model: 'mock',
|
||||
})
|
||||
const scheduled = waitForRetry(context, alwaysAgent, 1)
|
||||
alwaysAgent.followup({ content: [{ type: 'text', text: 'always' }], source: { kind: 'user' } })
|
||||
expect((await scheduled).data).toMatchObject({
|
||||
provider: 'other',
|
||||
mode: 'always',
|
||||
retry: 1,
|
||||
delayMs: 1,
|
||||
})
|
||||
const alwaysIdle = waitForIdle(context, alwaysAgent)
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
await alwaysIdle
|
||||
|
||||
expect(adapter.requests.map(request => request.provider)).toEqual(['mock', 'other', 'other'])
|
||||
})
|
||||
|
||||
it('selects an always policy from the provider chosen by agent/request', async () => {
|
||||
vi.useFakeTimers()
|
||||
const adapter = new ScriptedAdapter([
|
||||
new LlmError('rerouted auth failed', 'AUTH'),
|
||||
textResponse('rerouted recovery'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter, {
|
||||
other: alwaysConfig({ initialDelayMs: 1, maxDelayMs: 1, jitterRatio: 0 }),
|
||||
}, (ctx) => {
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => ({
|
||||
...await next(),
|
||||
provider: 'other',
|
||||
}))
|
||||
}))
|
||||
const agent = context.agentLoop.create(SessionId('retry-provider-rerouted'), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
const scheduled = waitForRetry(context, agent, 1)
|
||||
|
||||
agent.followup({ content: [{ type: 'text', text: 'reroute' }], source: { kind: 'user' } })
|
||||
expect((await scheduled).data).toMatchObject({ provider: 'other', mode: 'always' })
|
||||
const idle = waitForIdle(context, agent)
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
await idle
|
||||
|
||||
expect(adapter.requests.map(request => request.provider)).toEqual(['other', 'other'])
|
||||
})
|
||||
|
||||
it('keeps finite retry budgets scoped to the failed provider', async () => {
|
||||
vi.useFakeTimers()
|
||||
const adapter = new ScriptedAdapter([
|
||||
new LlmError('mock failed', 'SERVER'),
|
||||
new LlmError('other failed', 'SERVER'),
|
||||
textResponse('other recovered'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter, {
|
||||
mock: normalConfig({
|
||||
maxRetries: 1,
|
||||
backoff: { initialDelayMs: 1, maxDelayMs: 1 },
|
||||
}),
|
||||
other: normalConfig({
|
||||
maxRetries: 1,
|
||||
backoff: { initialDelayMs: 1, maxDelayMs: 1 },
|
||||
}),
|
||||
}, (ctx) => {
|
||||
ctx.on('agent/request', async (_agent, turn, _step, _signal, next) => ({
|
||||
...await next(),
|
||||
provider: turn === 1 ? 'mock' : 'other',
|
||||
}))
|
||||
}))
|
||||
const agent = context.agentLoop.create(SessionId('retry-provider-budgets'), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
const idle = waitForIdle(context, agent)
|
||||
|
||||
agent.followup({
|
||||
content: [{ type: 'text', text: 'switch provider after failure' }],
|
||||
source: { kind: 'user' },
|
||||
})
|
||||
await vi.runAllTimersAsync()
|
||||
await idle
|
||||
|
||||
expect(adapter.requests.map(request => request.provider)).toEqual(['mock', 'other', 'other'])
|
||||
expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => ({
|
||||
provider: event.data.provider,
|
||||
retry: event.data.retry,
|
||||
}))).toEqual([
|
||||
{ provider: 'mock', retry: 1 },
|
||||
{ provider: 'other', retry: 1 },
|
||||
])
|
||||
})
|
||||
|
||||
it.each(['thrown', 'in-band'] as const)(
|
||||
'uses the serving registration policy and resets changed-policy history after a %s failure',
|
||||
async (failureKind) => {
|
||||
vi.useFakeTimers()
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
const oldAdapter = new ScriptedAdapter([(async function * (): AsyncGenerator<StreamChunk> {
|
||||
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({
|
||||
content: [{ type: 'text', text: 'replace while in flight' }],
|
||||
source: { kind: 'user' },
|
||||
})
|
||||
await entered.promise
|
||||
|
||||
mounted.disposeAdapter()
|
||||
const replacement = new ScriptedAdapter([
|
||||
new LlmError('replacement failed', 'AUTH'),
|
||||
textResponse('replacement recovered'),
|
||||
])
|
||||
replacement.configureRetryPolicies({ mock: alwaysConfig({
|
||||
initialDelayMs: 3,
|
||||
maxDelayMs: 3,
|
||||
}) })
|
||||
context.llm.registerAdapter(['mock'], replacement)
|
||||
release.resolve(undefined)
|
||||
|
||||
const firstEvent = await scheduled
|
||||
expect(firstEvent.data).toMatchObject({
|
||||
provider: 'mock',
|
||||
mode: 'always',
|
||||
retry: 1,
|
||||
delayMs: 1,
|
||||
})
|
||||
const replacementScheduled = waitForRetry(context, agent, 1)
|
||||
const idle = waitForIdle(context, agent)
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
const replacementEvent = await replacementScheduled
|
||||
expect(replacementEvent.data).toMatchObject({
|
||||
provider: 'mock',
|
||||
mode: 'always',
|
||||
retry: 1,
|
||||
delayMs: 3,
|
||||
})
|
||||
expect(replacementEvent.data.policyKey).not.toBe(firstEvent.data.policyKey)
|
||||
await vi.advanceTimersByTimeAsync(3)
|
||||
await idle
|
||||
|
||||
expect(oldAdapter.requests).toHaveLength(1)
|
||||
expect(replacement.requests).toHaveLength(2)
|
||||
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([
|
||||
new LlmError('auth one', 'AUTH'),
|
||||
new LlmError('auth two', 'AUTH'),
|
||||
new LlmError('auth three', 'AUTH'),
|
||||
new LlmError('auth four', 'AUTH'),
|
||||
textResponse('eventually recovered'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter, { mock: alwaysConfig({
|
||||
initialDelayMs: 1,
|
||||
maxDelayMs: 4,
|
||||
jitterRatio: 0.1,
|
||||
}) }, undefined, { random: () => 1 }))
|
||||
const agent = context.agentLoop.create(SessionId('retry-always-unbounded'), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
const idle = waitForIdle(context, agent)
|
||||
|
||||
agent.followup({ content: [{ type: 'text', text: 'keep trying' }], source: { kind: 'user' } })
|
||||
await vi.runAllTimersAsync()
|
||||
await idle
|
||||
|
||||
const events = agent.session.events.filter(event => event.type === 'llm/retry')
|
||||
expect(adapter.requests).toHaveLength(5)
|
||||
expect(events.map(event => ({
|
||||
provider: event.data.provider,
|
||||
mode: event.data.mode,
|
||||
retry: event.data.retry,
|
||||
delayMs: event.data.delayMs,
|
||||
hasMax: 'maxRetries' in event.data,
|
||||
}))).toEqual([
|
||||
{ provider: 'mock', mode: 'always', retry: 1, delayMs: 1.1, hasMax: false },
|
||||
{ provider: 'mock', mode: 'always', retry: 2, delayMs: 2.2, hasMax: false },
|
||||
{ provider: 'mock', mode: 'always', retry: 3, delayMs: 4, hasMax: false },
|
||||
{ provider: 'mock', mode: 'always', retry: 4, delayMs: 4, hasMax: false },
|
||||
])
|
||||
})
|
||||
|
||||
it('keeps failed error text and partial output out of every retried model context', async () => {
|
||||
vi.useFakeTimers()
|
||||
const diagnostic = 'private provider diagnostic must not enter context'
|
||||
const adapter = new ScriptedAdapter([
|
||||
partialToolFailure(new LlmError(diagnostic, 'AUTH')),
|
||||
textResponse('recovered without leaked context'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter, { mock: alwaysConfig({
|
||||
initialDelayMs: 1,
|
||||
maxDelayMs: 1,
|
||||
}) }))
|
||||
const agent = context.agentLoop.create(SessionId('retry-always-context-isolation'), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
const scheduled = waitForRetry(context, agent, 1)
|
||||
|
||||
agent.followup({ content: [{ type: 'text', text: 'safe input' }], source: { kind: 'user' } })
|
||||
await scheduled
|
||||
const idle = waitForIdle(context, agent)
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
await idle
|
||||
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
expect(adapter.requests[1]?.messages).toEqual(adapter.requests[0]?.messages)
|
||||
const retriedContext = JSON.stringify(adapter.requests[1]?.messages)
|
||||
expect(retriedContext).not.toContain(diagnostic)
|
||||
expect(retriedContext).not.toContain('discarded partial output')
|
||||
expect(agent.session.events.some(event =>
|
||||
event.type === 'llm/retry' && event.data.failure.message === diagnostic,
|
||||
)).toBe(true)
|
||||
})
|
||||
|
||||
it('lets downstream specialized recovery run before always fallback', async () => {
|
||||
const adapter = new ScriptedAdapter([
|
||||
new LlmError('requires specialized recovery', 'AUTH'),
|
||||
textResponse('specialized recovery won'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter, { mock: alwaysConfig() }))
|
||||
context.on('agent/request-error', async () => ({ kind: 'retry' }))
|
||||
const agent = context.agentLoop.create(SessionId('retry-always-composition'), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
const idle = waitForIdle(context, agent)
|
||||
|
||||
agent.followup({ content: [{ type: 'text', text: 'recover' }], source: { kind: 'user' } })
|
||||
await idle
|
||||
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
|
||||
})
|
||||
|
||||
it.each([
|
||||
['synchronously', () => { throw new Error('downstream recovery failed') }],
|
||||
['asynchronously', async () => { throw new Error('downstream recovery failed') }],
|
||||
])('falls back to always retry when downstream recovery throws %s', async (_kind, failDownstream) => {
|
||||
vi.useFakeTimers()
|
||||
const adapter = new ScriptedAdapter([
|
||||
new LlmError('requires fallback', 'AUTH'),
|
||||
textResponse('always recovered'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter, { mock: alwaysConfig({
|
||||
initialDelayMs: 1,
|
||||
maxDelayMs: 1,
|
||||
}) }))
|
||||
context.on('agent/request-error', failDownstream)
|
||||
const agent = context.agentLoop.create(SessionId('retry-always-downstream-error'), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
const scheduled = waitForRetry(context, agent, 1)
|
||||
|
||||
agent.followup({ content: [{ type: 'text', text: 'recover' }], source: { kind: 'user' } })
|
||||
await scheduled
|
||||
const idle = waitForIdle(context, agent)
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
await idle
|
||||
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('aborts and drains a captured backoff before plugin disposal completes', async () => {
|
||||
vi.useFakeTimers()
|
||||
const adapter = new ScriptedAdapter([
|
||||
new LlmError('temporary', 'TRANSPORT'),
|
||||
textResponse('must not run'),
|
||||
])
|
||||
const mounted = await harness(adapter)
|
||||
const mounted = await harness(adapter, { mock: alwaysConfig() })
|
||||
context = mounted.ctx
|
||||
const agent = context.agentLoop.create(SessionId('retry-hmr'), { provider: 'mock', model: 'mock' })
|
||||
const scheduled = waitForRetry(context, agent, 1)
|
||||
agent.followup([{ type: 'text', text: 'go' }])
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
await scheduled
|
||||
const idle = waitForIdle(context, agent)
|
||||
|
||||
@@ -377,37 +789,122 @@ describe('bounded transient retry policy', () => {
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('does not make plugin disposal wait for a delegated recovery policy', async () => {
|
||||
it('drains delegated recovery before completing plugin disposal', async () => {
|
||||
const adapter = new ScriptedAdapter([new LlmError('bad key', 'AUTH')])
|
||||
const mounted = await harness(adapter)
|
||||
const mounted = await harness(adapter, { mock: alwaysConfig() })
|
||||
context = mounted.ctx
|
||||
const downstream = Promise.withResolvers<RequestErrorDecision>()
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
context.on('agent/request-error', () => {
|
||||
const order: string[] = []
|
||||
context.on('agent/request-error', async () => {
|
||||
entered.resolve(undefined)
|
||||
return downstream.promise
|
||||
await release.promise
|
||||
order.push('downstream')
|
||||
return { kind: 'retry' }
|
||||
})
|
||||
const agent = context.agentLoop.create(SessionId('retry-delegated-disposal'), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
const idle = waitForIdle(context, agent)
|
||||
agent.followup([{ type: 'text', text: 'go' }])
|
||||
const idle = waitForIdle(context, agent).then(() => { order.push('idle') })
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
await entered.promise
|
||||
|
||||
const disposing = mounted.retryFiber.dispose()
|
||||
const disposing = mounted.retryFiber.dispose().then(() => { order.push('disposed') })
|
||||
let timer: ReturnType<typeof setTimeout> | undefined
|
||||
const outcome = await Promise.race([
|
||||
disposing.then(() => 'disposed' as const),
|
||||
new Promise<'blocked'>((resolve) => { timer = setTimeout(() => { resolve('blocked') }, 100) }),
|
||||
])
|
||||
if (timer !== undefined) clearTimeout(timer)
|
||||
downstream.resolve({ action: 'fail' })
|
||||
expect(outcome).toBe('blocked')
|
||||
|
||||
release.resolve(undefined)
|
||||
await disposing
|
||||
await idle
|
||||
|
||||
expect(outcome).toBe('disposed')
|
||||
expect(order[0]).toBe('downstream')
|
||||
expect(order).toEqual(expect.arrayContaining(['disposed', 'idle']))
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
|
||||
})
|
||||
|
||||
it('drains delegated recovery before turn cancellation reaches idle', async () => {
|
||||
const adapter = new ScriptedAdapter([new LlmError('bad key', 'AUTH')])
|
||||
const mounted = await harness(adapter, { mock: alwaysConfig() })
|
||||
context = mounted.ctx
|
||||
const downstream = Promise.withResolvers<RequestErrorAction>()
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
const order: string[] = []
|
||||
context.on('agent/request-error', async () => {
|
||||
entered.resolve(undefined)
|
||||
const decision = await downstream.promise
|
||||
order.push('downstream')
|
||||
return decision
|
||||
})
|
||||
const agent = context.agentLoop.create(SessionId('retry-delegated-cancel'), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
const idle = waitForIdle(context, agent).then(() => { order.push('idle') })
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
await entered.promise
|
||||
|
||||
agent.cancel({ kind: 'user' })
|
||||
let timer: ReturnType<typeof setTimeout> | undefined
|
||||
const outcome = await Promise.race([
|
||||
idle.then(() => 'idle' as const),
|
||||
new Promise<'blocked'>((resolve) => { timer = setTimeout(() => { resolve('blocked') }, 100) }),
|
||||
])
|
||||
if (timer !== undefined) clearTimeout(timer)
|
||||
expect(outcome).toBe('blocked')
|
||||
|
||||
downstream.resolve({ kind: 'retry' })
|
||||
await idle
|
||||
|
||||
expect(order).toEqual(['downstream', 'idle'])
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
expect(agent.session.events.at(-1)).toMatchObject({
|
||||
type: 'turn/end',
|
||||
data: { reason: { kind: 'aborted' } },
|
||||
})
|
||||
})
|
||||
|
||||
it('handles synchronous cancellation while entering delegated recovery', async () => {
|
||||
const adapter = new ScriptedAdapter([new LlmError('bad key', 'AUTH')])
|
||||
const mounted = await harness(adapter, { mock: alwaysConfig() })
|
||||
context = mounted.ctx
|
||||
const downstream = Promise.withResolvers<RequestErrorAction>()
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
context.on('agent/request-error', (agent) => {
|
||||
agent.cancel({ kind: 'user' })
|
||||
entered.resolve(undefined)
|
||||
return downstream.promise
|
||||
})
|
||||
const agent = context.agentLoop.create(SessionId('retry-delegated-sync-cancel'), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
const idle = waitForIdle(context, agent)
|
||||
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
await entered.promise
|
||||
let timer: ReturnType<typeof setTimeout> | undefined
|
||||
const outcome = await Promise.race([
|
||||
idle.then(() => 'idle' as const),
|
||||
new Promise<'blocked'>((resolve) => { timer = setTimeout(() => { resolve('blocked') }, 100) }),
|
||||
])
|
||||
if (timer !== undefined) clearTimeout(timer)
|
||||
expect(outcome).toBe('blocked')
|
||||
|
||||
downstream.resolve({ kind: 'retry' })
|
||||
await idle
|
||||
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
expect(agent.session.events.at(-1)).toMatchObject({
|
||||
type: 'turn/end',
|
||||
data: { reason: { kind: 'aborted' } },
|
||||
})
|
||||
})
|
||||
|
||||
it('fails a captured callback after disposal without entering downstream policy', async () => {
|
||||
@@ -415,8 +912,10 @@ describe('bounded transient 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, _signal, next) => {
|
||||
return new Promise<RequestErrorDecision>((resolve) => {
|
||||
ctx.on('agent/request-error', (
|
||||
_agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next,
|
||||
) => {
|
||||
return new Promise<RequestErrorAction>((resolve) => {
|
||||
invokeCaptured = async () => { resolve(await next()) }
|
||||
captured.resolve(undefined)
|
||||
})
|
||||
@@ -424,7 +923,9 @@ describe('bounded transient 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()
|
||||
})
|
||||
@@ -433,7 +934,7 @@ describe('bounded transient retry policy', () => {
|
||||
model: 'mock',
|
||||
})
|
||||
const idle = waitForIdle(context, agent)
|
||||
agent.followup([{ type: 'text', text: 'go' }])
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
await captured.promise
|
||||
|
||||
await mounted.retryFiber.dispose()
|
||||
@@ -448,13 +949,13 @@ describe('bounded transient retry policy', () => {
|
||||
it('lets turn cancellation win during backoff without opening another step', async () => {
|
||||
vi.useFakeTimers()
|
||||
const adapter = new ScriptedAdapter([
|
||||
new LlmError('temporary', 'TIMEOUT'),
|
||||
new LlmError('permanent', 'AUTH'),
|
||||
textResponse('must not run'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter))
|
||||
;({ ctx: context } = await harness(adapter, { mock: alwaysConfig() }))
|
||||
const agent = context.agentLoop.create(SessionId('retry-cancel'), { provider: 'mock', model: 'mock' })
|
||||
const scheduled = waitForRetry(context, agent, 1)
|
||||
agent.followup([{ type: 'text', text: 'go' }])
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
await scheduled
|
||||
const idle = waitForIdle(context, agent)
|
||||
agent.cancel({ kind: 'user' })
|
||||
@@ -468,14 +969,19 @@ describe('bounded transient retry policy', () => {
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('lets an earlier recovery listener cancel before retry policy runs', async () => {
|
||||
it.each([
|
||||
['normal', normalConfig()],
|
||||
['always', alwaysConfig()],
|
||||
])('lets an earlier recovery listener cancel before %s retry policy runs', async (_mode, policy) => {
|
||||
vi.useFakeTimers()
|
||||
const adapter = new ScriptedAdapter([
|
||||
new LlmError('temporary', 'SERVER'),
|
||||
textResponse('must not run'),
|
||||
])
|
||||
;({ ctx: context } = await harness(adapter, {}, (ctx) => {
|
||||
ctx.on('agent/request-error', async (agent, _turn, _step, _error, _failure, _history, _signal, next) => {
|
||||
;({ ctx: context } = await harness(adapter, { mock: policy }, (ctx) => {
|
||||
ctx.on('agent/request-error', async (
|
||||
agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next,
|
||||
) => {
|
||||
agent.cancel({ kind: 'user' })
|
||||
return next()
|
||||
})
|
||||
@@ -483,7 +989,7 @@ describe('bounded transient retry policy', () => {
|
||||
const agent = context.agentLoop.create(SessionId('retry-pre-cancel'), { provider: 'mock', model: 'mock' })
|
||||
const idle = waitForIdle(context, agent)
|
||||
|
||||
agent.followup([{ type: 'text', text: 'go' }])
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
await idle
|
||||
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
@@ -507,7 +1013,7 @@ describe('bounded transient retry policy', () => {
|
||||
})
|
||||
const idle = waitForIdle(context, agent)
|
||||
|
||||
agent.followup([{ type: 'text', text: 'go' }])
|
||||
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||
await idle
|
||||
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
@@ -515,19 +1021,17 @@ describe('bounded transient retry policy', () => {
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it.each([
|
||||
[{ maxTransientRetries: -1 }, /maxTransientRetries/],
|
||||
[{ maxTransientRetries: 1.5 }, /maxTransientRetries/],
|
||||
[{ initialDelayMs: 0 }, /initialDelayMs/],
|
||||
[{ maxDelayMs: Number.POSITIVE_INFINITY }, /maxDelayMs/],
|
||||
[{ initialDelayMs: MAX_TIMER_DELAY_MS + 1 }, /initialDelayMs/],
|
||||
[{ maxDelayMs: MAX_TIMER_DELAY_MS + 1 }, /maxDelayMs/],
|
||||
[{ initialDelayMs: 20, maxDelayMs: 10 }, /less than or equal/],
|
||||
[{ jitterRatio: 1.1 }, /jitterRatio/],
|
||||
[{ retryableCodes: [] }, /must not be empty/],
|
||||
[{ retryableCodes: ['SERVER', 'SERVER'] }, /duplicates/],
|
||||
[{ retryableCodes: [''] }, /non-empty strings/],
|
||||
] as const)('fails direct composition for invalid config %#', (config, message) => {
|
||||
expect(() => { retry.apply(new Context(), config as retry.Config) }).toThrow(message)
|
||||
it('rejects retry policy configured on the executor instead of a provider', () => {
|
||||
expectTypeOf<{}>().toExtend<retry.Config>()
|
||||
expectTypeOf<{ retryPolicy: { mode: 'always' } }>().not.toExtend<retry.Config>()
|
||||
expect(() => {
|
||||
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: {} } as unknown as retry.Config)
|
||||
}).toThrow(/unknown key "retryPolciy"/)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -39,13 +39,17 @@ async function harness(
|
||||
apiKey: 'mock-key',
|
||||
baseURL,
|
||||
streamIdleTimeoutMs: options.streamIdleTimeoutMs ?? 1_000,
|
||||
retryPolicy: {
|
||||
mode: 'normal',
|
||||
maxRetries: 2,
|
||||
backoff: {
|
||||
initialDelayMs: options.initialDelayMs ?? 10,
|
||||
maxDelayMs: options.initialDelayMs ?? 10,
|
||||
jitterRatio: 0,
|
||||
},
|
||||
},
|
||||
})
|
||||
await ctx.plugin(Retry, {
|
||||
maxTransientRetries: 2,
|
||||
initialDelayMs: options.initialDelayMs ?? 10,
|
||||
maxDelayMs: options.initialDelayMs ?? 10,
|
||||
jitterRatio: 0,
|
||||
})
|
||||
await ctx.plugin(Retry)
|
||||
await ctx.plugin(AgentLoop, { agents: [] })
|
||||
return ctx
|
||||
}
|
||||
@@ -62,7 +66,7 @@ function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
|
||||
|
||||
function sendAndWait(ctx: Context, agent: Agent): Promise<void> {
|
||||
const idle = waitForIdle(ctx, agent)
|
||||
agent.followup([{ type: 'text', text: 'recover through the provider boundary' }])
|
||||
agent.followup({ content: [{ type: 'text', text: 'recover through the provider boundary' }], source: { kind: 'user' } })
|
||||
return idle
|
||||
}
|
||||
|
||||
@@ -102,8 +106,9 @@ describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {
|
||||
|
||||
expect(server).toBeDefined()
|
||||
expect(server?.requests).toHaveLength(1)
|
||||
expect(agent.session.events.filter(event => event.type === 'step/start').map(event => event.data.step))
|
||||
.toEqual([1, 2])
|
||||
expect(agent.session.events.filter(event => event.type === 'step/start')
|
||||
.map(event => [event.data.turn, event.data.step]))
|
||||
.toEqual([[1, 1], [2, 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')
|
||||
@@ -131,10 +136,11 @@ 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)
|
||||
expect(agent.session.events.filter(event =>
|
||||
event.type === 'assistant/chunk' && event.data.step === 1,
|
||||
event.type === 'assistant/chunk' && event.data.turn === 1,
|
||||
)).toHaveLength(failedChunkCount)
|
||||
expect(agent.session.events.filter(event => event.type === 'assistant/message').map(event => event.data.step))
|
||||
.toEqual([2])
|
||||
expect(agent.session.events.filter(event => event.type === 'assistant/message')
|
||||
.map(event => [event.data.turn, event.data.step]))
|
||||
.toEqual([[2, 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')
|
||||
@@ -157,8 +163,9 @@ describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {
|
||||
expect(server.requests[0]?.body).toEqual(server.requests[1]?.body)
|
||||
expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data.failure.code))
|
||||
.toEqual(['EMPTY_RESPONSE'])
|
||||
expect(agent.session.events.filter(event => event.type === 'assistant/message').map(event => event.data.step))
|
||||
.toEqual([2])
|
||||
expect(agent.session.events.filter(event => event.type === 'assistant/message')
|
||||
.map(event => [event.data.turn, event.data.step]))
|
||||
.toEqual([[2, 1]])
|
||||
expect(agent.session.events.at(-1)).toMatchObject({
|
||||
type: 'turn/end',
|
||||
data: { reason: { kind: 'completed' } },
|
||||
@@ -182,7 +189,7 @@ 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.step === 1,
|
||||
event.type === 'assistant/chunk' && event.data.turn === 1,
|
||||
)).toHaveLength(2)
|
||||
expect(agent.session.events.some(event => event.type === 'assistant/message')).toBe(false)
|
||||
expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
import { defineConfig } from 'tsdown'
|
||||
|
||||
/** Build the package root and invariant companion as independent bundles. */
|
||||
export default defineConfig([
|
||||
{
|
||||
entry: ['lib/types/index.js'],
|
||||
outDir: 'lib',
|
||||
format: ['esm'],
|
||||
platform: 'node',
|
||||
target: 'es2024',
|
||||
fixedExtension: false,
|
||||
dts: false,
|
||||
clean: false,
|
||||
},
|
||||
{
|
||||
entry: ['lib/types/invariant.js'],
|
||||
outDir: 'lib',
|
||||
format: ['esm'],
|
||||
platform: 'node',
|
||||
target: 'es2024',
|
||||
fixedExtension: false,
|
||||
dts: false,
|
||||
clean: false,
|
||||
},
|
||||
])
|
||||
Reference in New Issue
Block a user