From cc79cf553bd53bfaa95a7bcad19180161f1f83e6 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Mon, 19 Oct 2026 15:26:30 +0800 Subject: [PATCH] fix(llm-retry): retain recovery requests during activation --- packages/llm/llm-retry/README.i18n.yaml | 4 +- packages/llm/llm-retry/README.md | 2 + packages/llm/llm-retry/README.zh.md | 2 + packages/llm/llm-retry/src/index.ts | 9 +- .../llm-retry/tests/initialization.spec.ts | 82 +++++++++++++++++++ snapshots/session/headless.snapshot.ts | 8 ++ 6 files changed, 102 insertions(+), 5 deletions(-) create mode 100644 packages/llm/llm-retry/tests/initialization.spec.ts diff --git a/packages/llm/llm-retry/README.i18n.yaml b/packages/llm/llm-retry/README.i18n.yaml index 6dbc01a710..691b098e1c 100644 --- a/packages/llm/llm-retry/README.i18n.yaml +++ b/packages/llm/llm-retry/README.i18n.yaml @@ -39,8 +39,8 @@ en: 9b79cdd6597930e0 zh: 0838f3efbc1e83ac /deepseek-ai-dsh-llm-retry/understand-the-implementation/recovery-flow: - en: 85bace285b029101 - zh: 6e0237450ca6ccc0 + en: 4ba0a3ef069bd416 + zh: e1cf971729a8717c /deepseek-ai-dsh-llm-retry/understand-the-implementation/waterfall-composition: en: 388e8db333e03cb1 zh: 54b6446ac3584724 diff --git a/packages/llm/llm-retry/README.md b/packages/llm/llm-retry/README.md index 2393ffbfea..3b1455864e 100644 --- a/packages/llm/llm-retry/README.md +++ b/packages/llm/llm-retry/README.md @@ -81,6 +81,8 @@ The executor is built on one rule: **durable before wait, open-step boundaries.* ### Recovery flow +The recovery listener is installed before awaiting projection registration. Failures arriving during activation wait for the reader before applying the provider policy. Registration failure propagates to the agent; cancellation or plugin disposal while waiting prevents retry records and further attempts. + A failed step arrives on the waterfall with its provider and resolved policy. Always mode settles downstream recovery first and honors a downstream `retry` decision; normal mode first checks that the failure code is eligible and the budget is not exhausted. The plugin computes the delay — provider `Retry-After` when valid and within bounds, otherwise local bounded exponential backoff with symmetric jitter — appends the `llm/retry` event, waits on a cancellable timer, appends `llm/retry-started`, and returns `{ kind: 'retry' }`. The loop then re-runs the failed step inside the same open turn over the same durable history. ### Waterfall composition diff --git a/packages/llm/llm-retry/README.zh.md b/packages/llm/llm-retry/README.zh.md index eef7a17aae..bbad68655d 100644 --- a/packages/llm/llm-retry/README.zh.md +++ b/packages/llm/llm-retry/README.zh.md @@ -81,6 +81,8 @@ kind: "package-reference" ### 恢复流程 +恢复监听器在等待投影注册前安装。激活期间到达的失败会先等待读取句柄，再应用提供方策略。注册失败会传播给 agent；等待期间取消或释放插件，不会写入重试记录或发起后续尝试。 + 失败步骤连同其提供方与解析后的策略一起到达 waterfall。always mode 先结算下游恢复，并遵循下游的 `retry` 决定；normal mode 先检查失败 code 是否合格、预算是否未耗尽。插件计算延迟——有效且在边界内的提供方 `Retry-After`，否则带对称抖动的本地有界指数退避——追加 `llm/retry` 事件，在可取消定时器上等待，追加 `llm/retry-started`，然后返回 `{ kind: 'retry' }`。loop 随后在同一个打开的轮次内重跑失败步骤（仍基于同一份持久历史）。 ### Waterfall 组合 diff --git a/packages/llm/llm-retry/src/index.ts b/packages/llm/llm-retry/src/index.ts index 0c181260ce..45fa11c5d4 100644 --- a/packages/llm/llm-retry/src/index.ts +++ b/packages/llm/llm-retry/src/index.ts @@ -110,6 +110,8 @@ const llmRetryStateSchema: zod.ZodType = zod.record(zod.string(), /** * Install provider-routed normal or unbounded request recovery. + * Recovery admitted during activation awaits projection registration. A + * registration failure propagates; cancellation or disposal suppresses retries. * @param ctx - plugin context that owns the listener and active waits. * @param config - empty executor config; provider registrations own policy. * @param internals - non-serializable deterministic hooks for tests. @@ -117,7 +119,7 @@ const llmRetryStateSchema: zod.ZodType = zod.record(zod.string(), */ export async function apply(ctx: Context, config: Config = {}, internals: RetryInternals = {}): Promise { validateConfig(config) - const llmRetryProjectionHandle = await ctx.sessionProjections.register({ + const projectionReady = ctx.sessionProjections.register({ key: 'llmRetry', stateVersion: 1, stateSchema: llmRetryStateSchema, @@ -155,7 +157,6 @@ export async function apply(ctx: Context, config: Config = {}, internals: RetryI signal: AbortSignal, ): Promise { const fusedSignal = AbortSignal.any([signal, lifetime.signal]) - if (fusedSignal.aborted) return const eventData: LlmRetryEventData = policy.mode === 'normal' ? { retryId, @@ -191,8 +192,9 @@ export async function apply(ctx: Context, config: Config = {}, internals: RetryI next: () => Promise, ): Promise { if (policy === undefined) return next() + const llmRetryProjectionHandle = await projectionReady + if (signal.aborted || ctx.fiber.uid === null || lifetime.signal.aborted) return 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. @@ -251,4 +253,5 @@ export async function apply(ctx: Context, config: Config = {}, internals: RetryI lifetime.abort(new Error('llm-retry plugin disposed')) await Promise.allSettled([...active]) }, 'llm-retry: abort and drain active recovery') + await projectionReady } diff --git a/packages/llm/llm-retry/tests/initialization.spec.ts b/packages/llm/llm-retry/tests/initialization.spec.ts new file mode 100644 index 0000000000..f8b88d3f9f --- /dev/null +++ b/packages/llm/llm-retry/tests/initialization.spec.ts @@ -0,0 +1,82 @@ +/** Provider recovery overlapping retry projection registration. */ +import { Context } from '@deepseek-ai/cordis' +import AgentLoop from '@deepseek-ai/dsh-agent-loop' +import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit' +import { createUserMessage, resolveRetryPolicy } from '@deepseek-ai/dsh-llm' +import type { ResolvedRetryPolicy } from '@deepseek-ai/dsh-llm' +import * as retry from '@deepseek-ai/dsh-llm-retry' +import { SessionId } from '@deepseek-ai/dsh-session' +import { describe, expect, it, onTestFinished, vi } from 'vitest' +import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts' + +class RetryAdapter extends MockAdapter { + override providerRetryPolicy(): ResolvedRetryPolicy { + return resolveRetryPolicy({ + mode: 'normal', maxRetries: 1, + backoff: { initialDelayMs: 1, maxDelayMs: 1, jitterRatio: 0 }, + }, 'initialization test') + } +} + +describe('retry activation', () => { + it.each(['ready', 'cancelled', 'disposed', 'failed'] as const)('handles a %s registration before settling the first provider failure', async (outcome) => { + const ctx = new Context() + const registered = Promise.withResolvers() + const release = Promise.withResolvers() + onTestFinished(async () => { + release.resolve(undefined) + await ctx.fiber.dispose() + }) + await mountAgentLoopTestDependencies(ctx) + await ctx.plugin(AgentLoop, { agents: [] }) + const adapter = new RetryAdapter([ + [{ type: 'finish', reason: { kind: 'error', failure: { code: 'SERVER', message: 'temporary failure' } } }], + textResponse('recovered'), + ]) + ctx.llm.registerAdapter(['mock'], adapter) + const agent = await ctx.agentLoop.create(SessionId('initial-retry'), { provider: 'mock', model: 'mock' }) + const register = ctx.sessionProjections.register.bind(ctx.sessionProjections) + const spy = vi.spyOn(ctx.sessionProjections, 'register').mockImplementation(async (definition) => { + const handle = await register(definition) + if (definition.key === 'llmRetry') { + registered.resolve(undefined) + await release.promise + } + return handle + }) + onTestFinished(() => { spy.mockRestore() }) + const fiber = ctx.plugin(retry) + await registered.promise + const recovering = Promise.withResolvers() + ctx.on('agent/request-error', (_payload, next) => { + recovering.resolve(undefined) + return next() + }, { prepend: true }) + const errors: unknown[] = [] + ctx.on('agent/error', ({ error }) => { errors.push(error) }) + agent.followup(createUserMessage({ content: [{ type: 'text', text: 'recover this request' }], source: { kind: 'user' } })) + await recovering.promise + + if (outcome === 'cancelled') agent.cancel({ kind: 'user' }) + const disposal = outcome === 'disposed' ? fiber.dispose() : undefined + if (outcome === 'failed') { + const rejected = expect(Promise.resolve(fiber)).rejects.toThrow('retry registration failed') + release.reject(new Error('retry registration failed')) + await rejected + } else { + release.resolve(undefined) + await (disposal ?? fiber) + } + await agent.whenIdle() + + expect(adapter.requests).toHaveLength(outcome === 'ready' ? 2 : 1) + const log = agent.session.snapshotEvents() + expect(log.filter(event => event.type === 'llm/retry')).toHaveLength(outcome === 'ready' ? 1 : 0) + expect(log.filter(event => event.type === 'llm/retry-started')).toHaveLength(outcome === 'ready' ? 1 : 0) + if (outcome === 'ready') { + expect(adapter.requests[1]?.messages).toEqual(adapter.requests[0]?.messages) + expect(agent.session.deriveMessages().at(-1)?.content).toEqual([{ type: 'text', text: 'recovered' }]) + } + if (outcome === 'failed') expect(errors).toContainEqual(new Error('retry registration failed')) + }) +}) diff --git a/snapshots/session/headless.snapshot.ts b/snapshots/session/headless.snapshot.ts index 503b64ecd9..9c039898f0 100644 --- a/snapshots/session/headless.snapshot.ts +++ b/snapshots/session/headless.snapshot.ts @@ -1334,6 +1334,14 @@ describe('headless recorded-session snapshots', () => { }, inspect: async (cwd) => { actualLogs = await persistedSessions(cwd) + if (scenario.name === 'empty-response-retry-current') { + const events = parseSessionLog(actualLogs[0]!.content) + expect(events.filter(event => event.type === 'assistant/attempt' || event.type === 'llm/retry' + || event.type === 'llm/retry-started' || event.type === 'assistant/message') + .map(event => event.type)).toEqual([ + 'assistant/attempt', 'llm/retry', 'llm/retry-started', 'assistant/message', + ]) + } if (retainedToolInput !== undefined) { verifyToolResultWriterParity(primaryFixture, actualLogs[0]!.content) }