Merge pull request #11 from benavlabs/dispatch-hooks

Dispatch hooks

Author: igorbenav

docs/api/index.md

@@ -94,6 +94,8 @@ from fastroai import (
     FastroAIError,
     PipelineValidationError,
     CostBudgetExceededError,
+    DispatchSkippedError,
+    ErrorCategory,
 )
 ```
 
@@ -105,9 +107,12 @@ All FastroAI exceptions inherit from `FastroAIError`, so you can catch all libra
 FastroAIError                    # Base for all FastroAI errors
 ├── PipelineValidationError      # Invalid pipeline configuration
 ├── StepExecutionError           # Step failed during execution
-└── CostBudgetExceededError      # Cost budget exceeded
+├── CostBudgetExceededError      # Cost budget exceeded
+└── DispatchSkippedError         # Raised from on_before_dispatch to short-circuit (no retry)
 ```
 
+`ErrorCategory` (StrEnum: `TRANSIENT`, `PERMANENT`, `RESOURCE_EXHAUSTION`, `UNKNOWN`) is provided for callers that want to categorize exceptions inside `on_after_dispatch` (e.g., a circuit breaker that only counts transient failures toward its open threshold). The library doesn't auto-categorize.
+
 ```python
 try:
     result = await pipeline.execute(inputs, deps)

docs/changelog.md

@@ -6,6 +6,30 @@ The Changelog documents all notable changes made to FastroAI. This includes new
 
 ---
 
+## [0.6.0] - May 7, 2026
+
+#### Added
+- **Dispatch hooks on `FastroAgent`** by [@igorbenav](https://github.com/igorbenav)
+  - `on_before_dispatch` — async callback fired before each `agent.run()` attempt inside the retry loop. Raising `DispatchSkippedError` (or a subclass) short-circuits the dispatch — the retry loop propagates the error without retrying, and `on_after_dispatch` is not called.
+  - `on_after_dispatch` — async callback fired after each attempt with the exception that occurred, or `None` on success. Use for circuit-breaker recording, retry-budget tracking, alerting.
+  - Hooks fire **per attempt** (every retry triggers them again), letting downstream guards see each individual try rather than only the final outcome.
+  - Pre-flight rejections like `CostBudgetExceededError` skip both hooks — they raise before the dispatch loop is entered.
+
+- **`DispatchSkippedError`** exception — subclass of `FastroAIError`. Raise from `on_before_dispatch` to fail fast on circuit-breaker-open / kill-switch / rate-limit signals without burning a retry slot.
+
+- **`ErrorCategory`** StrEnum (`TRANSIENT`, `PERMANENT`, `RESOURCE_EXHAUSTION`, `UNKNOWN`) — provided for callers that want to categorize exceptions inside `on_after_dispatch` (e.g., a circuit breaker that only counts transient failures toward its open threshold). The library doesn't auto-categorize; callers map exceptions themselves.
+
+- **`AgentConfig.timeout_name`** — optional human-readable label for the configured timeout, surfaced in span attributes and `TimeoutError` messages emitted by the retry loop. Has no effect on enforcement; observability tooling (e.g. Logfire) uses it to distinguish which configured timeout fired without parsing call-site context.
+
+#### Documentation
+- Updated `docs/learn/6-when-things-go-wrong.md` with a Dispatch Hooks section showing the breaker pattern.
+- Updated error-hierarchy diagrams in `docs/api/index.md`, `docs/learn/6-when-things-go-wrong.md`, and `CLAUDE.md` to include `DispatchSkippedError`.
+- Updated `docs/guides/fastro-agent.md` configuration reference table with `timeout_name`, `on_before_dispatch`, `on_after_dispatch`.
+
+**Full Changelog**: https://github.com/benavlabs/fastroai/compare/v0.5.0...v0.6.0
+
+---
+
 ## [0.5.0] - May 6, 2026
 
 #### Fixed

docs/guides/fastro-agent.md

@@ -68,7 +68,10 @@ agent = FastroAgent(config=config)
 | `temperature` | `0.7` | Sampling temperature (0.0-2.0) |
 | `max_tokens` | `4096` | Maximum response tokens |
 | `timeout` | `None` | Per-request timeout in seconds. When set, forwarded to `ModelSettings.timeout` and enforced by the underlying model client. `None` means use the client's own default (typically 600s read on OpenAI). |
+| `timeout_name` | `None` | Human-readable label for this timeout, surfaced in span attributes and `TimeoutError` messages emitted by the retry loop. Has no effect on enforcement — observability only. |
 | `max_retries` | `3` | Retry attempts on failure |
+| `on_before_dispatch` | `None` | Async callback fired before each `agent.run()` attempt inside the retry loop. Raise `DispatchSkippedError` to short-circuit (no retry, no `on_after_dispatch`). Use for circuit breakers, kill switches, rate limiters. Lives on `FastroAgent`, not `AgentConfig`. |
+| `on_after_dispatch` | `None` | Async callback fired after each attempt with the exception (or `None` on success). Use for breaker recording, retry-budget tracking. Does not fire on pre-flight rejections like `CostBudgetExceededError`. Lives on `FastroAgent`, not `AgentConfig`. |
 
 Model names use PydanticAI's provider prefix format: `openai:gpt-4o`, `anthropic:claude-3-5-sonnet`, `google:gemini-1.5-pro`. The prefix tells PydanticAI which API client to use.
 

docs/learn/6-when-things-go-wrong.md

@@ -87,7 +87,8 @@ FastroAI uses a structured exception hierarchy:
 FastroAIError                    # Base for all FastroAI errors
 ├── PipelineValidationError      # Invalid pipeline config
 ├── StepExecutionError           # Step failed during execution
-└── CostBudgetExceededError      # Budget limit hit
+├── CostBudgetExceededError      # Budget limit hit
+└── DispatchSkippedError         # Short-circuit a dispatch (breaker open, kill switch)
 ```
 
 Catch `FastroAIError` to handle all library-specific errors:
@@ -125,6 +126,32 @@ except CostBudgetExceededError as e:
     print(f"Step: {e.step_id}")
 ```
 
+## Dispatch Hooks
+
+Some failures are best handled before they happen. If your downstream provider is rate-limiting you, retrying ten times in a row just makes it worse. If a circuit breaker has tripped, the next request shouldn't even be attempted. `FastroAgent` accepts two async callbacks that fire **per attempt** inside the retry loop:
+
+```python
+async def before():
+    if breaker.is_open():
+        raise DispatchSkippedError("breaker open")
+
+async def after(exc: Exception | None):
+    if exc is None:
+        breaker.record_success()
+    else:
+        breaker.record_failure(categorize(exc))
+
+agent = FastroAgent(
+    model="openai:gpt-4o",
+    on_before_dispatch=before,
+    on_after_dispatch=after,
+)
+```
+
+Raising `DispatchSkippedError` from `on_before_dispatch` short-circuits the dispatch entirely — the retry loop propagates the error without retrying, and `on_after_dispatch` is not called. Use it for guards that should fail fast: circuit breakers, kill switches, rate limiters that have already exhausted budget. Pre-flight rejections like `CostBudgetExceededError` skip both hooks (they raise before the dispatch loop is entered).
+
+For categorizing exceptions inside `on_after_dispatch`, the library exports `ErrorCategory` — a StrEnum with `TRANSIENT`, `PERMANENT`, `RESOURCE_EXHAUSTION`, and `UNKNOWN`. Callers map exceptions themselves; the library doesn't auto-categorize.
+
 ## Designing for Failure
 
 The goal isn't preventing all failures - it's ensuring failures don't cascade.

fastroai/__init__.py

@@ -34,10 +34,11 @@ async def classify(ctx: StepContext[None]) -> str:
     print(result.output)
 """
 
-__version__ = "0.1.0"
+from importlib.metadata import PackageNotFoundError
+from importlib.metadata import version as _version
 
 from .agent import AgentConfig, AgentStepWrapper, ChatResponse, FastroAgent, StreamChunk
-from .errors import CostBudgetExceededError, FastroAIError, PipelineValidationError
+from .errors import CostBudgetExceededError, DispatchSkippedError, ErrorCategory, FastroAIError, PipelineValidationError
 from .pipelines import (
     BasePipeline,
     BaseStep,
@@ -57,6 +58,11 @@ async def classify(ctx: StepContext[None]) -> str:
 from .tracing import LogfireTracer, NoOpTracer, SimpleTracer, Tracer
 from .usage import CostCalculator
 
+try:
+    __version__ = _version("fastroai")
+except PackageNotFoundError:
+    __version__ = "0.0.0+unknown"
+
 __all__ = [
     "__version__",
     # Agent
@@ -69,6 +75,8 @@ async def classify(ctx: StepContext[None]) -> str:
     "FastroAIError",
     "PipelineValidationError",
     "CostBudgetExceededError",
+    "DispatchSkippedError",
+    "ErrorCategory",
     # Pipelines
     "Pipeline",
     "PipelineResult",

fastroai/agent/agent.py

@@ -10,7 +10,7 @@
 import inspect
 import logging
 import time
-from collections.abc import AsyncGenerator, Callable
+from collections.abc import AsyncGenerator, Awaitable, Callable
 from typing import Any, Generic, TypeVar, cast
 
 from pydantic_ai import Agent
@@ -96,6 +96,8 @@ def __init__(
         output_type: type[OutputT] | None = None,
         toolsets: list[AbstractToolset] | None = None,
         cost_calculator: CostCalculator | None = None,
+        on_before_dispatch: Callable[[], Awaitable[None]] | None = None,
+        on_after_dispatch: Callable[[Exception | None], Awaitable[None]] | None = None,
         **kwargs: Any,
     ) -> None:
         """Initialize FastroAgent.
@@ -107,8 +109,23 @@ def __init__(
             output_type: Pydantic model for structured output. Defaults to str.
             toolsets: Tool sets available to the agent.
             cost_calculator: Cost calculator. Default uses standard pricing.
+            on_before_dispatch: Optional async callback fired immediately before
+                each `agent.run()` attempt inside the StepContext retry loop.
+                Raise `DispatchSkippedError` (or a subclass) to short-circuit
+                the dispatch — the retry loop propagates that exception without
+                retrying. Use this for guards that should fail fast: circuit
+                breakers, rate limiters, kill switches. Fires per attempt
+                (every retry triggers the hook again).
+            on_after_dispatch: Optional async callback fired immediately after
+                each `agent.run()` attempt completes. Receives the exception
+                that was raised, or `None` on success. Use this for outcome
+                signaling: breaker recording, retry-budget tracking, alerting.
+                Fires per attempt (every retry triggers the hook again).
+                Does NOT fire when `on_before_dispatch` raises (that
+                short-circuits dispatch entirely) or when a pre-flight guard
+                like `CostBudgetExceededError` rejects before dispatch.
             **kwargs: Passed to AgentConfig if config is None.
-                     Common: model, system_prompt, temperature, max_tokens.
+                     Common: model, system_prompt, temperature, max_tokens, timeout, timeout_name.
 
         Examples:
             ```python
@@ -132,12 +149,32 @@ def __init__(
             from pydantic_ai import Agent
             pydantic_agent = Agent(model="gpt-4o", output_type=MyType)
             agent = FastroAgent(agent=pydantic_agent)
+
+            # With dispatch hooks for circuit-breaker integration
+            async def before():
+                await breaker.check()  # may raise DispatchSkippedError subclass
+
+            async def after(exc):
+                if exc is None:
+                    await breaker.record_success()
+                else:
+                    await breaker.record_failure(categorize(exc))
+
+            agent = FastroAgent(
+                model="gpt-4o",
+                timeout=300,
+                timeout_name="my_step.dispatch",
+                on_before_dispatch=before,
+                on_after_dispatch=after,
+            )
             ```
         """
         self.config = config or AgentConfig(**kwargs)
         self.toolsets = toolsets or []
         self.cost_calculator = cost_calculator or CostCalculator()
         self._output_type = output_type
+        self._on_before_dispatch = on_before_dispatch
+        self._on_after_dispatch = on_after_dispatch
         model_explicitly_set = (config is not None and config.model != DEFAULT_MODEL) or "model" in kwargs
         self._fallback_model: str | None = self.config.model if model_explicitly_set else None
 

fastroai/agent/schemas.py

@@ -67,6 +67,16 @@ class AgentConfig(BaseModel):
             "by this value — total wall time is therefore (max_retries + 1) × timeout."
         ),
     )
+    timeout_name: str | None = Field(
+        default=None,
+        description=(
+            "Human-readable label for this timeout, surfaced in span attributes and "
+            "TimeoutError messages emitted by the retry loop. Used by observability "
+            "tooling (e.g. Logfire) to distinguish which configured timeout fired "
+            "without parsing call-site context. Has no effect on enforcement — the "
+            "actual deadline is still controlled by `timeout`."
+        ),
+    )
     max_retries: int = Field(default=DEFAULT_MAX_RETRIES, ge=0, description="Maximum retry attempts on failure.")
 
     def get_effective_system_prompt(self) -> str:

fastroai/errors.py

@@ -1,13 +1,19 @@
-"""FastroAI exception hierarchy.
+"""FastroAI exception hierarchy and error classification.
 
 Provides structured exceptions for clear error handling:
 - FastroAIError: Base exception for all FastroAI errors
 - PipelineValidationError: Invalid pipeline configuration
 - CostBudgetExceeded: Cost budget was exceeded
+- DispatchSkippedError: Short-circuit signal raised from on_before_dispatch hooks
+
+Plus the ErrorCategory enum, used by on_after_dispatch callbacks to classify
+exceptions for downstream consumers (e.g. circuit breakers).
 """
 
 from __future__ import annotations
 
+from enum import StrEnum
+
 
 class FastroAIError(Exception):
     """Base exception for all FastroAI errors.
@@ -90,3 +96,65 @@ def __init__(
         self.step_id = step_id
         location = f" in step '{step_id}'" if step_id else ""
         super().__init__(f"Cost budget exceeded{location}: {actual} microcents > {budget} microcents budget")
+
+
+class DispatchSkippedError(FastroAIError):
+    """Raise from `on_before_dispatch` to short-circuit the dispatch.
+
+    When raised, `agent.run()` is not called and the retry loop in
+    `StepContext._execute_with_config` does not retry — the exception
+    propagates immediately to the caller of `ctx.run()`. Use this for
+    guards that should fail fast: circuit breakers being the canonical
+    example. Subclass this for application-specific skip reasons.
+
+    The retry loop distinguishes `DispatchSkippedError` from other exceptions:
+    the former propagates without retry; the latter triggers exponential
+    backoff. This avoids burning ``retries × retry_delay`` seconds on a
+    sick provider that the guard already classified as "don't even try."
+
+    Examples:
+        ```python
+        class BreakerOpen(DispatchSkippedError):
+            def __init__(self, provider: str):
+                super().__init__(f"{provider} circuit breaker open")
+                self.provider = provider
+
+        async def before_dispatch() -> None:
+            if breaker.state == "open":
+                raise BreakerOpen("deepseek")
+
+        agent = FastroAgent(
+            ...,
+            on_before_dispatch=before_dispatch,
+        )
+        ```
+    """
+
+    pass
+
+
+class ErrorCategory(StrEnum):
+    """Bare-minimum vocabulary for error classification at the dispatch hooks layer.
+
+    Concrete categorization (which exceptions are TRANSIENT for which
+    provider) is the application's responsibility — fastroai stays
+    provider-agnostic. Used by `on_after_dispatch` callbacks to feed
+    downstream consumers (circuit breakers, retry budgets, alerting).
+
+    Categories:
+        TRANSIENT: Should retry / let breaker count toward opening.
+            Network errors, rate limits, 5xx responses, request timeouts.
+        PERMANENT: Should not retry / no breaker signal.
+            4xx responses (except 408/429), schema validation errors,
+            malformed input.
+        RESOURCE_EXHAUSTION: Out of memory / disk / quota.
+            Distinct from TRANSIENT because recovery typically needs
+            operational action (scale up, free disk) rather than a
+            retry-after delay.
+        UNKNOWN: Unclassified — caller decides how to treat.
+    """
+
+    TRANSIENT = "transient"
+    PERMANENT = "permanent"
+    RESOURCE_EXHAUSTION = "resource_exhaustion"
+    UNKNOWN = "unknown"

fastroai/pipelines/__init__.py

@@ -26,7 +26,13 @@ async def execute(self, context: StepContext[MyDeps]) -> str:
     print(result.output)  # "POSITIVE"
 """
 
-from ..errors import CostBudgetExceededError, FastroAIError, PipelineValidationError
+from ..errors import (
+    CostBudgetExceededError,
+    DispatchSkippedError,
+    ErrorCategory,
+    FastroAIError,
+    PipelineValidationError,
+)
 from .base import BaseStep, ConversationState, ConversationStatus, StepContext
 from .config import PipelineConfig, StepConfig
 from .decorators import step
@@ -51,6 +57,8 @@ async def execute(self, context: StepContext[MyDeps]) -> str:
     "PipelineValidationError",
     "StepExecutionError",
     "CostBudgetExceededError",
+    "DispatchSkippedError",
+    "ErrorCategory",
     # Pipeline
     "Pipeline",
     "PipelineResult",