Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(llm-retry): simplify policy history validation

Browse Source
This commit is contained in:
Turtle
2026-07-26 01:16:41 +08:00
parent fcd355bf45
commit 19a1aec89e
12 changed files with 186 additions and 663 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/llm/llm-retry/README.md
View File

@@ -8,7 +8,7 @@ Both modes use bounded exponential backoff with symmetric jitter. A valid `provi
Before waiting, the plugin appends a non-surface `llm/retry` event with the provider, mode, canonical resolved-policy key, failure, and scheduled delay. 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 `∞`. 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 a producer-canonical policy key consistent with its mode and finite budget, binds normal failures and every scheduled delay to that policy, has a unique step record and correct provider-policy retry number, and carries a bounded timer delay. Full jitter may schedule zero milliseconds at its lower boundary.
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-deepseek'

14
packages/llm/llm-retry/src/index.ts
View File

@@ -11,7 +11,6 @@ import type { Agent, RequestError, RequestErrorDecision } from '@deepseek-ai/dsh
import type { LlmFailure, ResolvedRetryPolicy } from '@deepseek-ai/dsh-llm'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import { providerForClosedStep } from './history.ts'
import { retryPolicyKey } from './policy-key.ts'
declare module '@deepseek-ai/dsh-session' {
interface SessionEventMap {
@@ -84,6 +83,19 @@ function localDelay(config: ResolvedRetryPolicy, retry: number, random: () => nu
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) => {

25
packages/llm/llm-retry/src/invariant.ts
View File

@@ -3,9 +3,9 @@
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 { parseRetryPolicyKey } from './policy-key.ts'
import type {} from './index.ts'
const PACKAGE_NAME = '@deepseek-ai/dsh-llm-retry'
@@ -54,11 +54,10 @@ function validateRetry(
fail('llm/retry retry must be a positive safe integer')
}
if (typeof provider !== 'string' || provider.length === 0) {
fail('llm/retry provider must be non-empty string')
fail('llm/retry provider must be a non-empty string')
}
const keyedPolicy = parseRetryPolicyKey(policyKey)
if (keyedPolicy === undefined) {
fail('llm/retry policyKey must encode a canonical resolved policy')
if (typeof policyKey !== 'string' || policyKey.length === 0) {
fail('llm/retry policyKey must be a non-empty string')
}
switch (mode) {
case 'normal': {
@@ -66,29 +65,17 @@ function validateRetry(
if (!Number.isSafeInteger(maxRetries) || maxRetries < 1 || retry > maxRetries) {
fail(`llm/retry retry ${retry} must not exceed a positive safe maxRetries ${maxRetries}`)
}
if (keyedPolicy.mode !== 'normal') {
fail(`llm/retry mode normal must match policyKey mode ${keyedPolicy.mode}`)
}
if (keyedPolicy.maxRetries !== maxRetries) {
fail(`llm/retry maxRetries ${maxRetries} must match policyKey`)
}
if (!keyedPolicy.retryableCodes.includes(failure.code)) {
fail(`llm/retry failure code ${failure.code} must be eligible under policyKey`)
}
break
}
case 'always':
if (keyedPolicy.mode !== 'always') {
fail(`llm/retry mode always must match policyKey mode ${keyedPolicy.mode}`)
}
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 > keyedPolicy.maxDelayMs) {
fail(`llm/retry delayMs must be a finite number within policyKey range 0..${keyedPolicy.maxDelayMs}`)
|| delayMs < 0 || delayMs > MAX_TIMER_DELAY_MS) {
fail(`llm/retry delayMs must be a finite number within 0..${MAX_TIMER_DELAY_MS}`)
}
const turnStartIndex = history.findLastIndex(prior =>

100
packages/llm/llm-retry/src/policy-key.ts
View File

@@ -1,100 +0,0 @@
/** Canonical durable identity for resolved retry policies. @module @deepseek-ai/dsh-llm-retry/policy-key */
import type { ResolvedRetryBackoff, ResolvedRetryPolicy } from '@deepseek-ai/dsh-llm'
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
function parseBackoff(
tuple: readonly unknown[],
offset: number,
): ResolvedRetryBackoff | undefined {
const initialDelayMs = tuple[offset]
const maxDelayMs = tuple[offset + 1]
const jitterRatio = tuple[offset + 2]
if (typeof initialDelayMs !== 'number' || !Number.isFinite(initialDelayMs)
|| initialDelayMs <= 0 || initialDelayMs > MAX_TIMER_DELAY_MS
|| typeof maxDelayMs !== 'number' || !Number.isFinite(maxDelayMs)
|| maxDelayMs <= 0 || maxDelayMs > MAX_TIMER_DELAY_MS
|| initialDelayMs > maxDelayMs
|| typeof jitterRatio !== 'number' || !Number.isFinite(jitterRatio)
|| jitterRatio < 0 || jitterRatio > 1) {
return undefined
}
return { initialDelayMs, maxDelayMs, jitterRatio }
}
/**
* Derive the canonical durable key for one fully resolved provider policy.
* Retryable-code order is normalized because eligibility uses set membership.
* @param policy - immutable policy captured from the serving registration.
* @returns canonical JSON tuple containing every behavior-affecting field.
*/
export function retryPolicyKey(policy: ResolvedRetryPolicy): string {
if (policy.mode === 'always') {
return JSON.stringify([
policy.mode,
policy.initialDelayMs,
policy.maxDelayMs,
policy.jitterRatio,
])
}
return JSON.stringify([
policy.mode,
policy.maxRetries,
[...policy.retryableCodes].sort(),
policy.initialDelayMs,
policy.maxDelayMs,
policy.jitterRatio,
])
}
/**
* Parse a producer-canonical policy key from durable input.
* @param value - untrusted persisted event field.
* @returns the resolved policy encoded by the key, or `undefined` for any non-canonical value.
*/
export function parseRetryPolicyKey(value: unknown): ResolvedRetryPolicy | undefined {
if (typeof value !== 'string' || value.length === 0) return undefined
let tuple: unknown
try {
tuple = JSON.parse(value) as unknown
} catch (_invalidPolicyKeyJson) {
return undefined
}
if (!Array.isArray(tuple)) return undefined
const items = tuple as readonly unknown[]
const mode = items[0]
let policy: ResolvedRetryPolicy
switch (mode) {
case 'always': {
if (items.length !== 4) return undefined
const backoff = parseBackoff(items, 1)
if (backoff === undefined) return undefined
policy = Object.freeze({ mode, ...backoff })
break
}
case 'normal': {
if (items.length !== 6) return undefined
const maxRetries = items[1]
const retryableCodes = items[2]
const backoff = parseBackoff(items, 3)
if (!Number.isSafeInteger(maxRetries) || (maxRetries as number) < 0
|| !Array.isArray(retryableCodes) || retryableCodes.length === 0
|| (retryableCodes as readonly unknown[])
.some(code => typeof code !== 'string' || code.length === 0)
|| new Set(retryableCodes).size !== retryableCodes.length
|| backoff === undefined) {
return undefined
}
policy = Object.freeze({
mode,
maxRetries: maxRetries as number,
retryableCodes: Object.freeze(retryableCodes as string[]),
...backoff,
})
break
}
default:
return undefined
}
return retryPolicyKey(policy) === value ? policy : undefined
}

398
packages/llm/llm-retry/tests/invariant.spec.ts
View File

@@ -28,13 +28,26 @@ function closeStep(ctx: Context, id: string, turn = 1, step = 1) {
}
const failure = { message: 'provider busy', code: 'RATE_LIMIT', status: 429 }
const normalPolicyKey = (maxRetries: number): string =>
`["normal",${maxRetries},["RATE_LIMIT"],1,10000,0]`
const alwaysPolicyKey = '["always",1,10000,0]'
const normal = { provider: 'mock', mode: 'normal' as const, policyKey: normalPolicyKey(2) }
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('has no provider without the requested closed step', () => {
it('has no provider without the requested closed step or a route marker', () => {
expect(providerForClosedStep([], 1, 1)).toBeUndefined()
expect(providerForClosedStep([{
type: 'step/end',
@@ -42,82 +55,31 @@ describe('llm-retry invariants', () => {
}] as never, 1, 1)).toBeUndefined()
})
it('inherits the latest provider across a turn boundary when the header is unchanged', () => {
expect(providerForClosedStep([
{ type: 'turn/start', data: { turn: 1 } },
{
type: 'request/header',
data: { header: { config: { provider: 'prior' } } },
},
{ type: 'turn/end', data: { turn: 1 } },
{ type: 'turn/start', data: { turn: 2 } },
{ type: 'step/end', data: { turn: 2, step: 1 } },
] as never, 2, 1)).toBe('prior')
})
it('accepts increasing retry records for successive closed steps and ignores unrelated events', async () => {
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, retry: 1, maxRetries: 2, delayMs: 500, failure,
})
session.append('llm/retry', { turn: 1, step: 1, ...normal })
session.append('step/start', { turn: 1, step: 2 })
session.append('step/end', { turn: 1, step: 2 })
session.append('llm/retry', {
turn: 1, step: 2, ...normal, retry: 2, maxRetries: 2, delayMs: 1_000, failure,
})
const zeroDelay = closeStep(ctx, 'retry-invariant-zero-delay')
zeroDelay.append('llm/retry', {
turn: 1, step: 1, ...normal, policyKey: normalPolicyKey(1),
retry: 1, maxRetries: 1, delayMs: 0, failure,
turn: 1, step: 2, ...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('accepts unbounded always records without serializing an infinite maximum', async () => {
const ctx = await setup()
const session = closeStep(ctx, 'retry-invariant-always')
expect(() => {
session.append('llm/retry', {
turn: 1,
step: 1,
provider: 'mock',
mode: 'always',
policyKey: alwaysPolicyKey,
retry: 1,
delayMs: 500,
failure,
})
}).not.toThrow()
expect(() => {
session.append('llm/retry', {
turn: 1,
step: 1,
provider: 'mock',
mode: 'always',
policyKey: alwaysPolicyKey,
retry: 1,
maxRetries: 2,
delayMs: 500,
failure,
} as never)
}).toThrow(/always mode must omit maxRetries/)
})
it('validates complete durable failures before either retry mode uses them', async () => {
it('validates the complete durable failure payload', async () => {
const ctx = await setup()
const complete = closeStep(ctx, 'retry-invariant-complete-failure')
expect(() => {
complete.append('llm/retry', {
turn: 1,
step: 1,
provider: 'mock',
mode: 'always',
policyKey: alwaysPolicyKey,
retry: 1,
delayMs: 1,
...always,
failure: {
message: 'provider busy',
code: 'RATE_LIMIT',
@@ -128,16 +90,8 @@ describe('llm-retry invariants', () => {
})
}).not.toThrow()
const normalNull = closeStep(ctx, 'retry-invariant-normal-null-failure')
expect(() => {
normalNull.append('llm/retry', {
turn: 1, step: 1, ...normal,
retry: 1, maxRetries: 2, delayMs: 1, failure: null,
} as never)
}).toThrow(/failure must be an object/)
const invalidFailures: readonly [string, unknown, RegExp][] = [
['always-null', null, /failure must be an object/],
['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/],
@@ -159,294 +113,124 @@ describe('llm-retry invariants', () => {
['request-id-empty', { message: 'failed', code: 'RATE_LIMIT', requestId: '' }, /failure\.requestId/],
]
for (const [name, invalidFailure, message] of invalidFailures) {
const session = closeStep(ctx, `retry-invariant-${name}`)
const session = closeStep(ctx, `retry-invariant-failure-${name}`)
expect(() => {
session.append('llm/retry', {
turn: 1,
step: 1,
provider: 'mock',
mode: 'always',
policyKey: alwaysPolicyKey,
retry: 1,
delayMs: 1,
failure: invalidFailure,
turn: 1, step: 1, ...always, failure: invalidFailure,
} as never)
}).toThrow(message)
}
})
it('binds event mode and finite budget to the canonical policy key', async () => {
const ctx = await setup()
const normalModeMismatch = closeStep(ctx, 'retry-invariant-normal-mode-key')
expect(() => {
normalModeMismatch.append('llm/retry', {
turn: 1, step: 1, ...normal, policyKey: alwaysPolicyKey,
retry: 1, maxRetries: 2, delayMs: 1, failure,
})
}).toThrow(/mode normal must match policyKey mode always/)
const alwaysModeMismatch = closeStep(ctx, 'retry-invariant-always-mode-key')
expect(() => {
alwaysModeMismatch.append('llm/retry', {
turn: 1,
step: 1,
provider: 'mock',
mode: 'always',
policyKey: normalPolicyKey(2),
retry: 1,
delayMs: 1,
failure,
})
}).toThrow(/mode always must match policyKey mode normal/)
const budgetMismatch = closeStep(ctx, 'retry-invariant-budget-key')
expect(() => {
budgetMismatch.append('llm/retry', {
turn: 1, step: 1, ...normal, policyKey: normalPolicyKey(3),
retry: 1, maxRetries: 2, delayMs: 1, failure,
})
}).toThrow(/maxRetries 2 must match policyKey/)
})
it('binds the failure code and scheduled delay to the canonical policy key', async () => {
const ctx = await setup()
const ineligibleFailure = closeStep(ctx, 'retry-invariant-failure-code-key')
expect(() => {
ineligibleFailure.append('llm/retry', {
turn: 1, step: 1, ...normal,
retry: 1, maxRetries: 2, delayMs: 1,
failure: { message: 'authentication failed', code: 'AUTH', status: 401 },
})
}).toThrow(/failure code AUTH must be eligible under policyKey/)
const overPolicyDelay = closeStep(ctx, 'retry-invariant-delay-key')
expect(() => {
overPolicyDelay.append('llm/retry', {
turn: 1,
step: 1,
provider: 'mock',
mode: 'always',
policyKey: '["always",1,1,0]',
retry: 1,
delayMs: 2,
failure,
})
}).toThrow(/within policyKey range 0\.\.1/)
})
it('rejects empty providers and unknown modes from hostile durable input', async () => {
const ctx = await setup()
const emptyProvider = closeStep(ctx, 'retry-invariant-empty-provider')
expect(() => {
emptyProvider.append('llm/retry', {
turn: 1,
step: 1,
provider: '',
mode: 'always',
policyKey: alwaysPolicyKey,
retry: 1,
delayMs: 1,
failure,
})
}).toThrow(/provider must be non-empty/)
const unknownMode = closeStep(ctx, 'retry-invariant-unknown-mode')
expect(() => {
unknownMode.append('llm/retry', {
turn: 1,
step: 1,
provider: 'mock',
mode: 'sometimes',
policyKey: alwaysPolicyKey,
retry: 1,
delayMs: 1,
failure,
} as never)
}).toThrow(/mode must be normal or always/)
const emptyPolicyKey = closeStep(ctx, 'retry-invariant-empty-policy-key')
expect(() => {
emptyPolicyKey.append('llm/retry', {
turn: 1,
step: 1,
provider: 'mock',
mode: 'always',
policyKey: '',
retry: 1,
delayMs: 1,
failure,
})
}).toThrow(/policyKey must encode a canonical resolved policy/)
})
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) => {
['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-bounds-${data.retry}-${data.maxRetries}-${data.delayMs}`)
const session = closeStep(ctx, `retry-invariant-${name}`)
expect(() => {
session.append('llm/retry', { turn: 1, step: 1, ...normal, ...data, failure })
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, ...normal, 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, ...normal, 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, ...normal, 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, ...normal, 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, ...normal, 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, ...normal, retry: 1, maxRetries: 2, delayMs: 1, failure,
})
closedTurn.append('llm/retry', { turn: 1, step: 1, ...normal })
}).toThrow(/inside an open turn/)
})
it('binds the policy provider to the failed step rather than a later header', async () => {
it('rejects a second retry record for the same step', async () => {
const ctx = await setup()
const session = closeStep(ctx, 'retry-invariant-duplicate')
session.append('llm/retry', { turn: 1, step: 1, ...normal })
expect(() => {
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('step/start', { turn: 1, step: 2 })
mismatch.append('step/end', { turn: 1, step: 2 })
expect(() => {
mismatch.append('llm/retry', { turn: 1, step: 2, ...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('step/start', { turn: 1, step: 2 })
reset.append('assistant/message', {
turn: 1,
step: 2,
content: [{ type: 'text', text: 'success' }],
provenance: { provider: 'mock', model: 'mock' },
}, { surfaceOp: 'append' })
reset.append('step/end', { turn: 1, step: 2 })
reset.append('step/start', { turn: 1, step: 3 })
reset.append('step/end', { turn: 1, step: 3 })
expect(() => {
reset.append('llm/retry', { turn: 1, step: 3, ...normal })
}).not.toThrow()
})
it('rejects a provider that does not match the failed request route', async () => {
const ctx = await setup()
const session = closeStep(ctx, 'retry-invariant-provider')
session.append('request/header', {
header: { config: { provider: 'other', model: 'mock' } },
reason: 'change',
})
expect(() => {
session.append('llm/retry', {
turn: 1,
step: 1,
provider: 'mock',
mode: 'always',
policyKey: alwaysPolicyKey,
retry: 1,
delayMs: 1,
failure,
})
}).not.toThrow()
const mismatch = closeStep(ctx, 'retry-invariant-provider-mismatch')
expect(() => {
mismatch.append('llm/retry', {
turn: 1,
step: 1,
provider: 'other',
mode: 'always',
policyKey: alwaysPolicyKey,
retry: 1,
delayMs: 1,
failure,
})
session.append('llm/retry', { turn: 1, step: 1, ...always, provider: 'other' })
}).toThrow(/does not match the failed request provider mock/)
})
it('accepts a current-turn retry under an unchanged prior provider route', async () => {
const ctx = await setup()
const session = closeStep(ctx, 'retry-invariant-prior-route')
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('step/start', { turn: 2, step: 1 })
session.append('step/end', { turn: 2, step: 1 })
expect(() => {
session.append('llm/retry', {
turn: 2,
step: 1,
provider: 'mock',
mode: 'always',
policyKey: alwaysPolicyKey,
retry: 1,
delayMs: 1,
failure,
})
}).not.toThrow()
})
it('rejects non-numeric durable delays', async () => {
const ctx = await setup()
const session = closeStep(ctx, 'retry-invariant-delay-type')
expect(() => {
session.append('llm/retry', {
turn: 1, step: 1, ...normal, retry: 1, maxRetries: 2, delayMs: '1', failure,
} as never)
}).toThrow(/delayMs must be a finite number/)
})
it('rejects duplicate and non-increasing retry records', async () => {
const ctx = await setup()
const duplicate = closeStep(ctx, 'retry-invariant-duplicate')
duplicate.append('llm/retry', {
turn: 1, step: 1, ...normal, policyKey: normalPolicyKey(3),
retry: 1, maxRetries: 3, delayMs: 1, failure,
})
expect(() => {
duplicate.append('llm/retry', {
turn: 1, step: 1, ...normal, policyKey: normalPolicyKey(3),
retry: 2, maxRetries: 3, delayMs: 1, failure,
})
}).toThrow(/duplicates the retry record/)
const nonIncreasing = closeStep(ctx, 'retry-invariant-non-increasing')
nonIncreasing.append('llm/retry', {
turn: 1, step: 1, ...normal, policyKey: normalPolicyKey(3),
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, ...normal, policyKey: normalPolicyKey(3),
retry: 1, maxRetries: 3, delayMs: 1, failure,
})
}).toThrow(/must equal provider policy retry 2/)
})
it('validates existing histories on late registration', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const session = ctx.sessions.create(SessionId('retry-invariant-late'))
session.append('step/end', { turn: 1, step: 1 })
session.append('llm/retry', {
turn: 1, step: 1, ...normal, 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/)
})

71
packages/llm/llm-retry/tests/policy-key.spec.ts
View File

@@ -1,71 +0,0 @@
import { describe, expect, it } from 'vitest'
import { resolveRetryPolicy } from '@deepseek-ai/dsh-llm'
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
import { parseRetryPolicyKey, retryPolicyKey } from '../src/policy-key.ts'
describe('retry policy durable key', () => {
it('includes every policy field while normalizing code-set order', () => {
const first = resolveRetryPolicy({
mode: 'normal',
maxRetries: 4,
retryableCodes: ['SERVER', 'RATE_LIMIT'],
backoff: { initialDelayMs: 3, maxDelayMs: 9, jitterRatio: 0.25 },
}, 'first')
const reordered = resolveRetryPolicy({
mode: 'normal',
maxRetries: 4,
retryableCodes: ['RATE_LIMIT', 'SERVER'],
backoff: { initialDelayMs: 3, maxDelayMs: 9, jitterRatio: 0.25 },
}, 'reordered')
const key = retryPolicyKey(first)
expect(key).toBe('["normal",4,["RATE_LIMIT","SERVER"],3,9,0.25]')
expect(retryPolicyKey(reordered)).toBe(key)
expect(parseRetryPolicyKey(key)).toEqual(reordered)
})
it('round-trips always mode', () => {
const policy = resolveRetryPolicy({
mode: 'always',
backoff: { initialDelayMs: 2, maxDelayMs: 8, jitterRatio: 1 },
}, 'always')
const key = retryPolicyKey(policy)
expect(key).toBe('["always",2,8,1]')
expect(parseRetryPolicyKey(key)).toEqual(policy)
})
it.each([
undefined,
'',
'{',
'{}',
'["sometimes",1,2,0]',
'["always",1,2]',
'["always","1",2,0]',
'["always",1e400,2,0]',
'["always",0,2,0]',
`["always",${MAX_TIMER_DELAY_MS + 1},${MAX_TIMER_DELAY_MS + 1},0]`,
'["always",1,"2",0]',
'["always",1,1e400,0]',
'["always",1,0,0]',
`["always",1,${MAX_TIMER_DELAY_MS + 1},0]`,
'["always",2,1,0]',
'["always",1,2,"0"]',
'["always",1,2,1e400]',
'["always",1,2,-0.1]',
'["always",1,2,1.1]',
'["normal",2,["SERVER"],1,2]',
'["normal","2",["SERVER"],1,2,0]',
'["normal",-1,["SERVER"],1,2,0]',
'["normal",2,"SERVER",1,2,0]',
'["normal",2,[],1,2,0]',
'["normal",2,[1],1,2,0]',
'["normal",2,[""],1,2,0]',
'["normal",2,["SERVER","SERVER"],1,2,0]',
'["normal",2,["SERVER"],2,1,0]',
'["normal",2,["SERVER","RATE_LIMIT"],1,2,0]',
])('rejects non-canonical durable input %#', (value) => {
expect(parseRetryPolicyKey(value)).toBeUndefined()
})
})

227
packages/llm/llm-retry/tests/retry.spec.ts
View File

@@ -181,19 +181,14 @@ describe('provider-routed retry policy', () => {
new LlmError('busy', 'RATE_LIMIT', { status: 429 }),
textResponse('done'),
])
;({ ctx: context } = await harness(adapter, {}, undefined, { random: () => 0.5 }))
;({ 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' }])
const event = await scheduled
@@ -203,7 +198,7 @@ describe('provider-routed retry policy', () => {
step: 1,
provider: 'mock',
mode: 'normal',
policyKey: '["normal",2,["EMPTY_RESPONSE","RATE_LIMIT","SERVER","TIMEOUT","TRANSPORT"],500,10000,0.1]',
policyKey: '["normal",2,["RATE_LIMIT","SERVER"],500,10000,0]',
retry: 1,
maxRetries: 2,
delayMs: 500,
@@ -227,45 +222,6 @@ describe('provider-routed retry policy', () => {
})
})
it('retries a later turn under its unchanged provider header', async () => {
vi.useFakeTimers()
const adapter = new ScriptedAdapter([
textResponse('first turn'),
new LlmError('busy on second turn', 'RATE_LIMIT'),
textResponse('second turn recovered'),
])
;({ ctx: context } = await harness(adapter, { mock: normalConfig({
backoff: { initialDelayMs: 1, maxDelayMs: 1 },
}) }))
const agent = context.agentLoop.create(SessionId('retry-later-turn'), {
provider: 'mock',
model: 'mock',
})
const firstIdle = waitForIdle(context, agent)
agent.followup([{ type: 'text', text: 'first' }])
await firstIdle
expect(agent.session.events.filter(event => event.type === 'request/header')).toHaveLength(1)
const scheduled = waitForRetry(context, agent, 1)
agent.followup([{ type: 'text', text: 'second' }])
expect((await scheduled).data).toMatchObject({
turn: 2,
step: 1,
provider: 'mock',
})
const secondIdle = waitForIdle(context, agent)
await vi.advanceTimersByTimeAsync(1)
await secondIdle
expect(adapter.requests).toHaveLength(3)
expect(agent.session.events.filter(event => event.type === 'request/header')).toHaveLength(1)
expect(agent.session.deriveMessages().at(-1)).toMatchObject({
role: 'assistant',
content: [{ type: 'text', text: 'second turn recovered' }],
})
})
it('retries an EMPTY_RESPONSE error finish under the default retryable codes', async () => {
vi.useFakeTimers()
const adapter = new ScriptedAdapter([
@@ -556,8 +512,50 @@ describe('provider-routed retry policy', () => {
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, config) => ({
...config,
provider: step === 1 ? 'mock' : 'other',
}))
}))
const agent = context.agentLoop.create(SessionId('retry-provider-budgets'), {
provider: 'mock',
model: 'mock',
})
const idle = waitForIdle(context, agent)
agent.followup([{ type: 'text', text: 'switch provider after failure' }])
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 when an in-flight route is replaced after a %s failure',
'uses the serving registration policy and resets changed-policy history after a %s failure',
async (failureKind) => {
vi.useFakeTimers()
const entered = Promise.withResolvers<undefined>()
@@ -590,23 +588,40 @@ describe('provider-routed retry policy', () => {
await entered.promise
mounted.disposeAdapter()
const replacement = new ScriptedAdapter([textResponse('replacement recovered')])
replacement.configureRetryPolicies({ mock: normalConfig({ maxRetries: 0 }) })
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)
expect((await scheduled).data).toMatchObject({
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(1)
expect(replacement.requests).toHaveLength(2)
expect(agent.session.deriveMessages().at(-1)).toMatchObject({
role: 'assistant',
content: [{ type: 'text', text: 'replacement recovered' }],
@@ -614,110 +629,6 @@ describe('provider-routed retry policy', () => {
},
)
it('starts a new retry history when a same-mode route replacement changes policy', async () => {
vi.useFakeTimers()
const oldAdapter = new ScriptedAdapter([
new LlmError('old route failed', 'AUTH'),
])
const mounted = await harness(oldAdapter, { mock: alwaysConfig({
initialDelayMs: 1,
maxDelayMs: 1,
}) })
context = mounted.ctx
const agent = context.agentLoop.create(SessionId('retry-policy-replacement'), {
provider: 'mock',
model: 'mock',
})
const first = waitForRetry(context, agent, 1)
agent.followup([{ type: 'text', text: 'replace policy between attempts' }])
expect((await first).data).toMatchObject({
mode: 'always',
retry: 1,
delayMs: 1,
})
mounted.disposeAdapter()
const replacement = new ScriptedAdapter([
new LlmError('replacement failed', 'AUTH'),
textResponse('replacement recovered'),
])
replacement.configureRetryPolicies({ mock: alwaysConfig({
initialDelayMs: 3,
maxDelayMs: 9,
}) })
context.llm.registerAdapter(['mock'], replacement)
const second = waitForRetry(context, agent, 1)
await vi.advanceTimersByTimeAsync(1)
expect((await second).data).toMatchObject({
mode: 'always',
retry: 1,
delayMs: 3,
})
const idle = waitForIdle(context, agent)
await vi.advanceTimersByTimeAsync(3)
await idle
expect(oldAdapter.requests).toHaveLength(1)
expect(replacement.requests).toHaveLength(2)
expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => ({
policyKey: event.data.policyKey,
retry: event.data.retry,
}))).toEqual([
{ policyKey: '["always",1,1,0]', retry: 1 },
{ policyKey: '["always",3,9,0]', retry: 1 },
])
})
it('continues retry history when a replacement only reorders retryable codes', async () => {
vi.useFakeTimers()
const oldAdapter = new ScriptedAdapter([
new LlmError('old route failed', 'SERVER'),
])
const mounted = await harness(oldAdapter, { mock: normalConfig({
maxRetries: 2,
retryableCodes: ['SERVER', 'RATE_LIMIT'],
backoff: { initialDelayMs: 1, maxDelayMs: 4 },
}) })
context = mounted.ctx
const agent = context.agentLoop.create(SessionId('retry-policy-code-order'), {
provider: 'mock',
model: 'mock',
})
const first = waitForRetry(context, agent, 1)
agent.followup([{ type: 'text', text: 'replace equivalent policy between attempts' }])
const firstEvent = await first
expect(firstEvent.data.delayMs).toBe(1)
mounted.disposeAdapter()
const replacement = new ScriptedAdapter([
new LlmError('replacement failed', 'SERVER'),
textResponse('replacement recovered'),
])
replacement.configureRetryPolicies({ mock: normalConfig({
maxRetries: 2,
retryableCodes: ['RATE_LIMIT', 'SERVER'],
backoff: { initialDelayMs: 1, maxDelayMs: 4 },
}) })
context.llm.registerAdapter(['mock'], replacement)
const second = waitForRetry(context, agent, 2)
await vi.advanceTimersByTimeAsync(1)
const secondEvent = await second
expect(secondEvent.data).toMatchObject({ retry: 2, delayMs: 2 })
expect(secondEvent.data.policyKey).toBe(firstEvent.data.policyKey)
const idle = waitForIdle(context, agent)
await vi.advanceTimersByTimeAsync(2)
await idle
expect(oldAdapter.requests).toHaveLength(1)
expect(replacement.requests).toHaveLength(2)
})
it('keeps always mode unbounded while preserving cancellable jittered backoff', async () => {
vi.useFakeTimers()
const adapter = new ScriptedAdapter([