feat: implement bounded LLM request recovery

This commit is contained in:
Tianyi Cui
2026-07-20 03:34:19 +08:00
parent 7cf966fc0e
commit 3b0b0cefeb
115 changed files with 3311 additions and 366 deletions

View File

@@ -0,0 +1,39 @@
# `@deepseek-ai/dsh-llm-retry`
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.
The default policy permits two retries for `RATE_LIMIT`, `SERVER`, `TIMEOUT`, and `TRANSPORT`, using bounded exponential backoff from 500 ms to 10 seconds with 10 percent jitter. Delay bounds must fit Node's supported timer range. A valid provider `retryAfterMs` replaces local backoff when it is within the configured cap; an over-cap instruction delegates to the next recovery policy instead.
Before waiting, the plugin appends a non-surface `llm/retry` event with the failure and scheduled delay. 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.
```yaml
- name: '@deepseek-ai/dsh-llm-retry'
config:
maxTransientRetries: 2
initialDelayMs: 500
maxDelayMs: 10000
jitterRatio: 0.1
retryableCodes: [RATE_LIMIT, SERVER, TIMEOUT, TRANSPORT]
```
## Model Experience
### Transient 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.
#### 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.
#### 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.
## 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.
- **`llm/retry` records scheduling, not completion** — later step and turn events establish success, exhaustion, or cancellation.

View File

@@ -0,0 +1,47 @@
{
"name": "@deepseek-ai/dsh-llm-retry",
"description": "Bounded transient LLM request retry policy for the DeepSeek Harness",
"version": "0.0.1",
"private": true,
"type": "module",
"main": "lib/index.js",
"types": "lib/types/index.d.ts",
"exports": {
".": {
"types": "./lib/types/index.d.ts",
"default": "./lib/index.js"
},
"./package.json": "./package.json"
},
"files": [
"lib/index.js",
"lib/types/**/*.d.ts",
"lib/types/**/*.d.ts.map",
"src"
],
"license": "BSD-3-Clause",
"peerDependencies": {
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-llm": "^0.0.1",
"@deepseek-ai/dsh-session": "^0.0.1",
"@deepseek-ai/dsh-timeout": "^0.0.1",
"cordis": "^4.0.0-rc.7"
},
"dependencies": {
"schemastery": "^3.18.0"
},
"devDependencies": {
"@cordisjs/plugin-include": "workspace:^",
"@cordisjs/plugin-loader": "workspace:^",
"@deepseek-ai/dsh-agent": "workspace:^",
"@deepseek-ai/dsh-agent-loop": "workspace:^",
"@deepseek-ai/dsh-llm": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
"@deepseek-ai/dsh-session-persistence-sqlite": "workspace:^",
"@deepseek-ai/dsh-system-prompt": "workspace:^",
"@deepseek-ai/dsh-timeout": "workspace:^",
"@deepseek-ai/dsh-tools": "workspace:^",
"cordis": "^4.0.0-rc.7"
}
}

View File

@@ -0,0 +1,211 @@
/**
* Bounded transient 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
*/
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'
declare module '@deepseek-ai/dsh-session' {
interface SessionEventMap {
/** Durable, non-surface record of one transient retry scheduled after a closed failed step. */
'llm/retry': {
turn: number
step: number
retry: number
maxRetries: number
delayMs: number
failure: LlmFailure
}
}
}
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(['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[]
}
/** 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]),
})
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')
}
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),
})
}
/** Non-serializable seams used to make timing policy deterministic in tests. */
export interface RetryInternals {
/** Random sample in the inclusive zero-to-one range used for jitter. */
random?: () => number
}
function localDelay(config: ResolvedConfig, 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 cancellableDelay(delayMs: number, signal: AbortSignal): Promise<boolean> {
if (signal.aborted) return Promise.resolve(false)
return new Promise((resolve) => {
const timer = setTimeout(() => {
signal.removeEventListener('abort', onAbort)
resolve(true)
}, delayMs)
function onAbort(): void {
clearTimeout(timer)
resolve(false)
}
signal.addEventListener('abort', onAbort, { once: true })
})
}
/**
* Install bounded transient request recovery.
* @param ctx - plugin context that owns the listener and active waits.
* @param config - retry budget, delay bounds, jitter, and eligible codes.
* @param internals - non-serializable deterministic seams for tests.
*/
export function apply(ctx: Context, config: Config = {}, internals: RetryInternals = {}): void {
const resolved = resolveConfig(config)
const random = internals.random ?? Math.random
const lifetime = new AbortController()
const active = new Set<Promise<RequestErrorDecision>>()
async function backoff(
agent: Agent,
turn: number,
step: number,
failure: LlmFailure,
retry: number,
delayMs: number,
signal: AbortSignal,
): Promise<RequestErrorDecision> {
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' }
}
const disposeListener = ctx.on('agent/request-error', (
agent: Agent,
turn: number,
step: number,
_error: RequestError,
failure: LlmFailure,
priorFailures: readonly LlmFailure[],
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()
const retry = priorTransientFailures + 1
let delayMs: number
if (failure.retryAfterMs !== undefined && Number.isFinite(failure.retryAfterMs) && failure.retryAfterMs > 0) {
if (failure.retryAfterMs > resolved.maxDelayMs) return next()
delayMs = failure.retryAfterMs
} else {
delayMs = localDelay(resolved, retry, random)
}
const tracked = backoff(agent, turn, step, failure, retry, delayMs, signal)
.finally(() => active.delete(tracked))
active.add(tracked)
return tracked
})
ctx.effect(() => async () => {
disposeListener()
lifetime.abort(new Error('llm-retry plugin disposed'))
await Promise.allSettled([...active])
}, 'llm-retry: abort and drain backoffs')
}

View File

@@ -0,0 +1,124 @@
import { mkdtemp, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { pathToFileURL } from 'node:url'
import { afterEach, describe, expect, it } from 'vitest'
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 SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import * as retry from '../src/index.ts'
let root: string | undefined
let context: Context | undefined
class TransientOnceAdapter extends LlmAdapter {
requests = 0
async * stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
this.requests += 1
if (this.requests === 1) throw new LlmError('temporary outage', 'SERVER')
yield { type: 'block-start', index: 0, blockType: 'text' }
yield { type: 'text-delta', index: 0, text: 'recovered' }
yield { type: 'block-end', index: 0, block: { type: 'text', text: 'recovered' } }
yield { type: 'finish', reason: { kind: 'stop' } }
}
}
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
if (root !== undefined) await rm(root, { recursive: true, force: true })
root = undefined
})
async function loadYaml(lines: readonly string[]): Promise<Context> {
root = await mkdtemp(join(tmpdir(), 'dsh-llm-retry-loader-'))
const configPath = join(root, 'cordis.yml')
await writeFile(configPath, [...lines, ''].join('\n'))
context = new Context()
context.baseUrl = pathToFileURL(root).href + '/'
await context.plugin(Loader)
context.loader.builtins.include = Include
const modules = new Map<string, unknown>([
['@deepseek-ai/dsh-llm', LlmService],
['@deepseek-ai/dsh-session', SessionStore],
['@deepseek-ai/dsh-system-prompt', SystemPrompt],
['@deepseek-ai/dsh-tools', ToolRegistry],
['@deepseek-ai/dsh-agent', AgentRegistry],
['@deepseek-ai/dsh-llm-retry', retry],
['@deepseek-ai/dsh-agent-loop', AgentLoop],
])
context.loader.internal = {
version: 'v2',
async import(specifier: string) {
if (!modules.has(specifier)) throw new Error(`unexpected Loader import: ${specifier}`)
return modules.get(specifier)
},
} as unknown as NonNullable<typeof context.loader.internal>
await context.loader.create({
name: 'cordis:include',
config: { path: pathToFileURL(configPath).href },
})
await context.loader.await()
return context
}
describe('real Loader composition', () => {
it('loads the flat policy and records recovery through the shipping loop', async () => {
const loaded = await loadYaml([
"- name: '@deepseek-ai/dsh-llm'",
"- name: '@deepseek-ai/dsh-session'",
"- name: '@deepseek-ai/dsh-system-prompt'",
"- 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'",
])
const unloaded = [...loaded.loader.entries()]
.filter(entry => entry.fiber === undefined && !entry.disabled)
.map(entry => entry.options.name)
expect(unloaded).toEqual([])
expect(loaded.agents).toBeInstanceOf(AgentRegistry)
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.send([{ type: 'text', text: 'recover' }])
await idle
expect(adapter.requests).toBe(2)
expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(1)
expect(agent.session.deriveMessages().at(-1)).toMatchObject({
role: 'assistant',
content: [{ type: 'text', text: 'recovered' }],
})
})
})

View File

@@ -0,0 +1,57 @@
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
import SessionPersistenceSqlite from '@deepseek-ai/dsh-session-persistence-sqlite'
import type {} from '../src/index.ts'
const dirs: string[] = []
afterEach(async () => {
for (const dir of dirs.splice(0)) await rm(dir, { recursive: true, force: true })
})
async function backend(kind: 'jsonl' | 'sqlite'): Promise<Context> {
const ctx = new Context()
await ctx.plugin(SessionStore)
if (kind === 'jsonl') {
const root = await mkdtemp(join(tmpdir(), 'dsh-llm-retry-jsonl-'))
dirs.push(root)
await ctx.plugin(SessionPersistenceJsonl, { root })
} else {
await ctx.plugin(SessionPersistenceSqlite, { path: ':memory:' })
}
return ctx
}
describe.each(['jsonl', 'sqlite'] as const)('%s retry-event persistence', (kind) => {
it('round-trips the event losslessly without adding a model message', async () => {
const ctx = await backend(kind)
try {
const session = ctx.sessions.create(SessionId(`retry-${kind}`))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('step/start', { turn: 1, step: 1 })
session.append('step/end', { turn: 1, step: 1 })
const event = session.append('llm/retry', {
turn: 1,
step: 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', reason: 'cancelled in backoff' } })
expect(session.deriveMessages()).toEqual([])
await ctx.sessions.flush(session)
const loaded = await ctx.sessionPersistence.load(session.id)
expect(loaded.events.find(item => item.type === 'llm/retry')).toEqual(event)
} finally {
await ctx.fiber.dispose()
}
})
})

View File

@@ -0,0 +1,453 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import type { Fiber } from 'cordis'
import LlmService, { CallId, LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import type { Agent, RequestErrorDecision } 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>
class ScriptedAdapter extends LlmAdapter {
readonly requests: GenerateOptions[] = []
constructor(private readonly entries: ScriptEntry[]) {
super()
}
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
this.requests.push(options)
const entry = this.entries.shift()
if (entry === undefined) throw new Error('retry test script exhausted')
if (entry instanceof Error) throw entry
yield* entry
}
}
async function* partialToolFailure(error: Error): AsyncGenerator<StreamChunk> {
const id = CallId('discarded-call')
yield { type: 'block-start', index: 0, blockType: 'text' }
yield { type: 'text-delta', index: 0, text: 'discarded partial output' }
yield { type: 'block-end', index: 0, block: { type: 'text', text: 'discarded partial output' } }
yield { type: 'block-start', index: 1, blockType: 'tool-call' }
yield { type: 'tool-call-delta', index: 1, id, name: 'danger', argumentsDelta: '{}' }
yield { type: 'block-end', index: 1, block: { type: 'tool-call', id, name: 'danger', arguments: '{}' } }
throw error
}
function textResponse(text: string): StreamChunk[] {
return [
{ type: 'block-start', index: 0, blockType: 'text' },
{ type: 'text-delta', index: 0, text },
{ type: 'block-end', index: 0, block: { type: 'text', text } },
{ type: 'finish', reason: { kind: 'stop' } },
]
}
async function harness(
adapter: LlmAdapter,
config: retry.Config = {},
beforeRetry?: (ctx: Context) => void,
internals: retry.RetryInternals = {},
): Promise<{ ctx: Context; retryFiber: Fiber }> {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
beforeRetry?.(ctx)
const resolvedConfig = Object.assign({
maxTransientRetries: 2,
initialDelayMs: 500,
maxDelayMs: 10_000,
jitterRatio: 0,
}, config)
const retryFiber = await ctx.plugin(Object.assign((inner: Context) => {
retry.apply(inner, resolvedConfig, internals)
}, { inject: retry.inject }))
await ctx.plugin(AgentLoop, { agents: [] })
ctx.llm.registerAdapter(['mock'], adapter)
return { ctx, retryFiber }
}
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()
}
})
})
}
function waitForRetry(ctx: Context, agent: Agent, retryNumber: number): Promise<Extract<SessionEvent, { type: 'llm/retry' }>> {
return new Promise((resolve) => {
const dispose = ctx.on('session/event', (session, event) => {
if (session === agent.session && event.type === 'llm/retry' && event.data.retry === retryNumber) {
dispose()
resolve(event)
}
})
})
}
let context: Context | undefined
afterEach(async () => {
vi.useRealTimers()
await context?.fiber.dispose()
context = undefined
})
describe('bounded transient 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))
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)
}
})
})
agent.send([{ type: 'text', text: 'go' }])
const event = await scheduled
expect(event.data).toEqual({
turn: 1,
step: 1,
retry: 1,
maxRetries: 2,
delayMs: 500,
failure: { message: 'busy', code: 'RATE_LIMIT', status: 429 },
})
expect(adapter.requests).toHaveLength(1)
await vi.advanceTimersByTimeAsync(499)
expect(adapter.requests).toHaveLength(1)
const idle = waitForIdle(context, agent)
await vi.advanceTimersByTimeAsync(1)
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.deriveMessages().at(-1)).toEqual({
role: 'assistant',
content: [{ type: 'text', text: 'done' }],
provenance: { provider: 'mock', model: 'mock' },
})
})
it('leaves partial failed chunks on their step without committing a message or tool side effect', async () => {
vi.useFakeTimers()
const adapter = new ScriptedAdapter([
partialToolFailure(new LlmError('stream interrupted', 'TRANSPORT')),
textResponse('recovered'),
])
;({ ctx: context } = await harness(adapter))
let toolExecutions = 0
context.tools.register(defineTool({
name: 'danger',
description: 'must not run for a failed provider attempt',
parameters: {},
async execute() {
toolExecutions += 1
return [{ type: 'text', text: 'unexpected' }]
},
}))
const agent = context.agentLoop.create(SessionId('retry-partial'), { provider: 'mock', model: 'mock' })
const scheduled = waitForRetry(context, agent, 1)
agent.send([{ type: 'text', text: 'go' }])
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,
)
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.some(event => event.type === 'tool/call')).toBe(false)
expect(toolExecutions).toBe(0)
expect(agent.session.deriveMessages().at(-1)).toMatchObject({
role: 'assistant',
content: [{ type: 'text', text: 'recovered' }],
provenance: { provider: 'mock', model: 'mock' },
})
})
it('applies bounded exponential jitter and stops after the configured budget', async () => {
vi.useFakeTimers()
const samples = [0, 1]
const adapter = new ScriptedAdapter([
new LlmError('busy one', 'SERVER'),
new LlmError('busy two', 'SERVER'),
new LlmError('busy three', 'SERVER'),
])
;({ ctx: context } = await harness(adapter, { 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.send([{ type: 'text', text: 'go' }])
expect((await first).data.delayMs).toBe(450)
const second = waitForRetry(context, agent, 2)
await vi.advanceTimersByTimeAsync(450)
expect((await second).data.delayMs).toBe(1_100)
const idle = waitForIdle(context, agent)
await vi.advanceTimersByTimeAsync(1_100)
await idle
expect(adapter.requests).toHaveLength(3)
expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(2)
expect(agent.session.events.at(-1)).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'error', failure: { message: 'busy three', code: 'SERVER' } } },
})
})
it('uses a bounded provider Retry-After verbatim and delegates an over-cap instruction', async () => {
vi.useFakeTimers()
const accepted = new ScriptedAdapter([
new LlmError('wait', 'RATE_LIMIT', { retryAfterMs: 2_000 }),
textResponse('done'),
])
;({ ctx: context } = await harness(accepted, { jitterRatio: 1 }))
const acceptedAgent = context.agentLoop.create(SessionId('retry-after-accepted'), { provider: 'mock', model: 'mock' })
const scheduled = waitForRetry(context, acceptedAgent, 1)
acceptedAgent.send([{ type: 'text', text: 'go' }])
expect((await scheduled).data.delayMs).toBe(2_000)
const acceptedIdle = waitForIdle(context, acceptedAgent)
await vi.advanceTimersByTimeAsync(2_000)
await acceptedIdle
expect(accepted.requests).toHaveLength(2)
await context.fiber.dispose()
const rejected = new ScriptedAdapter([
new LlmError('wait too long', 'RATE_LIMIT', { retryAfterMs: 10_001 }),
])
;({ ctx: context } = await harness(rejected))
const rejectedAgent = context.agentLoop.create(SessionId('retry-after-rejected'), { provider: 'mock', model: 'mock' })
const rejectedIdle = waitForIdle(context, rejectedAgent)
rejectedAgent.send([{ type: 'text', text: 'go' }])
await rejectedIdle
expect(rejected.requests).toHaveLength(1)
expect(rejectedAgent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
})
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.send([{ type: 'text', text: 'go' }])
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('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)
context = mounted.ctx
const agent = context.agentLoop.create(SessionId('retry-hmr'), { provider: 'mock', model: 'mock' })
const scheduled = waitForRetry(context, agent, 1)
agent.send([{ type: 'text', text: 'go' }])
await scheduled
const idle = waitForIdle(context, agent)
await mounted.retryFiber.dispose()
await idle
await vi.advanceTimersByTimeAsync(60_000)
expect(adapter.requests).toHaveLength(1)
expect(agent.session.events.filter(event => event.type === 'step/start')).toHaveLength(1)
expect(vi.getTimerCount()).toBe(0)
})
it('does not make plugin disposal wait for a delegated recovery policy', async () => {
const adapter = new ScriptedAdapter([new LlmError('bad key', 'AUTH')])
const mounted = await harness(adapter)
context = mounted.ctx
const downstream = Promise.withResolvers<RequestErrorDecision>()
const entered = Promise.withResolvers<undefined>()
context.on('agent/request-error', () => {
entered.resolve(undefined)
return downstream.promise
})
const agent = context.agentLoop.create(SessionId('retry-delegated-disposal'), {
provider: 'mock',
model: 'mock',
})
const idle = waitForIdle(context, agent)
agent.send([{ type: 'text', text: 'go' }])
await entered.promise
const disposing = mounted.retryFiber.dispose()
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' })
await disposing
await idle
expect(outcome).toBe('disposed')
expect(adapter.requests).toHaveLength(1)
})
it('fails a captured callback after disposal without entering downstream policy', async () => {
const adapter = new ScriptedAdapter([new LlmError('bad key', 'AUTH')])
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) => {
invokeCaptured = async () => { resolve(await next()) }
captured.resolve(undefined)
})
})
})
context = mounted.ctx
let downstreamCalls = 0
context.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, _signal, next) => {
downstreamCalls += 1
return next()
})
const agent = context.agentLoop.create(SessionId('retry-captured-disposal'), {
provider: 'mock',
model: 'mock',
})
const idle = waitForIdle(context, agent)
agent.send([{ type: 'text', text: 'go' }])
await captured.promise
await mounted.retryFiber.dispose()
if (invokeCaptured === undefined) throw new Error('request-error waterfall did not capture retry callback')
await invokeCaptured()
await idle
expect(downstreamCalls).toBe(0)
expect(adapter.requests).toHaveLength(1)
})
it('lets turn cancellation win during backoff without opening another step', async () => {
vi.useFakeTimers()
const adapter = new ScriptedAdapter([
new LlmError('temporary', 'TIMEOUT'),
textResponse('must not run'),
])
;({ ctx: context } = await harness(adapter))
const agent = context.agentLoop.create(SessionId('retry-cancel'), { provider: 'mock', model: 'mock' })
const scheduled = waitForRetry(context, agent, 1)
agent.send([{ type: 'text', text: 'go' }])
await scheduled
const idle = waitForIdle(context, agent)
agent.cancel('user cancelled during retry')
await idle
expect(adapter.requests).toHaveLength(1)
expect(agent.session.events.at(-1)).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'aborted', reason: 'user cancelled during retry' } },
})
expect(vi.getTimerCount()).toBe(0)
})
it('lets an earlier recovery listener cancel before retry policy runs', async () => {
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) => {
agent.cancel('cancelled by earlier recovery policy')
return next()
})
}))
const agent = context.agentLoop.create(SessionId('retry-pre-cancel'), { provider: 'mock', model: 'mock' })
const idle = waitForIdle(context, agent)
agent.send([{ type: 'text', text: 'go' }])
await idle
expect(adapter.requests).toHaveLength(1)
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: 'aborted', reason: 'cancelled by earlier recovery policy' } },
})
})
it('handles synchronous cancellation from the retry status event', async () => {
vi.useFakeTimers()
const adapter = new ScriptedAdapter([
new LlmError('temporary', 'SERVER'),
textResponse('must not run'),
])
;({ ctx: context } = await harness(adapter))
const agent = context.agentLoop.create(SessionId('retry-event-cancel'), { provider: 'mock', model: 'mock' })
context.on('session/event', (session, event) => {
if (session === agent.session && event.type === 'llm/retry') agent.cancel('cancelled by retry observer')
})
const idle = waitForIdle(context, agent)
agent.send([{ type: 'text', text: 'go' }])
await idle
expect(adapter.requests).toHaveLength(1)
expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(1)
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)
})
})

View File

@@ -0,0 +1,33 @@
{
"extends": "../../../tsconfig.base.json",
"compilerOptions": {
"rootDir": "src",
"outDir": "lib/types"
},
"include": [
"src"
],
"references": [
{
"path": "../../../vendor/cosmokit"
},
{
"path": "../../../vendor/cordis"
},
{
"path": "../../../vendor/schemastery"
},
{
"path": "../../llm/llm"
},
{
"path": "../../core/session"
},
{
"path": "../../core/agent"
},
{
"path": "../../util/timeout"
}
]
}