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

# Robustness

> Build resilient AI workflows with graceful degradation and debugging

Build production-ready workflows that continue when optional steps fail and leave an execution trace for debugging.

> [!NOTE]
>
> As of PraisonAI [#4334](https://github.com/MervinPraison/PraisonAI/pull/4334) (2026-08-25), `skip_on_failure=True` and `on_error="continue"` are honored across every failure-cascade path (`AgentTeam` sync + async, `AgentFlow`, `Process.sequential` / `asequential`). Previously these flags were inert on cascade paths and dependents were force-failed the moment any upstream dependency failed. If you were working around this by removing `context=[...]` dependencies, you can put them back.

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

researcher = Agent(name="researcher", instructions="Research topics thoroughly")
enricher = Agent(name="enricher", instructions="Add real-world examples")

workflow = AgentFlow(
    steps=[
        Task(description="Research AI trends", agent=researcher, max_retries=3, retry_delay=1.0),
        Task(description="Add examples", agent=enricher, skip_on_failure=True, retry_delay=0.5),
    ],
    history=True,
)
result = workflow.start("AI agents 2025")
```

The user starts a multi-step workflow; optional steps can fail without aborting the run, with execution history for debugging.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
    Input[Input] --> Task1[Task 1]
    Task1 --> |Success| Task2[Task 2]
    Task1 --> |Fail + skip_on_failure=True| Task2
    Task1 --> |Fail + on_error=continue| Task2
    Task2 --> |Retry with delay| Task2
    Task2 --> Output[Output]

    classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef tool fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef output fill:#10B981,stroke:#7C90A0,color:#fff

    class Input,Output agent
    class Task1,Task2 tool
```

## How It Works

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
    participant User
    participant Agent
    participant Feature as Robustness

    User->>Agent: Request
    Agent->>Feature: Process request
    Feature-->>Agent: Result    Agent-->>User: Response
```

## Quick Start

**1. Simple Usage**

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

researcher = Agent(name="researcher", instructions="Research topics thoroughly")
enricher = Agent(name="enricher", instructions="Add real-world examples")

workflow = AgentFlow(
    steps=[
        Task(description="Research AI trends", agent=researcher, max_retries=3, retry_delay=1.0),
        Task(description="Add examples", agent=enricher, skip_on_failure=True),
    ],
    history=True,
)
result = workflow.start("AI agents 2025")
history = workflow.get_history()
```

**2. With Configuration**

Full YAML with retries, optional steps, and execution history:

```yaml theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
name: Robust Workflow
history: true

steps:
  - agent: researcher
    action: Research AI trends
    max_retries: 3
    retry_delay: 1.0

  - agent: enricher
    action: Add examples
    skip_on_failure: true
    retry_delay: 0.5
```

***

## How Graceful Degradation Now Works

A dependent task that opts in still runs even when an upstream dependency in its `context` fails.

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

researcher = Agent(name="researcher", instructions="Research topics")
enricher   = Agent(name="enricher",   instructions="Add examples")

research = Task(name="research", action="Research AI trends", agent=researcher)
enrich   = Task(
    name="enrich",
    action="Add real-world examples",
    agent=enricher,
    context=[research],
    skip_on_failure=True,     # or on_error="continue"
)

team = AgentTeam(agents=[researcher, enricher], tasks=[research, enrich])
team.start()   # even if `research` fails, `enrich` still runs (in degraded mode)
```

***

## Equivalent Flags

Two flags opt into graceful degradation — pick either one.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
    Start{Need a dependent<br/>to keep running?} --> Legacy[skip_on_failure=True]
    Start --> Rich[on_error=continue]
    Legacy --> Continue[Dependent runs<br/>in degraded mode]
    Rich --> Continue
    Start -->|Default| Stop[on_error=stop<br/>workflow fails]

    classDef config fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef success fill:#10B981,stroke:#7C90A0,color:#fff
    classDef fail fill:#8B0000,stroke:#7C90A0,color:#fff

    class Legacy,Rich config
    class Continue success
    class Stop fail
```

* `skip_on_failure=True` — legacy boolean, still works, means "keep the workflow moving".
* `on_error="continue"` — richer three-state field (`"stop"` / `"continue"` / `"retry"`). Prefer this in new code.
* Either flag alone is enough; setting both is redundant but harmless.

***

## Configuration Options

| Param | Type | Default | Description |
| - | - | - | - |
| `skip_on_failure` | `bool` | `False` | Continue workflow if this task fails (legacy; equivalent to `on_error="continue"`). |
| `on_error` | `str` | `"stop"` | Failure behaviour: `"stop"` fails the workflow, `"continue"` keeps it running (equivalent to `skip_on_failure=True`), `"retry"` retries per `max_retries`. |
| `retry_delay` | `float` | `0.0` | Seconds between retries. |
| `max_retries` | `int` | `3` | Maximum retry attempts. |
| `history` | `bool` | `False` | Enable execution trace on `AgentFlow`. |

***

## Task Parameters

### skip\_on\_failure

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
task = Task(
    description="Optional data enrichment",
    skip_on_failure=True,
    agent=enricher,
)
```

### on\_error

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
task = Task(
    description="Optional data enrichment",
    on_error="continue",   # "stop" | "continue" | "retry"
    agent=enricher,
)
```

### retry\_delay

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
task = Task(
    description="API call with rate limiting",
    retry_delay=2.0,
    max_retries=5,
    agent=api_agent,
)
```

***

## Workflow History

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
workflow = AgentFlow(steps=[task1, task2], history=True)
result = workflow.start("input")

for entry in workflow.get_history():
    print(f"{entry['step']}: {'ok' if entry['success'] else 'failed'}")
    if entry.get("error"):
        print(f"  Error: {entry['error']}")
```

| Field | Type | Description |
| - | - | - |
| `step` | `str` | Step or agent name |
| `timestamp` | `str` | ISO timestamp |
| `success` | `bool` | Whether the step succeeded |
| `output` | `str` | Truncated output (500 chars) |
| `error` | `str` | Error message if failed |

***

## Conditional Branching

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlow, when

workflow = AgentFlow(steps=[
    scorer_agent,
    when(
        condition="{{score}} > 80",
        then_steps=[approve_agent],
        else_steps=[reject_agent],
    ),
])
```

> [!WARNING]
>
> `if_()` is deprecated. Use `when()` instead.

***

## Concurrency & Durability

The recipe wrapper serialises shared state and reuses connections so concurrent runs stay correct and cheap.

* **Concurrent-safe run history.** `RunHistory.store()` / `delete()` / `cleanup()` are serialised in-process with a reentrant lock and across processes with a POSIX `fcntl` advisory lock; `_save_index` is atomic (`tmp file + os.replace`). Concurrent completions from `recipe.arun_background()` cannot silently drop index entries, and a crash mid-write cannot truncate `index.json`. See [Run History → Thread- and process-safety](https://praison.ai/docs/cli/recipe-runs.md#thread--and-process-safety).
* **Connection pooling across the wrapper.** `JobHandle`, `JobExecutor` webhooks, and `LangfuseClient` each hold a single lazily-constructed shared HTTP client; the passthrough capability (`praisonai.capabilities.passthrough`) holds one shared sync `httpx.Client` and one async `httpx.AsyncClient` per running event loop (so pooled connections never cross loops — this is what makes `asyncio.run(apassthrough(...))` in a hot path safe). Poll loops, per-agent passthrough calls, and per-lookup Langfuse fetches no longer redo a TLS handshake per call. Deterministic release is provided via `JobHandle.close()` / `with ... as job:`, `LangfuseClient.close()` / `with ... as client:`, `passthrough.close_clients()` (drops every per-loop async client and closes the sync one) / `passthrough.aclose_clients()` (await-closes the current loop's async client from within its own loop), and `JobExecutor.stop()`.
* **Passthrough `api_base` SSRF guard.** On the direct HTTP fallback, `passthrough` / `apassthrough` resolve `api_base`, screen every A/AAAA record (rejecting `localhost`, loopback, private, link-local, multicast, reserved and unspecified addresses), and pin the connection to the screened IP. This is the same check used for job webhooks in `jobs/models.py`. See [Passthrough SSRF Guard](https://praison.ai/docs/features/passthrough-api-base-ssrf.md).

***

## Runtime Resilience

Three runtime guarantees keep long-running agents working after failures.

* **Hooks:** hooked async tools inherit the caller's ContextVars (session identity, cancel, handoff chain).
* **MCP stdio:** crashed stdio MCP servers auto-respawn up to 3 times; the dying call is not replayed. See [MCP stdio](https://praison.ai/docs/mcp/stdio.md#auto-respawn-on-crash).
* **Cancellation:** `/stop` interrupts LLM retry/backoff sleeps within \~200 ms. See [Cooperative Cancellation](https://praison.ai/docs/features/cooperative-tool-cancellation.md#cancel-during-llm-retry-backoff).

***

## Best Practices

> [accordion] Mark optional steps with skip\_on\_failure
>
> Enhancement and enrichment tasks should not block the main workflow path.

> [accordion] Use higher retry\_delay for rate-limited APIs
>
> Set `retry_delay=2.0` or more when calling external APIs with throttling.

> [accordion] Enable history during development
>
> Always set `history=True` while building workflows — inspect `get_history()` after failures.

> [accordion] Separate critical from optional in YAML
>
> Keep required steps strict (`skip_on_failure: false`) and flag enrichers as optional explicitly.

***

## Related

> [card] [Workflows](https://praison.ai/docs/features/workflows.md)
>
> Workflow patterns and orchestration

> [card] [Workflow Error Recovery](https://praison.ai/docs/features/workflow-error-recovery.md)
>
> Recover from workflow failures gracefully


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.