docs/pipeline/api-reference.md

# API Reference

Complete reference for all public classes, methods, and enums in the pipeline engine.

---

## `pipeline.context`

### `StepContext`

Frozen dataclass passed from step to step. The pipeline engine only reads `sample` and `metadata` — domain-specific fields are added by subclassing.

```python
@dataclass(frozen=True)
class StepContext:
    sample: Any = None
    metadata: MappingProxyType = field(
        default_factory=lambda: MappingProxyType({})
    )
```

| Method | Signature | Description |
|--------|-----------|-------------|
| `replace` | `(**changes: Any) -> StepContext` | Return a new context with the given fields replaced. Uses `dataclasses.replace` internally. |

**Behavior:**

- `metadata` is auto-coerced from `dict` to `MappingProxyType` in `__post_init__`
- Subclasses inherit `.replace()` — it works on all fields including subclass-defined ones

---

## `pipeline.protocol`

### `StepProtocol`

Structural protocol that every step (and Pipeline/Branch) must satisfy.

```python
@runtime_checkable
class StepProtocol(Protocol):
    requires: AbstractSet[str]
    provides: AbstractSet[str]

    def __call__(self, ctx: StepContext) -> StepContext: ...
```

| Attribute | Type | Description |
|-----------|------|-------------|
| `requires` | `AbstractSet[str]` | Metadata keys the step reads |
| `provides` | `AbstractSet[str]` | Metadata keys the step writes |
| `__call__` | `(StepContext) -> StepContext` | Execute the step |

**Notes:**

- `AbstractSet[str]` accepts both `set` and `frozenset`
- `@runtime_checkable` enables `isinstance(step, StepProtocol)` checks

---

### `SampleResult`

Outcome for one sample after the pipeline has run.

```python
@dataclass
class SampleResult:
    sample: Any
    output: StepContext | None
    error: Exception | None
    failed_at: str | None
    cause: Exception | None = None
```

| Field | Type | Description |
|-------|------|-------------|
| `sample` | `Any` | The original input sample |
| `output` | `StepContext \| None` | Final context (`None` if any step failed) |
| `error` | `Exception \| None` | The exception (`None` if succeeded) |
| `failed_at` | `str \| None` | Class name of the step that raised (`None` if succeeded) |
| `cause` | `Exception \| None` | Inner exception for `BranchError` failures (default `None`) |

**Notes:**

- Mutable — background threads update it in-place when background steps complete
- For background steps, `output`/`error` may be `None` until `wait_for_background()` completes

---

## `pipeline.pipeline`

### `Pipeline`

Ordered sequence of steps. Satisfies `StepProtocol` — can be nested inside other pipelines.

#### Constructor

```python
Pipeline(steps: list | None = None, hooks: list[PipelineHook] | None = None)
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `steps` | `list \| None` | `None` | Optional initial list of steps |
| `hooks` | `list[PipelineHook] \| None` | `None` | Observation-only hooks fired around each foreground step |

Validates step ordering and infers contracts at construction time.

#### Attributes

| Attribute | Type | Description |
|-----------|------|-------------|
| `requires` | `frozenset[str]` | Fields the pipeline needs from external context (auto-inferred) |
| `provides` | `frozenset[str]` | Fields the pipeline writes (auto-inferred, union of all steps) |

#### Methods

##### `then`

```python
def then(self, step: object) -> Pipeline
```

Append a step and return `self` for chaining. Validates ordering immediately.

| Parameter | Type | Description |
|-----------|------|-------------|
| `step` | `object` | Any object satisfying `StepProtocol` |

**Returns:** `self` (for method chaining)

**Raises:** `PipelineOrderError` if the step requires a field produced by a later step

---

##### `branch`

```python
def branch(
    self,
    *pipelines: object,
    merge: MergeStrategy | Callable = MergeStrategy.RAISE_ON_CONFLICT,
) -> Pipeline
```

Append a `Branch` step and return `self` for chaining. Shorthand for `.then(Branch(*pipelines, merge=merge))`.

**Returns:** `self` (for method chaining)

---

##### `run`

```python
def run(
    self,
    contexts: Iterable[StepContext],
    workers: int = 1,
    on_sample_done: Callable[[SampleResult], None] | None = None,
    cancel_token: CancellationToken | None = None,
) -> list[SampleResult]
```

Process contexts through the pipeline (sync entry point).

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `contexts` | `Iterable[StepContext]` | — | Input contexts to process |
| `workers` | `int` | `1` | Max concurrent samples in foreground steps |
| `on_sample_done` | `Callable \| None` | `None` | Callback after each sample's foreground steps complete (or fail). Must not block. |
| `cancel_token` | `CancellationToken \| None` | `None` | Cancellation signal. Checked before each step and each new sample. Pass a fresh token per invocation. |

**Returns:** `list[SampleResult]` — one result per input context

**Notes:** Calls `asyncio.run(self.run_async(...))` internally. For background steps, call `wait_for_background()` after this returns. When `cancel_token` is provided, also sets `cancel_token_var` so code inside steps (e.g. LLM clients) can read it.

---

##### `run_async`

```python
async def run_async(
    self,
    contexts: Iterable[StepContext],
    workers: int = 1,
    on_sample_done: Callable[[SampleResult], None] | None = None,
    cancel_token: CancellationToken | None = None,
) -> list[SampleResult]
```

Async entry point. Use `await pipe.run_async(contexts)` from coroutine contexts.

Same parameters and return type as `run()`.

---

##### `__call__`

```python
def __call__(self, ctx: StepContext) -> StepContext
```

Run all steps sequentially on a single context. Used when the pipeline is nested as a step inside another pipeline.

**Notes:** `async_boundary` markers are ignored in this mode — all steps run to completion.

---

##### `wait_for_background`

```python
def wait_for_background(self, timeout: float | None = None) -> None
```

Block until all background tasks complete.

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `timeout` | `float \| None` | `None` | Max seconds to wait. `None` = wait indefinitely. |

**Raises:** `TimeoutError` if timeout elapses before completion

---

##### `background_stats`

```python
def background_stats(self) -> dict[str, int]
```

Return a snapshot of background task progress. Thread-safe.

**Returns:** `{"active": int, "completed": int}`

---

## `pipeline.branch`

### `MergeStrategy`

Enum of built-in merge strategies for `Branch` outputs.

```python
class MergeStrategy(Enum):
    RAISE_ON_CONFLICT = "raise_on_conflict"
    LAST_WRITE_WINS = "last_write_wins"
    NAMESPACED = "namespaced"
```

| Value | Behavior |
|-------|----------|
| `RAISE_ON_CONFLICT` | Raises `ValueError` if two branches write different values to the same named field. Metadata merges with last-writer-wins. |
| `LAST_WRITE_WINS` | Last branch's value wins for every conflicting field. |
| `NAMESPACED` | Each branch's output stored at `metadata["branch_N"]`. No conflict possible. |

---

### `Branch`

Runs multiple pipelines in parallel, then merges their outputs. Satisfies `StepProtocol`.

#### Constructor

```python
Branch(
    *pipelines: object,
    merge: MergeStrategy | Callable = MergeStrategy.RAISE_ON_CONFLICT,
)
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `*pipelines` | `object` | — | Child pipelines to run in parallel (at least one required) |
| `merge` | `MergeStrategy \| Callable` | `RAISE_ON_CONFLICT` | Merge strategy or custom `fn(list[StepContext]) -> StepContext` |

**Raises:** `ValueError` if no pipelines are provided

#### Attributes

| Attribute | Type | Description |
|-----------|------|-------------|
| `requires` | `frozenset[str]` | Union of all children's requires |
| `provides` | `frozenset[str]` | Union of all children's provides |
| `pipelines` | `list` | The child pipelines |

#### Methods

##### `__call__`

```python
def __call__(self, ctx: StepContext) -> StepContext
```

Sync fan-out via `ThreadPoolExecutor`. All branches run to completion before any failure is raised.

**Raises:** `BranchError` if any branch fails

---

##### `__call_async__`

```python
async def __call_async__(self, ctx: StepContext) -> StepContext
```

Async fan-out via `asyncio.gather`. Sync children are wrapped with `asyncio.to_thread`.

**Raises:** `BranchError` if any branch fails

---

## `pipeline.protocol`  — Hooks

### `PipelineHook`

Observation-only protocol fired around each foreground step. Hooks cannot modify context — both methods return `None`.

```python
@runtime_checkable
class PipelineHook(Protocol):
    def before_step(self, step_name: str, ctx: StepContext) -> None: ...
    def after_step(self, step_name: str, ctx: StepContext) -> None: ...
```

| Method | Parameters | Description |
|--------|-----------|-------------|
| `before_step` | `step_name: str, ctx: StepContext` | Called before each foreground step executes |
| `after_step` | `step_name: str, ctx: StepContext` | Called after each foreground step completes |

**Notes:**

- `step_name` is `type(step).__name__` — hooks know what ran but cannot inspect or mutate the step instance
- Hooks fire for foreground steps only — background steps (after `async_boundary`) do not trigger hooks
- If a hook raises, the pipeline logs the error and continues — a broken hook never kills the pipeline
- For `Branch` steps, hooks fire once for `"Branch"` as a whole, not for inner steps

---

## `pipeline.errors`

### `CancellationToken`

Thread-safe cancellation signal. Create a fresh token per `run()` invocation.

```python
class CancellationToken:
    def cancel(self) -> None: ...

    @property
    def is_cancelled(self) -> bool: ...
```

| Method / Property | Description |
|-------------------|-------------|
| `cancel()` | Signal cancellation. Thread-safe, idempotent. |
| `is_cancelled` | `True` after `cancel()` has been called. |

---

### `cancel_token_var`

`ContextVar` set by `Pipeline.run_async()` so code inside steps (e.g. LLM clients) can read the current cancel token without parameter changes.

```python
cancel_token_var: ContextVar[CancellationToken | None]  # default: None
```

**Notes:**

- Set before steps run, reset after `run_async()` completes
- `asyncio.to_thread()` copies contextvars automatically — visible in sync steps too
- Read with `cancel_token_var.get(None)` — returns `None` when no pipeline is running

---

### `PipelineCancelled`

```python
class PipelineCancelled(Exception): ...
```

A `cancel_token` was triggered. Surfaces in `SampleResult.error` — never propagated to the caller of `run()`. Callers check `isinstance(result.error, PipelineCancelled)` to distinguish cancellation from step failures.

---

### `PipelineOrderError`

```python
class PipelineOrderError(Exception): ...
```

A step requires a field that no earlier step provides (but a later step does). Raised at **construction time**.

---

### `PipelineConfigError`

```python
class PipelineConfigError(Exception): ...
```

Invalid pipeline wiring. Raised at **construction time**. Examples:

- More than one `async_boundary = True` step in the same pipeline
- An `async_boundary = True` step inside a `Branch` child

---

### `BranchError`

```python
class BranchError(Exception):
    failures: list[BaseException]
```

One or more branch pipelines failed. All branches always run to completion before this is raised. Raised at **runtime**.

| Attribute | Type | Description |
|-----------|------|-------------|
| `failures` | `list[BaseException]` | One exception per failed branch |

---

## Step class attributes

Optional attributes a step class can declare to control pipeline behavior:

| Attribute | Type | Default | Description |
|-----------|------|---------|-------------|
| `requires` | `set[str] \| frozenset[str]` | *(required)* | Metadata keys the step reads |
| `provides` | `set[str] \| frozenset[str]` | *(required)* | Metadata keys the step writes |
| `async_boundary` | `bool` | `False` | Marks the foreground/background split point |
| `max_workers` | `int` | `1` | Max concurrent background threads for this step class |