Merge branch 'master' into codex/bump-pi-ai-0.82.1

2026-07-27 23:01:27 +08:00
parent a843c37749 5b5f089bd0
commit 676fd97319
533 changed files with 12197 additions and 10636 deletions
--- a/packages/llm/llm-retry/README.i18n.yaml
+++ b/packages/llm/llm-retry/README.i18n.yaml
@@ -1,6 +1,6 @@
 # Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
-#   pnpm run verify-translation-pairing --write
-README.md: 96dc2314bac59b36a97627e038ac614f3db5f9b3
-README.zh.md: cbee3291d688dfda1c4109fb87630c030bf4b45c
+#   pnpm run verify-translation-pairing --write packages/llm/llm-retry/README.md
+README.md: 596a46e5395a4b5be9d400a85ee7c54d613ec2e6
+README.zh.md: a6a48f203e4701688816d4365b03da1ae2cfad82
--- a/packages/llm/llm-retry/README.md
+++ b/packages/llm/llm-retry/README.md
@@ -2,13 +2,13 @@

 English | [中文](README.zh.md)

-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.
+Function plugin that retries selected transient model-request failures through the `agent/request-error` waterfall. It does not wrap `ctx.llm.stream()`: every adapter call remains one provider attempt, and every retry opens a fresh numbered turn.

 The default policy permits two retries for `EMPTY_RESPONSE`, `RATE_LIMIT`, `SERVER`, `TIMEOUT`, and `TRANSPORT`, using bounded exponential backoff from 500 ms to 10 seconds with 10 percent jitter. `EMPTY_RESPONSE` is the adapters' classification of a degenerate provider completion (a terminal stop with zero content blocks); the attempt produced nothing durable, so repeating it is safe. Delay bounds must fit Node's supported timer range. A valid `providerRetryAfterMs` 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.
+The recovery listener appends a non-surface `llm/retry` event after the failed step, waits for the backoff while the failed turn's signal remains live, then returns `{ kind: 'retry' }`. The loop closes that failed turn and opens a retry turn over the same durable history. The policy keeps its own retry count across that uninterrupted recovery chain and clears it at terminal `agent/settled`. Turn cancellation and plugin disposal abort the wait.

-The separately published `./invariant` companion checks that every retry record names the current open turn and its latest closed step, has a unique step record and increasing retry number, and carries a positive bounded retry budget and non-negative bounded timer delay. Full jitter may schedule zero milliseconds at its lower boundary.
+The separately published `./invariant` companion checks that every retry record appears inside an open turn after its failed step, matches its position in the current retry chain, and carries a positive bounded retry budget and non-negative bounded timer delay. Full jitter may schedule zero milliseconds at its lower boundary.

 ```yaml
 - name: '@deepseek-ai/dsh-llm-retry'
@@ -26,7 +26,7 @@ The separately published `./invariant` companion checks that every retry record

 #### 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.
+No retry event, delay, or failure prose is model-visible. The retry turn reconstructs the same explicit provider/model request from durable session history; failed chunks never enter derived messages.

 #### Token effect

@@ -38,6 +38,6 @@ The reconstructed request preserves the prior prefix and is eligible for provide

 ## 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.
+- **Agent turns 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.
+- **`llm/retry` records completed backoff, not request completion** — later step and turn events establish success, exhaustion, or cancellation.
--- a/packages/llm/llm-retry/README.zh.md
+++ b/packages/llm/llm-retry/README.zh.md
@@ -2,13 +2,13 @@

 [English](README.md) | 中文

-一个函数插件，在 agent loop 的已关闭步骤恢复 seam 上重试特定的短暂模型请求失败。它不包装 `ctx.llm.stream()`：每次适配器调用仍是一次提供方尝试，每次重试都会开启新的编号步骤。
+一个函数插件，通过 `agent/request-error` waterfall 重试特定的短暂模型请求失败。它不包装 `ctx.llm.stream()`：每次适配器调用仍是一次提供方尝试，每次重试都会开启新的编号轮次。

 默认策略允许为 `EMPTY_RESPONSE`、`RATE_LIMIT`、`SERVER`、`TIMEOUT` 和 `TRANSPORT` 重试两次，使用从 500 ms 到 10 秒的有界指数退避与 10% jitter。`EMPTY_RESPONSE` 是适配器对退化提供方完成的分类（携带零个内容块的终止 stop）；该尝试未产生持久内容，因此可安全重复。延迟边界必须适合 Node 支持的定时器范围。有效 `providerRetryAfterMs` 在已配置上限内时替换本地退避；超出上限的指令会委托给下一项恢复策略。

-等待之前，插件会追加一个非表层 `llm/retry` 事件，携带失败与计划延迟。取消与插件 dispose 会中止等待；dispose 会排空插件的活跃退避，dispose 前捕获的 callback 如果在之后调用，将快速失败。
+恢复 listener 会在失败步骤之后追加一个非表层 `llm/retry` 事件，在失败轮次的信号仍存活期间等待退避，然后返回 `{ kind: 'retry' }`。循环会关闭该失败轮次，并在同一持久历史上开启重试轮次。策略在这条不间断的恢复链中维护自己的重试计数，并在终态 `agent/settled` 时清零。轮次取消与插件 dispose 会中止等待。

-单独发布的 `./invariant` 配套模块会检查每个重试记录是否指向当前开启轮次及其最新已关闭步骤，是否拥有唯一步骤记录与递增重试编号，以及是否携带正数有界重试预算和非负有界定时器延迟。完整 jitter 可以在下界调度为零毫秒。
+单独发布的 `./invariant` 配套模块会检查每个重试记录是否出现在开启轮次内的失败步骤之后，是否与其在当前重试链中的位置匹配，以及是否携带正数有界重试预算和非负有界定时器延迟。完整 jitter 可以在下界调度为零毫秒。

 ```yaml
 - name: '@deepseek-ai/dsh-llm-retry'
@@ -26,7 +26,7 @@

 #### 模型看到的内容

-模型不会看到重试事件、延迟或失败文本。重试后，下一个编号步骤会从持久会话历史中重建相同的显式提供方／模型请求；失败 chunk 绝不会进入派生消息。
+模型不会看到重试事件、延迟或失败文本。重试轮次会从持久会话历史中重建相同的显式提供方／模型请求；失败 chunk 绝不会进入派生消息。

 #### Token 影响

@@ -38,6 +38,6 @@

 ## 已知限制与暂缓事项

- **Agent 步骤是唯一重试边界**：直接 `ctx.llm.stream()` 消费方仍只尝试一次，因为原始流无法将已发出 chunk 持久分隔为不同尝试。
+- **Agent 轮次是唯一重试边界**：直接 `ctx.llm.stream()` 消费方仍只尝试一次，因为原始流无法将已发出 chunk 持久分隔为不同尝试。
 - **有限插件预算可叠加**：该策略只统计已配置短暂 code；上下文溢出压缩只统计自身 code。未来如有 code 重叠的策略，必须记录并测试注册顺序行为。
- **`llm/retry` 记录调度，不是完成**：后续步骤与轮次事件用于确立成功、耗尽或取消。
+- **`llm/retry` 记录已完成的退避，不是请求完成**：后续步骤与轮次事件用于确立成功、耗尽或取消。
--- a/packages/llm/llm-retry/src/index.ts
+++ b/packages/llm/llm-retry/src/index.ts
@@ -1,13 +1,13 @@
 /**
- * Bounded transient model-request retry policy on the agent loop's closed-step
- * recovery seam. Each scheduled retry is durable before its cancellable wait.
+ * Bounded transient model-request retry policy on the agent request-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 { Agent, RequestError, RequestErrorAction } 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'
@@ -145,7 +145,8 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna
  const resolved = resolveConfig(config)
  const random = internals.random ?? Math.random
  const lifetime = new AbortController()
-  const active = new Set<Promise<RequestErrorDecision>>()
+  const active = new Set<Promise<RequestErrorAction>>()
+  const retries = new WeakMap<Agent, number>()

  async function backoff(
    agent: Agent,
@@ -155,9 +156,9 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna
    retry: number,
    delayMs: number,
    signal: AbortSignal,
-  ): Promise<RequestErrorDecision> {
+  ): Promise<RequestErrorAction> {
    const fusedSignal = AbortSignal.any([signal, lifetime.signal])
-    if (fusedSignal.aborted) return { action: 'fail' }
+    if (fusedSignal.aborted) return
    agent.session.append('llm/retry', {
      turn,
      step,
@@ -166,29 +167,41 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna
      delayMs,
      failure,
    })
-    if (!await cancellableDelay(delayMs, fusedSignal)) return { action: 'fail' }
-    return { action: 'retry' }
+    retries.set(agent, retry)
+    if (!await cancellableDelay(delayMs, fusedSignal)) return
+    return { kind: 'retry' }
  }

+  ctx.on('agent/settled', (agent) => {
+    retries.delete(agent)
+  })
+
+  // A completed model response ends the consecutive-failure sequence even
+  // when its tool calls keep the turn running into another request.
+  ctx.on('session/event', (session, event) => {
+    if (event.type !== 'assistant/message') return
+    const agent = ctx.agents.get(session.id)
+    if (agent?.session === session) retries.delete(agent)
+  })
+
  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>,
+    next: () => Promise<RequestErrorAction>,
  ) => {
    // 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 (lifetime.signal.aborted) return Promise.resolve<RequestErrorAction>(undefined)
    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 priorRetries = retries.get(agent) ?? 0
+    if (priorRetries >= resolved.maxTransientRetries) return next()

-    const retry = priorTransientFailures + 1
+    const retry = priorRetries + 1
    let delayMs: number
    if (failure.providerRetryAfterMs !== undefined
      && Number.isFinite(failure.providerRetryAfterMs)
--- a/packages/llm/llm-retry/src/invariant.ts
+++ b/packages/llm/llm-retry/src/invariant.ts
@@ -13,6 +13,34 @@ export const name = 'llm-retry-invariant'
 /** Service required before the companion can reserve package ownership. */
 export const inject = ['invariants']

+/** Find the first turn in the structured-failure retry chain containing `turn`. */
+function retryChainStart(history: readonly SessionEvent[], turn: number): number {
+  let startIndex = history.findLastIndex(
+    event => event.type === 'turn/start' && event.data.turn === turn,
+  )
+  while (startIndex >= 0) {
+    const start = history[startIndex]
+    if (start?.type !== 'turn/start' || start.data.trigger.kind !== 'retry') break
+
+    let endIndex = startIndex - 1
+    while (endIndex >= 0 && history[endIndex]?.type !== 'turn/end') endIndex -= 1
+    const end = history[endIndex]
+    if (end?.type !== 'turn/end'
+      || end.data.reason.kind !== 'error'
+      || end.data.reason.failure === undefined) break
+
+    const previousStart = history.findLastIndex(
+      (event, index) =>
+        index < endIndex
+        && event.type === 'turn/start'
+        && event.data.turn === end.data.turn,
+    )
+    if (previousStart < 0) break
+    startIndex = previousStart
+  }
+  return startIndex
+}
+
 /** Validate one retry record against the open turn and most recently closed step. */
 function validateRetry(
  history: readonly SessionEvent[],
@@ -59,14 +87,17 @@ function validateRetry(
    fail(`llm/retry names step ${step}, but the latest closed step is ${String(closedStep)}`)
  }

-  const priorRetries = currentTurnEvents
+  const chainStart = retryChainStart(history, turn)
+  const chain = history.slice(Math.max(chainStart, 0))
+  const lastSuccess = chain.findLastIndex(prior => prior.type === 'assistant/message')
+  const chainRetries = chain.slice(lastSuccess + 1)
    .filter((prior): prior is SessionEvent<'llm/retry'> => prior.type === 'llm/retry')
-  if (priorRetries.some(prior => prior.data.step === step)) {
+  if (chainRetries.some(prior => prior.data.turn === turn && prior.data.step === step)) {
    fail(`llm/retry duplicates the retry record for turn ${turn}/step ${step}`)
  }
-  const priorRetry = priorRetries[0]
-  if (priorRetry !== undefined && retry <= priorRetry.data.retry) {
-    fail(`llm/retry retry ${retry} must increase after retry ${priorRetry.data.retry}`)
+  const expectedRetry = chainRetries.length + 1
+  if (retry !== expectedRetry) {
+    fail(`llm/retry retry ${retry} must equal retry-chain position ${expectedRetry}`)
  }
 }

--- a/packages/llm/llm-retry/tests/invariant.spec.ts
+++ b/packages/llm/llm-retry/tests/invariant.spec.ts
@@ -13,32 +13,35 @@ async function setup(): Promise<Context> {
  return ctx
 }

+const failure = { message: 'provider busy', code: 'RATE_LIMIT', status: 429 }
+
 function closeStep(ctx: Context, id: string, turn = 1, step = 1) {
  const session = ctx.sessions.create(SessionId(id))
-  session.append('turn/start', { turn, trigger: { kind: 'message', source: { kind: 'user' } } })
+  session.append('turn/start', {
+    turn,
+    trigger: turn === 1
+      ? { kind: 'message', source: { kind: 'user' } }
+      : { kind: 'retry' },
+  })
  session.append('step/start', { turn, step })
  session.append('step/end', { turn, step })
  return session
 }

-const failure = { message: 'provider busy', code: 'RATE_LIMIT', status: 429 }
-
 describe('llm-retry invariants', () => {
-  it('accepts increasing retry records for successive closed steps and ignores unrelated events', async () => {
+  it('accepts increasing retry schedules for successive failed turns', async () => {
    const ctx = await setup()
    const session = closeStep(ctx, 'retry-invariant-valid')
    expect(() => {
      session.append('llm/retry', {
        turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 500, failure,
      })
-      session.append('step/start', { turn: 1, step: 2 })
-      session.append('step/end', { turn: 1, step: 2 })
+      session.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, failure } })
+      session.append('turn/start', { turn: 2, trigger: { kind: 'retry' } })
+      session.append('step/start', { turn: 2, step: 1 })
+      session.append('step/end', { turn: 2, step: 1 })
      session.append('llm/retry', {
-        turn: 1, step: 2, retry: 2, maxRetries: 2, delayMs: 1_000, failure,
-      })
-      const zeroDelay = closeStep(ctx, 'retry-invariant-zero-delay')
-      zeroDelay.append('llm/retry', {
-        turn: 1, step: 1, retry: 1, maxRetries: 1, delayMs: 0, failure,
+        turn: 2, step: 1, retry: 2, maxRetries: 2, delayMs: 0, failure,
      })
    }).not.toThrow()
    expect(() => { ctx.emit('tools/change') }).not.toThrow()
@@ -60,7 +63,51 @@ describe('llm-retry invariants', () => {
    }).toThrow(message)
  })

-  it('rejects retry records outside the matching closed-step boundary', async () => {
+  it('rejects a retry record appended after its turn already closed', async () => {
+    const ctx = await setup()
+    const closed = closeStep(ctx, 'retry-invariant-closed-turn')
+    closed.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, failure } })
+    expect(() => {
+      closed.append('llm/retry', {
+        turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
+      })
+    }).toThrow(/inside an open turn/)
+  })
+
+  it('starts a fresh chain when the turn before a retry trigger did not fail structurally', async () => {
+    const ctx = await setup()
+    const session = closeStep(ctx, 'retry-invariant-completed-predecessor')
+    session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+    session.append('turn/start', { turn: 2, trigger: { kind: 'retry' } })
+    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, retry: 1, maxRetries: 2, delayMs: 1, failure,
+      })
+    }).not.toThrow()
+  })
+
+  it('walks the chain across non-boundary events and stops at an unmatched turn start', async () => {
+    const ctx = await setup()
+    // The failed predecessor's turn/start is outside this log prefix (e.g. a
+    // truncated replay): the chain walk must stop rather than loop or throw.
+    const session = ctx.sessions.create(SessionId('retry-invariant-unmatched-start'))
+    session.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, failure } })
+    // A durable non-boundary record between the turns exercises the walk over
+    // non-turn/end events.
+    session.append('todo/write', { todos: [] })
+    session.append('turn/start', { turn: 2, trigger: { kind: 'retry' } })
+    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, retry: 1, maxRetries: 2, delayMs: 1, failure,
+      })
+    }).not.toThrow()
+  })
+
+  it('requires an open turn and its latest closed step', async () => {
    const ctx = await setup()
    const absent = ctx.sessions.create(SessionId('retry-invariant-no-turn'))
    expect(() => {
@@ -85,31 +132,15 @@ describe('llm-retry invariants', () => {
      })
    }).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, retry: 1, maxRetries: 2, delayMs: 1, failure,
-      })
-    }).toThrow(/latest closed step is undefined/)
-
    const wrongStep = closeStep(ctx, 'retry-invariant-wrong-step')
    expect(() => {
      wrongStep.append('llm/retry', {
        turn: 1, step: 2, retry: 1, maxRetries: 2, delayMs: 1, failure,
      })
    }).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, retry: 1, maxRetries: 2, delayMs: 1, failure,
-      })
-    }).toThrow(/inside an open turn/)
  })

-  it('rejects duplicate and non-increasing retry records', async () => {
+  it('rejects duplicate and out-of-sequence retry schedules', async () => {
    const ctx = await setup()
    const duplicate = closeStep(ctx, 'retry-invariant-duplicate')
    duplicate.append('llm/retry', {
@@ -119,30 +150,70 @@ describe('llm-retry invariants', () => {
      duplicate.append('llm/retry', {
        turn: 1, step: 1, retry: 2, maxRetries: 3, delayMs: 1, failure,
      })
-    }).toThrow(/duplicates the retry record/)
+    }).toThrow(/duplicates/)

    const nonIncreasing = closeStep(ctx, 'retry-invariant-non-increasing')
    nonIncreasing.append('llm/retry', {
      turn: 1, step: 1, retry: 1, maxRetries: 3, delayMs: 1, failure,
    })
-    nonIncreasing.append('step/start', { turn: 1, step: 2 })
-    nonIncreasing.append('step/end', { turn: 1, step: 2 })
+    nonIncreasing.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, failure } })
+    nonIncreasing.append('turn/start', { turn: 2, trigger: { kind: 'retry' } })
+    nonIncreasing.append('step/start', { turn: 2, step: 1 })
+    nonIncreasing.append('step/end', { turn: 2, step: 1 })
    expect(() => {
      nonIncreasing.append('llm/retry', {
-        turn: 1, step: 2, retry: 1, maxRetries: 3, delayMs: 1, failure,
+        turn: 2, step: 1, retry: 1, maxRetries: 3, delayMs: 1, failure,
      })
-    }).toThrow(/must increase/)
+    }).toThrow(/retry-chain position 2/)
+  })
+
+  it('resets retry numbering after a completed chain', async () => {
+    const ctx = await setup()
+    const session = closeStep(ctx, 'retry-invariant-reset')
+    session.append('llm/retry', {
+      turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
+    })
+    session.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, failure } })
+    session.append('turn/start', { turn: 2, trigger: { kind: 'retry' } })
+    session.append('step/start', { turn: 2, step: 1 })
+    session.append('step/end', { turn: 2, step: 1 })
+    session.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
+    session.append('turn/start', {
+      turn: 3,
+      trigger: { kind: 'message', source: { kind: 'user' } },
+    })
+    session.append('step/start', { turn: 3, step: 1 })
+    session.append('step/end', { turn: 3, step: 1 })
+
+    expect(() => {
+      session.append('llm/retry', {
+        turn: 3, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
+      })
+    }).not.toThrow()
  })

  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, retry: 1, maxRetries: 2, delayMs: 1, failure,
    })
    await ctx.plugin(InvariantService)
    await expect(ctx.plugin(RetryInvariant)).rejects.toThrow(/inside an open turn/)
  })
+
+  it('accepts a valid mixed pre-existing history on late registration', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const session = ctx.sessions.create(SessionId('retry-invariant-late-valid'))
+    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 })
+    session.append('llm/retry', {
+      turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
+    })
+    await ctx.plugin(InvariantService)
+    await expect(ctx.plugin(RetryInvariant)).resolves.toBeDefined()
+  })
 })
--- a/packages/llm/llm-retry/tests/loader-composition.spec.ts
+++ b/packages/llm/llm-retry/tests/loader-composition.spec.ts
@@ -7,7 +7,6 @@ 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'
@@ -32,17 +31,6 @@ class TransientOnceAdapter extends LlmAdapter {
  }
 }

-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
@@ -113,9 +101,8 @@ describe('real Loader composition', () => {
    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.followup([{ type: 'text', text: 'recover' }])
-    await idle
+    agent.followup({ content: [{ type: 'text', text: 'recover' }], source: { kind: 'user' } })
+    await agent.whenIdle()

    expect(adapter.requests).toBe(2)
    expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(1)
--- a/packages/llm/llm-retry/tests/persistence.spec.ts
+++ b/packages/llm/llm-retry/tests/persistence.spec.ts
@@ -43,7 +43,14 @@ describe.each(['jsonl', 'sqlite'] as const)('%s retry-event persistence', (kind)
        delayMs: 750,
        failure: { message: 'provider busy', code: 'RATE_LIMIT', status: 429 },
      })
-      session.append('turn/end', { turn: 1, reason: { kind: 'aborted' } })
+      session.append('turn/end', {
+        turn: 1,
+        reason: {
+          kind: 'error',
+          step: 1,
+          failure: { message: 'provider busy', code: 'RATE_LIMIT', status: 429 },
+        },
+      })

      expect(session.deriveMessages()).toEqual([])
      await ctx.sessions.flush(session)
--- a/packages/llm/llm-retry/tests/retry.spec.ts
+++ b/packages/llm/llm-retry/tests/retry.spec.ts
@@ -8,9 +8,8 @@ import type { SessionEvent } from '@deepseek-ai/dsh-session'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import ToolRegistry, { defineContentToolFixture } from '@deepseek-ai/dsh-tools'
 import AgentRegistry from '@deepseek-ai/dsh-agent'
-import type { Agent, RequestErrorDecision } from '@deepseek-ai/dsh-agent'
+import type { Agent } 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>
@@ -51,6 +50,16 @@ function textResponse(text: string): StreamChunk[] {
  ]
 }

+function toolResponse(callId: string, name: string): StreamChunk[] {
+  const id = CallId(callId)
+  return [
+    { type: 'block-start', index: 0, blockType: 'tool-call' },
+    { type: 'tool-call-delta', index: 0, id, name, argumentsDelta: '{}' },
+    { type: 'block-end', index: 0, block: { type: 'tool-call', id, name, arguments: '{}' } },
+    { type: 'finish', reason: { kind: 'tool-calls' } },
+  ]
+}
+
 /**
 * A degenerate empty provider completion as an error finish chunk. Both
 * adapters emit this shape and the EMPTY_RESPONSE code (the field the policy
@@ -127,6 +136,23 @@ afterEach(async () => {
  context = undefined
 })

+describe('config validation', () => {
+  it.each([
+    [{ maxTransientRetries: 1.5 }, /maxTransientRetries must be a non-negative integer/],
+    [{ maxTransientRetries: -1 }, /maxTransientRetries must be a non-negative integer/],
+    [{ initialDelayMs: 0 }, /initialDelayMs must be a positive finite number/],
+    [{ initialDelayMs: Number.NaN }, /initialDelayMs must be a positive finite number/],
+    [{ maxDelayMs: 0 }, /maxDelayMs must be a positive finite number/],
+    [{ initialDelayMs: 600, maxDelayMs: 500 }, /initialDelayMs must be less than or equal to maxDelayMs/],
+    [{ jitterRatio: Number.NaN }, /jitterRatio must be between 0 and 1/],
+    [{ retryableCodes: [] }, /retryableCodes must not be empty/],
+    [{ retryableCodes: ['SERVER', ''] }, /retryableCodes must contain only non-empty strings/],
+    [{ retryableCodes: ['SERVER', 'SERVER'] }, /retryableCodes must not contain duplicates/],
+  ] satisfies [retry.Config, RegExp][])('rejects invalid config %j at load', (config, message) => {
+    expect(() => { retry.apply(new Context(), config) }).toThrow(message)
+  })
+})
+
 describe('bounded transient retry policy', () => {
  it('records the scheduled delay before opening a fresh request attempt', async () => {
    vi.useFakeTimers()
@@ -148,7 +174,7 @@ describe('bounded transient retry policy', () => {
      })
    })

-    agent.followup([{ type: 'text', text: 'go' }])
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
    const event = await scheduled

    expect(event.data).toEqual({
@@ -168,8 +194,8 @@ describe('bounded transient retry policy', () => {
    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.events.filter(item => item.type === 'step/start').map(item => item.data))
+      .toEqual([{ turn: 1, step: 1 }, { turn: 2, step: 1 }])
    expect(agent.session.deriveMessages().at(-1)).toEqual({
      role: 'assistant',
      content: [{ type: 'text', text: 'done' }],
@@ -190,7 +216,7 @@ describe('bounded transient retry policy', () => {
    const agent = context.agentLoop.create(SessionId('retry-empty-response'), { provider: 'mock', model: 'mock' })
    const scheduled = waitForRetry(context, agent, 1)

-    agent.followup([{ type: 'text', text: 'go' }])
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
    const event = await scheduled
    expect(event.data.failure).toEqual({
      message: 'model returned a completed response with no content',
@@ -202,8 +228,9 @@ describe('bounded transient retry policy', () => {
    await idle

    expect(adapter.requests).toHaveLength(2)
-    expect(agent.session.events.filter(event => event.type === 'assistant/message').map(event => event.data.step))
-      .toEqual([2])
+    expect(agent.session.events.filter(event => event.type === 'assistant/message')
+      .map(event => [event.data.turn, event.data.step]))
+      .toEqual([[2, 1]])
    expect(agent.session.deriveMessages().at(-1)).toMatchObject({
      role: 'assistant',
      content: [{ type: 'text', text: 'recovered' }],
@@ -230,18 +257,20 @@ describe('bounded transient retry policy', () => {
    const agent = context.agentLoop.create(SessionId('retry-partial'), { provider: 'mock', model: 'mock' })
    const scheduled = waitForRetry(context, agent, 1)

-    agent.followup([{ type: 'text', text: 'go' }])
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
    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,
+      event.type === 'assistant/chunk' && event.data.turn === 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.filter(event => event.type === 'assistant/message').map(event => ({
+      turn: event.data.turn,
+      step: event.data.step,
+    }))).toEqual([{ turn: 2, step: 1 }])
    expect(agent.session.events.some(event => event.type === 'tool/call')).toBe(false)
    expect(toolExecutions).toBe(0)
    expect(agent.session.deriveMessages().at(-1)).toMatchObject({
@@ -265,7 +294,7 @@ describe('bounded transient retry policy', () => {
    const agent = context.agentLoop.create(SessionId('retry-exhausted'), { provider: 'mock', model: 'mock' })
    const first = waitForRetry(context, agent, 1)

-    agent.followup([{ type: 'text', text: 'go' }])
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
    expect((await first).data.delayMs).toBe(450)

    const second = waitForRetry(context, agent, 2)
@@ -284,6 +313,73 @@ describe('bounded transient retry policy', () => {
    })
  })

+  it('resets the retry budget for a later message', async () => {
+    vi.useFakeTimers()
+    const adapter = new ScriptedAdapter([
+      new LlmError('first busy', 'SERVER'),
+      textResponse('first done'),
+      new LlmError('second busy', 'SERVER'),
+      textResponse('second done'),
+    ])
+    ;({ ctx: context } = await harness(adapter, { maxTransientRetries: 1 }))
+    const agent = context.agentLoop.create(SessionId('retry-reset'), { provider: 'mock', model: 'mock' })
+
+    const firstRetry = waitForRetry(context, agent, 1)
+    agent.followup({ content: [{ type: 'text', text: 'first' }], source: { kind: 'user' } })
+    await firstRetry
+    const firstIdle = waitForIdle(context, agent)
+    await vi.advanceTimersByTimeAsync(500)
+    await firstIdle
+
+    const secondRetry = waitForRetry(context, agent, 1)
+    agent.followup({ content: [{ type: 'text', text: 'second' }], source: { kind: 'user' } })
+    await secondRetry
+    const secondIdle = waitForIdle(context, agent)
+    await vi.advanceTimersByTimeAsync(500)
+    await secondIdle
+
+    expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data.retry))
+      .toEqual([1, 1])
+    expect(adapter.requests).toHaveLength(4)
+  })
+
+  it('resets the retry budget after a successful tool-call response within the same drain', async () => {
+    vi.useFakeTimers()
+    const adapter = new ScriptedAdapter([
+      new LlmError('first busy', 'SERVER'),
+      toolResponse('work-1', 'work'),
+      new LlmError('second busy', 'SERVER'),
+      textResponse('done'),
+    ])
+    ;({ ctx: context } = await harness(adapter, { maxTransientRetries: 1 }))
+    context.tools.register(defineContentToolFixture({
+      name: 'work',
+      description: 'continue into another model step',
+      parameters: {},
+      async execute() {
+        return [{ type: 'text', text: 'worked' }]
+      },
+    }))
+    const agent = context.agentLoop.create(SessionId('retry-reset-after-success'), {
+      provider: 'mock',
+      model: 'mock',
+    })
+
+    const firstRetry = waitForRetry(context, agent, 1)
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
+    await firstRetry
+    const secondRetry = waitForRetry(context, agent, 1)
+    await vi.advanceTimersByTimeAsync(500)
+    await secondRetry
+    const idle = waitForIdle(context, agent)
+    await vi.advanceTimersByTimeAsync(500)
+    await idle
+
+    expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data.retry))
+      .toEqual([1, 1])
+    expect(adapter.requests).toHaveLength(4)
+  })
+
  it('accepts the zero-delay lower jitter bound', async () => {
    vi.useFakeTimers()
    const adapter = new ScriptedAdapter([
@@ -298,7 +394,7 @@ describe('bounded transient retry policy', () => {
    const agent = context.agentLoop.create(SessionId('retry-zero-delay'), { provider: 'mock', model: 'mock' })
    const scheduled = waitForRetry(context, agent, 1)

-    agent.followup([{ type: 'text', text: 'go' }])
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
    expect((await scheduled).data.delayMs).toBe(0)

    const idle = waitForIdle(context, agent)
@@ -316,7 +412,7 @@ describe('bounded transient retry policy', () => {
    ;({ 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.followup([{ type: 'text', text: 'go' }])
+    acceptedAgent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
    expect((await scheduled).data.delayMs).toBe(2_000)
    const acceptedIdle = waitForIdle(context, acceptedAgent)
    await vi.advanceTimersByTimeAsync(2_000)
@@ -330,7 +426,7 @@ describe('bounded transient retry policy', () => {
    ;({ ctx: context } = await harness(rejected))
    const rejectedAgent = context.agentLoop.create(SessionId('retry-after-rejected'), { provider: 'mock', model: 'mock' })
    const rejectedIdle = waitForIdle(context, rejectedAgent)
-    rejectedAgent.followup([{ type: 'text', text: 'go' }])
+    rejectedAgent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
    await rejectedIdle
    expect(rejected.requests).toHaveLength(1)
    expect(rejectedAgent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
@@ -342,13 +438,132 @@ describe('bounded transient retry policy', () => {
    ;({ ctx: context } = await harness(adapter))
    const agent = context.agentLoop.create(SessionId('retry-auth'), { provider: 'mock', model: 'mock' })
    const idle = waitForIdle(context, agent)
-    agent.followup([{ type: 'text', text: 'go' }])
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
    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('keeps the consumed budget when an unowned session logs an assistant message', async () => {
+    vi.useFakeTimers()
+    const adapter = new ScriptedAdapter([
+      new LlmError('busy one', 'SERVER'),
+      new LlmError('busy two', 'SERVER'),
+    ])
+    ;({ ctx: context } = await harness(adapter, { maxTransientRetries: 1 }))
+    const agent = context.agentLoop.create(SessionId('retry-foreign-session'), { provider: 'mock', model: 'mock' })
+    const scheduled = waitForRetry(context, agent, 1)
+
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
+    await scheduled
+
+    // A session no agent owns completes a response; the agent's consecutive-
+    // failure sequence must not reset from that foreign success.
+    const foreign = context.sessions.create(SessionId('retry-foreign-session-other'))
+    foreign.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    foreign.append('step/start', { turn: 1, step: 1 })
+    foreign.append('assistant/message', {
+      turn: 1,
+      step: 1,
+      content: [{ type: 'text', text: 'foreign' }],
+      provenance: { provider: 'mock', model: 'mock' },
+    }, { surfaceOp: 'append' })
+
+    const idle = waitForIdle(context, agent)
+    await vi.advanceTimersByTimeAsync(500)
+    await idle
+
+    expect(adapter.requests).toHaveLength(2)
+    expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(1)
+    expect(agent.session.events.at(-1)).toMatchObject({
+      type: 'turn/end',
+      data: { reason: { kind: 'error', failure: { message: 'busy two', code: 'SERVER' } } },
+    })
+  })
+
+  it('drops a scheduled retry when cancellation lands between its durable record and its wait', async () => {
+    vi.useFakeTimers()
+    const adapter = new ScriptedAdapter([
+      new LlmError('busy', 'SERVER'),
+      textResponse('must not run'),
+    ])
+    ;({ ctx: context } = await harness(adapter))
+    const agent = context.agentLoop.create(SessionId('retry-cancel-at-record'), { provider: 'mock', model: 'mock' })
+    // The durable record commits synchronously before the cancellable wait; a
+    // user cancel observed at that exact point must skip the wait entirely.
+    const dispose = context.on('session/event', (session, event) => {
+      if (session === agent.session && event.type === 'llm/retry') {
+        dispose()
+        agent.cancel({ kind: 'user' })
+      }
+    })
+    const idle = waitForIdle(context, agent)
+
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
+    await idle
+    await vi.advanceTimersByTimeAsync(60_000)
+
+    expect(adapter.requests).toHaveLength(1)
+    expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(1)
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('schedules nothing when an upstream recovery listener already cancelled the turn', async () => {
+    vi.useFakeTimers()
+    const adapter = new ScriptedAdapter([
+      new LlmError('busy', 'SERVER'),
+      textResponse('must not run'),
+    ])
+    ;({ ctx: context } = await harness(adapter, {}, (ctx) => {
+      // Registered before the retry plugin, so it wraps the policy: it cancels
+      // the turn, then delegates into a policy that sees an aborted signal.
+      ctx.on('agent/request-error', (agent, _turn, _step, _error, _failure, _signal, next) => {
+        agent.cancel({ kind: 'user' })
+        return next()
+      })
+    }))
+    const agent = context.agentLoop.create(SessionId('retry-upstream-cancel'), { provider: 'mock', model: 'mock' })
+    const idle = waitForIdle(context, agent)
+
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
+    await idle
+    await vi.advanceTimersByTimeAsync(60_000)
+
+    expect(adapter.requests).toHaveLength(1)
+    expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('does nothing when its captured listener resumes after plugin disposal', async () => {
+    vi.useFakeTimers()
+    const adapter = new ScriptedAdapter([
+      new LlmError('busy', 'SERVER'),
+      textResponse('must not run'),
+    ])
+    const holder: { dispose?: () => Promise<void> } = {}
+    const mounted = await harness(adapter, {}, (ctx) => {
+      // An upstream listener captured in the same waterfall disposes the retry
+      // plugin before delegating; the stale downstream callback must bail.
+      ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _signal, next) => {
+        await holder.dispose?.()
+        return next()
+      })
+    })
+    context = mounted.ctx
+    holder.dispose = () => mounted.retryFiber.dispose()
+    const agent = context.agentLoop.create(SessionId('retry-stale-listener'), { provider: 'mock', model: 'mock' })
+    const idle = waitForIdle(context, agent)
+
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
+    await idle
+    await vi.advanceTimersByTimeAsync(60_000)
+
+    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([
@@ -359,10 +574,9 @@ describe('bounded transient retry policy', () => {
    context = mounted.ctx
    const agent = context.agentLoop.create(SessionId('retry-hmr'), { provider: 'mock', model: 'mock' })
    const scheduled = waitForRetry(context, agent, 1)
-    agent.followup([{ type: 'text', text: 'go' }])
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
    await scheduled
    const idle = waitForIdle(context, agent)
-
    await mounted.retryFiber.dispose()
    await idle
    await vi.advanceTimersByTimeAsync(60_000)
@@ -372,157 +586,4 @@ describe('bounded transient retry policy', () => {
    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.followup([{ 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.followup([{ 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.followup([{ type: 'text', text: 'go' }])
-    await scheduled
-    const idle = waitForIdle(context, agent)
-    agent.cancel({ kind: 'user' })
-    await idle
-
-    expect(adapter.requests).toHaveLength(1)
-    expect(agent.session.events.at(-1)).toMatchObject({
-      type: 'turn/end',
-      data: { reason: { kind: 'aborted' } },
-    })
-    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({ kind: 'user' })
-        return next()
-      })
-    }))
-    const agent = context.agentLoop.create(SessionId('retry-pre-cancel'), { provider: 'mock', model: 'mock' })
-    const idle = waitForIdle(context, agent)
-
-    agent.followup([{ 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' } },
-    })
-  })
-
-  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({ kind: 'user' })
-    })
-    const idle = waitForIdle(context, agent)
-
-    agent.followup([{ 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)
-  })
 })
--- a/packages/llm/llm-retry/tests/transport-recovery.spec.ts
+++ b/packages/llm/llm-retry/tests/transport-recovery.spec.ts
@@ -62,7 +62,7 @@ function waitForIdle(ctx: Context, agent: Agent): Promise<void> {

 function sendAndWait(ctx: Context, agent: Agent): Promise<void> {
  const idle = waitForIdle(ctx, agent)
-  agent.followup([{ type: 'text', text: 'recover through the provider boundary' }])
+  agent.followup({ content: [{ type: 'text', text: 'recover through the provider boundary' }], source: { kind: 'user' } })
  return idle
 }

@@ -102,8 +102,9 @@ describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {

    expect(server).toBeDefined()
    expect(server?.requests).toHaveLength(1)
-    expect(agent.session.events.filter(event => event.type === 'step/start').map(event => event.data.step))
-      .toEqual([1, 2])
+    expect(agent.session.events.filter(event => event.type === 'step/start')
+      .map(event => [event.data.turn, event.data.step]))
+      .toEqual([[1, 1], [2, 1]])
    expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data.failure.code))
      .toEqual(['TRANSPORT'])
    expect(finalAssistantText(agent)).toBe('connected after retry')
@@ -131,10 +132,11 @@ describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {
    expect(server.requests).toHaveLength(2)
    expect(server.requests[0]?.body).toEqual(server.requests[1]?.body)
    expect(agent.session.events.filter(event =>
-      event.type === 'assistant/chunk' && event.data.step === 1,
+      event.type === 'assistant/chunk' && event.data.turn === 1,
    )).toHaveLength(failedChunkCount)
-    expect(agent.session.events.filter(event => event.type === 'assistant/message').map(event => event.data.step))
-      .toEqual([2])
+    expect(agent.session.events.filter(event => event.type === 'assistant/message')
+      .map(event => [event.data.turn, event.data.step]))
+      .toEqual([[2, 1]])
    expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data.failure.code))
      .toEqual(['TRANSPORT'])
    expect(finalAssistantText(agent)).toBe('recovered response')
@@ -157,8 +159,9 @@ describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {
    expect(server.requests[0]?.body).toEqual(server.requests[1]?.body)
    expect(agent.session.events.filter(event => event.type === 'llm/retry').map(event => event.data.failure.code))
      .toEqual(['EMPTY_RESPONSE'])
-    expect(agent.session.events.filter(event => event.type === 'assistant/message').map(event => event.data.step))
-      .toEqual([2])
+    expect(agent.session.events.filter(event => event.type === 'assistant/message')
+      .map(event => [event.data.turn, event.data.step]))
+      .toEqual([[2, 1]])
    expect(agent.session.events.at(-1)).toMatchObject({
      type: 'turn/end',
      data: { reason: { kind: 'completed' } },
@@ -182,7 +185,7 @@ describe('bounded retry through the real DeepSeek HTTP/SSE adapter', () => {

    expect(server.requests).toHaveLength(1)
    expect(agent.session.events.filter(event =>
-      event.type === 'assistant/chunk' && event.data.step === 1,
+      event.type === 'assistant/chunk' && event.data.turn === 1,
    )).toHaveLength(2)
    expect(agent.session.events.some(event => event.type === 'assistant/message')).toBe(false)
    expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
--- a/packages/llm/llm/src/types.ts
+++ b/packages/llm/llm/src/types.ts
@@ -235,8 +235,7 @@ export interface GenerateOptions {
  /**
   * Ordered conversation messages, exactly as the provider sees them (after
   * the `system` slot). A loop-built request assembles them as
-   * `EpochHeader.messagePrefix` + the derived history (dsh-agent-loop); a
-   * hand-built one-shot passes any list.
+   * the derived history (dsh-agent-loop); a hand-built one-shot passes any list.
   */
  messages: Message[]
  /** System prompt text (adapters map to the provider's system slot). */
--- a/packages/llm/token-meter/src/index.ts
+++ b/packages/llm/token-meter/src/index.ts
@@ -378,7 +378,6 @@ export class TokenMeterService extends Service {
  private _estimateHeader(header: EpochHeader | undefined): number {
    if (header === undefined) return 0
    let tokens = 0
-    for (const message of header.messagePrefix ?? []) tokens += this.estimateMessage(message)
    if (header.system !== undefined) {
      tokens += Math.ceil(header.system.length / CHARS_PER_TOKEN) + ROLE_OVERHEAD
    }
--- a/packages/llm/token-meter/tests/token-meter.spec.ts
+++ b/packages/llm/token-meter/tests/token-meter.spec.ts
@@ -183,7 +183,7 @@ describe('TokenMeterService pricing', () => {
    expect(snapshot.nodes).toHaveLength(1)
  })

-  it('prices header, prefix, tools, and surface when no reusable usage exists', () => {
+  it('prices header, tools, and surface when no reusable usage exists', () => {
    const service = meter()
    const session = new Session(SessionId('heuristic'))
    session.append('user/message', {
@@ -192,7 +192,6 @@ describe('TokenMeterService pricing', () => {
    }, { surfaceOp: 'append' })
    appendHeader(session, header('deepseek-v4-flash', {
      system: 'system',
-      messagePrefix: [textMessage('prefix')],
      tools: [{ name: 'read', description: 'read', parameters: { type: 'object' } }],
    }))
    const result = service.measure(session)
@@ -354,10 +353,6 @@ describe('replay anchors and surface folds', () => {
      ...anchoredHeader,
      config: { ...anchoredHeader.config, temperature: 0.2 },
    }).baseline.kind).toBe('estimated')
-    expect(service.measure(session, {
-      ...anchoredHeader,
-      messagePrefix: [textMessage('prefix')],
-    }).baseline.kind).toBe('estimated')
    expect(service.measure(session, {
      ...anchoredHeader,
      tools: [{ name: 'read', description: 'read', parameters: { type: 'object' } }],