Error Recovery

Contents

Overview

flowchart TD
  ERR["Streaming Error"]
  CAT["Categorize Error"]
  RATE["rate_limit"]
  CTX["context_overflow"]
  TIMEOUT["timeout"]
  AUTH["auth_failure"]
  PROV["provider_error"]
  TRANS["transient"]
  RETRY{"Retryable?"}
  ABORT{"Abort signal?"}
  SELECT["Select next fallback model"]
  PROBE{"Cooldown probe?"}
  ATTEMPT["Attempt stream on fallback"]
  SUCCESS["Success — resume streaming"]
  EXHAUST["All models exhausted"]
  FSE["FallbackSummaryError"]
  OBS["failover-observation.ts — log event"]

  ERR --> CAT
  CAT --> RATE
  CAT --> CTX
  CAT --> TIMEOUT
  CAT --> AUTH
  CAT --> PROV
  CAT --> TRANS
  RATE --> RETRY
  CTX --> RETRY
  TIMEOUT --> RETRY
  AUTH --> RETRY
  PROV --> RETRY
  TRANS --> RETRY
  RETRY -->|No| FSE
  RETRY -->|Yes| ABORT
  ABORT -->|Yes| FSE
  ABORT -->|No| SELECT
  SELECT --> PROBE
  PROBE -->|Yes, try recently-failed| ATTEMPT
  PROBE -->|No, skip to next| SELECT
  ATTEMPT -->|Success| SUCCESS
  ATTEMPT -->|Failure| SELECT
  SELECT -->|No more models| EXHAUST
  EXHAUST --> FSE
  ATTEMPT --> OBS
  SUCCESS --> OBS

The overview flowchart traces an error from initial detection through categorization, the fallback selection loop, and the two possible outcomes: successful recovery on an alternate model, or a FallbackSummaryError when all options are exhausted. Note how failover-observation.ts logs every attempt regardless of outcome, enabling post-session analysis of failover patterns.

Decision Points

How It Works

Error recovery operates as a state machine wrapping the streaming attempt loop. Each attempt either succeeds (returning the response payload) or fails (adding an entry to the fallback summary and advancing to the next model).

Phase 1 — Error Categorization

When a streaming error occurs, the system first categorizes it. Each provider wrapper translates its native error format into one of six canonical categories:

Phase 2 — Model Fallback Chain

The model-fallback.ts module maintains an ordered list of models to try. When the primary model fails, it walks the chain and attempts each one. The chain is configured per-deployment and typically follows the pattern: primary model → same-provider cheaper model → cross-provider equivalent → cross-provider cheaper model.

// model-fallback.ts — core fallback loop
interface FallbackAttempt {
  model: string;
  error: CategorizedError;
  durationMs: number;
  cooldownExpiry: Date | null;
}

async function attemptWithFallback(
  request: StreamRequest,
  fallbackChain: string[],
  signal: AbortSignal,
): Promise<StreamResponse> {
  const attempts: FallbackAttempt[] = [];

  for (const model of fallbackChain) {
    if (signal.aborted) break;

    const provider = getProviderForModel(model);
    if (provider.isInCooldown(model)) {
      if (!shouldAllowCooldownProbeForReason(provider, model)) {
        continue; // skip — cooldown not yet expired
      }
    }

    const start = Date.now();
    try {
      const response = await provider.stream(request, model, signal);
      logFailoverObservation(model, "success", attempts);
      return response;
    } catch (err) {
      const categorized = categorizeError(err, provider);
      attempts.push({
        model,
        error: categorized,
        durationMs: Date.now() - start,
        cooldownExpiry: categorized.cooldownExpiry,
      });
      provider.recordFailure(model, categorized);
      logFailoverObservation(model, "failure", attempts);
    }
  }

  throw new FallbackSummaryError(attempts);
}

Phase 3 — Cooldown Probing

When a provider is in cooldown but the cooldown window is nearly expired (remainingMs < PROBE_THRESHOLD_MS), the system may issue a "cooldown probe" — a retry attempt that would normally be skipped. This enables faster recovery when rate limits expire mid-fallback-chain.

// shouldAllowCooldownProbeForReason()
const PROBE_THRESHOLD_MS = 5000;

function shouldAllowCooldownProbeForReason(
  provider: Provider,
  model: string,
): boolean {
  const cooldown = provider.getCooldownState(model);
  if (!cooldown) return true; // not in cooldown

  const remainingMs = cooldown.expiresAt.getTime() - Date.now();
  if (remainingMs <= 0) return true; // cooldown expired

  // Only probe if nearly expired and the reason was rate_limit or transient
  if (remainingMs < PROBE_THRESHOLD_MS) {
    return cooldown.reason === "rate_limit" || cooldown.reason === "transient";
  }

  return false;
}

Phase 4 — FallbackSummaryError

When all models in the fallback chain have been tried and failed, the system creates a FallbackSummaryError. This error carries structured data for every attempt — which model was tried, what error it returned, how long it took, and when its cooldown expires. The soonestCooldownExpiry property tells the caller when the earliest retry opportunity will be available.

Abort Handling

The abort.ts module manages stream abort propagation. When the user cancels or the system determines the stream should stop, the abort signal is raised. However, to prevent data loss, the signal propagation includes a settle timeout (STEER_ABORT_SETTLE_TIMEOUT_MS = 5000) that allows in-flight chunks to drain before the connection is terminated.

Behavior Diagram

stateDiagram-v2
  [*] --> Streaming
  Streaming --> ErrorDetected : provider error
  ErrorDetected --> Categorizing : inspect error
  Categorizing --> CheckingRetryable : assign category
  CheckingRetryable --> FallbackSummary : not retryable
  CheckingRetryable --> CheckingAbort : retryable
  CheckingAbort --> FallbackSummary : abort signal raised
  CheckingAbort --> SelectingFallback : no abort
  SelectingFallback --> CooldownProbe : model in cooldown, nearly expired
  SelectingFallback --> AttemptingStream : model available
  SelectingFallback --> FallbackSummary : no models remaining
  CooldownProbe --> AttemptingStream : probe allowed
  CooldownProbe --> SelectingFallback : probe denied, skip model
  AttemptingStream --> Streaming : success
  AttemptingStream --> SelectingFallback : failure, try next
  FallbackSummary --> [*]

  state Streaming {
    [*] --> ReceivingChunks
    ReceivingChunks --> ReceivingChunks : chunk received
    ReceivingChunks --> StopReasonCheck : stream ended
    StopReasonCheck --> [*] : normal stop
    StopReasonCheck --> ErrorDetected : unexpected stop reason
  }

The state diagram shows two levels: the outer error-recovery state machine and the inner streaming state. Note that the StopReasonCheck state inside streaming can transition to ErrorDetected when attempt.stop-reason-recovery.ts identifies an unexpected stop reason (e.g., content_filter, premature length stop). This means error recovery handles not just explicit errors but also abnormal completions.

ACP/Subagent Lifecycle

Error recovery interacts with the broader subagent lifecycle. Each subagent (ACP or spawned child) has its own lifecycle that error recovery must respect:

Annotated Code Walkthrough

Walk through the critical path from error detection to fallback resolution.

// error-categorizer.ts
type ErrorCategory =
  | "rate_limit"
  | "context_overflow"
  | "timeout"
  | "auth_failure"
  | "provider_error"
  | "transient";

interface CategorizedError {
  category: ErrorCategory;
  retryable: boolean;
  cooldownExpiry: Date | null;
  originalError: Error;
  providerMessage: string;
}

function categorizeError(err: unknown, provider: Provider): CategorizedError {
  if (err instanceof TimeoutError) {
    return { category: "timeout", retryable: true, cooldownExpiry: null,
             originalError: err, providerMessage: "LLM idle timeout" };
  }

  const status = (err as any)?.status ?? 0;
  if (status === 429) {
    const retryAfter = parseRetryAfterHeader(err);
    return { category: "rate_limit", retryable: true,
             cooldownExpiry: new Date(Date.now() + retryAfter),
             originalError: err as Error,
             providerMessage: `Rate limited, retry after ${retryAfter}ms` };
  }

  if (status === 401 || status === 403) {
    return { category: "auth_failure", retryable: false,
             cooldownExpiry: null, originalError: err as Error,
             providerMessage: "Authentication failed" };
  }

  if (status >= 500) {
    return { category: "provider_error", retryable: true,
             cooldownExpiry: new Date(Date.now() + 1000),
             originalError: err as Error,
             providerMessage: `Provider returned ${status}` };
  }

  if (isContextOverflowError(err)) {
    return { category: "context_overflow", retryable: true,
             cooldownExpiry: null, originalError: err as Error,
             providerMessage: "Context window exceeded" };
  }

  return { category: "transient", retryable: true,
           cooldownExpiry: new Date(Date.now() + 2000),
           originalError: err as Error,
           providerMessage: String(err) };
}

// model-fallback.ts — FallbackSummaryError
class FallbackSummaryError extends Error {
  readonly attempts: FallbackAttempt[];
  readonly soonestCooldownExpiry: Date | null;
  readonly aborted: boolean;

  constructor(attempts: FallbackAttempt[], aborted = false) {
    const models = attempts.map(a => a.model).join(", ");
    super(`All fallback models exhausted: ${models}`);
    this.name = "FallbackSummaryError";
    this.attempts = attempts;
    this.aborted = aborted;

    // Find the earliest cooldown expiry across all attempts
    this.soonestCooldownExpiry = attempts
      .map(a => a.cooldownExpiry)
      .filter((d): d is Date => d !== null)
      .sort((a, b) => a.getTime() - b.getTime())[0] ?? null;
  }

  get retryableAt(): Date | null {
    return this.soonestCooldownExpiry;
  }
}

// failover-observation.ts
interface FailoverObservation {
  timestamp: Date;
  model: string;
  outcome: "success" | "failure";
  errorCategory?: ErrorCategory;
  durationMs: number;
  attemptIndex: number;
  totalAttempts: number;
}

function logFailoverObservation(
  model: string,
  outcome: "success" | "failure",
  attempts: FallbackAttempt[],
): void {
  const observation: FailoverObservation = {
    timestamp: new Date(),
    model,
    outcome,
    errorCategory: outcome === "failure"
      ? attempts[attempts.length - 1]?.error.category
      : undefined,
    durationMs: outcome === "failure"
      ? attempts[attempts.length - 1]?.durationMs ?? 0
      : 0,
    attemptIndex: attempts.length - 1,
    totalAttempts: attempts.length,
  };
  telemetry.record("failover", observation);
}

Edge Cases & Error Handling

Configuration

Related Topics

See also:

• Streaming overview — the parent L2 page covering the full streaming pipeline
• Chunk Assembly — how raw SSE events are normalized into unified payloads
• Backpressure & Flow Control — how the pipeline manages context window limits and compaction

Directory Map

src/agents/pi-embedded-runner/
├── fallback/
│   ├── model-fallback.ts                  ← Core fallback loop and FallbackSummaryError
│   ├── error-categorizer.ts               ← Error categorization into canonical categories
│   └── failover-observation.ts            ← Telemetry logging for fallback events
├── attempt/
│   └── attempt.stop-reason-recovery.ts    ← Handles unexpected stop reasons
├── abort.ts                               ← Stream abort signal management
├── auth/
│   └── auth-profiles.ts                   ← Auth token management and cooldown recovery
└── streaming/
    └── llm-idle-timeout.ts                ← Idle connection timeout detection