> ## Documentation Index
> Fetch the complete documentation index at: https://docs.praison.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Workflow Error Handling

> Handle workflow step failures with WorkflowStepError exception

Workflow error handling provides structured exception handling for step failures, enabling robust parallel execution and graceful error recovery.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
    subgraph "Error Handling Flow"
        A[📝 Step] --> B{🔍 Fails?}
        B -->|No| C[✅ Continue]
        B -->|Yes| D[⚠️ WorkflowStepError]
        D --> E[🛠️ Handle Error]
        E --> F[💡 Recovery]
    end
    
    classDef step fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef error fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef success fill:#10B981,stroke:#7C90A0,color:#fff
    
    class A step
    class B,D,E error
    class C,F success
```

## Quick Start

<Steps>
  <Step title="Basic Error Handling">
    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    from praisonaiagents import Agent, Workflow, WorkflowStepError

    agent = Agent(
        name="Research Agent", 
        instructions="Research topics that might fail"
    )

    workflow = Workflow(steps=[agent])

    try:
        result = workflow.start("Research invalid topic")
    except WorkflowStepError as e:
        print(f"Workflow failed: {e}")
        print(f"Root cause: {e.cause}")
    ```
  </Step>

  <Step title="Handling Multiple Errors">
    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    from praisonaiagents import parallel, WorkflowStepError

    workflow = Workflow(steps=[
        parallel([agent_a, agent_b, agent_c], on_failure="fail_all"),
    ])

    try:
        workflow.start("Process all branches")
    except WorkflowStepError as e:
        print(f"Workflow failed: {e}")
        print(f"Root cause: {e.cause}")
        for err in e.errors:
            print(f"  Branch {err['step']}: {err['error']}")
    ```
  </Step>
</Steps>

***

## How It Works

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
    participant User
    participant Workflow
    participant Step
    participant ErrorHandler
    
    User->>Workflow: start()
    Workflow->>Step: execute
    Step-->>Workflow: Exception
    Workflow->>ErrorHandler: WorkflowStepError
    ErrorHandler-->>User: Structured Error
```

| Component             | Role                                            |
| --------------------- | ----------------------------------------------- |
| **WorkflowStepError** | Main exception class for workflow failures      |
| **cause**             | Original exception that triggered the failure   |
| **errors**            | List of multiple errors (for parallel failures) |

***

## Configuration Options

| Attribute | Type                | Default | Description                                                                            |
| --------- | ------------------- | ------- | -------------------------------------------------------------------------------------- |
| `cause`   | `Exception \| None` | `None`  | The underlying exception that triggered the failure (first error in `fail_all` mode)   |
| `errors`  | `List[dict]`        | `[]`    | List of `{"step": int, "error": Exception}` for `fail_all` mode. Empty for `fail_fast` |

***

## Failed Task Propagation

When tasks fail after exhausting retries, dependent tasks are automatically skipped instead of running with `None` context:

### How It Works

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
    subgraph "Task Dependency Flow"
        A[📋 Task A] --> B{🔄 Retries?}
        B -->|Success| C[✅ Task A Success]
        B -->|Failed| D[❌ Task A Failed]
        C --> E[▶️ Run Task B]
        D --> F[⏭️ Skip Task B]
        F --> G[❌ Task B Failed]
    end
    
    classDef success fill:#10B981,stroke:#7C90A0,color:#fff
    classDef failed fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef process fill:#F59E0B,stroke:#7C90A0,color:#fff
    
    class C,E success
    class D,F,G failed
    class A,B process
```

### Example

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent, Task, PraisonAIAgents

agent = Agent(name="Worker", instructions="Process data")

# Primary task that might fail
fetch_data = Task(
    description="Fetch data from unreliable API",
    agent=agent,
    max_retries=3
)

# Dependent task - will be skipped if fetch_data fails
process_data = Task(
    description="Process the fetched data",
    agent=agent,
    context=[fetch_data]  # Depends on fetch_data
)

workflow = PraisonAIAgents(
    agents=[agent], 
    tasks=[fetch_data, process_data]
)

result = workflow.start()

# Check task statuses
print(f"Fetch data status: {fetch_data.status}")
print(f"Process data status: {process_data.status}")

if fetch_data.status == "failed":
    # process_data.status will also be "failed" (skipped)
    print("Primary task failed, dependent task was skipped")
```

### Failure Propagation Rules

1. **Failed Task**: When a task fails after `max_retries`, its `status` is set to `"failed"`
2. **Dependent Detection**: Tasks with `context=[failed_task]` are identified as dependents
3. **Skip Execution**: Dependent tasks are marked as `"failed"` without execution
4. **No None Propagation**: Dependent tasks don't receive `None` values from failed dependencies

### Process Integration

This behavior works consistently across all process types:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# Sequential process - stops at first failure
workflow = PraisonAIAgents(
    agents=[agent],
    tasks=[task_a, task_b, task_c],
    process="sequential"  # Stops if task_a fails
)

# Workflow process - skips dependents of failed tasks  
workflow = PraisonAIAgents(
    agents=[agent],
    tasks=[fetch, process, save],
    process="workflow"  # Skips process+save if fetch fails
)
```

***

## Common Patterns

### Pattern 1: Single Step Recovery

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent, Workflow, WorkflowStepError

def with_retry():
    for attempt in range(3):
        try:
            workflow = Workflow(steps=[unreliable_agent])
            return workflow.start("Task")
        except WorkflowStepError as e:
            if attempt == 2:  # Last attempt
                raise
            print(f"Attempt {attempt + 1} failed: {e}")
```

### Pattern 2: Parallel Error Analysis

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
def analyze_parallel_failures(workflow_errors):
    """Analyze which parallel branches failed and why."""
    failed_branches = []
    for error_info in workflow_errors.errors:
        step_idx = error_info['step']
        error = error_info['error']
        failed_branches.append({
            'branch': step_idx,
            'error_type': type(error).__name__,
            'message': str(error)
        })
    return failed_branches
```

### Pattern 3: Graceful Degradation

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
def robust_workflow(input_data):
    """Run workflow with fallback strategies."""
    try:
        # Try optimal path
        return run_full_workflow(input_data)
    except WorkflowStepError as e:
        if "timeout" in str(e).lower():
            # Fallback to simpler workflow
            return run_simple_workflow(input_data)
        else:
            # Log and re-raise for other errors
            logger.error(f"Workflow failed: {e}")
            raise
```

***

## Best Practices

<AccordionGroup>
  <Accordion title="Always Catch Specific Errors">
    Catch `WorkflowStepError` specifically rather than generic `Exception` to handle workflow failures appropriately while allowing other errors to bubble up.

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    try:
        result = workflow.start("Task")
    except WorkflowStepError as e:
        # Handle workflow-specific failures
        handle_workflow_error(e)
    except Exception as e:
        # Handle unexpected errors
        logger.exception("Unexpected error")
        raise
    ```
  </Accordion>

  <Accordion title="Inspect Error Details">
    Use the `cause` and `errors` attributes to understand what specifically went wrong and implement targeted recovery strategies.

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    except WorkflowStepError as e:
        if isinstance(e.cause, TimeoutError):
            # Retry with longer timeout
            retry_with_timeout()
        elif isinstance(e.cause, ConnectionError):
            # Switch to backup service
            use_backup_service()
    ```
  </Accordion>

  <Accordion title="Log Error Context">
    Include workflow context in error logs to help with debugging and monitoring.

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    except WorkflowStepError as e:
        logger.error(
            "Workflow failed",
            extra={
                "workflow_id": workflow.id,
                "step_count": len(workflow.steps),
                "error_count": len(e.errors),
                "root_cause": str(e.cause)
            }
        )
    ```
  </Accordion>

  <Accordion title="Design for Partial Success">
    When using parallel execution, design your aggregation logic to handle partial results gracefully.

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    def smart_aggregator(ctx):
        """Aggregate results even with some failures."""
        outputs = ctx.variables.get("parallel_outputs", [])
        valid_results = [o for o in outputs if not o.startswith("Error:")]
        
        if len(valid_results) >= 2:  # Minimum threshold
            return aggregate_partial_results(valid_results)
        else:
            raise WorkflowStepError("Insufficient successful results")
    ```
  </Accordion>
</AccordionGroup>

***

## Related

<CardGroup cols={2}>
  <Card title="Workflow Parallel" icon="arrows-split-up-and-left" href="/features/workflow-parallel">
    Parallel execution with failure strategies
  </Card>

  <Card title="Workflow Patterns" icon="diagram-project" href="/features/workflow-patterns">
    Common workflow implementation patterns
  </Card>
</CardGroup>