Merge origin/master into worktree/explicit-turn-signal

packages/core/agent-loop/src/loop.ts

@@ -1,16 +1,16 @@
 /**
  * Drives one agent across queued durable turns. Turn failures are contained so
  * later work can run; the session log, not this driver, owns conversation state.
- * See docs/rfc/implemented/architecture/2026-06-18-agent-lifecycle-and-ownership-seams.md.
+ * See .agents/notes/implemented/architecture/2026-06-18-agent-lifecycle-and-ownership-seams.md.
  * @module dsh-agent-loop/loop
  */
 
 import type { Context } from 'cordis'
-import { isDeepStrictEqual } from 'node:util'
 import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
-import { assertNever, BlockAssembler, HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
+import { isDeepStrictEqual } from 'node:util'
+import { BlockAssembler, HarnessError, assertNever, deepFreeze, errorChain, isLlmAdapterFailure } from '@deepseek-ai/dsh-llm'
 import { agentEvents, agentInterruptReasonOf, assembleContextFor } from '@deepseek-ai/dsh-agent'
-import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
+import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision, RequestError, RequestErrorDecision } from '@deepseek-ai/dsh-agent'
 import { canonicalHeader } from '@deepseek-ai/dsh-session'
 import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
 import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
@@ -19,28 +19,32 @@ import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
 import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
 import type {} from '@deepseek-ai/dsh-tools'
 import { executeToolCalls } from './tool-calls.ts'
-import type { ReactLoopAgent } from './agent.ts'
 import type { Inbox } from './inbox.ts'
 import type { TurnCancellation } from './cancellation.ts'
 
-/** An Error with an optional machine-readable code (e.g., from LlmError or a throwing plugin). */
-type CodedError = Error & { code?: string }
-
 /** Normalize thrown values while preserving an existing error code. */
-function toError(error: unknown): CodedError {
+function toError(error: unknown): RequestError {
   return error instanceof Error ? error : new HarnessError(String(error), 'UNKNOWN', { cause: error })
 }
 
+/** Distinguishes final model-request failures from failures in later step processing. */
+class TerminalModelRequestFailure extends Error {
+  constructor(readonly requestError: RequestError) {
+    super(requestError.message, { cause: requestError })
+    this.name = 'TerminalModelRequestFailure'
+  }
+}
+
 /** Convert terminal failure finishes into step errors; unknown extensible finishes remain successful. */
-function finishError(finish: FinishReason): CodedError | undefined {
+function finishError(finish: FinishReason): RequestError | undefined {
   switch (finish.kind) {
     case 'error': {
-      const error: CodedError = new Error(finish.message)
+      const error: RequestError = new Error(finish.message)
       if (finish.code !== undefined) error.code = finish.code
       return error
     }
     case 'aborted': {
-      const error: CodedError = new Error('model stream aborted')
+      const error: RequestError = new Error('model stream aborted')
       error.code = 'ABORTED'
       return error
     }
@@ -53,9 +57,12 @@ function finishError(finish: FinishReason): CodedError | undefined {
 /**
  * Build the `{ message, code? }` part of an error payload, omitting the
  * `code` key entirely when absent (exactOptionalPropertyTypes-correct).
+ * The durable message renders the full cause chain: `turn/end` is the single
+ * durable record of an in-turn failure, so a wrapper message alone (e.g.
+ * `fetch failed`) would lose the diagnosis the session log exists to keep.
  */
-function errorData(err: CodedError): { message: string; code?: string } {
-  return { message: err.message, ...typeof err.code === 'string' ? { code: err.code } : {} }
+function errorData(err: RequestError): { message: string; code?: string } {
+  return { message: errorChain(err), ...typeof err.code === 'string' ? { code: err.code } : {} }
 }
 
 /** Map a successful max-token finish onto the turn reason; other successful finishes add nothing. */
@@ -88,10 +95,10 @@ function interruptionTurnEndReason(handle: LoopHandle, signal: AbortSignal): Tur
     case 'user':
     case 'parent':
       return { kind: 'aborted' }
-    /* v8 ignore next 2 -- the private holder requests disposed only after lifecycle state flips, which returns above */
+    /* v8 ignore next 2 -- the private holder requests disposed only after lifecycle state flips, which returns above. */
     case 'disposed':
       return { kind: 'disposed' }
-    /* v8 ignore next 2 -- AgentInterruptReason is closed and the public helper filters unsupported reasons */
+    /* v8 ignore next 2 -- AgentInterruptReason is closed and the public helper filters unsupported reasons. */
     default:
       return assertNever(reason, 'AgentInterruptReason')
   }
@@ -115,20 +122,25 @@ export interface LoopHandle {
   isPreRunCancelled(): boolean
   /** Clear the cause-less pre-run marker without affecting replacement work. */
   clearPreRunCancel(): void
-  /** Settle idle waiters when pre-running cancellation skips a turn, without emitting `agent/status`. */
+  /** Settle idle waiters before pre-running cancellation publishes idle. */
   settleIdle(): void
   /** Run an active tool-call batch, accepting post-tool context into the FIFO drained before settlement. */
   readonly withToolBatch: <T>(run: (acceptContext: (context: HookContext) => void) => Promise<T>) => Promise<T>
 }
 
 /**
- * Drive queued batches as durable turns until disposal. Plugin failures end the
- * current turn without terminating the driver.
- * @param ctx - the plugin context the loop reaches events (agent/…, session/flush) and services (systemPrompt, llm, tools) through.
- * @param agent - the agent this invocation drives for its whole lifetime (its inbox, session, and options).
+ * Drive queued messages as independent durable turns until disposal. Plugin
+ * failures end the current turn without terminating the driver. The caller
+ * establishes the `ctx.agents.withInitiator()` boundary before entry; package-private
+ * orchestration recovers that exact Agent and captures its Session locally.
+ * @param ctx - the plugin context the loop reaches its initiating Agent,
+ * events (agent/…, session/flush), and services (systemPrompt, llm, tools)
+ * through.
  * @param handle - the bridge to status, turn cancellation ownership, disposal, and pre-run cancellation state.
+ * @throws when no initiating Agent is active.
  */
-export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle): Promise<void> {
+export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
+  const agent = ctx.agents.requireInitiator()
   // Per-instance prefix and request-header state; conversation history remains in the session log.
   const transmission = createTransmissionLog()
 
@@ -137,29 +149,42 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
   const events = agentEvents(ctx, agent)
 
   while (!handle.isDisposed()) {
-    await handle.inbox.waitForQueued(handle.disposed)
-    if (handle.isDisposed()) break
-
-    // Cancellation between wake and `running` skips only the cancelled work;
-    // a replacement prompt still runs and owns the eventual idle transition.
+    // An idle listener can enqueue and cancel replacement work before the next
+    // wait is installed. Consume that empty marker before parking the driver.
     if (handle.isPreRunCancelled()) {
       handle.clearPreRunCancel()
       if (!handle.inbox.hasQueued) {
         handle.settleIdle()
+        handle.setStatus('idle')
+        continue
+      }
+    }
+
+    await handle.inbox.waitForQueued(handle.disposed)
+    if (handle.isDisposed()) break
+
+    // Cancellation between wake and `running` skips only the cancelled work;
+    // a replacement prompt still runs before the eventual idle transition.
+    if (handle.isPreRunCancelled()) {
+      handle.clearPreRunCancel()
+      if (!handle.inbox.hasQueued) {
+        // Settle before publishing idle: the already-idle path has no status
+        // transition, while an idle listener can register waiters for new work.
+        handle.settleIdle()
+        handle.setStatus('idle')
         continue
       }
     }
 
     let cancellation = handle.installTurnCancellation()
     handle.setStatus('running')
-
     if (handle.isDisposed()) {
       handle.clearTurnCancellation(cancellation)
       break
     }
 
-    // A synchronous running listener may cancel old work and enqueue a
-    // replacement. The replacement receives a fresh, non-aborted turn owner.
+    // A synchronous `running` listener can cancel before `runTurn`; balance the
+    // status only when no replacement prompt was queued by that listener.
     if (cancellation.signal.aborted) {
       handle.clearTurnCancellation(cancellation)
       if (!handle.inbox.hasQueued) {
@@ -173,11 +198,11 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
     const turn = lastTurnNumber(session) + 1
     let terminalStopped = false
     try {
-      terminalStopped = await runTurn(ctx, events, agent, handle, turn, transmission, cancellation.signal)
+      terminalStopped = await runTurn(ctx, events, handle, turn, transmission, cancellation.signal)
     } catch (error: unknown) {
       // Pre-turn failure has no durable boundary to close; report it without appending outside a turn.
       const err = toError(error)
-      ctx.logger.warn(`agent "${agent.id}": turn ${turn} failed before it started: ${err.message}`)
+      ctx.logger.warn(`agent "${agent.id}": turn ${turn} failed before it started: ${errorChain(err)}`)
       try {
         events.emit('agent/error', turn, 0, err)
       } catch { /* contained: a throwing agent/error listener must not kill the driver */ }
@@ -195,20 +220,28 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
 }
 
 async function runTurn(
-  ctx: Context, events: AgentEventDispatch, agent: ReactLoopAgent, handle: LoopHandle, turn: number, transmission: TransmissionLog,
+  ctx: Context, events: AgentEventDispatch, handle: LoopHandle, turn: number, transmission: TransmissionLog,
   signal: AbortSignal,
 ): Promise<boolean> {
+  const agent = ctx.agents.requireInitiator()
   const { session } = agent
+  const drainSteering = (): boolean => {
+    const messages = handle.inbox.drainSteering()
+    for (const message of messages) {
+      session.append('steering/message', { turn, content: message.content, source: message.source }, { surfaceOp: 'append' })
+    }
+    return messages.length > 0
+  }
 
-  // Drain before opening the turn, but append only after `turn/start`.
-  const queued = handle.inbox.drainQueued()
-  const first = queued[0]
+  // Claim one queued message before opening its turn, but append it only after `turn/start`.
+  const message = handle.inbox.dequeueQueued()
   /* v8 ignore next 3 -- invariant guard: runLoop only calls runTurn when hasQueued */
-  if (!first) throw new Error('runTurn invariant violated: no queued message at turn start')
-  const trigger: TurnTrigger = { kind: 'message', source: first.source }
+  if (!message) throw new Error('runTurn invariant violated: no queued message at turn start')
+  const trigger: TurnTrigger = { kind: 'message', source: message.source }
 
   let reason: TurnEndReason = { kind: 'completed' }
   let step = 0
+  let requestRetryAttempt = 0
   let stepOpen = false
   let errorReported = false
   let terminalStopped = false
@@ -221,7 +254,7 @@ async function runTurn(
   }
 
   // Record the durable turn failure once and contain the live error notification.
-  const failTurn = (err: CodedError): void => {
+  const failTurn = (err: RequestError): void => {
     if (errorReported) return
     errorReported = true
     reason = { kind: 'error', step, ...errorData(err) }
@@ -244,71 +277,51 @@ async function runTurn(
     // veto leaves no turn/start in the log and therefore owes no turn/end.
     session.append('turn/start', { turn, trigger })
     interruptionCheckpoint(signal)
-    // Each drained queued message runs the `agent/prompt-submit` waterfall before
-    // it becomes a `user/message` — a hook can rewrite the prompt or block it.
+    // The claimed message runs the `agent/prompt-submit` waterfall before it
+    // becomes a `user/message` — a hook can rewrite the prompt or block it.
     // Recorded INSIDE the turn (after turn/start) so every event is turn-enclosed;
     // turn/end is now owed, so a throwing prompt-submit listener (the waterfall
     // throws) is caught below and the turn still closes.
-    let anyAllowed = false
-    // Seeded with a floor (only observable if the batch were empty, which
-    // runTurn never allows — it is called with ≥1 queued message); each `block`
-    // decision carries a required `reason` and overwrites it, so a fully-blocked
-    // batch always reports the last vetoing reason.
-    let lastBlockReason = 'prompt blocked by hook'
-    for (const message of queued) {
-      const decision = await events.waterfall(
-        'agent/prompt-submit', message.content, message.source, signal,
-        () => Promise.resolve<PromptDecision>({ kind: 'allow' }),
-      )
-      interruptionCheckpoint(signal)
-      if (decision.kind === 'block') {
-        lastBlockReason = decision.reason
-        // Record the veto durably: `PromptDecision.reason` is the durable record
-        // of why a prompt was blocked, but a fully-blocked batch's `rejected`
-        // turn/end only preserves the LAST reason, and a MIXED batch (this prompt
-        // blocked, another allowed) does not end `rejected` at all — so without
-        // this append a blocked prompt would vanish from the log whenever any
-        // sibling prompt is allowed. `prompt/blocked` sits in the open turn in
-        // place of the `user/message` this prompt would have become.
-        session.append('prompt/blocked', { content: message.content, source: message.source, reason: decision.reason })
-        continue
-      }
-      anyAllowed = true
+    const promptDecision = await events.waterfall(
+      'agent/prompt-submit', message.content, message.source, signal,
+      () => Promise.resolve<PromptDecision>({ kind: 'allow' }),
+    )
+    interruptionCheckpoint(signal)
+    if (promptDecision.kind === 'block') {
+      session.append('prompt/blocked', { content: message.content, source: message.source, reason: promptDecision.reason })
+      reason = { kind: 'rejected', reason: promptDecision.reason }
+    } else {
       // `allow.content` REPLACES the prompt bytes (a rewrite); absent keeps them.
-      const content = decision.content ?? message.content
+      const content = promptDecision.content ?? message.content
       session.append('user/message', { content, source: message.source }, { surfaceOp: 'append' })
       // Every `allow.additionalContexts` entry is a separate context/message the
       // next request also sees. The turn is open, so inject() appends each one
-      // into THIS turn without flattening provenance, framing, or metadata.
-      for (const context of decision.additionalContexts ?? []) {
+      // into THIS turn without flattening provenance or metadata.
+      for (const context of promptDecision.additionalContexts ?? []) {
         agent.inject(context.content, {
           source: context.source,
-          ...context.envelope !== undefined ? { envelope: context.envelope } : {},
           ...context.meta !== undefined ? { meta: context.meta } : {},
         })
       }
     }
 
     while (true) {
-      // A fully blocked batch closes its zero-step turn as rejected.
-      if (!anyAllowed) {
-        reason = { kind: 'rejected', reason: lastBlockReason }
-        break
-      }
+      // A blocked prompt closes its zero-step turn as rejected.
+      if (promptDecision.kind === 'block') break
       step += 1
 
       // Steering from the previous round's continuation listeners joins before
       // the request.
-      drainSteering(agent, handle.inbox, turn)
+      drainSteering()
 
-      // Assemble once before pre-step so pressure checks and the request share the same prompt.
+      // Assemble once before pre-step so listener work and the request share one prompt value.
       const assembly = await ctx.systemPrompt.assemble(assembleContextFor(agent, signal))
       interruptionCheckpoint(signal)
       const fullSystemPrompt = renderPrompt(assembly)
 
-      // Compose the request-only prefix once per loop instance before pressure
-      // checks. It precedes all derived history and is recorded only in the
-      // request header, not as session history.
+      // Compose the request-only prefix once per loop instance before the first
+      // request boundary. It precedes all derived history and is recorded only
+      // in the request header, not as session history.
       if (transmission.sessionPrefix === undefined) {
         const emptyPrefix: Message[] = deepFreeze([])
         const composed = await events.waterfall(
@@ -320,8 +333,8 @@ async function runTurn(
         transmission.sessionPrefix = deepFreeze(structuredClone(composed))
       }
 
-      // Await surface mutations outside the step; pressure checks receive the pending prefix.
-      await events.serial('agent/pre-step', turn, step, fullSystemPrompt, transmission.sessionPrefix, signal)
+      // Await surface mutations outside the step before snapshotting history.
+      await events.serial('agent/pre-step', turn, step, signal)
       interruptionCheckpoint(signal)
 
       // Snapshot the exact log prefix before step/start: the reconstruction
@@ -333,15 +346,67 @@ async function runTurn(
       // pre-commit veto throws before this assignment; post-commit observers
       // are contained inside Session.append().
       stepOpen = true
+
       // A synchronous step/start observer can cancel after the step opened.
       interruptionCheckpoint(signal)
 
-      let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
+      let stepOutcome:
+        | { hadToolCalls: boolean; finish: FinishReason }
+        | { requestError: RequestError }
+        | { error: RequestError }
       try {
         stepOutcome = await runStep(
-          ctx, events, agent, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, signal)
+          ctx, events, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, signal)
       } catch (error: unknown) {
-        stepOutcome = { error: toError(error) }
+        if (error instanceof TerminalModelRequestFailure) {
+          stepOutcome = { requestError: error.requestError }
+        } else {
+          stepOutcome = { error: toError(error) }
+        }
+      }
+
+      if ('requestError' in stepOutcome) {
+        // Recovery observes a balanced failed step and the original provider
+        // error while the failed step's signal remains the active owner.
+        closeStep()
+        const interrupted = interruptionTurnEndReason(handle, signal)
+        if (interrupted !== undefined) {
+          reason = interrupted
+          break
+        }
+
+        const defaultDecision: RequestErrorDecision = { action: 'fail' }
+        let recoveryDecision: RequestErrorDecision = defaultDecision
+        try {
+          recoveryDecision = await events.waterfall(
+            'agent/request-error', turn, step, stepOutcome.requestError,
+            requestRetryAttempt, signal,
+            () => Promise.resolve(defaultDecision),
+          )
+        } catch (recoveryError: unknown) {
+          ctx.logger.warn(
+            `agent "${agent.id}": request recovery failed at turn ${turn}, step ${step}: ${errorChain(recoveryError)}`,
+          )
+        }
+        // Cancellation and disposal always win over either a recovery decision
+        // or a recovery-listener failure.
+        const recoveryInterrupted = interruptionTurnEndReason(handle, signal)
+        if (recoveryInterrupted !== undefined) {
+          reason = recoveryInterrupted
+          break
+        }
+        switch (recoveryDecision.action) {
+          case 'retry':
+            requestRetryAttempt += 1
+            continue
+          case 'fail':
+            failTurn(stepOutcome.requestError)
+            break
+          /* v8 ignore next -- closed-union exhaustiveness guard */
+          default:
+            assertNever(recoveryDecision, 'agent request-error decision')
+        }
+        break
       }
 
       if ('error' in stepOutcome) {
@@ -350,21 +415,43 @@ async function runTurn(
         // starts a fresh turn instead of being silently consumed.
         closeStep()
         const { error } = stepOutcome
-        const interruption = interruptionTurnEndReason(handle, signal)
-        if (interruption === undefined) failTurn(error)
-        else reason = interruption
+        const interrupted = interruptionTurnEndReason(handle, signal)
+        if (interrupted === undefined) failTurn(error)
+        else reason = interrupted
         break
       }
 
+      requestRetryAttempt = 0
+
       // Preserve max-token completion unless a later disposal, abort, or error wins.
       const stepReason = stepFinishReason(stepOutcome.finish)
       if (stepReason) reason = stepReason
 
       // Steering that arrived during streaming/tool execution.
-      const steered = drainSteering(agent, handle.inbox, turn)
+      const steered = drainSteering()
+
+      try {
+        await events.serial('agent/post-step', turn, step, signal)
+      } catch (error: unknown) {
+        stepOutcome = { error: toError(error) }
+      }
+
+      if ('error' in stepOutcome) {
+        closeStep()
+        const interrupted = interruptionTurnEndReason(handle, signal)
+        if (interrupted === undefined) failTurn(stepOutcome.error)
+        else reason = interrupted
+        break
+      }
+
+      const postStepInterrupted = interruptionTurnEndReason(handle, signal)
+      if (postStepInterrupted !== undefined) {
+        reason = postStepInterrupted
+        closeStep()
+        break
+      }
 
       closeStep()
-      interruptionCheckpoint(signal)
 
       const defaultDecision: ContinuationDecision = { action: stepOutcome.hadToolCalls || steered ? 'continue' : 'stop' }
       let decision: ContinuationDecision
@@ -375,9 +462,9 @@ async function runTurn(
         )
         interruptionCheckpoint(signal)
       } catch (error: unknown) {
-        const interruption = interruptionTurnEndReason(handle, signal)
-        if (interruption === undefined) failTurn(toError(error))
-        else reason = interruption
+        const interrupted = interruptionTurnEndReason(handle, signal)
+        if (interrupted === undefined) failTurn(toError(error))
+        else reason = interrupted
         break
       }
 
@@ -397,9 +484,11 @@ async function runTurn(
         interruptionCheckpoint(signal)
         terminalStop = stop !== undefined
       } catch (error: unknown) {
-        const interruption = interruptionTurnEndReason(handle, signal)
-        if (interruption === undefined) failTurn(toError(error))
-        else reason = interruption
+        // A broken terminal policy is an ordinary continuation failure: fail
+        // this turn closed while leaving the driver alive for later turns.
+        const interrupted = interruptionTurnEndReason(handle, signal)
+        if (interrupted === undefined) failTurn(toError(error))
+        else reason = interrupted
         break
       }
       if (terminalStop) {
@@ -409,11 +498,7 @@ async function runTurn(
         shouldContinue = false
       }
 
-      if (!shouldContinue || handle.isDisposed()) {
-        /* v8 ignore next -- disposal during continuation-decision window is a narrow race; error-path disposal is covered elsewhere */
-        if (handle.isDisposed()) reason = { kind: 'disposed' }
-        break
-      }
+      if (!shouldContinue) break
     }
 
     // Normal / inline-error loop exit: close the turn.
@@ -423,9 +508,9 @@ async function runTurn(
     const turnStartLogged = session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
     if (!turnStartLogged) throw error
     closeStep()
-    const interruption = interruptionTurnEndReason(handle, signal)
-    if (interruption === undefined) failTurn(toError(error))
-    else reason = interruption
+    const interrupted = interruptionTurnEndReason(handle, signal)
+    if (interrupted === undefined) failTurn(toError(error))
+    else reason = interrupted
     closeTurn()
   }
 
@@ -435,7 +520,7 @@ async function runTurn(
   } catch (error: unknown) {
     // The turn is closed, so report the failed flush live rather than append outside a turn.
     const err = toError(error)
-    ctx.logger.warn(`agent "${agent.id}": session/flush failed at turn ${turn}: ${err.message}`)
+    ctx.logger.warn(`agent "${agent.id}": session/flush failed at turn ${turn}: ${errorChain(err)}`)
     try {
       events.emit('agent/error', turn, step, err)
     } catch {
@@ -445,15 +530,6 @@ async function runTurn(
   return terminalStopped
 }
 
-/** Drain the steering queue into the session. Returns whether any arrived. */
-function drainSteering(agent: ReactLoopAgent, inbox: Inbox, turn: number): boolean {
-  const messages = inbox.drainSteering()
-  for (const message of messages) {
-    agent.session.append('steering/message', { turn, content: message.content, source: message.source }, { surfaceOp: 'append' })
-  }
-  return messages.length > 0
-}
-
 /**
  * Run one committed step: transform call config, log the request header, build
  * the request from the cached prefix plus the step-boundary snapshot, stream and
@@ -463,7 +539,6 @@ function drainSteering(agent: ReactLoopAgent, inbox: Inbox, turn: number): boole
 async function runStep(
   ctx: Context,
   events: AgentEventDispatch,
-  agent: ReactLoopAgent,
   handle: LoopHandle,
   turn: number,
   step: number,
@@ -473,6 +548,7 @@ async function runStep(
   transmission: TransmissionLog,
   signal: AbortSignal,
 ): Promise<{ hadToolCalls: boolean; finish: FinishReason }> {
+  const agent = ctx.agents.requireInitiator()
   const { session, options } = agent
 
   // Seed the first request from agent options and later requests from the logged header;
@@ -483,7 +559,9 @@ async function runStep(
     : { provider: options.provider ?? '', model: options.model ?? '' }))
 
   // Listener replacements are recorded in the request header before dispatch.
-  const config = await events.waterfall('agent/request', turn, step, seedConfig, signal, () => Promise.resolve(seedConfig))
+  const config = await events.waterfall(
+    'agent/request', turn, step, seedConfig, signal, () => Promise.resolve(seedConfig),
+  )
   interruptionCheckpoint(signal)
   if (!config.provider || !config.model) {
     throw new Error(`agent "${agent.id}" has no provider/model: set AgentOptions.provider and AgentOptions.model or supply both via the agent/request waterfall`)
@@ -518,27 +596,67 @@ async function runStep(
   // --- Model call (streaming-first; raw chunks are the replay record) ---
   const assembler = new BlockAssembler()
   const chunkSeqs: number[] = []
-  for await (const chunk of ctx.llm.stream(request)) {
-    interruptionCheckpoint(signal)
-    const chunkEvent = session.append('assistant/chunk', { turn, step, chunk })
-    chunkSeqs.push(chunkEvent.seq)
-    assembler.push(chunk)
+  const stream = ctx.llm.stream(request)
+  try {
+    for await (const chunk of stream) {
+      interruptionCheckpoint(signal)
+      const chunkEvent = session.append('assistant/chunk', { turn, step, chunk })
+      chunkSeqs.push(chunkEvent.seq)
+      assembler.push(chunk)
+    }
+  } catch (error: unknown) {
+    if (isLlmAdapterFailure(stream, error)) throw new TerminalModelRequestFailure(error)
+    throw error
   }
   interruptionCheckpoint(signal)
 
   // Normalize failure finish chunks into the same path as thrown stream errors.
   const stepError = finishError(assembler.finish)
-  if (stepError) throw stepError
+  if (stepError) throw new TerminalModelRequestFailure(stepError)
+
+  const recordAssistantMessage = (
+    assembledContent: ContentBlock[],
+    message: Message,
+    preserveReplayState = true,
+  ): void => {
+    session.append(
+      'assistant/message',
+      {
+        turn,
+        step,
+        content: message.content,
+        provenance: assistantProvenance(
+          header.config,
+          assembler.replayState,
+          preserveReplayState && isDeepStrictEqual(message.content, assembledContent),
+        ),
+        ...assembler.usage === undefined ? {} : { usage: assembler.usage },
+      },
+      { surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
+    )
+  }
+
+  // A rejected result still records the successful provider call without retaining rejected output.
+  const processStepResult = async (assembledContent: ContentBlock[], message: Message): Promise<Message> => {
+    try {
+      const processed = await events.waterfall(
+        'agent/step-result', turn, step, message, signal, () => Promise.resolve(message),
+      )
+      interruptionCheckpoint(signal)
+      return processed
+    } catch (error: unknown) {
+      recordAssistantMessage(assembledContent, { ...message, content: [] }, false)
+      throw error
+    }
+  }
 
   if (assembler.finish.kind === 'max-tokens') {
     const assembled = assembler.message()
     const assembledContent = structuredClone(assembled.content)
     let message: Message = withoutToolCalls(assembled)
-    message = withoutToolCalls(await processStepResult(
-      events, session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs, signal,
-    ))
+    message = withoutToolCalls(await processStepResult(assembledContent, message))
     // Preserve usage even when max-token truncation produced no content.
-    recordAssistantMessage(session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs)
+    recordAssistantMessage(assembledContent, message)
     return { hadToolCalls: false, finish: assembler.finish }
   }
 
@@ -546,89 +664,23 @@ async function runStep(
   const assembled = assembler.message()
   const assembledContent = structuredClone(assembled.content)
   let message: Message = assembled
-  message = await processStepResult(
-    events, session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs, signal,
-  )
+  message = await processStepResult(assembledContent, message)
 
   // Every successful call records its completion anchor, including explicit
   // empty chunk provenance for a contentless, usage-less provider response.
-  recordAssistantMessage(session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs)
+  recordAssistantMessage(assembledContent, message)
 
   // Dispatch may overlap; policy, durable results, and result context stay model-ordered.
   const toolCalls = message.content.filter(block => block.type === 'tool-call')
   if (toolCalls.length === 0) return { hadToolCalls: false, finish: assembler.finish }
   return handle.withToolBatch(async (acceptContext) => {
     await executeToolCalls(
-      ctx, agent, turn, step, toolCalls, signal, handle.maxParallelToolCalls, acceptContext,
+      ctx, turn, step, toolCalls, signal, handle.maxParallelToolCalls, acceptContext,
     )
     return { hadToolCalls: true, finish: assembler.finish }
   })
 }
 
-/** Preserve successful-call accounting without retaining output that result processing rejected. */
-async function processStepResult(
-  events: AgentEventDispatch,
-  session: Session,
-  turn: number,
-  step: number,
-  config: LlmCallConfig,
-  assembledContent: ContentBlock[],
-  message: Message,
-  assembler: BlockAssembler,
-  chunkSeqs: number[],
-  signal: AbortSignal,
-): Promise<Message> {
-  try {
-    const processed = await events.waterfall(
-      'agent/step-result', turn, step, message, signal, () => Promise.resolve(message),
-    )
-    interruptionCheckpoint(signal)
-    return processed
-  } catch (error: unknown) {
-    recordAssistantMessage(
-      session,
-      turn,
-      step,
-      config,
-      assembledContent,
-      { ...message, content: [] },
-      assembler,
-      chunkSeqs,
-      false,
-    )
-    throw error
-  }
-}
-
-/** Record one content-or-usage assistant message with replay-safe provenance. */
-function recordAssistantMessage(
-  session: Session,
-  turn: number,
-  step: number,
-  config: LlmCallConfig,
-  assembledContent: ContentBlock[],
-  message: Message,
-  assembler: BlockAssembler,
-  chunkSeqs: number[],
-  preserveReplayState = true,
-): void {
-  session.append(
-    'assistant/message',
-    {
-      turn,
-      step,
-      content: message.content,
-      provenance: assistantProvenance(
-        config,
-        assembler.replayState,
-        preserveReplayState && isDeepStrictEqual(message.content, assembledContent),
-      ),
-      ...assembler.usage === undefined ? {} : { usage: assembler.usage },
-    },
-    { surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
-  )
-}
-
 /** Build durable assistant provenance, dropping replay state after any content rewrite. */
 function assistantProvenance(config: LlmCallConfig, replayState: unknown, contentUnchanged: boolean): NonNullable<Message['provenance']> {
   return {