Merge remote-tracking branch 'origin/master' into codex/enforce-tool-cancellation

# Conflicts:
#	packages/core/agent-loop/tests/tool-calls.spec.ts

packages/core/agent-loop/README.md

@@ -54,7 +54,7 @@ The driver owns one agent for its lifetime and runs inside `ctx.agents.withIniti
 
 Every provider call that reaches a successful finish appends exactly one `assistant/message` completion anchor, including content-less calls and `max-tokens` finishes. A successful `agent/step-result` stores its transformed content; a rejected result records empty content before the original failure continues. The anchor retains exact chunk provenance (`[]` for a stream with no chunks) and usage when available, while empty content stays out of derived message history.
 
-Plugin failure ends the current turn, not the loop. Cancellation clears pending work and aborts the current step without leaking to the next prompt. Terminal continuation stops remain authoritative through turn close and durability flush.
+Plugin failure ends the current turn, not the loop. Only final adapter dispatch/iteration failures and terminal in-band error or aborted finishes enter `agent/request-error`; middleware, result processing, tools, and `agent/post-step` remain ordinary turn failures. Recovery observes a closed failed step, and a retry rebuilds the request from the durable log in a new numbered step. Cancellation clears pending work and aborts the current step without leaking to the next prompt; undispatched model tool calls receive synthetic `tool/call` and aborted result pairs. Terminal continuation stops remain authoritative through turn close and durability flush.
 
 Within a step, exclusive calls form barriers; parallel-safe calls use a bounded rolling pool and are reclassified before start. Only dispatch/body overlaps. Policy, durable results, and result context remain model-ordered. Abort stops new calls, drains started results, then drains accepted batch context before the turn closes through the normal abort path.
 
@@ -62,7 +62,7 @@ Within a step, exclusive calls form barriers; parallel-safe calls use a bounded
 
 Everything that goes beyond "call the model, run the tools, repeat" belongs to plugins listening on the event taxonomy:
 - Hooks and policy: the relevant `agent/*` checkpoints plus the guarded `tools/pre-execute` → `tools/execute` → `tools/post-execute` → `tools/result` pipeline; exact signatures and modes live in the [generated event catalog](../../../docs/cordis-catalog/events.md)
-- Compaction: `agent/pre-step`
+- Compaction: pressure on `agent/post-step`; canonical context overflow on `agent/request-error`
 - Sandbox, permission, plan mode: `tools/pre-execute` for extensible deny/ask, `tools.guard()` for monotonic owner policy, `tools/post-execute` for result decisions, and `tools/result` for final observation
 - Sub-agents: implemented outside the loop as `ctx.subagents` providers; in-process providers use `ctx.agents.create()` and owned `AgentHandle` teardown, while generic [`ctx.tasks`](../../tasks/tasks/) plus [`dsh-tool-subagent`](../../subagent/tool-subagent/) own background collection.
 - Persistence: `session/event` + `session/flush`
@@ -82,6 +82,12 @@ Everything that goes beyond "call the model, run the tools, repeat" belongs to p
 
 **Token effect**: Input grows with every surface message until a compaction replacement shadows older nodes; a multi-step tool turn resends the accumulated prefix and history each step.
 
+### Undispatched calls after cancellation
+
+**What the model sees**: If a later request replays an aborted step, each tool call that cancellation prevented from dispatching has the error result text `Error: tool call skipped because the step was aborted before execution`.
+
+**Token effect**: One fixed error result per skipped call remains in history until compaction shadows it.
+
 ## Known Limitations and Deferred Work
 
 - **Classification is unary** — calls whose safety depends on comparing siblings or resources must remain exclusive ([rationale](../../../docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md)).

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

@@ -8,9 +8,9 @@
 import type { Context } from 'cordis'
 import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
 import { isDeepStrictEqual } from 'node:util'
-import { BlockAssembler, HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
+import { BlockAssembler, HarnessError, assertNever, deepFreeze, isLlmAdapterFailure } from '@deepseek-ai/dsh-llm'
 import { agentEvents, 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'
@@ -21,24 +21,29 @@ import type {} from '@deepseek-ai/dsh-tools'
 import { executeToolCalls } from './tool-calls.ts'
 import type { Inbox } from './inbox.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
     }
@@ -52,7 +57,7 @@ function finishError(finish: FinishReason): CodedError | undefined {
  * Build the `{ message, code? }` part of an error payload, omitting the
  * `code` key entirely when absent (exactOptionalPropertyTypes-correct).
  */
-function errorData(err: CodedError): { message: string; code?: string } {
+function errorData(err: RequestError): { message: string; code?: string } {
   return { message: err.message, ...typeof err.code === 'string' ? { code: err.code } : {} }
 }
 
@@ -186,6 +191,7 @@ async function runTurn(
 
   let reason: TurnEndReason = { kind: 'completed' }
   let step = 0
+  let requestRetryAttempt = 0
   let stepOpen = false
   let errorReported = false
   let terminalStopped = false
@@ -198,7 +204,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) }
@@ -284,7 +290,7 @@ async function runTurn(
       const abort = new AbortController()
       handle.setAbort(abort)
 
-      // 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))
       const fullSystemPrompt = renderPrompt(assembly)
 
@@ -295,9 +301,9 @@ async function runTurn(
         break
       }
 
-      // 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(
@@ -314,8 +320,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, abort.signal)
+      // Await surface mutations outside the step before snapshotting history.
+      await events.serial('agent/pre-step', turn, step, abort.signal)
 
       // Interruption landing during the pre-step seam: do not open an empty step.
       if (handle.isCancelled() || handle.isDisposed()) {
@@ -345,14 +351,69 @@ async function runTurn(
         break
       }
 
-      let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
+      let stepOutcome:
+        | { hadToolCalls: boolean; finish: FinishReason }
+        | { requestError: RequestError }
+        | { error: RequestError }
       try {
         stepOutcome = await runStep(
           ctx, events, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
       } catch (error: unknown) {
-        stepOutcome = { error: toError(error) }
-      } finally {
+        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()
+        if (handle.isDisposed() || abort.signal.aborted) {
+          handle.setAbort(undefined)
+          reason = handle.isDisposed()
+            ? { kind: 'disposed' }
+            : { kind: 'aborted', reason: String(abort.signal.reason) }
+          break
+        }
+
+        const defaultDecision: RequestErrorDecision = { action: 'fail' }
+        let recoveryDecision: RequestErrorDecision = defaultDecision
+        try {
+          recoveryDecision = await events.waterfall(
+            'agent/request-error', turn, step, stepOutcome.requestError,
+            requestRetryAttempt, abort.signal,
+            () => Promise.resolve(defaultDecision),
+          )
+        } catch (recoveryError: unknown) {
+          ctx.logger.warn(
+            `agent "${agent.id}": request recovery failed at turn ${turn}, step ${step}: ${toError(recoveryError).message}`,
+          )
+        }
         handle.setAbort(undefined)
+
+        // Cancellation and disposal always win over either a recovery decision
+        // or a recovery-listener failure.
+        // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+        if (handle.isDisposed() || abort.signal.aborted) {
+          reason = handle.isDisposed()
+            ? { kind: 'disposed' }
+            : { kind: 'aborted', reason: String(abort.signal.reason) }
+          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) {
@@ -360,7 +421,9 @@ async function runTurn(
         // runLoop re-enqueues it as a queued message, so an abort-then-steer
         // starts a fresh turn instead of being silently consumed.
         closeStep()
+        handle.setAbort(undefined)
         const { error } = stepOutcome
+        /* v8 ignore next -- narrow race: disposal while non-request step work throws. */
         if (handle.isDisposed()) {
           reason = { kind: 'disposed' }
         } else if (abort.signal.aborted) {
@@ -372,6 +435,8 @@ async function runTurn(
         break
       }
 
+      requestRetryAttempt = 0
+
       // Preserve max-token completion unless a later disposal, abort, or error wins.
       const stepReason = stepFinishReason(stepOutcome.finish)
       if (stepReason) reason = stepReason
@@ -379,7 +444,38 @@ async function runTurn(
       // Steering that arrived during streaming/tool execution.
       const steered = drainSteering()
 
+      try {
+        await events.serial('agent/post-step', turn, step, abort.signal)
+      } catch (error: unknown) {
+        stepOutcome = { error: toError(error) }
+      }
+
+      if ('error' in stepOutcome) {
+        closeStep()
+        handle.setAbort(undefined)
+        /* v8 ignore next -- narrow race: disposal while a post-step listener throws. */
+        if (handle.isDisposed()) {
+          reason = { kind: 'disposed' }
+        } else if (abort.signal.aborted) {
+          /* v8 ignore next -- signal.reason always set by cancellation or disposal. */
+          reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') }
+        } else {
+          failTurn(stepOutcome.error)
+        }
+        break
+      }
+
+      if (handle.isDisposed() || abort.signal.aborted) {
+        reason = handle.isDisposed()
+          ? { kind: 'disposed' }
+          : { kind: 'aborted', reason: String(abort.signal.reason) }
+        closeStep()
+        handle.setAbort(undefined)
+        break
+      }
+
       closeStep()
+      handle.setAbort(undefined)
 
       const defaultDecision: ContinuationDecision = { action: stepOutcome.hadToolCalls || steered ? 'continue' : 'stop' }
       let decision: ContinuationDecision
@@ -529,17 +625,23 @@ 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)) {
-    /* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
-    if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
-    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) {
+      /* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
+      if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
+      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
   }
 
   // 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[],

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

@@ -4,14 +4,15 @@
  * Dispatch may overlap, while policy, results, and result context remain
  * model-ordered. Abort stops replenishment and drains started calls.
  *
- * Each started call records `tool/call`; `tool/result` commits in model order,
- * preserving derived history when audit events interleave with earlier results.
+ * Each advertised call records a balanced `tool/call`/`tool/result` pair. Calls
+ * skipped after abort receive synthetic error results so replay stays valid.
  * @module dsh-agent-loop/tool-calls
  */
 
 import type { Context } from 'cordis'
 import { assertNever, type ToolCallBlock } from '@deepseek-ai/dsh-llm'
 import type { HookContext } from '@deepseek-ai/dsh-agent'
+import type { Session } from '@deepseek-ai/dsh-session'
 import { TOOL_REGISTRY_SCHEDULER, type ToolExecutionInput, type ToolExecutionMode, type ToolExecutionResult, type ToolRunContext } from '@deepseek-ai/dsh-tools'
 
 /** One tool call after argument parsing, ready to schedule. */
@@ -27,10 +28,17 @@ interface Slot {
   needsPost: boolean
 }
 
+/** One scheduler group outcome, including a drained cancellation. */
+interface GroupOutcome {
+  consumed: number
+  aborted: boolean
+}
+
 /**
  * Schedule one assistant step's tool calls by their live concurrency mode.
- * Started calls receive ordered results. Abort drains them and rethrows after
- * accepting their context into the batch FIFO owned by the caller.
+ * Started calls receive ordered results. Abort drains them, records synthetic
+ * results for unstarted calls, and returns with the signal still aborted after
+ * accepting started-call context into the batch FIFO owned by the caller.
  * The committed step's AgentLoop driver boundary supplies the initiating Agent
  * that becomes each explicit {@link ToolExecutionInput.agent}.
  *
@@ -52,6 +60,7 @@ export async function executeToolCalls(
   acceptContext: (context: HookContext) => void,
 ): Promise<void> {
   const agent = ctx.agents.requireInitiator()
+  const { session } = agent
 
   // Inputs are distinct because tools/execute wrappers may replace `exec.signal`.
   const planned: PlannedCall[] = toolCalls.map(block => ({
@@ -72,7 +81,14 @@ export async function executeToolCalls(
     const first = planned[next]!
     const mode = ctx.tools.executionMode(first.exec).kind
     const group = mode === 'parallel' ? planned.slice(next) : [first]
-    next += await runGroup(ctx, turn, step, group, mode, signal, maxParallel, acceptContext)
+    const outcome = await runGroup(
+      ctx, turn, step, group, mode, signal, maxParallel, acceptContext,
+    )
+    next += outcome.consumed
+    if (outcome.aborted) {
+      for (const call of planned.slice(next)) appendSkippedToolCall(session, turn, step, call.block)
+      return
+    }
   }
 }
 
@@ -90,7 +106,8 @@ function parseArguments(raw: string): unknown {
  * before start; an exclusive reclassification waits for the current pool to
  * drain and remains for the caller's next barrier. Results and contexts commit
  * in model order. Abort stops starts, drains and commits started calls, accepts
- * their contexts into the owning batch, and throws.
+ * their contexts into the owning batch, records results for skipped calls, and
+ * returns an aborted outcome.
  */
 async function runGroup(
   ctx: Context,
@@ -101,33 +118,8 @@ async function runGroup(
   signal: AbortSignal,
   maxParallel: number,
   acceptContext: (context: HookContext) => void,
-): Promise<number> {
+): Promise<GroupOutcome> {
   const { session } = ctx.agents.requireInitiator()
-  const appendToolCall = (block: ToolCallBlock): number => {
-    return session.append('tool/call', {
-      turn,
-      step,
-      callId: block.id,
-      name: block.name,
-      arguments: block.arguments,
-    }).seq
-  }
-  const appendToolResult = (block: ToolCallBlock, result: ToolExecutionResult, callSeq: number): void => {
-    session.append('tool/result', {
-      turn,
-      step,
-      // Correlation stays with the loop's authoritative model-transcript call id;
-      // registry results deliberately do not duplicate it.
-      callId: block.id,
-      content: result.content,
-      isError: result.isError,
-      ...result.error ? { error: result.error } : {},
-      // Persist presentation payloads so UI bridges reproduce result cards on replay.
-      ...result.meta !== undefined ? { meta: result.meta } : {},
-    }, { surfaceOp: 'append', sourceEventSeqs: [callSeq] })
-  }
-  /* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
-  if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
   const slots: (Slot | undefined)[] = group.map(() => undefined)
   // Started slots retain their tool/call seq for result provenance.
   const callSeqs: number[] = group.map(() => -1)
@@ -146,7 +138,7 @@ async function runGroup(
         ? await ctx.tools[TOOL_REGISTRY_SCHEDULER].finalize(slot.exec, slot.result)
         : ctx.tools[TOOL_REGISTRY_SCHEDULER].finish(slot.exec, slot.result)
       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
-      appendToolResult(call!.block, result, callSeqs[committed]!)
+      appendToolResult(session, turn, step, call!.block, result, callSeqs[committed]!)
       for (const context of result.additionalContexts ?? []) acceptContext(context)
       committed++
     }
@@ -157,7 +149,7 @@ async function runGroup(
   const startCall = async (index: number): Promise<void> => {
     // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
     const call = group[index]!
-    callSeqs[index] = appendToolCall(call.block)
+    callSeqs[index] = appendToolCall(session, turn, step, call.block)
     started++
     const prepared = await ctx.tools[TOOL_REGISTRY_SCHEDULER].prepare(call.exec)
     switch (prepared.kind) {
@@ -205,17 +197,57 @@ async function runGroup(
     inFlight.delete(settledIndex)
     await commitReady()
     // Abort may arrive while a tool or ordered commit awaits.
-    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+
     if (signal.aborted) aborted = true
     await fillPool()
   }
 
   if (aborted) {
-    // Started calls and accepted context settle before the turn records the abort.
-    /* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
-    throw new Error(String(signal.reason ?? 'aborted'))
+    // Started calls and accepted context settle first; every remaining model
+    // call then receives an ordered synthetic result before the turn aborts.
+    for (const call of group.slice(started)) appendSkippedToolCall(session, turn, step, call.block)
+    return { consumed: group.length, aborted: true }
   }
   /* v8 ignore next -- unreachable: a non-aborted group commits every started call */
   if (committed !== started) throw new Error('tool-call scheduler: uncommitted settled calls')
-  return started
+  return { consumed: started, aborted: false }
+}
+
+/** Append the durable call/result pair for a model call skipped after cancellation. */
+function appendSkippedToolCall(session: Session, turn: number, step: number, block: ToolCallBlock): void {
+  const callSeq = appendToolCall(session, turn, step, block)
+  appendToolResult(session, turn, step, block, {
+    content: [{ type: 'text', text: 'Error: tool call skipped because the step was aborted before execution' }],
+    isError: true,
+    error: { name: 'AbortError', code: 'ABORTED' },
+  }, callSeq)
+}
+
+/** Append a started call and return its provenance sequence. */
+function appendToolCall(session: Session, turn: number, step: number, block: ToolCallBlock): number {
+  const event = session.append('tool/call', { turn, step, callId: block.id, name: block.name, arguments: block.arguments })
+  return event.seq
+}
+
+/** Append a model-ordered result linked to its call event. */
+function appendToolResult(
+  session: Session,
+  turn: number,
+  step: number,
+  block: ToolCallBlock,
+  result: ToolExecutionResult,
+  callSeq: number,
+): void {
+  session.append('tool/result', {
+    turn, step,
+    // Correlation stays with the loop's authoritative model-transcript call id;
+    // registry results deliberately do not duplicate it.
+    callId: block.id,
+    content: result.content,
+    isError: result.isError,
+    ...result.error ? { error: result.error } : {},
+    // The tool's private presentation payload (e.g. a result-time diff),
+    // persisted so a UI bridge reproduces the card on replay.
+    ...result.meta !== undefined ? { meta: result.meta } : {},
+  }, { surfaceOp: 'append', sourceEventSeqs: [callSeq] })
 }

packages/core/agent-loop/tests/cancel.spec.ts

@@ -12,11 +12,10 @@ import { Context } from 'cordis'
 import LlmService, { type Message } from '@deepseek-ai/dsh-llm'
 import SessionStore, { SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
-import ToolRegistry from '@deepseek-ai/dsh-tools'
+import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
 import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
-
 import AgentLoop from '@deepseek-ai/dsh-agent-loop'
-import { MockAdapter, textResponse } from './mock-adapter.ts'
+import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
 
 function driverDone(agent: Agent): Promise<void> {
   return (agent as Agent & { done: Promise<void> }).done
@@ -144,6 +143,59 @@ describe('Agent.cancel()', () => {
     expect(reasons).toEqual([{ kind: 'aborted', reason: 'cancelled' }])
   })
 
+  it('cancel from an assistant/message observer skips execution but balances replay', async () => {
+    const adapter = new MockAdapter([
+      toolCallResponse('c1', 'danger', {}),
+      textResponse('recovered after cancellation'),
+    ])
+    const ctx = await harness(adapter)
+    let executions = 0
+    ctx.tools.register(defineTool({
+      name: 'danger',
+      description: 'must not run after cancellation',
+      parameters: {},
+      async execute() {
+        executions += 1
+        return [{ type: 'text', text: 'ran' }]
+      },
+    }))
+    const agent = ctx.agentLoop.create(SessionId('cancel-after-assistant-message'), { provider: 'mock', model: 'mock' })
+    const dispose = ctx.on('session/event', (session, event) => {
+      if (session === agent.session && event.type === 'assistant/message') {
+        agent.cancel('cancelled after assistant message')
+      }
+    })
+
+    const reasons: TurnEndReason[] = []
+    ctx.on('session/event', (_session, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+    dispose()
+
+    expect(executions).toBe(0)
+    expect(reasons).toEqual([{ kind: 'aborted', reason: 'cancelled after assistant message' }])
+    const call = agent.session.events.find(event => event.type === 'tool/call')
+    const result = agent.session.events.find(event => event.type === 'tool/result')
+    expect(call?.type === 'tool/call' ? call.data.callId : undefined).toBe('c1')
+    expect(result?.type === 'tool/result' ? result.data : undefined).toMatchObject({
+      callId: 'c1',
+      isError: true,
+      error: { name: 'AbortError', code: 'ABORTED' },
+    })
+
+    send(agent, 'continue safely')
+    await waitForIdle(ctx, agent)
+    const replayedResult = adapter.requests[1]!.messages
+      .flatMap(message => message.content)
+      .find(block => block.type === 'tool-result')
+    expect(replayedResult).toMatchObject({ toolCallId: 'c1', isError: true })
+    expect(reasons).toEqual([
+      { kind: 'aborted', reason: 'cancelled after assistant message' },
+      { kind: 'completed' },
+    ])
+  })
+
   it('a prompt sent AFTER a cancelled turn settles runs normally (marker reset)', async () => {
     const adapter = new MockAdapter(['hang', textResponse('second reply')])
     const ctx = await harness(adapter)

packages/core/agent-loop/tests/contract-regressions.spec.ts

@@ -204,7 +204,7 @@ describe('successful provider completion survives agent/step-result failure', ()
 })
 
 describe('abort during tool execution ends the turn', () => {
-  it('aborting the in-flight step inside a tool prevents both remaining tools and the next model step', async () => {
+  it('balances an aborted tool batch through context, steering, and post-step before closing', async () => {
     const adapter = new MockAdapter([
       // model asks for two tool calls in one step
       [
@@ -223,16 +223,24 @@ describe('abort during tool execution ends the turn', () => {
       name: 'aborter',
       description: '',
       parameters: {},
-      async execute() {
+      async execute(_args, exec) {
         executed.push('aborter')
-        // Fire the in-flight step's AbortController directly (the loop registers
-        // it on the agent). This is the bare step-abort path — distinct from
-        // cancel(), which would also clear the inbox; here the subject is the
-        // loop's response to its running step being aborted mid-tool.
+        exec.agent?.steer(
+          [{ type: 'text', text: 'steering before abort' }],
+          { source: { kind: 'plugin', plugin: 'abort-test' } },
+        )
+        // Exercise bare step abort without `cancel()` clearing queued work.
         ;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('user interrupt')
         return [{ type: 'text', text: 'done' }]
       },
     }))
+    ctx.on('tools/post-execute', async exec => ({
+      kind: 'accept',
+      additionalContexts: [{
+        content: [{ type: 'text', text: `context for ${exec.callId}` }],
+        source: { kind: 'plugin', plugin: 'abort-test' },
+      }],
+    }))
     ctx.tools.register(defineTool({
       name: 'second',
       description: '',
@@ -244,14 +252,69 @@ describe('abort during tool execution ends the turn', () => {
     }))
 
     const reasons: TurnEndReason[] = []
-    ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
+    const order: string[] = []
+    ctx.on('session/event', (session, event) => {
+      if (session !== agent.session) return
+      switch (event.type) {
+        case 'assistant/message': order.push('assistant/message'); break
+        case 'tool/call': order.push(`tool/call:${event.data.callId}`); break
+        case 'tool/result': {
+          const outcome = event.data.error?.code === 'ABORTED' ? 'aborted' : 'completed'
+          order.push(`tool/result:${event.data.callId}:${outcome}`)
+          break
+        }
+        case 'context/message': order.push('context/message'); break
+        case 'steering/message': order.push('steering/message'); break
+        case 'step/end': order.push('step/end'); break
+        case 'turn/end': {
+          reasons.push(event.data.reason)
+          order.push(`turn/end:${event.data.reason.kind}`)
+          break
+        }
+      }
+    })
+    let postSteps = 0
+    ctx.on('agent/post-step', (subject, turn, step, signal) => {
+      if (subject !== agent) return
+      postSteps += 1
+      expect({ turn, step, aborted: signal.aborted }).toEqual({ turn: 1, step: 1, aborted: true })
+      order.push('agent/post-step')
+    })
 
     send(agent, 'go')
     await waitForIdle(ctx, agent)
 
-    expect(executed).toEqual(['aborter'])           // second tool never ran
-    expect(adapter.requests).toHaveLength(1)        // no follow-up model call
+    expect(executed).toEqual(['aborter'])
+    expect(adapter.requests).toHaveLength(1)
+    expect(postSteps).toBe(1)
+    expect(order).toEqual([
+      'assistant/message',
+      'tool/call:c1',
+      'tool/result:c1:aborted',
+      'tool/call:c2',
+      'tool/result:c2:aborted',
+      'context/message',
+      'steering/message',
+      'agent/post-step',
+      'step/end',
+      'turn/end:aborted',
+    ])
     expect(reasons).toEqual([{ kind: 'aborted', reason: 'user interrupt' }])
+    const calls = agent.session.events.filter(event => event.type === 'tool/call')
+    const results = agent.session.events.filter(event => event.type === 'tool/result')
+    expect(calls.map(event => event.data.callId)).toEqual([CallId('c1'), CallId('c2')])
+    expect(results).toHaveLength(2)
+    expect(results[0]!.data).toMatchObject({
+      callId: CallId('c1'),
+      content: [{ type: 'text', text: 'Error: tool call aborted' }],
+      isError: true,
+      error: { name: 'AbortError', code: 'ABORTED' },
+    })
+    expect(results[1]!.data).toMatchObject({
+      callId: CallId('c2'),
+      isError: true,
+      error: { name: 'AbortError', code: 'ABORTED' },
+    })
   })
 
   it('records context accepted before a tool-step abort in the same turn', async () => {

packages/core/agent-loop/tests/interception.spec.ts

@@ -115,14 +115,11 @@ describe('agent/prompt-submit', () => {
     expect(ctxMsg?.type === 'context/message' && ctxMsg.data.source).toEqual({ kind: 'plugin', plugin: 'test' })
     expect(ctxMsg?.type === 'context/message' && ctxMsg.data.envelope).toBe('raw')
     expect(ctxMsg?.type === 'context/message' && ctxMsg.data.meta).toEqual(meta)
-    // both the prompt and the injected context reach the model
     const sent = JSON.stringify(adapter.requests[0]!.messages)
     expect(sent).toContain('extra ctx')
   })
 
-  it('a prompt-submit rewrite + additionalContexts is VISIBLE to the agent/pre-step seam (merged ordering)', async () => {
-    // Prompt rewrites and injected context land before `agent/pre-step`, so a
-    // compaction listener measures the current surface before the single derive.
+  it('runs pre-step after prompt rewrites and injected context become durable', async () => {
     const adapter = new MockAdapter([textResponse('ok')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -134,8 +131,6 @@ describe('agent/prompt-submit', () => {
         additionalContexts: [{ content: [{ type: 'text', text: 'injected ctx' }], source: { kind: 'plugin', plugin: 'test' } }],
       }))
 
-    // The pre-step seam (where compaction lives) derives the surface it would act
-    // on. Capture what it sees on the first step.
     let preStepDerived: string | undefined
     ctx.on('agent/pre-step', (subject, _turn, step) => {
       if (subject === agent && step === 1) preStepDerived = JSON.stringify(subject.session.deriveMessages())
@@ -144,8 +139,6 @@ describe('agent/prompt-submit', () => {
     send(agent, 'ORIGINAL prompt')
     await waitForIdle(ctx, agent)
 
-    // The pre-step seam ran and saw BOTH the rewrite (not the original) and the
-    // injected context — i.e. the prompt-submit effects landed before it.
     expect(preStepDerived).toBeDefined()
     expect(preStepDerived).toContain('REWRITTEN prompt')
     expect(preStepDerived).toContain('injected ctx')
@@ -379,7 +372,7 @@ describe('agent/session-prefix', () => {
     expect(agent.session.deriveMessages()[0]).toEqual({ role: 'user', content: [{ type: 'text', text: 'go' }] })
   })
 
-  it('composes before the first pre-step and hands the prefix to the seam (pressure gates see the real value)', async () => {
+  it('composes before the first pre-step and records the prefix on the request header', async () => {
     const adapter = new MockAdapter([textResponse('ok')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -390,20 +383,15 @@ describe('agent/session-prefix', () => {
       order.push('compose')
       return [reminder, ...await next()]
     })
-    const seen: (readonly Message[])[] = []
-    ctx.on('agent/pre-step', (_agent, _turn, _step, _system, sessionPrefix) => {
+    ctx.on('agent/pre-step', () => {
       order.push('pre-step')
-      seen.push(sessionPrefix)
     })
 
     send(agent, 'hi')
     await waitForIdle(ctx, agent)
 
-    // Composition precedes the pre-step seam, and the seam receives THIS
-    // instance's composed prefix — a token-pressure gate (compaction) counts
-    // what the request will actually carry, never a stale logged prefix.
     expect(order).toEqual(['compose', 'pre-step'])
-    expect(seen[0]).toEqual([reminder])
+    expect(agent.session.requestHeader()?.messagePrefix).toEqual([reminder])
   })
 
   it('the canonical prepend pattern composes contributions in registration order', async () => {

packages/core/agent-loop/tests/loop.spec.ts

@@ -577,10 +577,6 @@ describe('agent loop', () => {
   })
 
   it('agent/pre-step fires once per step before the step is opened', async () => {
-    // Two steps (a tool call, then a final text turn) → two model calls → two
-    // pre-step fires, each carrying the assembled full system prompt, BEFORE
-    // the step is opened and its request is derived (the request the adapter
-    // sees reflects any surface state at fire time).
     const adapter = new MockAdapter([
       toolCallResponse('c1', 'echo', {}, 'calling echo'),
       textResponse('done'),
@@ -592,21 +588,19 @@ describe('agent loop', () => {
     }))
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
 
-    const fires: { turn: number; step: number; fullSystemPrompt: string }[] = []
-    ctx.on('agent/pre-step', (subject, turn, step, fullSystemPrompt) => {
-      if (subject === agent) fires.push({ turn, step, fullSystemPrompt })
+    const fires: { turn: number; step: number; signal: AbortSignal }[] = []
+    ctx.on('agent/pre-step', (subject, turn, step, signal) => {
+      if (subject === agent) fires.push({ turn, step, signal })
     })
 
     send(agent, 'go')
     await waitForIdle(ctx, agent)
 
-    // One fire per step, in order, each with the assembled system prompt
-    // (here just the loop's own harness-identity section — no persona set).
-    const HARNESS = 'You are an AI agent powered by the DeepSeek Harness SDK.'
-    expect(fires).toEqual([
-      { turn: 1, step: 1, fullSystemPrompt: HARNESS },
-      { turn: 1, step: 2, fullSystemPrompt: HARNESS },
+    expect(fires.map(({ turn, step }) => ({ turn, step }))).toEqual([
+      { turn: 1, step: 1 },
+      { turn: 1, step: 2 },
     ])
+    expect(fires.every(({ signal }) => signal instanceof AbortSignal)).toBe(true)
   })
 
   it('agent/pre-step fires BEFORE the step it precedes opens (events land outside the step)', async () => {

packages/core/agent-loop/tests/request-recovery.spec.ts

@@ -0,0 +1,516 @@
+import { describe, expect, it, vi } from 'vitest'
+import { Context } from 'cordis'
+import LlmService, {
+  CallId,
+  CONTEXT_WINDOW_EXCEEDED_CODE,
+  LlmAdapter,
+  LlmError,
+} from '@deepseek-ai/dsh-llm'
+import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
+import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
+import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
+import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
+import type { PostToolDecision } from '@deepseek-ai/dsh-tools'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
+import type { Agent } from '@deepseek-ai/dsh-agent'
+import AgentLoop from '@deepseek-ai/dsh-agent-loop'
+import { maxTokensResponse, textResponse, toolCallResponse } from './mock-adapter.ts'
+
+class FailureScriptAdapter extends LlmAdapter {
+  requests: GenerateOptions[] = []
+
+  constructor(private readonly entries: (Error | StreamChunk[])[]) {
+    super()
+  }
+
+  async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
+    this.requests.push(options)
+    const entry = this.entries.shift()
+    if (entry === undefined) throw new Error('failure script exhausted')
+    if (entry instanceof Error) throw entry
+    yield* entry
+  }
+}
+
+class IteratorConstructionFailureAdapter extends LlmAdapter {
+  stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
+    return {
+      [Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
+        throw new LlmError('iterator construction failed', 'ITERATOR_CONSTRUCTION')
+      },
+    }
+  }
+}
+
+class SynchronousDispatchFailureAdapter extends LlmAdapter {
+  constructor(private readonly error: Error) {
+    super()
+  }
+
+  stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
+    throw this.error
+  }
+}
+
+class IteratorResultGetterFailureAdapter extends LlmAdapter {
+  constructor(
+    private readonly field: 'done' | 'value',
+    private readonly error: Error,
+  ) {
+    super()
+  }
+
+  stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
+    const result = this.field === 'done' ? {} : { done: false }
+    Object.defineProperty(result, this.field, { get: () => { throw this.error } })
+    return {
+      [Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
+        return { next: () => Promise.resolve(result as unknown as IteratorResult<StreamChunk>) }
+      },
+    }
+  }
+}
+
+const streamListenerFailureCases: readonly [string, (ctx: Context) => void][] = [
+  ['synchronous listener throw', (ctx) => {
+    ctx.on('llm/stream', () => { throw new Error('synchronous stream listener failed') })
+  }],
+  ['invalid listener iterable', (ctx) => {
+    ctx.on('llm/stream', () => ({}) as AsyncIterable<StreamChunk>)
+  }],
+  ['listener wrapper iteration failure', (ctx) => {
+    ctx.on('llm/stream', (_options, next) => (async function * () {
+      for await (const chunk of next()) {
+        yield chunk
+        throw new Error('stream listener wrapper failed')
+      }
+    })())
+  }],
+]
+
+async function harness(adapter?: LlmAdapter): Promise<Context> {
+  const ctx = new Context()
+  await ctx.plugin(LlmService)
+  await ctx.plugin(SessionStore)
+  await ctx.plugin(SystemPrompt)
+  await ctx.plugin(ToolRegistry)
+  await ctx.plugin(AgentRegistry)
+  await ctx.plugin(AgentLoop, { agents: [] })
+  if (adapter) ctx.llm.registerAdapter(['mock'], adapter)
+  return ctx
+}
+
+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()
+      }
+    })
+  })
+}
+
+function send(agent: Agent): void {
+  agent.send([{ type: 'text', text: 'go' }])
+}
+
+function contextError(message = 'context too large'): LlmError {
+  return new LlmError(message, CONTEXT_WINDOW_EXCEEDED_CODE)
+}
+
+describe('agent post-step and request-error lifecycle', () => {
+  it('fires post-step after results, buffered context, and steering but before step/end', async () => {
+    const twoCalls: StreamChunk[] = [
+      { type: 'block-start', index: 0, blockType: 'tool-call' },
+      { type: 'block-end', index: 0, block: { type: 'tool-call', id: CallId('call-1'), name: 'work', arguments: '{}' } },
+      { type: 'block-start', index: 1, blockType: 'tool-call' },
+      { type: 'block-end', index: 1, block: { type: 'tool-call', id: CallId('call-2'), name: 'work', arguments: '{}' } },
+      { type: 'usage', usage: { inputTokens: 10, outputTokens: 5 } },
+      { type: 'finish', reason: { kind: 'tool-calls' } },
+    ]
+    const adapter = new FailureScriptAdapter([twoCalls, textResponse('done')])
+    const ctx = await harness(adapter)
+    ctx.tools.register(defineTool({
+      name: 'work',
+      description: 'do work',
+      parameters: {},
+      async execute(_args, exec) {
+        if (exec.callId === CallId('call-2')) {
+          exec.agent?.steer([{ type: 'text', text: 'steered' }], { source: { kind: 'plugin', plugin: 'test' } })
+        }
+        return [{ type: 'text', text: 'worked' }]
+      },
+    }))
+    ctx.on('tools/post-execute', async (exec, _result): Promise<PostToolDecision> => ({
+      kind: 'accept',
+      additionalContexts: [{
+        content: [{ type: 'text', text: `context for ${exec.callId}` }],
+        source: { kind: 'plugin', plugin: 'test' },
+      }],
+    }))
+    const agent = ctx.agentLoop.create(SessionId('post-step-order'), { provider: 'mock', model: 'mock' })
+    const order: string[] = []
+    ctx.on('session/event', (_session, event) => {
+      if (
+        event.type === 'assistant/message' || event.type === 'tool/call'
+        || event.type === 'tool/result' || event.type === 'context/message'
+        || event.type === 'steering/message' || event.type === 'step/end'
+      ) {
+        if (!('step' in event.data) || event.data.step === 1) order.push(event.type)
+      }
+    })
+    ctx.on('agent/post-step', (subject, turn, step, signal) => {
+      if (subject !== agent || step !== 1) return
+      expect({ turn, step, aborted: signal.aborted }).toEqual({ turn: 1, step: 1, aborted: false })
+      subject.inject([{ type: 'text', text: 'listener mutation' }], { source: { kind: 'plugin', plugin: 'post-step' } })
+      order.push('agent/post-step')
+    })
+
+    send(agent)
+    await waitForIdle(ctx, agent)
+
+    expect(order).toEqual([
+      'assistant/message',
+      'tool/call',
+      'tool/result',
+      'tool/call',
+      'tool/result',
+      'context/message',
+      'context/message',
+      'steering/message',
+      'context/message',
+      'agent/post-step',
+      'step/end',
+    ])
+  })
+
+  it('fires post-step for max-tokens and lets cancellation override that success', async () => {
+    const adapter = new FailureScriptAdapter([maxTokensResponse('partial')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('cancel-post-step-max-tokens'), { provider: 'mock', model: 'mock' })
+    let entered!: () => void
+    const postStepEntered = new Promise<void>((resolve) => { entered = resolve })
+    ctx.on('agent/post-step', async (_agent, turn, step, signal) => {
+      expect({ turn, step }).toEqual({ turn: 1, step: 1 })
+      entered()
+      await new Promise<void>((resolve) => {
+        signal.addEventListener('abort', () => { resolve() }, { once: true })
+      })
+    })
+
+    send(agent)
+    const idle = waitForIdle(ctx, agent)
+    await postStepEntered
+    agent.cancel('cancelled during max-tokens post-step')
+    await idle
+
+    expect(agent.session.events.find(event => event.type === 'assistant/message')).toMatchObject({
+      data: { usage: { inputTokens: 10, outputTokens: 7 } },
+    })
+    expect(agent.session.events.at(-1)).toMatchObject({
+      type: 'turn/end',
+      data: { reason: { kind: 'aborted', reason: 'cancelled during max-tokens post-step' } },
+    })
+  })
+
+  it('closes the successful step as disposed when disposal lands during post-step', async () => {
+    const adapter = new FailureScriptAdapter([
+      toolCallResponse('dispose-call', 'work', {}),
+      textResponse('must not continue'),
+    ])
+    const ctx = await harness(adapter)
+    ctx.tools.register(defineTool({
+      name: 'work',
+      description: 'do work',
+      parameters: {},
+      async execute() { return [{ type: 'text', text: 'worked' }] },
+    }))
+    const agent = ctx.agentLoop.create(SessionId('dispose-post-step'), { provider: 'mock', model: 'mock' })
+    let entered!: () => void
+    const postStepEntered = new Promise<void>((resolve) => { entered = resolve })
+    ctx.on('agent/post-step', async (_agent, turn, step, signal) => {
+      expect({ turn, step }).toEqual({ turn: 1, step: 1 })
+      entered()
+      await new Promise<void>((resolve) => {
+        signal.addEventListener('abort', () => { resolve() }, { once: true })
+      })
+    })
+
+    send(agent)
+    await postStepEntered
+    await ctx.fiber.dispose()
+
+    expect(adapter.requests).toHaveLength(1)
+    const boundaries = agent.session.events.filter(event =>
+      event.type === 'step/start' || event.type === 'step/end',
+    )
+    expect(boundaries.map(event => event.type)).toEqual(['step/start', 'step/end'])
+    expect(boundaries.map(event => event.data)).toEqual([
+      { turn: 1, step: 1 },
+      { turn: 1, step: 1 },
+    ])
+    expect(agent.session.events.at(-1)).toMatchObject({
+      type: 'turn/end',
+      data: { reason: { kind: 'disposed' } },
+    })
+  })
+
+  it.each([
+    ['thrown', contextError()],
+    ['in-band', [{ type: 'finish', reason: { kind: 'error', message: 'too large', code: CONTEXT_WINDOW_EXCEEDED_CODE } }] satisfies StreamChunk[]],
+  ] as const)('recovers a %s request failure in a new reconstructable step', async (_style, failure) => {
+    const adapter = new FailureScriptAdapter([failure, textResponse('recovered')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId(`recover-${_style}`), { provider: 'mock', model: 'mock' })
+    const attempts: number[] = []
+    ctx.on('agent/request-error', async (subject, turn, step, error, attempt) => {
+      expect(subject).toBe(agent)
+      expect({ turn, step, code: error.code }).toEqual({ turn: 1, step: 1, code: CONTEXT_WINDOW_EXCEEDED_CODE })
+      attempts.push(attempt)
+      subject.session.append('context/message', {
+        content: [{ type: 'text', text: 'RECOVERY SURFACE MUTATION' }],
+        source: { kind: 'plugin', plugin: 'test-recovery' },
+      }, { surfaceOp: 'append' })
+      return { action: 'retry' }
+    })
+
+    send(agent)
+    await waitForIdle(ctx, agent)
+
+    expect(attempts).toEqual([0])
+    expect(adapter.requests).toHaveLength(2)
+    expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('RECOVERY SURFACE MUTATION')
+    const starts = agent.session.events.filter(event => event.type === 'step/start')
+    const ends = agent.session.events.filter(event => event.type === 'step/end')
+    expect(starts.map(event => event.data.step)).toEqual([1, 2])
+    expect(ends.map(event => event.data.step)).toEqual([1, 2])
+    const recovery = agent.session.events.find(event => event.type === 'context/message')!
+    expect(ends[0]!.seq).toBeLessThan(recovery.seq)
+    expect(recovery.seq).toBeLessThan(starts[1]!.seq)
+  })
+
+  it.each(streamListenerFailureCases)('does not offer %s to request recovery', async (_name, install) => {
+    const ctx = await harness(new FailureScriptAdapter([textResponse('unused')]))
+    const agent = ctx.agentLoop.create(SessionId(`stream-plugin-${_name.replaceAll(' ', '-')}`), { provider: 'mock', model: 'mock' })
+    let recoveries = 0
+    install(ctx)
+    ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _attempt, _signal, next) => {
+      recoveries += 1
+      return next()
+    })
+
+    send(agent)
+    await waitForIdle(ctx, agent)
+
+    expect(recoveries).toBe(0)
+    expect(agent.session.events.at(-1)).toMatchObject({ type: 'turn/end', data: { reason: { kind: 'error' } } })
+  })
+
+  it('does not offer a nested model-call failure as the outer request failure', async () => {
+    const outer = new FailureScriptAdapter([textResponse('outer adapter must not run')])
+    const nested = new FailureScriptAdapter([contextError('nested overflow')])
+    const ctx = await harness(outer)
+    ctx.llm.registerAdapter(['nested'], nested)
+    ctx.on('llm/stream', (options, next) => {
+      if (options.provider !== 'mock') return next()
+      return (async function* () {
+        yield* ctx.llm.stream({
+          provider: 'nested',
+          model: 'nested',
+          messages: [],
+          ...options.signal === undefined ? {} : { signal: options.signal },
+        })
+        yield* next()
+      })()
+    })
+    const agent = ctx.agentLoop.create(SessionId('nested-stream-not-recoverable'), { provider: 'mock', model: 'mock' })
+    let recoveries = 0
+    ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _attempt, _signal, next) => {
+      recoveries += 1
+      return next()
+    })
+
+    send(agent)
+    await waitForIdle(ctx, agent)
+
+    expect(nested.requests).toHaveLength(1)
+    expect(outer.requests).toHaveLength(0)
+    expect(recoveries).toBe(0)
+    expect(agent.session.events.at(-1)).toMatchObject({
+      type: 'turn/end',
+      data: { reason: { kind: 'error', message: 'nested overflow', code: CONTEXT_WINDOW_EXCEEDED_CODE } },
+    })
+  })
+
+  it.each(['prompt-submit', 'prompt-assembly', 'pre-step', 'request'] as const)(
+    'does not offer %s middleware failures to request recovery',
+    async (boundary) => {
+      const adapter = new FailureScriptAdapter([textResponse('unused')])
+      const ctx = await harness(adapter)
+      if (boundary === 'prompt-submit') {
+        ctx.on('agent/prompt-submit', () => { throw new Error('prompt submit failed') })
+      } else if (boundary === 'prompt-assembly') {
+        ctx.on('system-prompt/assemble', () => { throw new Error('prompt assembly failed') })
+      } else if (boundary === 'pre-step') {
+        ctx.on('agent/pre-step', () => { throw new Error('pre-step failed') })
+      } else {
+        ctx.on('agent/request', () => { throw new Error('request middleware failed') })
+      }
+      const agent = ctx.agentLoop.create(SessionId(`${boundary}-not-recoverable`), { provider: 'mock', model: 'mock' })
+      let recoveries = 0
+      ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _attempt, _signal, next) => {
+        recoveries += 1
+        return next()
+      })
+
+      send(agent)
+      await waitForIdle(ctx, agent)
+
+      expect(recoveries).toBe(0)
+      expect(adapter.requests).toHaveLength(0)
+      expect(agent.session.events.at(-1)).toMatchObject({ type: 'turn/end', data: { reason: { kind: 'error' } } })
+    },
+  )
+
+  it('does not offer result, tool, or post-step plugin failures to request recovery', async () => {
+    for (const failure of ['result', 'tool', 'post-step'] as const) {
+      const adapter = new FailureScriptAdapter([
+        failure === 'tool' ? toolCallResponse(`call-${failure}`, 'work', {}) : textResponse('done'),
+        ...(failure === 'tool' ? [textResponse('done')] : []),
+      ])
+      const ctx = await harness(adapter)
+      if (failure === 'result') ctx.on('agent/step-result', () => { throw new Error('result failed') })
+      if (failure === 'post-step') ctx.on('agent/post-step', () => { throw new Error('post-step failed') })
+      if (failure === 'tool') {
+        vi.spyOn(ctx.tools, 'execute').mockRejectedValue(new Error('tool service failed'))
+      }
+      const agent = ctx.agentLoop.create(SessionId(`${failure}-not-recoverable`), { provider: 'mock', model: 'mock' })
+      let recoveries = 0
+      ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _attempt, _signal, next) => {
+        recoveries += 1
+        return next()
+      })
+      send(agent)
+      await waitForIdle(ctx, agent)
+      expect(recoveries, failure).toBe(0)
+    }
+  })
+
+  it.each([
+    ['synchronous dispatch', (error: Error) => new SynchronousDispatchFailureAdapter(error)],
+    ['done getter', (error: Error) => new IteratorResultGetterFailureAdapter('done', error)],
+    ['value getter', (error: Error) => new IteratorResultGetterFailureAdapter('value', error)],
+  ] as const)('preserves original Error identity for adapter %s', async (_name, makeAdapter) => {
+    const original = contextError(`${_name} overflow`)
+    const ctx = await harness(makeAdapter(original))
+    const agent = ctx.agentLoop.create(SessionId(`identity-${_name.replaceAll(' ', '-')}`), { provider: 'mock', model: 'mock' })
+    let seen: Error | undefined
+    ctx.on('agent/request-error', async (_agent, _turn, _step, error, _attempt, _signal, next) => {
+      seen = error
+      return next()
+    })
+
+    send(agent)
+    await waitForIdle(ctx, agent)
+
+    expect(seen).toBe(original)
+  })
+
+  it('classifies iterator construction and explicit NO_ADAPTER as model-request failures', async () => {
+    for (const scenario of ['iterator', 'no-adapter'] as const) {
+      const ctx = scenario === 'iterator' ? await harness(new IteratorConstructionFailureAdapter()) : await harness()
+      const agent = ctx.agentLoop.create(SessionId(`request-boundary-${scenario}`), { provider: 'mock', model: 'mock' })
+      let seen = ''
+      ctx.on('agent/request-error', async (_agent, _turn, _step, error, _attempt, _signal, next) => {
+        seen = error.code ?? ''
+        return next()
+      })
+      send(agent)
+      await waitForIdle(ctx, agent)
+      expect(seen).toBe(scenario === 'iterator' ? 'ITERATOR_CONSTRUCTION' : 'NO_ADAPTER')
+    }
+  })
+
+  it('tracks consecutive retry attempts and resets after a successful request', async () => {
+    const capped = new FailureScriptAdapter([contextError('first overflow'), contextError('second overflow')])
+    const cappedCtx = await harness(capped)
+    const cappedAgent = cappedCtx.agentLoop.create(SessionId('retry-cap'), { provider: 'mock', model: 'mock' })
+    const cappedAttempts: number[] = []
+    cappedCtx.on('agent/request-error', async (_agent, _turn, _step, _error, attempt, _signal, next) => {
+      cappedAttempts.push(attempt)
+      return attempt < 1 ? { action: 'retry' } : next()
+    })
+    send(cappedAgent)
+    await waitForIdle(cappedCtx, cappedAgent)
+    expect(cappedAttempts).toEqual([0, 1])
+
+    const reset = new FailureScriptAdapter([
+      contextError('first overflow'),
+      toolCallResponse('retry-reset-call', 'work', {}),
+      contextError('later overflow'),
+    ])
+    const resetCtx = await harness(reset)
+    resetCtx.tools.register(defineTool({
+      name: 'work',
+      description: 'continue',
+      parameters: {},
+      async execute() { return [{ type: 'text', text: 'worked' }] },
+    }))
+    const resetAgent = resetCtx.agentLoop.create(SessionId('retry-reset'), { provider: 'mock', model: 'mock' })
+    const resetAttempts: { step: number; attempt: number }[] = []
+    resetCtx.on('agent/request-error', async (_agent, _turn, step, _error, attempt, _signal, next) => {
+      resetAttempts.push({ step, attempt })
+      return resetAttempts.length === 1 ? { action: 'retry' } : next()
+    })
+    send(resetAgent)
+    await waitForIdle(resetCtx, resetAgent)
+    expect(resetAttempts).toEqual([{ step: 1, attempt: 0 }, { step: 3, attempt: 0 }])
+  })
+
+  it('preserves the original provider error when recovery throws', async () => {
+    const adapter = new FailureScriptAdapter([contextError('original overflow')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('recovery-throws'), { provider: 'mock', model: 'mock' })
+    ctx.on('agent/request-error', () => { throw new Error('recovery exploded') })
+
+    send(agent)
+    await waitForIdle(ctx, agent)
+
+    expect(agent.session.events.at(-1)).toMatchObject({
+      type: 'turn/end',
+      data: { reason: { kind: 'error', message: 'original overflow', code: CONTEXT_WINDOW_EXCEEDED_CODE } },
+    })
+  })
+
+  it.each(['cancel', 'dispose'] as const)('keeps %s live through request recovery', async (action) => {
+    const adapter = new FailureScriptAdapter([contextError()])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId(`${action}-recovery`), { provider: 'mock', model: 'mock' })
+    let entered!: () => void
+    const recoveryEntered = new Promise<void>((resolve) => { entered = resolve })
+    ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _attempt, signal) => {
+      entered()
+      await new Promise<void>((resolve) => {
+        signal.addEventListener('abort', () => { resolve() }, { once: true })
+      })
+      return { action: 'retry' }
+    })
+
+    send(agent)
+    const idle = waitForIdle(ctx, agent)
+    await recoveryEntered
+    if (action === 'cancel') {
+      agent.cancel('cancelled during recovery')
+      await idle
+    } else {
+      await ctx.fiber.dispose()
+    }
+
+    expect(adapter.requests).toHaveLength(1)
+    expect(agent.session.events.at(-1)).toMatchObject({
+      type: 'turn/end',
+      data: { reason: action === 'cancel' ? { kind: 'aborted', reason: 'cancelled during recovery' } : { kind: 'disposed' } },
+    })
+  })
+})

packages/core/agent-loop/tests/tool-calls.spec.ts

@@ -469,8 +469,16 @@ describe('tool-call scheduler: abort handling', () => {
     await waitForIdle(ctx, agent)
 
     expect(gated.started).toEqual([])
-    expect(events(agent).filter(e => e.type === 'tool/call')).toEqual([])
-    expect(events(agent).filter(e => e.type === 'tool/result')).toEqual([])
+    expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
+      .toEqual([CallId('c1'), CallId('c2')])
+    expect(events(agent).filter(e => e.type === 'tool/result').map(e => ({
+      callId: e.data.callId,
+      isError: e.data.isError,
+      error: e.data.error,
+    }))).toEqual([
+      { callId: CallId('c1'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } },
+      { callId: CallId('c2'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } },
+    ])
   })
 
   it('skips dispatch and stops starting siblings when abort fires during ordered pre-execute', async () => {
@@ -494,10 +502,15 @@ describe('tool-call scheduler: abort handling', () => {
 
     expect(gated.started).toEqual([])
     expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
-      .toEqual([CallId('c1')])
-    const results = events(agent).filter(e => e.type === 'tool/result')
-    expect(results.map(e => e.data.callId)).toEqual([CallId('c1')])
-    expect(results[0]?.data.error).toEqual({ name: 'AbortError', code: 'ABORTED' })
+      .toEqual([CallId('c1'), CallId('c2')])
+    expect(events(agent).filter(e => e.type === 'tool/result').map(e => ({
+      callId: e.data.callId,
+      isError: e.data.isError,
+      error: e.data.error,
+    }))).toEqual([
+      { callId: CallId('c1'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } },
+      { callId: CallId('c2'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } },
+    ])
   })
 
   it('stops replenishing after abort, commits started results, and drains accepted additional contexts', async () => {
@@ -523,12 +536,17 @@ describe('tool-call scheduler: abort handling', () => {
 
     expect(gated.started).toEqual(['1', '2'])
     expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
-      .toEqual([CallId('c1'), CallId('c2')])
+      .toEqual([CallId('c1'), CallId('c2'), CallId('c3'), CallId('c4')])
     expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.callId))
-      .toEqual([CallId('c1'), CallId('c2')])
+      .toEqual([CallId('c1'), CallId('c2'), CallId('c3'), CallId('c4')])
+    expect(events(agent).filter(e => e.type === 'tool/result').slice(-2).map(e => e.data))
+      .toEqual([
+        expect.objectContaining({ callId: CallId('c3'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } }),
+        expect.objectContaining({ callId: CallId('c4'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } }),
+      ])
     const settled = events(agent).filter(e => e.type === 'tool/result' || e.type === 'context/message')
     expect(settled.map(e => e.type))
-      .toEqual(['tool/result', 'tool/result', 'context/message', 'context/message'])
+      .toEqual(['tool/result', 'tool/result', 'tool/result', 'tool/result', 'context/message', 'context/message'])
     expect(settled.filter(e => e.type === 'context/message')
       .map(e => (e.data.content[0] as { text: string }).text))
       .toEqual(['ctx-c1', 'ctx-c2'])
@@ -564,6 +582,8 @@ describe('tool-call scheduler: abort handling', () => {
 
     expect(exclusive).toEqual([])
     expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
-      .toEqual([CallId('c1'), CallId('c2')])
+      .toEqual([CallId('c1'), CallId('c2'), CallId('c3')])
+    expect(events(agent).filter(e => e.type === 'tool/result').at(-1)?.data)
+      .toMatchObject({ callId: CallId('c3'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } })
   })
 })