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

fix(llm): preserve serving retry policy

Browse Source
This commit is contained in:
Turtle
2026-07-25 13:30:34 +08:00
parent a623ed5183
commit 015ba14bae
27 changed files with 278 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -2,7 +2,7 @@
Function plugin that applies exact-provider retry policy on the agent loop's closed-step recovery seam. It does not wrap `ctx.llm.stream()`: every adapter call remains one provider attempt, and every retry opens a fresh numbered step.
Each provider adapter owns an optional nested `retryPolicy`, captured when its route registers on `ctx.llm`. Omission uses normal mode: two retries for `RATE_LIMIT`, `SERVER`, `TIMEOUT`, and `TRANSPORT`. A normal policy can change its finite budget, eligible codes, and backoff. Always mode asks downstream recovery first, then retries every model-request failure without an attempt limit; success, cancellation, or plugin disposal stops it.
Each provider adapter owns an optional nested `retryPolicy`, captured when its route registers on `ctx.llm` and carried with each call that reaches that registration's final adapter boundary. An in-flight failure retains that serving policy if the route is later disposed or replaced; a failure before any final adapter is selected has no provider policy and delegates. Omission uses normal mode: two retries for `RATE_LIMIT`, `SERVER`, `TIMEOUT`, and `TRANSPORT`. A normal policy can change its finite budget, eligible codes, and backoff. Always mode asks downstream recovery first, then retries every model-request failure without an attempt limit; success, cancellation, or plugin disposal stops it.
Both modes use bounded exponential backoff with symmetric jitter. A valid `providerRetryAfterMs` at or below `maxDelayMs` replaces local backoff without jitter. An over-cap provider delay makes normal mode delegate, while always mode uses its configured local backoff so it cannot terminate on that instruction.

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

@@ -37,13 +37,13 @@ declare module '@deepseek-ai/dsh-session' {
}
export const name = 'llm-retry'
export const inject = ['agents', 'llm']
export const inject = ['agents']
/** This policy executor has no config; providers own `retryPolicy`. */
export type Config = Readonly<Record<never, never>>
export type Config = Readonly<Record<string, never>>
/** Runtime schema for {@link Config}. */
export const Config: z<Config> = z.object({})
export const Config = z.object({}) as unknown as z<Config>
function validateConfig(config: Config): void {
const [key] = Object.keys(config)
@@ -170,6 +170,7 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna
_error: RequestError,
failure: LlmFailure,
priorFailures: readonly LlmFailure[],
policy: ResolvedRetryPolicy | undefined,
signal: AbortSignal,
next: () => Promise<RequestErrorDecision>,
) => {
@@ -177,15 +178,15 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna
// removed. Lifetime cancellation must prevent that stale callback from
// entering a downstream policy after disposal.
if (lifetime.signal.aborted) return Promise.resolve<RequestErrorDecision>({ action: 'fail' })
// Bind policy to the header in force when this step closed. Downstream
// recovery may append later state before an always fallback runs.
if (policy === undefined) return next()
// The call-local policy belongs to the registration that served this
// failure. Recover only the durable provider identity from the header;
// downstream recovery may append later state before an always fallback.
const provider = providerForClosedStep(agent.session.events, turn, step)
/* v8 ignore next 3 -- agent-loop closes only steps whose request header was recorded */
if (provider === undefined) {
throw new Error(`llm-retry: no request provider for closed turn ${turn}/step ${step}`)
}
const policy = ctx.llm.providerRetryPolicy(provider)
if (policy.mode === 'always') {
const downstream = await downstreamUntilAbort(
next,

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

@@ -1,4 +1,4 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { afterEach, describe, expect, expectTypeOf, it, vi } from 'vitest'
import { Context } from 'cordis'
import type { Fiber } from 'cordis'
import LlmService, { CallId, LlmAdapter, LlmError, resolveRetryPolicy } from '@deepseek-ai/dsh-llm'
@@ -79,7 +79,7 @@ async function harness(
policies: Readonly<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)
@@ -92,8 +92,8 @@ async function harness(
retry.apply(inner, {}, internals)
}, { inject: retry.inject }))
await ctx.plugin(AgentLoop, { agents: [] })
ctx.llm.registerAdapter(['mock', 'other'], adapter)
return { ctx, retryFiber }
const disposeAdapter = ctx.llm.registerAdapter(['mock', 'other'], adapter)
return { ctx, retryFiber, disposeAdapter }
}
function normalConfig(
@@ -413,6 +413,28 @@ describe('provider-routed retry policy', () => {
expect(vi.getTimerCount()).toBe(0)
})
it('delegates when no final adapter served the failed request', async () => {
const adapter = new ScriptedAdapter([textResponse('must not run')])
const mounted = await harness(adapter, { mock: alwaysConfig() })
context = mounted.ctx
mounted.disposeAdapter()
const agent = context.agentLoop.create(SessionId('retry-no-serving-policy'), {
provider: 'mock',
model: 'mock',
})
const idle = waitForIdle(context, agent)
agent.followup([{ type: 'text', text: 'missing route' }])
await idle
expect(adapter.requests).toHaveLength(0)
expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
expect(agent.session.events.at(-1)).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'error', failure: { code: 'NO_ADAPTER' } } },
})
})
it('selects policy by the failed request provider', async () => {
vi.useFakeTimers()
const adapter = new ScriptedAdapter([
@@ -481,6 +503,64 @@ describe('provider-routed retry policy', () => {
expect(adapter.requests.map(request => request.provider)).toEqual(['other', 'other'])
})
it.each(['thrown', 'in-band'] as const)(
'uses the serving registration policy when an in-flight route is replaced after a %s failure',
async (failureKind) => {
vi.useFakeTimers()
const entered = Promise.withResolvers<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([{ type: 'text', text: 'replace while in flight' }])
await entered.promise
mounted.disposeAdapter()
const replacement = new ScriptedAdapter([textResponse('replacement recovered')])
replacement.configureRetryPolicies({ mock: normalConfig({ maxRetries: 0 }) })
context.llm.registerAdapter(['mock'], replacement)
release.resolve(undefined)
expect((await scheduled).data).toMatchObject({
provider: 'mock',
mode: 'always',
retry: 1,
delayMs: 1,
})
const idle = waitForIdle(context, agent)
await vi.advanceTimersByTimeAsync(1)
await idle
expect(oldAdapter.requests).toHaveLength(1)
expect(replacement.requests).toHaveLength(1)
expect(agent.session.deriveMessages().at(-1)).toMatchObject({
role: 'assistant',
content: [{ type: 'text', text: 'replacement recovered' }],
})
},
)
it('keeps always mode unbounded while preserving cancellable jittered backoff', async () => {
vi.useFakeTimers()
const adapter = new ScriptedAdapter([
@@ -719,7 +799,9 @@ describe('provider-routed retry policy', () => {
const captured = Promise.withResolvers<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) => {
ctx.on('agent/request-error', (
_agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next,
) => {
return new Promise<RequestErrorDecision>((resolve) => {
invokeCaptured = async () => { resolve(await next()) }
captured.resolve(undefined)
@@ -728,7 +810,9 @@ describe('provider-routed retry policy', () => {
})
context = mounted.ctx
let downstreamCalls = 0
context.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, _signal, next) => {
context.on('agent/request-error', async (
_agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next,
) => {
downstreamCalls += 1
return next()
})
@@ -782,7 +866,9 @@ describe('provider-routed retry policy', () => {
textResponse('must not run'),
])
;({ ctx: context } = await harness(adapter, { mock: policy }, (ctx) => {
ctx.on('agent/request-error', async (agent, _turn, _step, _error, _failure, _history, _signal, next) => {
ctx.on('agent/request-error', async (
agent, _turn, _step, _error, _failure, _history, _retryPolicy, _signal, next,
) => {
agent.cancel({ kind: 'user' })
return next()
})
@@ -823,14 +909,16 @@ describe('provider-routed retry policy', () => {
})
it('rejects retry policy configured on the executor instead of a provider', () => {
expectTypeOf<{}>().toExtend<retry.Config>()
expectTypeOf<{ retryPolicy: { mode: 'always' } }>().not.toExtend<retry.Config>()
expect(() => {
retry.apply(new Context(), { retryPolicy: { mode: 'always' } })
retry.apply(new Context(), { retryPolicy: { mode: 'always' } } as unknown as retry.Config)
}).toThrow(/retryPolicy belongs under each provider/)
})
it('rejects unknown executor config', () => {
expect(() => {
retry.apply(new Context(), { retryPolciy: {} })
retry.apply(new Context(), { retryPolciy: {} } as unknown as retry.Config)
}).toThrow(/unknown key "retryPolciy"/)
})
})

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

@@ -15,7 +15,7 @@ An adapter registry plus a single streaming call surface, interceptable via a wa
- `ctx.llm.resolveModelContext(provider: string, model: string): Promise<LlmModelContext | undefined>` Resolve authoritative context capacity for one exact route from its owning adapter.
- `ctx.llm.stream(options: GenerateOptions): AsyncIterable<StreamChunk>` Stream one model call as raw chunks (token-level deltas). Consumers assemble the chunks into blocks/messages with `BlockAssembler`.
`LlmService` preserves errors from final adapter selection, synchronous dispatch, iterator construction, and iteration, and binds their provenance to the exact stream handle returned for that model call. `isLlmAdapterFailure(stream, value)` reports only errors from that call's final adapter boundary; `llmFailureOf(stream, value)` returns the adjacent immutable `LlmFailure`. Nested model calls, `llm/stream` middleware, and downstream consumer failures remain unclassified for the outer call. Classification never replaces or mutates the adapter's original coded `Error`.
`LlmService` preserves errors from final adapter selection, synchronous dispatch, iterator construction, and iteration, and binds their provenance to the exact stream handle returned for that model call. `isLlmAdapterFailure(stream, value)` reports only errors from that call's final adapter boundary; `llmFailureOf(stream, value)` returns the adjacent immutable `LlmFailure`; `llmRetryPolicyOf(stream)` returns the immutable policy of the exact registration selected at that boundary, even if the route is later disposed or replaced. A call that never reaches a final adapter has no serving policy. Nested model calls, `llm/stream` middleware, and downstream consumer failures remain unclassified for the outer call. Classification never replaces or mutates the adapter's original coded `Error`.
Provider and model metadata is a discovery surface, not a routing whitelist. `registerAdapter()` still owns provider exclusivity and captures the adapter's retry policy for each route, while an adapter may accept model ids absent from `listModels()`; consumers must not reject a request because its model is unlisted. Returned selector metadata is detached and invalid or duplicate adapter entries fail with `INVALID_ADAPTER` or `INVALID_CATALOG`.

29
packages/llm/llm/src/adapter-failure.ts
View File

@@ -6,9 +6,15 @@
import { HarnessError } from './error.ts'
import type { LlmFailure, StreamChunk } from './types.ts'
import type { ResolvedRetryPolicy } from './retry-policy.ts'
/** Errors and normalized facts proven to originate in one model call's final adapter boundary. */
export type AdapterFailureScope = WeakMap<Error, LlmFailure>
/** Call-local facts captured when one model call enters its final adapter boundary. */
export interface AdapterFailureScope {
/** Errors and normalized facts proven to originate in this call's final adapter boundary. */
readonly failures: WeakMap<Error, LlmFailure>
/** Immutable policy of the exact adapter registration selected for this call. */
retryPolicy?: ResolvedRetryPolicy
}
/** Call-local failure scopes keyed by the exact stream handle returned to a consumer. */
const adapterFailureScopes = new WeakMap<AsyncIterable<StreamChunk>, AdapterFailureScope>()
@@ -52,7 +58,7 @@ export function markLlmAdapterFailure(
message: errorMessage(error),
code: harnessErrorCode(error),
})
failures.set(error, failure)
failures.failures.set(error, failure)
return error
}
@@ -124,7 +130,7 @@ export function isLlmAdapterFailure(
value: unknown,
): value is Error & { code?: string } {
const failures = adapterFailureScopes.get(stream)
return value instanceof Error && failures !== undefined && failures.has(value)
return value instanceof Error && failures !== undefined && failures.failures.has(value)
}
/**
@@ -139,5 +145,18 @@ export function llmFailureOf(
value: unknown,
): LlmFailure | undefined {
const failures = adapterFailureScopes.get(stream)
return value instanceof Error ? failures?.get(value) : undefined
return value instanceof Error ? failures?.failures.get(value) : undefined
}
/**
* Read the retry policy of the exact registration selected at this call's
* final adapter boundary. The policy remains available after that registration
* is disposed or replaced; absence means no final adapter served the call.
* @param stream - the exact stream returned by the model call.
* @returns the immutable serving-registration policy, or `undefined`.
*/
export function llmRetryPolicyOf(
stream: AsyncIterable<StreamChunk>,
): ResolvedRetryPolicy | undefined {
return adapterFailureScopes.get(stream)?.retryPolicy
}

8
packages/llm/llm/src/index.ts
View File

@@ -33,7 +33,7 @@ export * from './retry-policy.ts'
export { BlockAssembler } from './assembler.ts'
export { callConfigEquals, deepFreeze, isAgentLoopRequest, markAgentLoopRequest } from './call-config.ts'
export type { LlmCallConfig } from './call-config.ts'
export { isLlmAdapterFailure, llmFailureOf } from './adapter-failure.ts'
export { isLlmAdapterFailure, llmFailureOf, llmRetryPolicyOf } from './adapter-failure.ts'
declare module 'cordis' {
interface Context {
@@ -337,7 +337,9 @@ export class LlmService extends Service {
): AsyncGenerator<StreamChunk> {
let iterator: AsyncIterator<StreamChunk>
try {
const adapter = this.registration(options.provider).adapter
const registration = this.registration(options.provider)
failures.retryPolicy = registration.retryPolicy
const adapter = registration.adapter
const stream = adapter.stream(this.forAdapter(options, adapter))
iterator = stream[Symbol.asyncIterator]()
} catch (error: unknown) {
@@ -386,7 +388,7 @@ export class LlmService extends Service {
* @returns the chunk stream, possibly wrapped by `llm/stream` listeners.
*/
stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
const failures: AdapterFailureScope = new WeakMap<Error, LlmFailure>()
const failures: AdapterFailureScope = { failures: new WeakMap<Error, LlmFailure>() }
const stream = this.ctx.waterfall(this, 'llm/stream', options, () => this.adapterStream(options, failures))
return bindAdapterFailureScope(stream, failures)
}

47
packages/llm/llm/tests/service.spec.ts
View File

@@ -10,6 +10,7 @@ import LlmService, {
LlmAdapter,
LlmError,
llmFailureOf,
llmRetryPolicyOf,
ProviderRequestId,
resolveRetryPolicy,
StreamChunk,
@@ -181,6 +182,51 @@ describe('LlmService', () => {
)
})
it('keeps the serving registration policy on an in-flight call after route replacement', async () => {
const oldPolicy = resolveRetryPolicy({ mode: 'always' }, 'old retryPolicy')
const newPolicy = resolveRetryPolicy({ mode: 'normal', maxRetries: 0 }, 'new retryPolicy')
const entered = Promise.withResolvers<undefined>()
const release = Promise.withResolvers<undefined>()
const failure = new LlmError('old route failed', 'AUTH')
const oldAdapter = new class extends LlmAdapter {
override providerRetryPolicy(): typeof oldPolicy {
return oldPolicy
}
async * stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
entered.resolve(undefined)
await release.promise
throw failure
}
}()
const newAdapter = new class extends ScriptedAdapter {
override providerRetryPolicy(): typeof newPolicy {
return newPolicy
}
}(SCRIPT)
const ctx = new Context()
await ctx.plugin(LlmService)
const disposeOld = ctx.llm.registerAdapter(['route'], oldAdapter)
const stream = ctx.llm.stream({ provider: 'route', model: 'model', messages: [] })
const outcome = (async (): Promise<unknown> => {
try {
for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) {
return error
}
return undefined
})()
await entered.promise
disposeOld()
ctx.llm.registerAdapter(['route'], newAdapter)
release.resolve(undefined)
expect(await outcome).toBe(failure)
expect(llmRetryPolicyOf(stream)).toBe(oldPolicy)
expect(ctx.llm.providerRetryPolicy('route')).toBe(newPolicy)
})
it('throws NO_ADAPTER for unregistered providers', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
@@ -195,6 +241,7 @@ describe('LlmService', () => {
expect((caught as LlmError).code).toBe('NO_ADAPTER')
expect((caught as LlmError).message).toContain('no adapter registered')
expect(isLlmAdapterFailure(stream, caught)).toBe(true)
expect(llmRetryPolicyOf(stream)).toBeUndefined()
})
it.each(['done', 'value'] as const)('tags a throwing IteratorResult.%s getter without replacing its Error', async (field) => {