feat(core): add post-step request recovery (PR3 phase 1)

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

@@ -7,37 +7,42 @@
 
 import type { Context } from 'cordis'
 import type { FinishReason, GenerateOptions, LlmCallConfig, Message, TokenUsage } from '@deepseek-ai/dsh-llm'
-import { BlockAssembler, HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
+import { BlockAssembler, HarnessError, 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'
 import type { TransmissionLog } from './request-log.ts'
 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 type { ToolExecutionResult } from '@deepseek-ai/dsh-tools'
 import type { ReactLoopAgent } from './agent.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 a terminal failure finish 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
     }
@@ -51,10 +56,19 @@ 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 } : {} }
 }
 
+/** Build the durable result for a model-requested call skipped after cancellation. */
+function skippedToolResult(): ToolExecutionResult {
+  return {
+    content: [{ type: 'text', text: 'Error: tool call skipped because the step was aborted before execution' }],
+    isError: true,
+    error: { name: 'AbortError', code: 'ABORTED' },
+  }
+}
+
 /** Map a successful max-token finish onto the turn reason; other successful finishes add nothing. */
 function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
   switch (finish.kind) {
@@ -168,6 +182,7 @@ async function runTurn(
 
   let reason: TurnEndReason = { kind: 'completed' }
   let step = 0
+  let requestRetryAttempt = 0
   let stepOpen = false
   let errorReported = false
   let terminalStopped = false
@@ -180,7 +195,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) }
@@ -322,14 +337,65 @@ 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, agent, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
       } catch (error: unknown) {
-        stepOutcome = { error: toError(error) }
-      } finally {
+        if (isLlmAdapterFailure(error)) {
+          stepOutcome = { requestError: error }
+        } else 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
+        }
+        if (recoveryDecision.action === 'retry') {
+          requestRetryAttempt += 1
+          continue
+        }
+        failTurn(stepOutcome.requestError)
+        break
       }
 
       if ('error' in stepOutcome) {
@@ -337,7 +403,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) {
@@ -349,6 +417,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
@@ -356,7 +426,38 @@ async function runTurn(
       // Steering that arrived during streaming/tool execution.
       const steered = drainSteering(agent, handle.inbox, turn)
 
+      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
@@ -523,7 +624,7 @@ async function runStep(
 
   // 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)
 
   if (assembler.finish.kind === 'max-tokens') {
     let message: Message = withoutToolCalls(assembler.message())
@@ -553,25 +654,30 @@ async function runStep(
   const toolCalls = message.content.filter(block => block.type === 'tool-call')
   // Buffer context until all results are appended to preserve call/result adjacency.
   const pendingContext: HookContext[] = []
+  let aborted = signal.aborted
   for (const call of toolCalls) {
-    /* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
-    if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
     const callEvent = session.append('tool/call', { turn, step, callId: call.id, name: call.name, arguments: call.arguments })
-    let parsedArguments: unknown
-    try {
-      parsedArguments = call.arguments ? JSON.parse(call.arguments) : {}
-    } catch {
-      parsedArguments = call.arguments
+    let result: ToolExecutionResult
+    if (aborted || signal.aborted) {
+      aborted = true
+      result = skippedToolResult()
+    } else {
+      let parsedArguments: unknown
+      try {
+        parsedArguments = call.arguments ? JSON.parse(call.arguments) : {}
+      } catch {
+        parsedArguments = call.arguments
+      }
+      // TODO(pre-tool-input-rewrite): Keep logged history and live presentation aligned;
+      // see docs/rfc/proposed/feature/2026-06-30-pre-tool-input-rewrite.md.
+      result = await ctx.tools.execute({
+        callId: call.id,
+        name: call.name,
+        arguments: parsedArguments,
+        agent,
+        signal,
+      })
     }
-    // TODO(pre-tool-input-rewrite): Keep logged history and live presentation aligned;
-    // see docs/rfc/proposed/feature/2026-06-30-pre-tool-input-rewrite.md.
-    const result = await ctx.tools.execute({
-      callId: call.id,
-      name: call.name,
-      arguments: parsedArguments,
-      agent,
-      signal,
-    })
     session.append('tool/result', {
       turn, step,
       // Correlation comes from the immutable execution input; the result does
@@ -584,13 +690,12 @@ async function runStep(
       ...result.meta !== undefined ? { meta: result.meta } : {},
     }, { surfaceOp: 'append', sourceEventSeqs: [callEvent.seq] })
     if (result.additionalContext) pendingContext.push(result.additionalContext)
-    // The signal may flip while the tool is awaited.
-    /* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
-    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
-    if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
-    /* v8 ignore stop */
+    if (signal.aborted) aborted = true
   }
 
+  /* v8 ignore next -- signal.reason always set by cancellation or disposal. */
+  if (aborted) throw new Error(String(signal.reason ?? 'aborted'))
+
   // Append buffered context after the complete result batch.
   for (const context of pendingContext) {
     agent.inject(context.content, { source: context.source })