docs/pipeline/error-handling.md

# Error Handling

The pipeline engine guarantees that every sample produces a `SampleResult` — nothing is dropped silently. One failing sample never blocks others. Retry logic is the responsibility of individual steps, not the pipeline.

---

## SampleResult

Every sample that enters `run()` produces exactly one `SampleResult`:

```python
@dataclass
class SampleResult:
    sample: Any                      # the original input
    output: StepContext | None       # final context (None if failed)
    error: Exception | None          # the exception (None if succeeded)
    failed_at: str | None            # step class name where error occurred
    cause: Exception | None = None   # inner exception for BranchError
```

| Field | On success | On failure |
|-------|-----------|------------|
| `sample` | original input | original input |
| `output` | final `StepContext` | `None` |
| `error` | `None` | the exception |
| `failed_at` | `None` | class name of the failing step (e.g. `"Tokenize"`) |
| `cause` | `None` | inner exception when `failed_at == "Branch"` |

!!! note "Background steps"
    For steps after an `async_boundary`, `output` and `error` may still be `None` when `run()` returns. Call `pipe.wait_for_background()` to block until all background work completes and results are finalized.

---

## Construction-time errors

These are caught **before any data flows** — they surface immediately when you build the pipeline.

### PipelineOrderError

Raised when a step requires a field that is produced by a **later** step in the pipeline:

```python
from pipeline import Pipeline
from pipeline.errors import PipelineOrderError

class Uppercase:
    requires = frozenset({"tokens"})
    provides = frozenset({"upper_tokens"})
    def __call__(self, ctx): ...

class Tokenize:
    requires = frozenset()
    provides = frozenset({"tokens"})
    def __call__(self, ctx): ...

try:
    Pipeline().then(Uppercase()).then(Tokenize())
except PipelineOrderError as e:
    print(e)
    # Uppercase requires {"tokens"} but it is provided by a later step
```

!!! tip
    `PipelineOrderError` is always a bug — reorder your steps. Fields not produced by **any** step in the pipeline are treated as external inputs and do not trigger this error.

### PipelineConfigError

Raised for invalid pipeline wiring:

**Multiple async boundaries:**

```python
from pipeline.errors import PipelineConfigError

class StepA:
    requires = frozenset()
    provides = frozenset({"a"})
    async_boundary = True
    def __call__(self, ctx): ...

class StepB:
    requires = frozenset({"a"})
    provides = frozenset({"b"})
    async_boundary = True
    def __call__(self, ctx): ...

try:
    Pipeline().then(StepA()).then(StepB())
except PipelineConfigError:
    print("Only one async_boundary per pipeline is allowed")
```

**Async boundary inside a Branch child:**

```python
try:
    Pipeline().branch(
        Pipeline().then(StepA()),  # async_boundary = True inside branch
    )
except PipelineConfigError:
    print("async_boundary inside Branch children is not allowed")
```

---

## Runtime errors

### Foreground failures

When a step before the `async_boundary` (or in a pipeline with no boundary) raises an exception, the pipeline catches it per-sample and records it in the `SampleResult`:

```python
class Boom:
    requires = frozenset()
    provides = frozenset()

    def __call__(self, ctx):
        raise RuntimeError(f"Failed on {ctx.sample!r}")

pipe = Pipeline().then(Tokenize()).then(Boom())

results = pipe.run([
    StepContext(sample="good"),
    StepContext(sample="also good"),
])

for r in results:
    if r.error:
        print(f"Sample '{r.sample}' failed at {r.failed_at}: {r.error}")
    else:
        print(f"Sample '{r.sample}' succeeded")
```

```
Sample 'good' failed at Boom: Failed on 'good'
Sample 'also good' failed at Boom: Failed on 'also good'
```

Each sample is processed independently — one failure does not prevent others from running.

### Background failures

When a step **after** the `async_boundary` raises, the caller has already moved on. The exception is captured and attached to the `SampleResult` in-place:

```python
pipe = Pipeline().then(Tokenize()).then(BrokenBackgroundStep())

results = pipe.run(samples)
# results returned immediately — background still running

pipe.wait_for_background(timeout=10.0)

# Now check for background failures
for r in results:
    if r.error:
        print(f"Background failure at {r.failed_at}: {r.error}")
```

### PipelineCancelled

When a `cancel_token` is triggered, remaining steps and samples are cancelled with `PipelineCancelled`:

```python
from pipeline import CancellationToken, PipelineCancelled

token = CancellationToken()
# ... later, from another thread or endpoint:
token.cancel()

results = pipe.run(contexts, cancel_token=token)

for r in results:
    if isinstance(r.error, PipelineCancelled):
        print(f"Sample '{r.sample}' was cancelled before {r.failed_at}")
    elif r.error:
        print(f"Sample '{r.sample}' failed at {r.failed_at}: {r.error}")
```

Cancellation is checked **between** steps — a running step always completes. `PipelineCancelled` follows the same `SampleResult` pattern as step errors: it is caught per-sample, not propagated.

### BranchError

When one or more branch pipelines fail, a `BranchError` is raised with the full list of failures:

```python
from pipeline.errors import BranchError

results = pipe.run(contexts)

for r in results:
    if isinstance(r.error, BranchError):
        print(f"{len(r.error.failures)} branch(es) failed:")
        for f in r.error.failures:
            print(f"  {type(f).__name__}: {f}")
    elif r.error:
        print(f"Step failure at {r.failed_at}: {r.error}")
```

All branches run to completion before `BranchError` is raised — no branch is cancelled when another fails. The `SampleResult.cause` field carries the inner exception from the failing branch.

---

## Inspecting results

The standard pattern after `run()`:

```python
results = pipe.run(contexts)
pipe.wait_for_background()  # if using async_boundary

succeeded = [r for r in results if r.error is None]
failed = [r for r in results if r.error is not None]

print(f"{len(succeeded)} succeeded, {len(failed)} failed")

for r in failed:
    print(f"  Sample: {r.sample}")
    print(f"  Failed at: {r.failed_at}")
    print(f"  Error: {r.error}")
```

---

## Background monitoring

### `wait_for_background()`

Blocks until all background tasks complete:

```python
# Wait indefinitely
pipe.wait_for_background()

# Wait with timeout — raises TimeoutError if not done
pipe.wait_for_background(timeout=30.0)
```

Completed threads are removed from the tracking list after this call.

### `background_stats()`

Returns a snapshot of background task progress. Thread-safe — can be called from any thread while the pipeline is running:

```python
stats = pipe.background_stats()
print(stats)
# {'active': 2, 'completed': 8}
```