Error Handling & Recovery

ChatFlowError (Base)
├── DependencyError (Missing optional extra dependencies)
├── ToolExecutionError (Uncaught error during tool execution)
├── ToolSchemaError (Invalid parameter type annotation) [v0.4.0rc1]
├── AgentHandoffError (Unknown handoff target or transfer loop)
├── ProviderError (LLM API status error, network failure, or timeout)
├── ReentrantTurnError (Concurrent turn invocation on locked session) [v0.4.0rc1]
└── SessionDeletedError (Message dispatched to a deleted session) [v0.4.0rc1]

import logging
from chatflow_agent import Agent, Runner, tool

logger = logging.getLogger("app.tools")

@tool
def process_refund(order_id: str, amount: float) -> str:
    """Issue a refund for an order.

    Args:
        order_id: Order identifier formatted as ORD-XXXX.
        amount: Positive monetary value to refund.
    """
    if not order_id.startswith("ORD-"):
        raise ValueError("Invalid order format. Order ID must begin with 'ORD-'.")
    if amount <= 0:
        raise ValueError("Refund amount must be strictly greater than 0.")
    return f"Refund of ${amount:.2f} processed for order {order_id}."

agent = Agent(
    name="RefundBot",
    model="gpt-4o",
    tools=[process_refund],
    instructions="Process refunds accurately. If an order ID format is invalid, clarify with the user.",
)

runner = Runner(starting_agent=agent)

Turn Start
    ├── Capture Checkpoint: (history, active_agent, metadata)
    ├── Append Inbound User Message
    ├── Execute LLM Turns & Tools
    │     ├── Success: Commit final assistant response to SessionStore
    │     └── Failure / CancelledError: Restore (history, active_agent, metadata)
    └── Release Session Lock

import asyncio
from chatflow_agent import Agent, Runner
from chatflow_agent.exceptions import ProviderError

def build_resilient_runner(agent: Agent) -> Runner:
    """Configure a runner with bounded handoffs and tool iterations."""
    return Runner(
        starting_agent=agent,
        max_tool_iterations=6,
        max_handoffs_per_turn=3,
    )

async def execute_turn_with_recovery(
    runner: Runner,
    session_id: str,
    user_message: str,
) -> str:
    """Execute a turn with explicit exception handling and graceful fallback."""
    try:
        response = await runner.run_async(
            session_id=session_id,
            user_message=user_message,
        )
        return response.content
    except ProviderError as err:
        # State was automatically rolled back to before the turn
        return "Our AI inference service is temporarily degraded. Please retry in a few moments."
    except TimeoutError:
        # Turn timeout exceeded; checkpoint restored
        return "The request took too long to complete. Please try a simpler question."
    except Exception as err:
        return "An unexpected system error occurred. Our operations team has been notified."

# [no-exec: illustrative turn timeout call]
import asyncio

# In v0.3.x, enforce turn timeouts via asyncio.wait_for
response = await asyncio.wait_for(
    runner.run_async(
        session_id="customer-1029",
        user_message="Analyze this quarterly financial audit report...",
    ),
    timeout=45.0,
)