fix(llm-retry): retain recovery requests during activation

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

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

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 组合

packages/llm/llm-retry/src/index.ts

@@ -110,6 +110,8 @@ const llmRetryStateSchema: zod.ZodType<LlmRetryState> = 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<LlmRetryState> = zod.record(zod.string(),
  */
 export async function apply(ctx: Context, config: Config = {}, internals: RetryInternals = {}): Promise<void> {
   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<RequestErrorAction> {
     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<RequestErrorAction>,
   ): Promise<RequestErrorAction> {
     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
 }

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<undefined>()
+    const release = Promise.withResolvers<undefined>()
+    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<undefined>()
+    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'))
+  })
+})

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)
             }