# interlock — full documentation

> A modern circuit breaker for Python: sync and async in a single class,
> sliding-window failure-rate and slow-call detection, a type-safe decorator
> API, and transparent per-host integrations for httpx2, aiohttp and requests.
> Zero-dependency core (standard library only); integrations ship as optional
> extras.

This file inlines every documentation page in reading order. It is generated
from the Markdown sources by ``scripts/build_llms_full.py`` — edit the pages in
``docs/``, not this file.


---

<!-- source: docs/getting-started.md -->

# Getting started

## Install

=== "uv"

    ```bash
    uv add interlock-cb
    ```

=== "pip"

    ```bash
    pip install interlock-cb
    ```

=== "poetry"

    ```bash
    poetry add interlock-cb
    ```

The core is pure standard library. External integrations are optional extras —
add the ones you need (same names with `pip install` / `poetry add`):

```bash
uv add 'interlock-cb[fastapi]'  # CircuitOpenError -> 503 + Retry-After
uv add 'interlock-cb[litestar]' # same for Litestar
uv add 'interlock-cb[httpx2]'   # per-host httpx2 transport
uv add 'interlock-cb[httpx]'    # per-host httpx transport
uv add 'interlock-cb[aiohttp]'  # per-host aiohttp client middleware
uv add 'interlock-cb[requests]' # per-host requests session adapter
uv add 'interlock-cb[tenacity]' # retry x breaker composition helpers
uv add 'interlock-cb[redis]'    # shared distributed state
uv add 'interlock-cb[otel]'     # OpenTelemetry metrics listener
```

## Create a breaker

A breaker is named and configured once, then reused:

```python
from interlock import CircuitBreaker, Config

breaker = CircuitBreaker(
    name='payments',
    config=Config(failure_rate_threshold=0.5, minimum_number_of_calls=20),
)
```

The defaults follow resilience4j: trip at a 50% failure rate over at least 10
calls, stay open for 60s, then admit up to 10 probe calls (one at a time) and
decide from their outcomes. See [Configuration](guides/configuration.md) for
every option.

For an existing production dependency, start with
`initial_state=State.METRICS_ONLY`: calls are always admitted while their
outcomes populate the window. Tune thresholds from real traffic, then deploy a
new breaker in the default `CLOSED` state. See
[Safe production rollout](guides/states.md#safe-rollout).

## Three ways to protect work

All three run over the same `call()` primitive.

### Decorator

```python
@breaker
def charge(amount: int) -> str:
    return gateway.charge(amount)
```

The decorator preserves the wrapped signature and its sync/async nature — type
checkers still see `charge` as `(int) -> str`.

### `breaker.call`

```python
result = breaker.call(gateway.charge, 100)
```

`call` inspects the callable to pick the sync or the async path. Where that is
already known — a synchronous client, an awaited handler — `call_sync` and
`call_async` skip the inspection:

```python
result = breaker.call_sync(gateway.charge, 100)
result = await breaker.call_async(client.get, url)
```

`call_sync` never awaits, so passing a coroutine function to it records the
coroutine's creation rather than its outcome. Reach for these on a hot path that
calls the same shape every time; `call` is the right default everywhere else.

### Context manager

```python
with breaker:
    gateway.charge(100)
```

!!! note "Contract difference"
    The decorator and `call` see a callable, so result-based classification and
    slow-call detection both apply. The context manager sees only the block —
    its exception and duration — so classification by **return value** is not
    available there. Need result-based classification? Use the decorator or
    `call`.

## Async

The same instance handles async. The decorator and `call` detect a coroutine
function; the instance is also an async context manager:

```python
@breaker
async def fetch(url: str) -> bytes:
    return await client.get(url)


result = await breaker.call(client.get, url)

async with breaker:
    await client.get(url)
```

## Handle rejections

When the circuit is not closed, the call is rejected with `CircuitOpenError`:

```python
from interlock import CircuitOpenError

try:
    breaker.call(gateway.charge, 100)
except CircuitOpenError as exc:
    # exc.breaker_name, exc.retry_after (seconds, may be None), exc.last_failure
    raise
```

## Inspect state

```python
breaker.state  # State.CLOSED / OPEN / HALF_OPEN / ...
breaker.snapshot()  # WindowSnapshot: total_calls, failed_calls, slow_calls,
# .failure_rate, .slow_call_rate
```

## Next steps

- [Runnable demo](demo.md) — watch a breaker trip and recover, then debug it
- [Configuration](guides/configuration.md)
- [States & manual control](guides/states.md)
- [Failure classification](guides/failure-classification.md)
- [Observability](guides/observability.md)
- [Integrations](integrations/index.md) — FastAPI, Litestar, httpx2, httpx, aiohttp, requests, tenacity, Redis
- [Resilience pipeline](guides/pipeline.md) — compose timeout, bulkhead, retry and fallback around the breaker

---

<!-- source: docs/demo.md -->

# Runnable demo

Three self-contained scripts in [`examples/`](https://github.com/bagowix/interlock/tree/main/examples)
show a breaker doing its job — standard library plus `interlock-cb` only, no
network, no services to stand up. The output is deterministic: what you see
below is exactly what you get, so set a breakpoint anywhere and step through.

```bash
pip install interlock-cb          # or: uv add interlock-cb

python examples/lifecycle.py      # one breaker through its full state cycle
python examples/two_clients.py    # two clients, one outage, no collateral damage
python examples/pipeline.py       # timeout + breaker + fallback around a hanging service
```

Both scripts print through an `EventListener` — every line tagged
`[listener]` comes from the breaker itself, not from the demo code. See
[Observability](guides/observability.md) for the hook reference.

## `lifecycle.py` — one breaker, full cycle

A fake payment gateway is healthy, goes down, and recovers. The breaker is
configured tightly so the whole story fits in seven calls: trip at 50%
failures over a 4-call window, stay open 1 second, close after 2 good probes.

??? example "lifecycle.py — full source"

    ```python
    """Walk one breaker through its full lifecycle: CLOSED -> OPEN -> HALF_OPEN -> CLOSED.

    Zero dependencies, no network — run it directly and watch every transition:

        python examples/lifecycle.py

    The gateway is deterministic (healthy, then down, then recovered), so the
    output is always the same. Tweak the ``Config`` values and re-run to see how
    the thresholds change the story. Explained line by line in
    https://bagowix.github.io/interlock/demo/.
    """

    import time

    from interlock import CircuitBreaker, CircuitOpenError, Config, Outcome, State


    class GatewayError(Exception):
        """The fake dependency's failure mode."""


    class FlakyGateway:
        """A payment gateway you can switch between healthy and down."""

        def __init__(self) -> None:
            self.healthy = True

        def charge(self, amount: int) -> str:
            if not self.healthy:
                raise GatewayError('503 from gateway')

            return f'charged ${amount}'


    class PrintListener:
        """An EventListener that narrates everything the breaker does."""

        def on_state_change(self, *, name: str, old: State, new: State) -> None:
            print(f'  [listener] {name}: state {old.name} -> {new.name}')

        def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None:
            print(f'  [listener] {name}: recorded {outcome.name} ({duration:.3f}s)')

        def on_rejected(self, *, name: str) -> None:
            print(f'  [listener] {name}: call rejected — circuit is open')

        def on_reset(self, *, name: str) -> None:
            print(f'  [listener] {name}: manual reset')


    gateway = FlakyGateway()
    breaker = CircuitBreaker(
        name='payments-api',
        config=Config(
            failure_rate_threshold=0.5,  # trip at >= 50% failures ...
            minimum_number_of_calls=4,  # ... once the window holds 4 calls
            window_size=4,
            wait_duration_in_open=1.0,  # stay OPEN for 1s, then allow probes
            permitted_calls_in_half_open=2,  # 2 good probes close the circuit
        ),
        listener=PrintListener(),
    )


    @breaker
    def charge(amount: int) -> str:
        """The protected call: the decorator preserves this signature."""
        return gateway.charge(amount)


    def attempt(number: int, amount: int) -> None:
        """Make one charge attempt and print what came back."""
        try:
            print(f'call {number}: {charge(amount)}')
        except GatewayError as exc:
            print(f'call {number}: dependency failed — {exc}')
        except CircuitOpenError as exc:
            print(
                f'call {number}: REJECTED in ~0ms, the gateway was never called '
                f'(retry_after={exc.retry_after:.1f}s)'
            )


    def main() -> None:
        """Run the four phases of the lifecycle."""
        print('phase 1 — healthy dependency, breaker CLOSED')
        attempt(1, 10)
        attempt(2, 20)

        print()
        print('phase 2 — the gateway goes down; failures fill the window')
        gateway.healthy = False
        attempt(3, 30)
        attempt(4, 40)  # 2 failures / 4 calls = 50% -> the breaker trips

        print()
        print('phase 3 — circuit OPEN: calls fail fast, the gateway gets a break')
        attempt(5, 50)

        print()
        print('phase 4 — after wait_duration_in_open the breaker probes')
        gateway.healthy = True  # ops fixed the gateway meanwhile
        time.sleep(1.1)
        attempt(6, 60)  # probe 1 (OPEN -> HALF_OPEN on admission)
        attempt(7, 70)  # probe 2 -> both succeeded -> HALF_OPEN -> CLOSED

        print()
        print(f'final state: {breaker.state.name}, window reset: {breaker.snapshot()}')


    if __name__ == '__main__':
        main()
    ```

Running it prints:

```text
phase 1 — healthy dependency, breaker CLOSED
  [listener] payments-api: recorded SUCCESS (0.000s)
call 1: charged $10
  [listener] payments-api: recorded SUCCESS (0.000s)
call 2: charged $20
```

Every completed call is recorded into the sliding window — the listener's
`on_call` hook fires with the classified `Outcome` and the duration.

```text
phase 2 — the gateway goes down; failures fill the window
  [listener] payments-api: recorded FAILURE (0.000s)
call 3: dependency failed — 503 from gateway
  [listener] payments-api: recorded FAILURE (0.000s)
  [listener] payments-api: state CLOSED -> OPEN
call 4: dependency failed — 503 from gateway
```

Call 4 is the moment the window reaches `minimum_number_of_calls=4` with 2
failures out of 4 — exactly the 50% threshold — so recording it trips the
circuit: `on_state_change` fires *inside* call 4's bookkeeping, before the
demo's own `call 4:` line prints.

```text
phase 3 — circuit OPEN: calls fail fast, the gateway gets a break
  [listener] payments-api: call rejected — circuit is open
call 5: REJECTED in ~0ms, the gateway was never called (retry_after=1.0s)
```

Call 5 never reaches the gateway. `CircuitOpenError` is raised immediately
and carries `retry_after` — the estimate until the next probe is allowed.
This is the entire point of a breaker: while the dependency is down, callers
spend no timeouts on it and it gets quiet time to recover.

```text
phase 4 — after wait_duration_in_open the breaker probes
  [listener] payments-api: state OPEN -> HALF_OPEN
  [listener] payments-api: recorded SUCCESS (0.000s)
call 6: charged $60
  [listener] payments-api: recorded SUCCESS (0.000s)
  [listener] payments-api: state HALF_OPEN -> CLOSED
call 7: charged $70

final state: CLOSED, window reset: WindowSnapshot(total_calls=0, failed_calls=0, slow_calls=0)
```

The transition to `HALF_OPEN` is lazy: it happens when call 6 asks for
admission after the 1-second wait, not on a timer (set
`Config.auto_transition=True` for the eager variant). Calls 6 and 7 are the
two permitted probes; both succeed, so recording the second one closes the
circuit and starts a fresh window.

**Things to try:** set `gateway.healthy = False` before phase 4 and watch the
failed probe reopen the circuit; raise `permitted_calls_in_half_open`; put a
breakpoint in `PrintListener.on_state_change` and inspect
`breaker.snapshot()` at each transition.

## `two_clients.py` — isolation under a partial outage

One asyncio app talks to two dependencies. A [`Registry`](reference.md) hands
each its own breaker, so when `recommendations` goes down during rounds 3–6,
only its circuit opens — the app serves a cached fallback — while `payments`
keeps charging as if nothing happened.

??? example "two_clients.py — full source"

    ```python
    """Two guarded clients in one event loop: one dependency dies, the other keeps serving.

    Zero dependencies, no network — run it directly:

        python examples/two_clients.py

    A ``Registry`` hands out one independent breaker per dependency. The
    ``recommendations`` service goes down during rounds 3-6: its breaker opens
    and the app falls back to a cached list, while ``payments`` — its own
    breaker untouched — keeps charging without a hiccup. Explained line by line
    in https://bagowix.github.io/interlock/demo/.
    """

    import asyncio

    from interlock import CircuitOpenError, Config, Outcome, Registry, State


    class RecsDownError(Exception):
        """The recommendations service's failure mode."""


    class PrintListener:
        """An EventListener that narrates state changes and rejections."""

        def on_state_change(self, *, name: str, old: State, new: State) -> None:
            print(f'  [listener] {name}: state {old.name} -> {new.name}')

        def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None:
            pass  # per-call noise is off; see examples/lifecycle.py for it

        def on_rejected(self, *, name: str) -> None:
            print(f'  [listener] {name}: call rejected — circuit is open')

        def on_reset(self, *, name: str) -> None:
            print(f'  [listener] {name}: manual reset')


    registry = Registry(
        config=Config(
            failure_rate_threshold=0.5,
            minimum_number_of_calls=4,
            window_size=4,
            wait_duration_in_open=1.0,
            permitted_calls_in_half_open=2,
        ),
        listener=PrintListener(),
    )

    outage = False


    @registry.get('recommendations')
    async def fetch_recommendations(user: str) -> list[str]:
        """Call the flaky recommendations service."""
        if outage:
            raise RecsDownError('recommendations service timed out')
        return [f'{user}-pick-1', f'{user}-pick-2']


    @registry.get('payments')
    async def charge(user: str, amount: int) -> str:
        """Call the payments service, which stays healthy throughout."""
        return f'charged {user} ${amount}'


    async def round_trip(number: int, user: str) -> None:
        """One request round: hit both dependencies concurrently."""
        print(f'round {number}:')
        payment, recs = await asyncio.gather(
            charge(user, 25),
            fetch_recommendations(user),
            return_exceptions=True,
        )
        print(f'  payments        -> {payment}')
        if isinstance(recs, CircuitOpenError):
            print('  recommendations -> rejected instantly -> fallback: cached picks')
        elif isinstance(recs, RecsDownError):
            print(f'  recommendations -> failed ({recs}) -> fallback: cached picks')
        else:
            print(f'  recommendations -> {recs}')


    async def main() -> None:
        """Run eight rounds across the outage and the recovery."""
        global outage  # noqa: PLW0603 - a module-level switch keeps the demo flat

        print('rounds 1-2 — both dependencies healthy')
        await round_trip(1, 'alice')
        await round_trip(2, 'bob')

        print()
        print('rounds 3-6 — recommendations goes down; payments must not care')
        outage = True
        await round_trip(3, 'carol')
        await round_trip(4, 'dave')  # 50% failures -> recommendations breaker opens
        await round_trip(5, 'erin')  # rejected in ~0ms: no timeout is burned
        await round_trip(6, 'frank')

        print()
        print('rounds 7-8 — the outage is over; the breaker probes and closes')
        outage = False
        await asyncio.sleep(1.1)  # let wait_duration_in_open elapse
        await round_trip(7, 'grace')
        await round_trip(8, 'heidi')

        print()
        for name in ('payments', 'recommendations'):
            print(f'final state of {name}: {registry.get(name).state.name}')


    if __name__ == '__main__':
        asyncio.run(main())
    ```

The interesting part of the output:

```text
round 4:
  [listener] recommendations: state CLOSED -> OPEN
  payments        -> charged dave $25
  recommendations -> failed (recommendations service timed out) -> fallback: cached picks
round 5:
  [listener] recommendations: call rejected — circuit is open
  payments        -> charged erin $25
  recommendations -> rejected instantly -> fallback: cached picks
```

Round 4's failure is the second in a 4-call window — the `recommendations`
breaker trips. From round 5 on the difference matters: the round-4 request
*waited* for the dependency to fail, the round-5 request is rejected in
microseconds, so the user still gets their page (with cached picks) at full
speed. Notice what is absent: the `payments` breaker never logs a single
state change for the whole run.

```text
rounds 7-8 — the outage is over; the breaker probes and closes
round 7:
  [listener] recommendations: state OPEN -> HALF_OPEN
  payments        -> charged grace $25
  recommendations -> ['grace-pick-1', 'grace-pick-2']
round 8:
  [listener] recommendations: state HALF_OPEN -> CLOSED
  payments        -> charged heidi $25
  recommendations -> ['heidi-pick-1', 'heidi-pick-2']
```

Recovery is gradual by design: two successful probes (rounds 7 and 8) must
complete before the circuit closes again.

**Things to try:** extend the outage past round 7 and watch a failed probe
send the circuit straight back to `OPEN`; give the two breakers different
configs via `registry.get(name, config=...)`; replace the hand-rolled
fallback with a [tenacity retry](integrations/tenacity.md) that waits
exactly `retry_after`.

## `pipeline.py` — composition against a quiet death

The nastiest failure mode gets the [v2 pipeline](guides/pipeline.md)
treatment: the inventory service never raises — it *hangs*. On its own that
trips nothing and starves every caller. Three composed strategies turn it
into a non-event: a timeout makes hangs classifiable, the breaker counts
them, a fallback keeps serving.

??? example "pipeline.py — full source"

    ```python
    """Compose timeout + breaker + fallback around one dependency that dies quietly.

    Zero dependencies, no network — run it directly:

        python examples/pipeline.py

    The inventory service never raises: it *hangs*. Alone, that starves callers
    and trips nothing. The pipeline turns the hang into a failure chain: the
    timeout cancels the attempt, the breaker counts it and opens, the fallback
    serves a cached snapshot — and while the circuit is open, requests cost
    ~0.0s instead of a timeout each. Add `.retry(...)` between the fallback and
    the breaker for the full stack (needs `interlock-cb[tenacity]`). Explained
    line by line in https://bagowix.github.io/interlock/demo/.
    """

    import asyncio
    import time

    from interlock import (
        CallTimeoutError,
        CircuitBreaker,
        CircuitOpenError,
        Config,
        Outcome,
        Pipeline,
        State,
    )

    CACHED_SNAPSHOT = ['widget (cached)', 'gadget (cached)']


    class PrintListener:
        """Narrates the breaker's transitions and the pipeline's decisions."""

        def on_state_change(self, *, name: str, old: State, new: State) -> None:
            print(f'  [listener] {name}: state {old.name} -> {new.name}')

        def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None:
            pass  # per-call noise is off; see examples/lifecycle.py for it

        def on_rejected(self, *, name: str) -> None:
            print(f'  [listener] {name}: call rejected — circuit is open')

        def on_reset(self, *, name: str) -> None:
            print(f'  [listener] {name}: manual reset')

        def on_fallback(self, *, name: str, error: BaseException) -> None:
            print(f'  [listener] {name}: fallback served instead of {type(error).__name__}')


    events = PrintListener()
    hanging = False
    breaker = CircuitBreaker(
        name='inventory',
        config=Config(
            failure_rate_threshold=0.5,  # trip at >= 50% failures ...
            minimum_number_of_calls=4,  # ... once the window holds 4 calls
            window_size=4,
            wait_duration_in_open=1.0,  # stay OPEN for 1s, then allow probes
            permitted_calls_in_half_open=2,  # 2 good probes close the circuit
        ),
        listener=events,
    )

    pipeline = (
        Pipeline.builder()
        .fallback(
            lambda _exc: CACHED_SNAPSHOT,
            on=(CircuitOpenError, CallTimeoutError),
            name='inventory',
            listener=events,
        )
        .circuit_breaker(breaker)
        .timeout(0.2)  # a hanging attempt becomes CallTimeoutError after 0.2s
        .build()
    )


    @pipeline
    async def fetch_inventory() -> list[str]:
        """The protected call: the dependency hangs instead of erroring."""
        if hanging:
            await asyncio.sleep(5)  # never finishes within the timeout
        return ['widget', 'gadget']


    async def request(number: int) -> None:
        """Serve one request and print what it cost."""
        start = time.perf_counter()
        items = await fetch_inventory()
        elapsed = time.perf_counter() - start
        print(f'request {number}: {items} in ~{elapsed:.1f}s')


    async def main() -> None:
        """Run the outage story end to end."""
        global hanging  # noqa: PLW0603 - a module-level switch keeps the demo flat

        print('phase 1 — healthy and fast, breaker CLOSED')
        await request(1)
        await request(2)

        print()
        print('phase 2 — the service starts hanging; timeouts become failures')
        hanging = True
        await request(3)  # waits the full 0.2s timeout, then serves the cache
        await request(4)  # 2 timeouts / 4 calls = 50% -> the breaker trips

        print()
        print('phase 3 — circuit OPEN: no timeout is burned, the cache is instant')
        await request(5)

        print()
        print('phase 4 — the service recovers; probes close the circuit')
        hanging = False
        await asyncio.sleep(1.1)  # let wait_duration_in_open elapse
        await request(6)  # probe 1 (OPEN -> HALF_OPEN on admission)
        await request(7)  # probe 2 -> both succeeded -> HALF_OPEN -> CLOSED

        print()
        print(f'final state: {breaker.state.name}')


    if __name__ == '__main__':
        asyncio.run(main())
    ```

The interesting part of the output:

```text
phase 2 — the service starts hanging; timeouts become failures
  [listener] inventory: fallback served instead of CallTimeoutError
request 3: ['widget (cached)', 'gadget (cached)'] in ~0.2s
  [listener] inventory: state CLOSED -> OPEN
  [listener] inventory: fallback served instead of CallTimeoutError
request 4: ['widget (cached)', 'gadget (cached)'] in ~0.2s

phase 3 — circuit OPEN: no timeout is burned, the cache is instant
  [listener] inventory: call rejected — circuit is open
  [listener] inventory: fallback served instead of CircuitOpenError
request 5: ['widget (cached)', 'gadget (cached)'] in ~0.0s
```

Watch the latency column. Requests 3–4 each pay the full 0.2 s timeout —
that is the *detection* cost while the window fills. Request 4 tips the
failure rate to 50% and the circuit opens; from request 5 on the rejection
is immediate and the cached snapshot costs ~0.0 s. The user never saw an
error: every response during the outage came from the fallback, and the
listener logged each substitution.

```text
phase 4 — the service recovers; probes close the circuit
  [listener] inventory: state OPEN -> HALF_OPEN
request 6: ['widget', 'gadget'] in ~0.0s
  [listener] inventory: state HALF_OPEN -> CLOSED
request 7: ['widget', 'gadget'] in ~0.0s
```

**Things to try:** raise `wait_duration_in_open` and watch how long the
cache serves; put a `.retry(...)` step between the fallback and the breaker
(needs `interlock-cb[tenacity]`) and see attempts in the listener via
`on_retry`; drop the `.timeout(...)` step and watch the outage become
invisible again — no strategy ever fires.

## Where to go next

These demos hand-roll their guarded clients so you can debug the mechanics.
In real code you usually don't have to: the [integrations](integrations/index.md)
apply the same per-dependency pattern to httpx2, httpx, aiohttp, requests and
FastAPI transparently, and the [resilience pipeline](guides/pipeline.md)
composes the breaker with timeout, bulkhead, retry and fallback declaratively.

---

<!-- source: docs/comparison.md -->

# Comparison

Picking a circuit breaker is mostly about which failure model, runtime and
coordination story you need. This page compares interlock-cb with the
established Python circuit breakers — honestly. interlock-cb is the youngest
of the five (first released in 2026); [pybreaker](https://github.com/danielfm/pybreaker),
[circuitbreaker](https://github.com/fabfuel/circuitbreaker),
[aiobreaker](https://github.com/arlyon/aiobreaker) and
[purgatory](https://github.com/mardiros/purgatory) have carried production
traffic for years, and for many projects one of them is exactly the right
choice.

## Feature table

| Feature | interlock-cb | pybreaker | circuitbreaker | aiobreaker | purgatory |
|---|:---:|:---:|:---:|:---:|:---:|
| Core states (closed / open / half-open) | ✅ | ✅ | ✅ | ✅ | ✅ |
| Choose which exceptions count as failures | ✅ | ✅ | ✅ | ✅ | ✅ |
| Zero-dependency core | ✅ | ✅ | ✅ | ✅ | ✅ |
| `async` / `await` (asyncio) | ✅ | Tornado | ✅ | ✅ | ✅ |
| Sync and async in **one** breaker class | ✅ | — | ✅ | ✅ | separate factories |
| Trip condition | failure **rate** over a window | consecutive count | consecutive count | consecutive count | consecutive count |
| Time-based sliding window | ✅ | — | — | — | — |
| Slow-call detection | ✅ | — | — | — | — |
| Result-based failure classification | ✅ | — | — | — | — |
| Event / state-change listeners | ✅ | ✅ | — | ✅ | ✅ |
| Shared state across processes (Redis) | ✅ | ✅ | — | ✅ | ✅ |
| Globally budgeted recovery probes | ✅ | — | — | — | — |
| Fallback function | ✅ | — | ✅ | — | — |
| Composable resilience pipeline (retry × breaker × bulkhead × timeout × fallback) | ✅ | — | — | — | — |
| Fully typed API (`py.typed`) | ✅ | — | — | — | ✅ |
| Signature-preserving decorator (`ParamSpec`) | ✅ | — | — | — | — |
| HTTP client integrations (httpx2 / httpx / aiohttp / requests) | ✅ | — | — | — | — |
| Retry composition helpers (tenacity) | ✅ | — | — | — | — |
| OpenTelemetry metrics | ✅ | — | — | — | — |
| Operator overrides (force-open / disable / shadow mode) | ✅ | — | — | — | — |
| Years of production use | new | ✅ | ✅ | ✅ | ✅ |
| Latest release (as of August 2026) | 2.6.0 · 2026 | 1.4.1 · 2025 | 2.1.3 | 1.2.0 · 2021 | 3.0.1 · 2024 |
| Python | ≥ 3.11 | ≥ 3.9 | ≥ 3.8 | ≥ 3.6 | ≥ 3.9 |

<sub>Compared against pybreaker 1.4.1, circuitbreaker 2.1.3, aiobreaker 1.2.0
and purgatory 3.0.1, as documented in July 2026. Something out of date or
unfair? Please
[open a PR](https://github.com/bagowix/interlock/pulls).</sub>

## The four established libraries, honestly

**pybreaker** is the original Python circuit breaker and the most proven one:
a small, stable sync breaker with listeners and optional Redis-backed state,
maintained for well over a decade. Its async support targets Tornado, not
asyncio, and it trips on a consecutive-failure count rather than a failure
rate. If you run a synchronous stack and want the most battle-tested option,
start here.

**circuitbreaker** has the smallest API of the five: one `@circuit` decorator
that also handles async functions, plus the only built-in **fallback
function** in this table. There are no listeners, no shared state and no rate
window — which is precisely its appeal when you need a guard on a handful of
call sites and nothing else.

**aiobreaker** is pybreaker ported to native asyncio, with the same
listener and Redis-storage features. Its last release was in 2021; for new
projects, prefer an actively maintained alternative.

**purgatory** brings a fully typed sync + async breaker with Redis storage
and event hooks. Sync and async live in separate factory classes
(`SyncCircuitBreakerFactory` / `AsyncCircuitBreakerFactory`), and tripping is
a consecutive-failure threshold with a TTL on the open state.

## Where interlock-cb differs

- **Rate over a window, not a streak.** A consecutive-failure counter resets
  on any single success, so a dependency failing 90% of requests under load
  can keep a breaker closed indefinitely. interlock trips on the failure
  *rate* across a count- or time-based sliding window
  ([configuration](guides/configuration.md)).
- **Slow calls are failures too.** A dependency that answers in 30 s can be
  worse than one that errors fast. `slow_call_duration_threshold` +
  `slow_call_rate_threshold` trip the breaker on latency degradation alone.
- **One class, both runtimes.** The same `CircuitBreaker` instance guards sync
  and async callables — decorator, context manager, or `call()` — and the
  decorator preserves the wrapped signature for type checkers.
- **Coordination built for fleets.** With the Redis storage, tripping is
  atomic across racing instances and half-open probes are budgeted globally,
  with graceful degradation to local state when Redis is down
  ([Redis integration](integrations/redis.md)).
- **Meets your stack where it is.** Per-host breakers ship for httpx2, httpx,
  aiohttp and requests; tenacity glue composes retries correctly; FastAPI and
  Litestar map rejections to `503 + Retry-After`
  ([integrations overview](integrations/index.md)).

The honest trade-off: interlock-cb requires Python ≥ 3.11 and has not had
years in production. Reach for an established
library if that maturity matters more than the feature gap; choose
interlock-cb when you want rate-based windows, slow-call detection,
coordinated state and a fully typed API.

---

<!-- source: docs/migration.md -->

# Migrating from pybreaker / circuitbreaker

Already using [pybreaker](https://github.com/danielfm/pybreaker) or
[circuitbreaker](https://github.com/fabfuel/circuitbreaker)? Moving to
interlock-cb is mostly mechanical — swap the import, translate the constructor
arguments, and keep the rest of your call sites. This page maps every concept
across, one library at a time.

If you have not decided *whether* to move yet, read the
[Comparison](comparison.md) first — this page assumes you have.

## The one conceptual change

Every established Python breaker trips on a **consecutive-failure count**:
`fail_max` / `failure_threshold` failures *in a row* open the circuit, and a
single success resets the counter. interlock trips on a **failure rate over a
sliding window** instead ([configuration](guides/configuration.md)).

That difference is the whole reason to migrate — a dependency failing 40% of
requests under load never trips a streak counter, because successes keep
resetting it — but it means the threshold numbers do **not** carry over
one-to-one. Everything else (timing, decorator, `call`, listeners) has a direct
equivalent.

Translate the trip condition like this:

| Old (streak) | interlock (rate over a window) |
|---|---|
| `fail_max=5` / `failure_threshold=5` | `minimum_number_of_calls` = how many calls to observe before trusting a rate; `failure_rate_threshold` = the fraction that trips |
| "open on 5 failures in a row" | e.g. `Config(minimum_number_of_calls=5, failure_rate_threshold=0.8)` — trip when ≥ 80% of the last window is failing |
| `reset_timeout` / `recovery_timeout` | `wait_duration_in_open` (seconds) — direct equivalent |

There is no exact arithmetic conversion, because the two models answer
different questions. A safe way to pick numbers is to run in
[shadow mode](guides/states.md#safe-rollout) first (see
[Roll out incrementally](#roll-out-incrementally) below) and read the real
rates off `breaker.snapshot()` before enforcing.

---

## From pybreaker

### Constructor

```python
# before — pybreaker
import pybreaker

breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    exclude=[ValueError],
    name='payments',
)
```

```python
# after — interlock
from interlock import CircuitBreaker, Config

breaker = CircuitBreaker(
    name='payments',
    config=Config(
        minimum_number_of_calls=5,
        failure_rate_threshold=0.8,
        wait_duration_in_open=60.0,
    ),
)
```

| pybreaker | interlock | Notes |
|---|---|---|
| `fail_max` | `minimum_number_of_calls` + `failure_rate_threshold` | Streak → rate; see [above](#the-one-conceptual-change). |
| `reset_timeout` | `Config.wait_duration_in_open` | Direct, in seconds. |
| `success_threshold` | `Config.permitted_calls_in_half_open` | Probes admitted before the breaker re-decides. interlock decides on the probe *rate*, not a fixed success count. |
| `exclude=[...]` | a `FailureClassifier` | See [below](#exclude-failureclassifier). |
| `name=` | `name=` | Same. |
| `state_storage=CircuitRedisStorage(...)` | `storage=RedisStorage(...)` | See [below](#redis-shared-state). |
| `listeners=[...]` | `listener=` | See [below](#listeners). |

### Decorator and `call`

Both keep working with the same shape — only the object changes:

```python
# before                         # after
@breaker                         @breaker
def charge(amount): ...          def charge(amount): ...

breaker.call(charge, 100)        breaker.call(charge, 100)
```

interlock's decorator additionally **preserves the wrapped signature** for type
checkers, and the same instance also works as a `with` block and on `async`
functions — no separate class ([getting started](getting-started.md#async)).

### `exclude` → `FailureClassifier`

pybreaker's `exclude` lists exceptions that should *not* count as failures.
interlock expresses the same policy as a
[classifier](guides/failure-classification.md):

```python
# before — pybreaker
breaker = pybreaker.CircuitBreaker(exclude=[ValueError])
```

```python
# after — interlock
class IgnoreValueError:
    def is_failure(self, *, result: object, exception: Exception | None) -> bool:
        if isinstance(exception, ValueError):
            return False  # business error, not a dependency problem
        return exception is not None


breaker = CircuitBreaker(name='payments', classifier=IgnoreValueError())
```

The classifier is strictly more capable: it also sees the **return value**, so
you can count an HTTP `503` response object as a failure — something no
exclude-list can do.

### Listeners

pybreaker's `CircuitBreakerListener` (`before_call` / `state_change` /
`failure` / `success`) maps onto interlock's
[`EventListener`](guides/observability.md):

| pybreaker | interlock |
|---|---|
| `state_change(cb, old, new)` | `on_state_change(*, name, old, new)` |
| `failure(cb, exc)` / `success(cb)` | `on_call(*, name, outcome, duration)` — `outcome.is_failure` distinguishes them |
| `before_call(cb, func, ...)` | — (no per-call pre-hook; use `on_call` after the fact) |
| — | `on_rejected(*, name)` fires when an open circuit rejects a call |

```python
# after — interlock
from interlock import State, Outcome


class Payments:
    def on_state_change(self, *, name: str, old: State, new: State) -> None:
        print(f'{name}: {old} -> {new}')

    def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None: ...
    def on_rejected(self, *, name: str) -> None: ...
    def on_reset(self, *, name: str) -> None: ...


breaker = CircuitBreaker(name='payments', listener=Payments())
```

A built-in `LoggingEventListener` covers the common case with zero code.

### Redis (shared state)

```python
# before — pybreaker
import redis
import pybreaker

breaker = pybreaker.CircuitBreaker(
    state_storage=pybreaker.CircuitRedisStorage(pybreaker.STATE_CLOSED, redis.StrictRedis()),
)
```

```python
# after — interlock
import redis
from interlock import CircuitBreaker
from interlock.integrations.redis import RedisStorage

breaker = CircuitBreaker(
    name='payments',
    storage=RedisStorage(redis.Redis(host='redis.internal')),
)
```

interlock's coordination is stronger than shared counters: tripping is atomic
across racing instances, half-open probes are **budgeted globally** (N
instances send at most `permitted_calls_in_half_open` probes *in total*), and a
Redis outage degrades to local state instead of failing your calls
([Redis integration](integrations/redis.md)).

### State inspection and the open error

```python
breaker.current_state  # pybreaker → 'open' / 'half-open' / 'closed'
breaker.state  # interlock → State.OPEN / HALF_OPEN / CLOSED
```

```python
# before                                    # after
from pybreaker import CircuitBreakerError    from interlock import CircuitOpenError

try:                                         try:
    breaker.call(charge, 100)                    breaker.call(charge, 100)
except CircuitBreakerError:                  except CircuitOpenError as exc:
    ...                                          # exc.retry_after, exc.breaker_name,
                                                 # exc.last_failure
    ...
```

`CircuitOpenError` carries a `retry_after` estimate (seconds until the next
probe), which the [FastAPI](integrations/fastapi.md) /
[Litestar](integrations/litestar.md) extras turn into `503 + Retry-After`
automatically.

---

## From circuitbreaker

### Decorator

```python
# before — circuitbreaker
from circuitbreaker import circuit


@circuit(failure_threshold=5, recovery_timeout=30, expected_exception=ConnectionError)
def external_call(): ...
```

```python
# after — interlock
from interlock import CircuitBreaker, Config

breaker = CircuitBreaker(
    name='external_call',
    config=Config(
        minimum_number_of_calls=5,
        failure_rate_threshold=0.8,
        wait_duration_in_open=30.0,
    ),
    classifier=OnlyConnectionError(),  # see below
)


@breaker
def external_call(): ...
```

| circuitbreaker | interlock | Notes |
|---|---|---|
| `failure_threshold` | `minimum_number_of_calls` + `failure_rate_threshold` | Streak → rate. |
| `recovery_timeout` | `Config.wait_duration_in_open` | Direct, in seconds. |
| `expected_exception` | a `FailureClassifier` | Only these count as failures; see [below](#expected_exception-failureclassifier). |
| `fallback_function` | `FallbackStrategy` in a pipeline | See [below](#fallback_function-fallbackstrategy). |
| `name=` | `name=` | interlock requires a name explicitly. |

interlock separates the breaker object from the decorator: build one
`CircuitBreaker` and apply it with `@breaker`, rather than configuring a fresh
circuit at each decoration site. To reuse config across many call sites, share
a [`Registry`](guides/configuration.md#sharing-config-with-a-registry).

### `expected_exception` → `FailureClassifier`

circuitbreaker's `expected_exception` is the *inverse* of pybreaker's
`exclude` — it names the exceptions that **do** count. Same classifier tool:

```python
class OnlyConnectionError:
    def is_failure(self, *, result: object, exception: Exception | None) -> bool:
        return isinstance(exception, ConnectionError)
```

### Class-based breakers

```python
# before — circuitbreaker
from circuitbreaker import CircuitBreaker


class ApiBreaker(CircuitBreaker):
    FAILURE_THRESHOLD = 10
    RECOVERY_TIMEOUT = 60
    EXPECTED_EXCEPTION = ConnectionError
```

There is no subclassing in interlock — the same intent is a reusable `Config`
value (and a classifier), which you pass wherever you need it:

```python
# after — interlock
from interlock import Config

API_CONFIG = Config(
    minimum_number_of_calls=10,
    failure_rate_threshold=0.8,
    wait_duration_in_open=60.0,
)
```

### `fallback_function` → `FallbackStrategy`

circuitbreaker calls `fallback_function` when the circuit is open. interlock
keeps the breaker a pure gate and layers the fallback with the v2.0
[pipeline](guides/pipeline.md), so the substitution is explicit about *which*
failures it stands in for:

```python
# before — circuitbreaker
@circuit(fallback_function=lambda: [])
def recommendations(): ...
```

```python
# after — interlock
from interlock import CircuitBreaker, CircuitOpenError, Pipeline

breaker = CircuitBreaker(name='recommendations')

pipeline = (
    Pipeline.builder()
    .fallback(lambda exc: [], on=(CircuitOpenError,))
    .circuit_breaker(breaker)
    .build()
)


@pipeline
def recommendations(): ...
```

The pipeline is also where you compose retries, bulkheads and timeouts around
the same breaker — none of which the old decorator offers.

### Async and monitoring

`@circuit` handles async functions; so does interlock's `@breaker` — with the
**same instance**, no separate class or import. circuitbreaker's
`CircuitBreakerMonitor` (enumerate all circuits, check `all_closed()`) has no
direct analogue; hold your breakers in a [`Registry`](guides/configuration.md)
and iterate that, or observe transitions through the
[listener](guides/observability.md).

`CircuitBreakerError` → `CircuitOpenError`, exactly as in the
[pybreaker section](#state-inspection-and-the-open-error).

---

## aiobreaker and purgatory

[aiobreaker](https://github.com/arlyon/aiobreaker) is pybreaker ported to
asyncio — follow the [pybreaker section](#from-pybreaker); its `fail_max` /
`timeout_duration` / `CircuitBreakerListener` map the same way, and interlock's
single class removes the need for an asyncio-specific breaker at all.

[purgatory](https://github.com/mardiros/purgatory) splits sync and async into
`SyncCircuitBreakerFactory` / `AsyncCircuitBreakerFactory` with a
`default_threshold` (consecutive) and a `default_ttl` on the open state. Map the
factory + `get_breaker(name)` pattern onto a single interlock
[`Registry`](guides/configuration.md#sharing-config-with-a-registry) whose
`registry.get(name)` returns one dual sync/async breaker; `default_ttl` becomes
`wait_duration_in_open`, and `default_threshold` follows the same streak → rate
translation.

---

## What actually changes at runtime

After migrating, expect these behavioural differences — all intended:

- **Trips reflect the rate, not a streak.** A dependency that fails
  intermittently under load will now trip where a streak counter never did.
  Conversely, a single burst of failures below `minimum_number_of_calls` will
  *not* trip — the window has to fill first.
- **Slow calls can trip too, but only when you opt in.**
  `slow_call_rate_threshold` defaults to `1.0`, so latency alone never trips
  until you tune it down ([configuration](guides/configuration.md#why-slow-calls-matter)).
- **Half-open is a budgeted probe round, not a single trial call.** Up to
  `permitted_calls_in_half_open` probes run (with a concurrency cap), and the
  breaker re-decides from their rate ([states](guides/states.md)).

## Roll out incrementally

You do not have to trust new threshold numbers on day one. Ship the breaker in
[shadow mode](guides/states.md#safe-rollout) — it records real
failure and slow-call rates without rejecting anything — tune against live data,
then switch to enforcing:

```python
breaker.metrics_only()  # observe production, reject nothing
# ... read breaker.snapshot().failure_rate / .slow_call_rate over real traffic ...
breaker.reset()  # start enforcing with a clean window
```

## Next steps

- [Configuration](guides/configuration.md) — pick your thresholds and window
- [Failure classification](guides/failure-classification.md) — port `exclude` / `expected_exception`
- [Observability](guides/observability.md) — port your listeners
- [Resilience pipeline](guides/pipeline.md) — port `fallback_function`, add retries and timeouts
- [Redis integration](integrations/redis.md) — port shared state

---

<!-- source: docs/correctness.md -->

# Correctness and testing

interlock-cb is young — [Comparison](comparison.md) says so plainly. What it
can offer instead of years in production is a bar most libraries in this space
do not hold themselves to, and a way to check that the bar is real rather than
promised. Every claim below links to the config, workflow or test file that
enforces it, so it can be verified against the repository at any commit, not
just taken on trust.

## Types

- **Three independent checkers, strict mode**: [mypy](https://mypy-lang.org/)
  (`strict = true`), [pyright](https://microsoft.github.io/pyright/)
  (`typeCheckingMode = "strict"`) and
  [pyrefly](https://pyrefly.org/) (`preset = "strict"`) all run over
  `interlock/` and `tests/typing_surface.py` on every pull request
  ([`pyproject.toml`](https://github.com/bagowix/interlock/blob/main/pyproject.toml),
  [`ci.yml`](https://github.com/bagowix/interlock/blob/main/.github/workflows/ci.yml)).
  Three implementations rather than one because they disagree often enough for
  it to matter — each catches inference gaps the others miss.
- **The public surface is asserted, not just checked.**
  [`tests/typing_surface.py`](https://github.com/bagowix/interlock/blob/main/tests/typing_surface.py)
  uses `assert_type` on `call()`'s overloads and the decorator's signature
  preservation. It is never executed; it is included in the mypy/pyright/pyrefly
  scope, so a regression in what a type checker *infers* at a call site fails
  CI even though nothing here runs at test time.
- **Public API breakage is diffed, not just typed.**
  [`api-compatibility.yml`](https://github.com/bagowix/interlock/blob/main/.github/workflows/api-compatibility.yml)
  runs [`griffe check`](https://mkdocstrings.github.io/griffe/) against the
  latest release tag on every pull request — covering `interlock/__init__.py`'s
  re-exports and `interlock/integrations/*`, which is public without being
  re-exported. A detected breakage fails the build unless the PR carries the
  `breaking-change` label.
- Ships `py.typed`; `ParamSpec` + `TypeVar` decorators preserve both the
  wrapped signature and its sync/async nature.

## Coverage

- **100% branch coverage, enforced, not aspirational**:
  `fail_under = 100` under `[tool.coverage.report]`
  ([`pyproject.toml`](https://github.com/bagowix/interlock/blob/main/pyproject.toml)),
  checked with `branch = true` so a branch taken only one way still fails the
  gate. Reported through [Codecov](https://codecov.io/gh/bagowix/interlock)
  (project **and** patch targets at 100% —
  [`codecov.yml`](https://github.com/bagowix/interlock/blob/main/.github/codecov.yml)).
- **What coverage doesn't prove** is exactly why the next four sections exist —
  see [Mutation testing](#mutation-testing) below.

## Property-based and model-based tests

Coverage proves every line ran; it says nothing about whether a test would
notice if the logic were wrong. Two Hypothesis suites target the state machine
specifically, both driven by an injected `FakeClock`
([`tests/conftest.py`](https://github.com/bagowix/interlock/blob/main/tests/conftest.py)),
never real time:

- [`test_state_machine_properties.py`](https://github.com/bagowix/interlock/blob/main/tests/test_state_machine_properties.py) —
  `@given` properties over hand-written sequences, checking four invariants:
  the minimum-calls gate, a saturated window, the `OPEN` wait, and the probe
  caps.
- [`test_state_machine_model.py`](https://github.com/bagowix/interlock/blob/main/tests/test_state_machine_model.py) —
  a Hypothesis `RuleBasedStateMachine` that generates the *sequence itself*:
  interleaved outcomes, clock advances, admissions and operator overrides,
  checked against an independently predicted state, generation, window and
  probe budget after every step. This is the suite that catches
  order-dependent bugs — an override landing mid-probe-round, a probe settling
  a generation late — that a fixed sequence cannot reach. Reaching those
  interleavings is not left to chance: the rule mix and a `target()` objective
  steer the search into `HALF_OPEN`, and the share of examples that get there
  is measured whenever a rule changes, because a generated sequence that never
  reaches the interesting state proves nothing.

When the model finds a counterexample, the shrunk sequence gets pinned as a
named regression test next to the model, so the reproducer survives the model
being changed later.

## Mutation testing

100% branch coverage proves every line executes; it does not prove an
assertion would catch a wrong value. [`mutmut`](https://mutmut.readthedocs.io/)
closes that gap for the two modules where a surviving mutant is a real bug:
[`interlock/_state_machine.py`](https://github.com/bagowix/interlock/blob/main/interlock/_state_machine.py)
(threshold arithmetic, transition ordering, probe admission) and
[`interlock/_engine.py`](https://github.com/bagowix/interlock/blob/main/interlock/_engine.py)
(lock scope, dispatch, recording order) —
[`[tool.mutmut]`](https://github.com/bagowix/interlock/blob/main/pyproject.toml)
in `pyproject.toml`.

Baseline: **526 of 549 mutants killed (95.8%)**. Every survivor is an
equivalent mutant in one of five enumerated classes — an unread sentinel
default, a comparison that only ever uses `!=`, a value used solely as a
boolean — documented with the reasoning in
[`CONTRIBUTING.md`](https://github.com/bagowix/interlock/blob/main/CONTRIBUTING.md#mutation-testing).
A survivor outside those classes is treated as a missing test, not as grounds
to raise the budget.

Run weekly and on demand
([`mutation.yml`](https://github.com/bagowix/interlock/blob/main/.github/workflows/mutation.yml)),
never as a pull-request gate — the signal is a slowly-moving score, not
something that should block an unrelated change. The hypothesis suites are
excluded from the run: `mutmut` maps tests to functions from a single stats
pass, and a randomised test that reaches a branch only sometimes would make
both that mapping and the score irreproducible.

## An I/O-free state machine, driven by an injected clock

[`interlock/_state_machine.py`](https://github.com/bagowix/interlock/blob/main/interlock/_state_machine.py)
never touches a socket, a file, or `time.time()` — all time comes from a
`Clock` passed in at construction. Every test in the suite that exercises
timed transitions does so with `FakeClock.advance(seconds)`
([`tests/conftest.py`](https://github.com/bagowix/interlock/blob/main/tests/conftest.py)),
never `sleep`; `test_examples.py` is the one deliberate exception, since it
runs the `examples/` scripts themselves as a subprocess smoke test. This is
what makes the property and model-based suites above practical at all — a
`RuleBasedStateMachine` generating thousands of transition sequences would be
too slow to run against a real clock.

## Concurrency, including free-threaded CPython

- The `threading.Lock` in
  [`interlock/_engine.py`](https://github.com/bagowix/interlock/blob/main/interlock/_engine.py)
  covers only the two await-free critical sections (admission and recording);
  the protected callable itself runs outside it.
- [`tests/test_concurrency.py`](https://github.com/bagowix/interlock/blob/main/tests/test_concurrency.py)
  drives one breaker from many real threads at once and checks window counts
  add up, `snapshot()` is never torn, the `HALF_OPEN` caps hold, and a
  concurrent trip emits exactly one `CLOSED → OPEN` event.
- **CI runs the full matrix on `3.14t`, the free-threaded build, as a required
  job** — Python 3.11, 3.12, 3.13, 3.14 and 3.14t, `fail-fast: false`
  ([`ci.yml`](https://github.com/bagowix/interlock/blob/main/.github/workflows/ci.yml)).
  A GIL-enabled interpreter cannot falsify a thread-safety claim resting on one
  lock: with `Engine._lock` removed, `test_concurrency.py` fails on 3.14t and
  passes everywhere else. This is the reason the job exists rather than being
  aspirational — it is the only Python build in the matrix that can actually
  disprove the claim.

## Storage contract and coordinated (distributed) breakers

- [`tests/test_storage_contract.py`](https://github.com/bagowix/interlock/blob/main/tests/test_storage_contract.py)
  runs a single behavioural contract suite against the in-memory reference
  storage
  ([`tests/inmemory_storage.py`](https://github.com/bagowix/interlock/blob/main/tests/inmemory_storage.py)) —
  the same contract every `Storage` / `AsyncStorage` implementation, including
  Redis, is expected to satisfy.
- [`tests/test_coordination.py`](https://github.com/bagowix/interlock/blob/main/tests/test_coordination.py)
  covers trip propagation, the global probe budget, coordinated close, and
  degradation-and-recovery across a shared storage — fully deterministic, on a
  shared `FakeClock` with a manually driven `poll_once()`.
- [`tests/test_redis_storage.py`](https://github.com/bagowix/interlock/blob/main/tests/test_redis_storage.py)
  runs against in-process `fakeredis` by default (so `uv run pytest` needs no
  server) and against a real Redis service container in CI, which is the
  authoritative check for Lua-script atomicity under concurrency. `3.14t`
  specifically requires the real server: `fakeredis`'s Lua engine (`lupa`)
  re-enables the GIL on import, which `filterwarnings = "error"` turns into a
  collection failure.

## Lower-bound dependency versions are tested, not guessed

Every optional extra declares a minimum version in `pyproject.toml`
(`httpx>=0.27.0`, `httpx2>=2.4.0`, `redis>=5.0.0`, and so on). The
`extras-min` job re-pins each
one to exactly that floor and runs the integration suite against it
([`ci.yml`](https://github.com/bagowix/interlock/blob/main/.github/workflows/ci.yml)) —
so a lower bound is a tested claim, not an untested guess about how far back
compatibility actually reaches. The latest versions of each extra are covered
separately by the main matrix.

## Warnings are errors

`filterwarnings = ["error"]`
([`pyproject.toml`](https://github.com/bagowix/interlock/blob/main/pyproject.toml))
turns every `DeprecationWarning` and `RuntimeWarning` raised during a test run
into a failure, with one narrow, commented exception for a third-party import
warning on CPython 3.11. This is what makes the `3.14t` / `fakeredis` /
`lupa` interaction above a hard collection failure instead of a silent GIL
re-enable, and it means an internal deprecation (`InterlockDeprecationWarning`)
must be explicitly asserted in the test that triggers it, never silenced.

## Linting

`ruff` runs with `select = ["ALL"]`
([`pyproject.toml`](https://github.com/bagowix/interlock/blob/main/pyproject.toml)),
including the `S` (flake8-bandit) security ruleset and `flake8-tidy-imports`
bans on legacy `typing` aliases. Ignored rules are listed individually in
`pyproject.toml`, each implicitly scoped to what it actually silences, not
disabled wholesale.

## Hot-path performance

[CodSpeed](https://codspeed.io) measures CPU instructions (not wall-clock
time, so results stay stable on shared CI runners) over the call paths, the
state machine, the sliding windows and the pipeline on every pull request
([`codspeed.yml`](https://github.com/bagowix/interlock/blob/main/.github/workflows/codspeed.yml)).
A regression is a signal, not silence — this is a performance floor, not a
correctness one, and is listed here for completeness rather than as a
correctness claim.

## Supply chain

Correctness of the code is only half the trust question; the other half is
whether what gets published is what was reviewed. See
[`SECURITY.md`](https://github.com/bagowix/interlock/blob/main/SECURITY.md#supply-chain)
for the full breakdown — trusted publishing with no long-lived PyPI token,
Sigstore-backed [PEP 740](https://peps.python.org/pep-0740/) provenance on
every release artifact, every GitHub Actions `uses:` pinned to a full commit
SHA and audited by [zizmor](https://docs.zizmor.sh) on every pull request, and
CodeQL default setup over both Python and the workflows themselves. The
[OpenSSF Scorecard](https://scorecard.dev/viewer/?uri=github.com/bagowix/interlock)
badge in the README is the standing summary of this section, refreshed weekly
and on every push to `main`. The project also holds the
[OpenSSF Best Practices](https://www.bestpractices.dev/projects/13932) badge —
the self-certified checklist covering release process, vulnerability
reporting, static/dynamic analysis and secure-development practices, most of
it satisfied by the mechanisms this page already documents.

## Known limits

An honest boundary is more persuasive than a claimed clean sweep:

- **Mutation testing covers two modules.** `_state_machine.py` and `_engine.py`
  are where a surviving mutant is unambiguously a bug; the rest of the package
  (integrations, the pipeline strategies, the registry) is covered by branch
  coverage and example-based tests only, not mutation testing.
- **Mutation testing excludes the Hypothesis suites**, for reproducibility —
  see [Mutation testing](#mutation-testing) above. A branch reached only by a
  property or model-based test can show up as a mutation survivor even when a
  property test does in fact cover it.
- **Free-threaded coverage is one test file's job.** `test_concurrency.py` is
  the suite specifically designed to fail without the engine's lock; the rest
  of the suite runs on 3.14t too, but is not designed to detect races the way
  that file is.
- **No production track record.** interlock-cb was first released in 2026 —
  see [Comparison](comparison.md). None of the above substitutes for years of
  real-world traffic; it is what can be verified today in place of that
  history.
- **Fuzzing is not part of this suite.** The OpenSSF Scorecard `Fuzzing` check
  is expected to stay red — the input surface here is typed function calls and
  config values, not a parser or protocol decoder, so structure-aware fuzzing
  does not apply the way it would to a format library.
- **Branch protection is not machine-verifiable from the public Scorecard
  run.** The `Branch-Protection` check needs a PAT with read access to
  repository settings that the workflow does not have; the settings themselves
  are configured in GitHub, not in a file this page can link to.

---

<!-- source: docs/guides/configuration.md -->

# Configuration

`Config` is an immutable (frozen) dataclass validated on construction. Pass it
to a `CircuitBreaker` or share it across a `Registry`. All fields are
keyword-only.

```python
from interlock import Config
from interlock import WindowType

config = Config(
    failure_rate_threshold=0.5,
    minimum_number_of_calls=20,
    slow_call_duration_threshold=2.0,
    slow_call_rate_threshold=1.0,
    permitted_calls_in_half_open=10,
    max_concurrent_probes=1,
    wait_duration_in_open=30.0,
    window_type=WindowType.COUNT_BASED,
    window_size=100,
)
```

## Fields

| Field | Default | Meaning |
|-------|---------|---------|
| `failure_rate_threshold` | `0.5` | Trip when the failure rate reaches this fraction. Range `(0, 1]`. |
| `minimum_number_of_calls` | `10` | Minimum calls in the window before a rate is trusted. Guards against `1/1 = 100%`. |
| `slow_call_duration_threshold` | `60.0` | Calls at or above this many seconds are **slow**. |
| `slow_call_rate_threshold` | `1.0` | Trip when the slow-call rate reaches this fraction. Range `(0, 1]`. |
| `permitted_calls_in_half_open` | `10` | Probe calls allowed while `HALF_OPEN`. |
| `max_concurrent_probes` | `1` | Cap on **simultaneous** probes in `HALF_OPEN`. Must be in `[1, permitted_calls_in_half_open]`. |
| `wait_duration_in_open` | `60.0` | Seconds to stay `OPEN` before the first probe is allowed. |
| `auto_transition` | `False` | When `True`, a timer moves the breaker `OPEN → HALF_OPEN` once the wait elapses, instead of waiting for the next call. See [States](states.md#proactive-transition-auto_transition). |
| `window_type` | `COUNT_BASED` | `COUNT_BASED` or `TIME_BASED`. |
| `window_size` | `100` | Last N calls (count-based) or last N seconds (time-based). |

Validation raises `ValueError` eagerly for out-of-range or inconsistent values,
so a misconfigured breaker fails at construction rather than in production.

## Windows

- **Count-based** keeps the last `window_size` calls. Predictable memory,
  independent of traffic rate. The default.
- **Time-based** keeps calls from the last `window_size` seconds. The right
  choice for high-throughput services where "last N calls" is a moving target.

```python
from interlock import Config, WindowType

# Trip on a 50% failure rate observed over the last 30 seconds.
Config(window_type=WindowType.TIME_BASED, window_size=30)
```

## Why slow calls matter

A dependency that answers slowly but never errors will never trip a
failure-rate breaker, yet it still exhausts your timeouts and threads.
Slow-call detection treats latency as a first-class failure signal. By default
`slow_call_rate_threshold=1.0` means slowness alone never trips the breaker
until you tune it down — safe to leave on while you observe.

## Sharing config with a Registry

```python
from interlock import Config, Registry, State

registry = Registry(
    config=Config(minimum_number_of_calls=20),
    initial_state=State.METRICS_ONLY,
)

payments = registry.get('payments')  # shared default
search = registry.get('search', config=Config(window_size=500))  # per-name override
```

The override applies only when the breaker is first created; later `get` calls
with the same name return the existing instance and ignore the `config`
argument. `initial_state` is also assigned once, inside the registry lock,
before a newly created breaker can serve traffic. See
[Safe rollout](states.md#safe-rollout) for using
`METRICS_ONLY` in production.

---

<!-- source: docs/guides/states.md -->

# States & manual control

A breaker has three core states plus three operator overrides.

## Core lifecycle

```mermaid
stateDiagram-v2
    CLOSED --> OPEN: failure/slow rate crosses threshold
    OPEN --> HALF_OPEN: first call after wait_duration_in_open (or timer, if auto_transition)
    HALF_OPEN --> CLOSED: probe round passes
    HALF_OPEN --> OPEN: probe round fails
```

- **`CLOSED`** — traffic flows; outcomes are recorded. When the failure rate (or
  slow-call rate) crosses its threshold over at least `minimum_number_of_calls`,
  the breaker trips to `OPEN`.
- **`OPEN`** — calls are rejected immediately with `CircuitOpenError`. After
  `wait_duration_in_open` seconds, the **next** call lazily moves the breaker to
  `HALF_OPEN`. Enable [`auto_transition`](#proactive-transition-auto_transition)
  to have a timer make that move on its own.
- **`HALF_OPEN`** — up to `permitted_calls_in_half_open` probe calls are
  admitted, with a cap on how many run concurrently, so a barely-recovered
  dependency is not hit by the full parallel load at once. Once the round
  completes, the breaker decides from the probes' outcomes using the same
  thresholds as `CLOSED`: rates below the thresholds close it, at or above
  re-open it. Calls beyond the probe caps are rejected while the round runs.

## Proactive transition (`auto_transition`)

By default the `OPEN → HALF_OPEN` move is **lazy**: it happens on the first call
after `wait_duration_in_open` elapses. A low-traffic service can therefore sit in
`OPEN` longer than necessary, and — since nothing changes until that call — the
state-change event is not emitted, leaving a blind spot on dashboards.

Set `auto_transition=True` to arm a timer that performs the move on its own when
the wait elapses, emitting `on_state_change` without waiting for a call:

```python
from interlock import CircuitBreaker, Config

breaker = CircuitBreaker(
    name='payments',
    config=Config(wait_duration_in_open=30.0, auto_transition=True),
)
# 30s after opening, the breaker moves to HALF_OPEN and emits the event,
# even if no call arrives.
```

The lazy path stays authoritative: the timer only flips the state (it admits no
probe), so the first real call still becomes the first probe. If a call arrives
exactly as the timer fires, a lock ensures the transition and its event happen
exactly once. The timer is cancelled automatically on `reset()`, `force_open()`,
or when a call makes the move first.

The timer is a daemon thread, used uniformly for sync and async breakers (the
breaker's critical sections are guarded by a `threading.Lock`, never an event
loop), so a pending timer never blocks interpreter shutdown.

## Operator overrides

Three special states are set manually and stay until you `reset()`:

| Method | State | Behaviour |
|--------|-------|-----------|
| `breaker.force_open()` | `FORCED_OPEN` | Reject all traffic regardless of metrics. |
| `breaker.disable()` | `DISABLED` | Admit all traffic but record no outcome — thresholds are never evaluated and `snapshot()` gets nothing new. Listener `on_call` events still fire. |
| `breaker.metrics_only()` | `METRICS_ONLY` | Admit all traffic, record metrics, but never trip. |
| `breaker.reset()` | `CLOSED` | Return to closed with a fresh, empty window. In coordinated mode, resume the cached shared state instead. |

```python
breaker.metrics_only()  # observe in production without enforcing
# ... inspect breaker.snapshot() until thresholds look right ...
breaker.reset()  # start enforcing with a clean window
```

### What an override does to your metrics

Two observability surfaces are in play, and an override does not move them
together:

- the sliding **window** — what `snapshot()` reports and what the thresholds
  read. `METRICS_ONLY` keeps filling it (that is the whole point of shadow
  mode); `DISABLED` records nothing and `FORCED_OPEN` admits nothing to record,
  so neither feeds it. A count-based window then keeps its last contents
  unchanged; a time-based one drains as its buckets expire;
- the **`EventListener`**, which observes calls rather than the window.
  `on_call` fires whenever an admitted call settles, whatever the state —
  `DISABLED` included; `on_rejected` fires for every rejected call,
  `FORCED_OPEN` included.

So `disable()` is not a way to silence a listener: `LoggingEventListener`, the
`OTelEventListener` or a Prometheus exporter keeps reporting outcomes and
durations for a disabled breaker, with the classifier still deciding success
from failure. That is deliberate — dashboards going dark the moment an operator
disables a breaker looks exactly like an outage. To stop the events, drop the
listener instead (construct the breaker without one).

Switching a rollout from `metrics_only()` to `disable()` therefore keeps
listener-exported dashboards alive, and only stops threshold evaluation and
`snapshot()`.

### Safe rollout

Shadow mode is the key to introducing a breaker without risk: it records the
exact failure and slow-call rates real traffic produces, so you can tune
thresholds against live data before letting the breaker reject anything. It
costs almost nothing to leave on.

Set the mode at construction when no call may be admitted first:

```python
from interlock import CircuitBreaker, Registry, State

breaker = CircuitBreaker(name='payments', initial_state=State.METRICS_ONLY)
registry = Registry(initial_state=State.METRICS_ONLY)
```

`Registry` applies the state while holding its creation lock and publishes the
breaker only afterwards. Every name created later therefore starts in shadow
mode too. Construction is not a state transition, so it does not emit a
synthetic `CLOSED → METRICS_ONLY` listener event.

Only stable states are valid at construction: `CLOSED`, `FORCED_OPEN`,
`DISABLED` and `METRICS_ONLY`. `OPEN` and `HALF_OPEN` require timing, probe and
failure history, so passing either as `initial_state` raises `ValueError`.

For a production rollout:

1. Deploy with `initial_state=State.METRICS_ONLY` and an `EventListener` that
   exports call outcomes.
2. Observe failure and slow-call rates, then tune `Config` against real traffic.
3. Deploy a new breaker, registry or transport with `initial_state=State.CLOSED`
   (the default). The enforcing instance starts with a fresh window.

Prefer a new deployment for step 3. Calling `reset()` enforces immediately for
an existing breaker, but a registry configured with `METRICS_ONLY` would still
apply that original initial state to hosts first seen later.

For local diagnosis, `registry.get_existing(name)` returns a cached breaker or
`None` without creating one. Inspect its `state` and `snapshot()`; use listeners
rather than polling snapshots for production metrics.

When the names are not known in advance — the HTTP transports create one
breaker per host, lazily — `registry.items()` lists every breaker created so
far, and `registry.names()` just their names. Both return a point-in-time copy,
so they also drive bulk operator actions:

```python
for _, breaker in registry.items():
    breaker.metrics_only()
```

## Coordinated state (optional)

With a shared [storage](../integrations/redis.md), `OPEN` and `HALF_OPEN` can
also be *adopted* from other instances: a trip anywhere in the fleet gates
admission everywhere, and the HALF_OPEN probe budget is shared globally.
`breaker.state` then reports the effective state — the shared one when it
governs admission, the local one otherwise (including while the storage is
unreachable).

Local operator overrides always take precedence over a healthy shared view:
`force_open()` rejects locally, while `disable()` and `metrics_only()` admit
locally without claiming a shared HALF_OPEN probe. `reset()` clears that local
override and freshens local metrics; it does not change the cluster. The
instance immediately resumes the cached shared `OPEN` or `HALF_OPEN` state.

## Observing transitions

Every transition (and reset) is delivered to the breaker's
[`EventListener`](observability.md), so you can log or export state changes
without polling `breaker.state`.

---

<!-- source: docs/guides/failure-classification.md -->

# Failure classification

What counts as a failure is a separate concern from *when to trip* (thresholds,
in [Config](configuration.md)). It is decided by a `FailureClassifier`.

## Default policy

By default, a call is a failure exactly when it **raises**, and any returned
value is a success:

```python
from interlock import CircuitBreaker

breaker = CircuitBreaker(name='svc')  # DefaultFailureClassifier
```

This is right for code that signals errors by raising. It is *not* enough when
failure is encoded in a **return value** — for example an HTTP response object
whose `503` status means the dependency is unhealthy.

## Classify by result

A classifier implements one method. The `result`/`exception` pair is mutually
exclusive: when `exception` is not `None` the call raised; otherwise `result`
holds the return value. `exception` is always an `Exception` — a
`BaseException` such as `CancelledError` says nothing about the dependency, so
the breaker releases the call without classifying it.

```python
from interlock import CircuitBreaker


class StatusClassifier:
    def is_failure(self, *, result: object, exception: Exception | None) -> bool:
        if exception is not None:
            return True
        return getattr(result, 'status_code', 200) >= 500


breaker = CircuitBreaker(name='api', classifier=StatusClassifier())
result = breaker.call(client.get, url)  # a 503 response now counts as a failure
```

Result-based classification needs the return value, so it works with the
**decorator** and **`call`**, but not the context manager (which only sees
exceptions and duration).

## Ignore expected errors

Business errors — a `404`, a validation failure — should not open the circuit.
Encode that by treating only the exceptions you care about as failures:

```python
class IgnoreNotFound:
    def is_failure(self, *, result: object, exception: Exception | None) -> bool:
        if isinstance(exception, NotFoundError):
            return False  # expected, not a dependency problem
        return exception is not None
```

An ignored exception is recorded as a **success** — the sliding window has only
two outcomes — and still propagates to the caller. It therefore dilutes the
failure rate rather than being invisible to it.

## HTTP out of the box

For HTTP clients, you do not need to write this yourself — the
[httpx2](../integrations/httpx2.md), [httpx](../integrations/httpx.md),
[aiohttp](../integrations/aiohttp.md), and
[requests](../integrations/requests.md) integrations ship
`HttpStatusClassifier`, which treats the canonical retryable statuses
(`429, 500, 502, 503, 504`) and every non-excluded transport exception as
failures.

Errors the *caller* caused are excluded for the same reason as a `404`: a
scheme-less URL or a local protocol violation is a bug in your code, not
evidence that the dependency is unhealthy, and counting it would open the
circuit of a host that is answering fine. The default set differs per
integration, because each client library raises different things inside the
guarded call:

| Integration | Excluded by default |
|---|---|
| [httpx2](../integrations/httpx2.md), [httpx](../integrations/httpx.md) | `UnsupportedProtocol`, `LocalProtocolError` |
| [requests](../integrations/requests.md) | `InvalidURL` (and its `InvalidProxyURL` subclass) |
| [aiohttp](../integrations/aiohttp.md) | nothing — it rejects malformed URLs before the middleware chain runs, so every handler exception counts unless you exclude it |

All four take `excluded_exceptions=(...)` to replace that set — `()` counts
every exception as a failure. Entries must be `Exception` subclasses;
anything else raises `TypeError` at construction.

---

<!-- source: docs/guides/observability.md -->

# Observability

A breaker reports everything it does through an `EventListener`. The same
hooks back logging, metrics, and any custom sink.

## The hooks

```python
from typing import Protocol

from interlock import Outcome, State


class CoreEventListener(Protocol):
    def on_state_change(self, *, name: str, old: State, new: State) -> None: ...
    def on_call(self, *, name: str, outcome: Outcome, duration: float) -> None: ...
    def on_rejected(self, *, name: str) -> None: ...
    def on_reset(self, *, name: str) -> None: ...


class StorageEventListener(Protocol):
    def on_storage_degraded(self, *, name: str, error: BaseException) -> None: ...
    def on_storage_recovered(self, *, name: str) -> None: ...
    def on_storage_write_dropped(self, *, name: str) -> None: ...


class PipelineEventListener(Protocol):
    def on_retry(self, *, name: str, attempt: int, delay: float) -> None: ...
    def on_bulkhead_rejected(self, *, name: str) -> None: ...
    def on_fallback(self, *, name: str, error: BaseException) -> None: ...


class EventListener(
    CoreEventListener,
    StorageEventListener,
    PipelineEventListener,
    Protocol,
): ...
```

Core listeners are called **outside** the breaker's lock, so a slow listener
never serialises throughput. `on_call` runs after the protected call completes;
transition, rejection, and reset hooks run immediately after their corresponding
breaker operation. Storage hooks run from the coordinated breaker's background
lane when its backend degrades, recovers, or drops a queued write.

Pipeline hooks run inside their strategy's execution path: before retry backoff,
when bulkhead admission fails, or before a fallback returns its substitute.

The protocols follow the owner of each event. `CoreEventListener` observes the
breaker itself, `StorageEventListener` observes breakers coordinated through a
shared [storage](../integrations/redis.md), and `PipelineEventListener` observes
[pipeline strategies](pipeline.md) given a `listener=`. `EventListener` combines
all three for sinks that observe the complete library.

The `name` namespace follows the same boundary: core and storage hooks receive
the **breaker name**, while pipeline hooks receive the **strategy name**. A
listener can therefore use `name` directly as its breaker or strategy label
without rediscovering the distinction from the call site.

Every hook is dispatched only if present, and each protocol supplies no-op
implementations for subclasses. A listener can override just the hooks it cares
about and keeps working when a later interlock version adds a new hook.

## Listener failures are isolated

Observability is optional; the protected call is not. A listener runs on the
breaker's own paths, so a bug in one — a metrics exporter with a stale label,
a logging handler with a full queue — would otherwise become a new failure
source. interlock guarantees it cannot:

- an `Exception` raised by a hook is logged to the `interlock` logger at
  `ERROR` (with the traceback) and then ignored;
- the protected call keeps its result, and a failing call keeps *its own*
  exception — a listener never masks the dependency's error;
- state transitions, probe accounting and the coordinated-mode background lane
  continue unaffected;
- a raising hook is never reported as storage degradation.

`BaseException` is **not** caught: `KeyboardInterrupt` and
`asyncio.CancelledError` propagate from a hook exactly as they do everywhere
else in interlock.

This covers `EventListener` hooks only. User-supplied *policy* callbacks are
part of the call's behaviour rather than observations of it, and keep raising
as before: a [`FailureClassifier`](failure-classification.md), a
[fallback](pipeline.md) function, a tenacity `before_sleep` hook.

To be notified when your own listener misbehaves, watch the `interlock`
logger:

```python
import logging

logging.getLogger('interlock').setLevel(logging.ERROR)
```

Attach one per breaker, or share one across a `Registry`:

```python
breaker = CircuitBreaker(name='payments', listener=my_listener)
registry = Registry(listener=my_listener)  # every breaker reports here
```

## Logging (zero dependencies)

`LoggingEventListener` is built in. State changes and rejections log at
`WARNING`, resets at `INFO`, and individual calls at `DEBUG`:

```python
from interlock import CircuitBreaker, LoggingEventListener

breaker = CircuitBreaker(name='payments', listener=LoggingEventListener())
```

Pass your own logger to control routing:

```python
import logging

LoggingEventListener(logging.getLogger('myapp.breakers'))
```

## OpenTelemetry metrics

The OTel listener lives in the `interlock-cb[otel]` extra and is imported
explicitly, so the core stays dependency-free:

```bash
uv add 'interlock-cb[otel]'
```

Supports `opentelemetry-api>=1.20.0` — the listener only calls
`get_meter`/`create_histogram`/`create_counter`, stable across any
`opentelemetry-distro`/SDK release from that version on.

```python
from interlock import CircuitBreaker
from interlock.integrations.otel import OTelEventListener

breaker = CircuitBreaker(name='payments', listener=OTelEventListener())
```

It records five instruments on the `interlock` meter (or a meter you pass in):

| Instrument | Type | Labels |
|------------|------|--------|
| `interlock.call.duration` | histogram (s) | `breaker`, `outcome` |
| `interlock.call.rejected` | counter | `breaker` |
| `interlock.state.changes` | counter | `breaker`, `from`, `to` |
| `interlock.reset` | counter | `breaker` |
| `interlock.storage.events` | counter | `breaker`, `event` (`degraded`/`recovered`/`write_dropped`), `error` |

## Custom listeners

For a partial listener that strict type checkers can verify, inherit the narrowest
protocol for its owner and override only the hooks you need. Every inherited hook
is a no-op:

```python
from interlock import CoreEventListener


class RejectionCounter(CoreEventListener):
    def __init__(self) -> None:
        self.rejected = 0

    def on_rejected(self, *, name: str) -> None:
        self.rejected += 1
```

Use `StorageEventListener` for storage-only sinks, `PipelineEventListener` for
strategy-only sinks, or `EventListener` when one object handles every group.
Inheritance is optional for a listener that structurally implements the relevant
protocol. At runtime, dispatch is still by name and skips any missing hook,
including on older listener objects that inherit none of them.

---

<!-- source: docs/guides/timeout.md -->

# Timeout

A circuit breaker without a timeout is unsafe. A call that hangs forever is
never counted as slow or failed — it just holds a resource indefinitely.
`timeout` bounds an awaited block and turns a hang into a `CallTimeoutError`,
which a surrounding breaker records as a (slow) failure.

```python
from interlock import timeout

async with timeout(2.0):
    await client.get(url)  # raises CallTimeoutError after 2 seconds
```

## Composing with a breaker

Compose `timeout` with a breaker manually — pipeline composition is a v2
feature. Put the timeout *inside* the protected callable so the breaker observes
the `CallTimeoutError`:

```python
from interlock import CircuitBreaker, timeout

breaker = CircuitBreaker(name='search')


@breaker
async def search(q: str) -> bytes:
    async with timeout(2.0):
        return await client.get('/search', params={'q': q})
```

Now a request that exceeds 2 seconds raises `CallTimeoutError`; the breaker
counts it as a failure and, once the failure rate crosses the threshold, opens
the circuit — converting slow hangs into fast rejections.

## Synchronous code

`timeout` relies on asyncio cancelling the coroutine in place, which has no
synchronous equivalent: a blocking call cannot be interrupted from outside its
own thread, and `signal.SIGALRM` only works in the main thread, so it breaks in
threaded servers. `sync_timeout` instead runs the callable in a daemon worker
thread and joins it with a deadline. It is a decorator, so it wraps a *callable*
rather than a block:

```python
from interlock import CircuitBreaker, sync_timeout

breaker = CircuitBreaker(name='search')


@breaker
@sync_timeout(2.0)
def search(q: str) -> bytes:
    return client.get('/search', params={'q': q}).content
```

A call that exceeds 2 seconds raises `CallTimeoutError`, which the breaker
records exactly as with the async path. The decorator preserves the wrapped
function's signature, arguments and return value.

!!! warning "The worker keeps running after a timeout"
    Python cannot forcibly kill a thread. After `sync_timeout` raises, the
    worker thread keeps running in the background until the call returns on its
    own — it cannot be cancelled, so it may still hold the resource it was
    waiting on. The caller is unblocked immediately, but the underlying work is
    not stopped. Prefer the async `timeout` wherever you control an event loop;
    reach for `sync_timeout` only in genuinely synchronous code.

## Why not bake it in?

interlock keeps retry, fallback and timeout as explicit, observable features
rather than hidden magic inside the breaker. You decide the deadline at the call
site, and the failure it produces flows through the same classification and
metrics as any other.

---

<!-- source: docs/guides/retries.md -->

# Retries and circuit breakers

Retries and circuit breakers pull in opposite directions: a retry *adds*
load to a struggling dependency, a breaker *sheds* it. Combined carelessly
they cancel each other out — retries hammer a dependency the breaker is
trying to protect, or the breaker's window never sees the real failure rate.
This guide fixes the composition; the ready-made tenacity helpers live in the
[tenacity integration](../integrations/tenacity.md).

## Which goes on the outside?

Both orders are valid — they answer different questions. What changes is what
the breaker's sliding window *sees*:

| Order | What the window sees | When to choose |
|---|---|---|
| **Retry outside → breaker inside** (recommended) | Every attempt individually — honest failure rate, the breaker trips as early as the dependency deserves | Default. Also the order used by Polly and resilience4j |
| Breaker outside → retry inside | One aggregated outcome per *operation* (all attempts folded into it) | When thresholds are tuned per business operation, not per request |

With retry outside, a rejected attempt is also visible to the retry loop —
which is exactly where the two failure modes below come from.

## Failure mode 1: retrying an open circuit

`CircuitOpenError` is not a transient error. The breaker rejects instantly,
so an exponential backoff loop around it burns its attempt budget in
milliseconds, never reaches the dependency, and buries the real signal in log
noise. Stop retrying the moment the circuit opens:

```python
from tenacity import Retrying, stop_after_attempt, wait_exponential_jitter

from interlock.integrations.tenacity import retry_unless_open

retrying = Retrying(
    retry=retry_unless_open(TimeoutError, ConnectionError),
    wait=wait_exponential_jitter(),
    stop=stop_after_attempt(5),
    reraise=True,
)
```

## Failure mode 2: blind waiting

Sometimes waiting *is* the right call — a nightly job would rather sleep
than fail. But `2^n` seconds is the wrong amount: the breaker already knows
when it will allow the next probe (`CircuitOpenError.retry_after`). Wait
exactly that long:

```python
from tenacity import AsyncRetrying, retry_if_exception_type, stop_after_attempt
from tenacity import wait_exponential_jitter

from interlock import CircuitOpenError
from interlock.integrations.tenacity import wait_probe

retrying = AsyncRetrying(
    retry=retry_if_exception_type((TimeoutError, CircuitOpenError)),
    wait=wait_probe(wait_exponential_jitter()),
    stop=stop_after_attempt(10),
    reraise=True,
)
```

Pick one mode per call site. Fail fast at request/latency-sensitive
boundaries; be patient in background work.

## Retrying on HTTP statuses

The HTTP integrations classify statuses for the *breaker* without raising —
a `503` response is returned to you, recorded as a failure. tenacity,
however, is exception-driven. Do **not** reach for `retry_if_result`: with
aiohttp a retried-away response is never released and leaks its connection.
Turn bad statuses into exceptions instead, then retry exceptions:

```python
import requests
from tenacity import Retrying, stop_after_attempt, wait_exponential_jitter

from interlock.integrations.requests import CircuitBreakerAdapter
from interlock.integrations.tenacity import retry_unless_open

session = requests.Session()
session.mount('https://', CircuitBreakerAdapter())


def fetch_orders() -> dict:
    response = session.get('https://api.example.com/orders')
    response.raise_for_status()
    return response.json()


retrying = Retrying(
    retry=retry_unless_open(requests.HTTPError, requests.ConnectionError),
    wait=wait_exponential_jitter(),
    stop=stop_after_attempt(5),
    reraise=True,
)

orders = retrying(fetch_orders)
```

The breaker still classifies by status (no exception needed), the retry loop
reacts to `raise_for_status()` — each tool sees the signal in its native
form. To align which statuses trip the breaker, pass
`HttpStatusClassifier(failure_statuses={...})` to the integration.

## Anti-patterns

- **Unbounded retries.** Always set a `stop` condition. A breaker caps
  concurrent damage, not the lifetime of a stubborn loop.
- **Retries without a breaker.** N clients × M retries is an N·M-fold
  amplification aimed at a dependency that is already failing — the classic
  retry storm. The breaker inside the loop is what breaks it.
- **Retrying non-transient errors.** A `404` or a validation error will not
  succeed on attempt five. List transient exception types explicitly in
  `retry_unless_open(...)` rather than retrying everything.
- **Nested retry layers.** urllib3's `max_retries`, your service mesh and
  tenacity each multiply attempts. Budget them together — one deliberate
  retry layer beats three accidental ones.

## Declarative composition

Everything on this page stays valid with the v2 [resilience
pipeline](pipeline.md) — `RetryStrategy` packages the same predicates and
wait strategies as a layer, so the manual recipe becomes::

```python
pipeline = (
    Pipeline.builder()
    .retry(attempts=4)  # retry outside — the recommended order
    .circuit_breaker(breaker)
    .timeout(2.0)
    .build()
)
```

---

<!-- source: docs/guides/pipeline.md -->

# Resilience pipeline

v2.0 turns interlock's primitives into composable **strategies**: timeout,
bulkhead, circuit breaker, retry and fallback applied around one call in an
explicit order, mirroring [Polly's](https://www.pollydocs.org/pipelines)
`ResiliencePipeline` semantics.

The pipeline is an additive layer — the standalone
[`CircuitBreaker`](../getting-started.md) remains a first-class primitive, and
existing v1 code keeps working unchanged. Reach for a pipeline when one
concern is not enough.

## At a glance

```python
from interlock import CircuitBreaker, CircuitOpenError, Pipeline

breaker = CircuitBreaker(name='recommendations')

pipeline = (
    Pipeline.builder()
    .fallback(lambda exc: [], on=(CircuitOpenError,))  # outermost
    .retry(attempts=4)  # requires interlock-cb[tenacity]
    .circuit_breaker(breaker)
    .bulkhead(8)
    .timeout(2.0)  # innermost
    .build()
)


@pipeline
async def fetch_picks(user: str) -> list[str]:
    return await client.get_picks(user)
```

One pipeline serves sync and async callables alike: the decorator and
`pipeline.call(fn, ...)` detect the callable's nature and dispatch, exactly
like `CircuitBreaker.call`. The decorator preserves the wrapped signature for
type checkers.

## Order is explicit — first is outermost

Strategies apply in declaration order: the first strategy sees everything the
inner layers produce. `Pipeline(A, B)` means `A(B(call))`. There is no hidden
"correct" order in the code; the recommended one is a documented default:

| Layer (outer → inner) | Why here |
|---|---|
| `fallback(...)` | Substitutes a value for whatever the inner stack gave up on — including rejections raised by the strategies themselves |
| `retry(...)` | Each attempt below is a complete guarded call: the breaker sees honest per-attempt statistics and stops the retry loop the moment the circuit opens |
| `circuit_breaker(...)` | Counts timeouts and failures of every attempt; open circuit rejects before the bulkhead slot or a connection is touched |
| `bulkhead(...)` | Inside retry — otherwise every backoff-and-retry cycle would multiply slot occupancy |
| `timeout(...)` | Innermost: bounds a single attempt, so one hung attempt cannot eat the whole retry budget |

Deviating is legitimate — e.g. a breaker *outside* retry counts one
aggregated outcome per operation instead of one per attempt (see
[Retries and circuit breakers](retries.md) for that trade-off) — but do it
deliberately.

## The strategies

### `CircuitBreakerStrategy`

Wraps a standalone [`CircuitBreaker`](../reference.md) without touching it:
the window, events, manual controls and the breaker's own listener behave
exactly as in direct use, and the same instance can still be called directly.
An open circuit raises `CircuitOpenError` before any inner layer runs.

```python
from interlock import CircuitBreakerStrategy, Pipeline

pipeline = Pipeline(CircuitBreakerStrategy(breaker))
```

### `TimeoutStrategy`

Bounds every attempt using the v1 primitives: `asyncio.timeout` on the async
path (the attempt is cancelled), `sync_timeout` on the sync path — which
inherits its [worker-thread limitation](timeout.md): the caller gets
`CallTimeoutError` on time, but Python cannot kill the overrunning thread.

### `BulkheadStrategy`

Caps how many calls run through the layer concurrently. With no free slot the
call fails immediately (`max_wait=0`, the default) or waits up to `max_wait`
seconds, then raises `BulkheadFullError`:

```python
from interlock import BulkheadStrategy, Pipeline

pipeline = Pipeline(BulkheadStrategy(8, max_wait=0.5))
```

`BulkheadFullError` is deliberately not `CircuitOpenError`: a full bulkhead
means *this process* is saturated, not that the dependency is unhealthy — the
right reaction is shedding load, not backing off. Sync calls share a
`threading.Semaphore`, async calls an `asyncio.Semaphore`; one configuration,
two independent pools.

### `FallbackStrategy`

Substitutes an explicit value for selected failures — nothing silent:

```python
from interlock import CircuitOpenError, FallbackStrategy, Pipeline

cached: list[str] = []
strategy = FallbackStrategy(lambda exc: cached, on=(CircuitOpenError,))
```

- The substitution happens **only** for exception types named in `on`;
  anything else propagates.
- The `fallback` callable receives the exception it stands in for.
- `on` accepts `Exception` subclasses exclusively — cancellation and
  `KeyboardInterrupt` always propagate.
- The strategy's own result type is the honest union `T | F`, not `Any`. At
  the pipeline level the substitute is expected to be shaped like the call's
  result (the same contract as Polly and resilience4j).
- A fallback never masks shadow-mode statistics: a `metrics_only` breaker
  below it keeps recording every failure.

### `RetryStrategy` (the `tenacity` extra)

interlock ships no retry engine; the strategy delegates all policy to
[tenacity](https://tenacity.readthedocs.io/) and packages the
[retry × breaker glue](../integrations/tenacity.md) for the pipeline:

```python
from interlock.integrations.tenacity import RetryStrategy

strategy = RetryStrategy(attempts=4)  # fail-fast: never retries CircuitOpenError
```

Attempts are always capped, the original exception is re-raised when the
budget runs out, and the default predicate stops retrying the moment the
circuit opens. For the *patient* mode (wait exactly until the breaker's next
probe) pass `wait=wait_probe(...)` — see the
[tenacity integration](../integrations/tenacity.md). The builder step
`.retry(...)` imports the extra lazily, so the pipeline core stays
zero-dependency.

## Two usage forms, not three

A pipeline works as a decorator and as `pipeline.call(fn, *args, **kwargs)` —
the same signature-preserving contracts as the breaker's. There is
deliberately **no context manager**: a `with` block cannot be re-run, so a
retry layer inside it is semantically impossible. This is the same honesty as
the v1 breaker's context manager not supporting result-based classification —
rather than a form that silently ignores half the strategies, the form does
not exist.

## Migrating from v1 (nothing breaks)

The v1 API is untouched — migration is wrapping, not rewriting:

```python
# v1: the breaker guards the call directly
result = breaker.call(fetch_orders, user_id)

# v2: the same breaker, now composed with a timeout
pipeline = Pipeline(CircuitBreakerStrategy(breaker), TimeoutStrategy(2.0))
result = pipeline.call(fetch_orders, user_id)
```

The manual composition recipe from the
[retries guide](retries.md) — `Retrying` wrapped around `breaker.call` —
keeps working and remains the most flexible form; the pipeline is that recipe
made declarative.

## Observability

`RetryStrategy`, `BulkheadStrategy` and `FallbackStrategy` (and their builder
steps) accept `name=` and `listener=`. Three optional
[`EventListener`](observability.md) hooks make the pipeline's decisions
visible — `on_retry(name, attempt, delay)`, `on_bulkhead_rejected(name)` and
`on_fallback(name, error)`:

```python
from interlock import LoggingEventListener, Pipeline

events = LoggingEventListener()

pipeline = (
    Pipeline.builder()
    .fallback(lambda exc: [], on=(CircuitOpenError,), name='recs', listener=events)
    .circuit_breaker(breaker)  # the breaker keeps its own listener
    .bulkhead(8, name='recs', listener=events)
    .timeout(2.0)
    .build()
)
```

Strategy hooks go through the same dispatcher as the breaker's own: dispatched
only if defined (listeners written before v2.0 keep working unchanged), and a
hook that raises is logged and ignored rather than failing the call — see
[listener failures are isolated](observability.md#listener-failures-are-isolated).
A `fallback` function is *not* a hook: it shapes the result, so its errors
propagate. `LoggingEventListener` logs retries at INFO and bulkhead rejections
/ fallbacks at WARNING; `OTelEventListener` counts all three in the
`interlock.pipeline.events` counter.

## Custom strategies

Any object with `execute` / `execute_async` is a strategy — the `Strategy`
protocol is structural:

```python
import time
from collections.abc import Awaitable, Callable
from typing import TypeVar

T = TypeVar('T')


class Measuring:
    """Times every layer below it."""

    def execute(self, call: Callable[[], T]) -> T:
        start = time.perf_counter()
        try:
            return call()
        finally:
            print(f'took {time.perf_counter() - start:.3f}s')

    async def execute_async(self, call: Callable[[], Awaitable[T]]) -> T:
        start = time.perf_counter()
        try:
            return await call()
        finally:
            print(f'took {time.perf_counter() - start:.3f}s')


pipeline = Pipeline.builder().add(Measuring()).timeout(2.0).build()
```

The contract, in full:

- Run the zero-argument next layer, return its result, let exceptions
  propagate. Never swallow `BaseException` — cancellation must cross every
  layer untouched.
- `execute_async` always receives a real coroutine function, so
  detect-dispatching primitives (like `breaker.call`) treat it as async.

---

<!-- source: docs/integrations/index.md -->

# Integrations

interlock plugs into the HTTP client, framework or retry library you already
use — you configure thresholds once and the breaker applies **per host** (or
per named dependency) with no decorators in call sites.

## Supported integrations

| Integration | Extra | What you get |
|---|---|---|
| [FastAPI](fastapi.md) | `interlock-cb[fastapi]` | `Depends`-injected breakers and a `CircuitOpenError → 503 + Retry-After` handler |
| [Litestar](litestar.md) | `interlock-cb[litestar]` | `Provide`-injected breakers and a `CircuitOpenError → 503 + Retry-After` handler (Litestar ≥ 2.23) |
| [httpx2](httpx2.md) | `interlock-cb[httpx2]` | `CircuitBreakerTransport` / `AsyncCircuitBreakerTransport` — per-host breaker at the transport level |
| [httpx](httpx.md) | `interlock-cb[httpx]` | `CircuitBreakerTransport` / `AsyncCircuitBreakerTransport` — per-host transport for httpx clients |
| [aiohttp](aiohttp.md) | `interlock-cb[aiohttp]` | `CircuitBreakerMiddleware` — per-host breaker as a client middleware (aiohttp ≥ 3.12) |
| [requests](requests.md) | `interlock-cb[requests]` | `CircuitBreakerAdapter` — per-host breaker mounted on a `Session` |
| [LLM SDKs](llm.md) | — (recipe) | Guard OpenAI / Anthropic SDK calls with a breaker + bounded retries |
| [tenacity](tenacity.md) | `interlock-cb[tenacity]` | Retry × breaker glue: stop retrying when the circuit opens, or wait exactly until the next probe |
| [Redis](redis.md) | `interlock-cb[redis]` | Shared breaker state across processes with graceful degradation |
| [Flask / Django](frameworks.md) | — (recipe) | Map `CircuitOpenError` to `503 + Retry-After` in other web frameworks |

## How integrations are built

Every integration follows the same rules, so learning one means knowing all:

- **Native extension points only.** A transport (httpx2/httpx), a client middleware
  (aiohttp), an adapter (requests), an exception handler (FastAPI). No
  monkey-patching, no private APIs — an integration survives minor releases
  of its host library.
- **One breaker per host.** HTTP integrations key breakers by request host:
  a failing `api.a` never trips `api.b`. Breakers are created lazily in a
  shared [`Registry`](../reference.md).
- **Safe rollout before enforcement.** Pass
  `initial_state=State.METRICS_ONLY` to record real traffic without rejecting
  it. The state is applied before each lazy breaker serves its first request;
  deploy a new integration with `CLOSED` after tuning thresholds.
- **One classification model.** Responses are classified by an
  `HttpStatusClassifier` — by default the canonical retryable set
  (`429, 500, 502, 503, 504`) plus every non-excluded transport exception
  counts as a failure, while `4xx` client mistakes do not. Exceptions the
  *caller* caused say nothing about the dependency's health, so each
  integration excludes the ones its own library raises inside the guarded
  call: `UnsupportedProtocol` / `LocalProtocolError` for httpx2 and httpx,
  `InvalidURL` for requests, nothing for aiohttp (it rejects malformed URLs
  before the middleware chain runs). Pass
  `HttpStatusClassifier(failure_statuses={...}, excluded_exceptions=(...))` or
  your own `FailureClassifier` to change the policy.
- **One rejection signal.** An open circuit always raises
  [`CircuitOpenError`](../reference.md) — carrying the breaker name, a
  `retry_after` estimate and the last recorded failure — *before* a
  connection is attempted.
- **Zero-dependency core.** Integrations live in `interlock.integrations.*`
  as optional extras; `import interlock` itself never pulls anything beyond
  the standard library.
- **Explicit ownership and teardown.** Transports and adapters close their
  native connection resources together with their breaker registry. The
  aiohttp middleware exposes `aclose()` because `ClientSession` does not own
  middleware resources.

## Support tiers

- **Tier 1 — shipped code.** Modules under `interlock.integrations.*`,
  covered by the test suite and CI against both the minimum supported and the
  latest version of the host library. Semver applies.
- **Tier 2 — recipes.** Documented, runnable patterns (LLM SDKs,
  Flask/Django) that need no dedicated glue code. They can graduate to Tier 1
  when demand shows up.

Missing an integration — gRPC, SQLAlchemy, Kafka, Celery?
[Open an issue](https://github.com/bagowix/interlock/issues): the next wave
is prioritised by demand.

---

<!-- source: docs/integrations/fastapi.md -->

# FastAPI

The `interlock-cb[fastapi]` extra protects a route's outgoing dependency with a
shared `Registry` and turns a tripped breaker into a clean
`503 Service Unavailable` response with a `Retry-After` header.

=== "uv"

    ```bash
    uv add 'interlock-cb[fastapi]'
    ```

=== "pip"

    ```bash
    pip install 'interlock-cb[fastapi]'
    ```

=== "poetry"

    ```bash
    poetry add 'interlock-cb[fastapi]'
    ```

## Usage

Install the exception handler once, then inject a per-name breaker into any route
with `Depends`:

```python
from typing import Annotated

from fastapi import Depends, FastAPI
from interlock import CircuitBreaker, Registry
from interlock.integrations.fastapi import breaker_dependency, install_exception_handler

app = FastAPI()
registry = Registry()
install_exception_handler(app)

orders_db = breaker_dependency('orders-db', registry=registry)


@app.get('/orders')
async def orders(breaker: Annotated[CircuitBreaker, Depends(orders_db)]) -> list[dict]:
    return await breaker.call(fetch_orders)
```

When `fetch_orders` fails often enough, the breaker opens. The next request is
rejected with `CircuitOpenError` *before* `fetch_orders` runs, and the installed
handler converts it into:

```http
HTTP/1.1 503 Service Unavailable
Retry-After: 30
Content-Type: application/json

{"detail": "Circuit 'orders-db' is open"}
```

## How it works

- **`breaker_dependency(name, *, registry)`** returns a FastAPI dependency that
  yields the named breaker from the shared `Registry`. The breaker is created
  lazily on first use and reused on every later request, so all requests to that
  route share one breaker (and one view of the dependency's health).
- **`install_exception_handler(app)`** registers a handler for
  `CircuitOpenError`. It responds `503` and sets `Retry-After` to the breaker's
  `retry_after` estimate, rounded up to whole seconds (per RFC 7231). The header
  is omitted when there is no estimate (for example after `force_open()`).

You protect the *outgoing* call (`breaker.call(...)`) rather than the route
itself: only the dependency you wrap counts toward the breaker, and the breaker's
own admission logic (probes, half-open) keeps working.

## Sharing breakers across routes

Reuse the same `name` (and the same `registry`) to share one breaker across
several routes that all depend on the same downstream:

```python
orders_db = breaker_dependency('orders-db', registry=registry)


@app.get('/orders')
async def list_orders(breaker: Annotated[CircuitBreaker, Depends(orders_db)]) -> list[dict]:
    return await breaker.call(fetch_orders)


@app.get('/orders/{order_id}')
async def get_order(order_id: int, breaker: Annotated[CircuitBreaker, Depends(orders_db)]) -> dict:
    return await breaker.call(fetch_order, order_id)
```

Pass `config`, `clock`, `classifier` or `listener` to the `Registry` to tune
every breaker it creates, or override per name via `registry.get(name, config=...)`.

## Custom responses

For a different response shape, register your own handler instead of
`install_exception_handler`:

```python
from fastapi import Request, Response
from interlock import CircuitOpenError


@app.exception_handler(CircuitOpenError)
async def on_open(request: Request, exc: CircuitOpenError) -> Response: ...
```

---

<!-- source: docs/integrations/litestar.md -->

# Litestar

The `interlock-cb[litestar]` extra (Litestar ≥ 2.23) protects a route's
outgoing dependency with a shared `Registry` and turns a tripped breaker into
a clean `503 Service Unavailable` response with a `Retry-After` header.

=== "uv"

    ```bash
    uv add 'interlock-cb[litestar]'
    ```

=== "pip"

    ```bash
    pip install 'interlock-cb[litestar]'
    ```

=== "poetry"

    ```bash
    poetry add 'interlock-cb[litestar]'
    ```

## Usage

Litestar wires exception handlers and dependencies at construction time —
declare both on the app (or a router / controller) and annotate the handler
parameter with `NamedDependency`:

```python
from litestar import Litestar, get
from litestar.di import NamedDependency

from interlock import CircuitBreaker, CircuitOpenError, Registry
from interlock.integrations.litestar import breaker_dependency, circuit_open_handler

registry = Registry()


@get('/orders')
async def orders(breaker: NamedDependency[CircuitBreaker]) -> list[dict]:
    return await breaker.call(fetch_orders)


app = Litestar(
    route_handlers=[orders],
    dependencies={'breaker': breaker_dependency('orders-db', registry=registry)},
    exception_handlers={CircuitOpenError: circuit_open_handler},
)
```

When `fetch_orders` fails often enough, the breaker opens. The next request is
rejected with `CircuitOpenError` *before* `fetch_orders` runs, and the handler
converts it into:

```http
HTTP/1.1 503 Service Unavailable
Retry-After: 30
Content-Type: application/json

{"detail": "Circuit 'orders-db' is open"}
```

## How it works

- **`breaker_dependency(name, *, registry)`** returns a Litestar
  [`Provide`](https://docs.litestar.dev/latest/usage/dependency-injection.html)
  that yields the named breaker from the shared `Registry`. The breaker is
  created lazily on first use and reused on every later request, so all
  requests sharing the dependency share one breaker (and one view of the
  downstream's health). Declare it at any layer — app, router, controller or
  handler.
- **`circuit_open_handler`** maps `CircuitOpenError` to `503` and sets
  `Retry-After` to the breaker's `retry_after` estimate, rounded up to whole
  seconds (per RFC 7231). The header is omitted when there is no estimate
  (for example after `force_open()`).

You protect the *outgoing* call (`breaker.call(...)`) rather than the route
itself: only the dependency you wrap counts toward the breaker, and the
breaker's own admission logic (probes, half-open) keeps working.

## Sharing breakers across routes

Reuse the same `name` (and the same `registry`) wherever routes depend on the
same downstream — one `dependencies={...}` declaration on the app covers them
all, and every route sees the same circuit state.

Pass `config`, `clock`, `classifier` or `listener` to the `Registry` to tune
every breaker it creates, or override per name via
`registry.get(name, config=...)`.

## Custom responses

For a different response shape, register your own handler instead:

```python
from litestar import Request, Response

from interlock import CircuitOpenError


def on_open(request: Request, exc: CircuitOpenError) -> Response[dict[str, str]]: ...


app = Litestar(..., exception_handlers={CircuitOpenError: on_open})
```

---

<!-- source: docs/integrations/httpx2.md -->

# httpx2

The `interlock-cb[httpx2]` extra wraps an [httpx2](https://pypi.org/project/httpx2/)
transport so a circuit breaker is applied **per host** transparently — no
decorators or `call` wrappers in your request code.

`interlock-cb` is listed in httpx2's official
[third-party packages directory](https://github.com/pydantic/httpx2/blob/main/docs/third_party_packages.md#interlock-cb).

=== "uv"

    ```bash
    uv add 'interlock-cb[httpx2]'
    ```

=== "pip"

    ```bash
    pip install 'interlock-cb[httpx2]'
    ```

=== "poetry"

    ```bash
    poetry add 'interlock-cb[httpx2]'
    ```

## Synchronous client

```python
import httpx2
from interlock.integrations.httpx2 import CircuitBreakerTransport

transport = CircuitBreakerTransport(httpx2.HTTPTransport())
client = httpx2.Client(transport=transport)

response = client.get('https://api.example.com/v1/users')
```

## Asynchronous client

```python
import httpx2
from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport

transport = AsyncCircuitBreakerTransport(httpx2.AsyncHTTPTransport())
client = httpx2.AsyncClient(transport=transport)

response = await client.get('https://api.example.com/v1/users')
```

Use the client as a context manager. Context entry and exit are delegated to
the wrapped transport, including for custom transports that acquire resources
in `__enter__` or `__aenter__`. Closing the client closes both the wrapped
connection pool and every breaker created by the transport.

## Safe production rollout

Start in shadow mode when introducing the integration to existing traffic:

```python
import httpx2

from interlock import State
from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport

transport = AsyncCircuitBreakerTransport(
    httpx2.AsyncHTTPTransport(),
    initial_state=State.METRICS_ONLY,
    listener=metrics_listener,
)
```

Every host created later starts in `METRICS_ONLY` before its first request: it
records outcomes but never raises `CircuitOpenError`. Use the listener for
production metrics. For local diagnosis,
`transport.registry.get_existing(host)` returns an existing breaker without
creating one; inspect its `state` and `snapshot()`. Hosts are only known at
runtime, so `transport.registry.items()` lists the breakers created so far and
`names()` just the names they were created under. Both are point-in-time copies:
a breaker created afterwards is not in them.

After tuning thresholds, deploy a new transport with the default
`initial_state=State.CLOSED`. Do not reset only the currently known hosts: a
transport configured for shadow mode would still create future hosts in
`METRICS_ONLY`. See the complete [safe-rollout guide](../guides/states.md#safe-rollout).

## Per-host isolation

Each host gets its own breaker, created lazily and cached. A failing
`api.a.example.com` trips only its own breaker; requests to
`api.b.example.com` are unaffected. Per-instance, per-host state is usually more
correct than global state — each host's health is observed independently.

When a host's breaker is open, its requests raise `CircuitOpenError` before
reaching the network.

## Custom breaker keys

Pass `name_resolver` when the request host is not the logical dependency
identity. The callback receives the native `httpx2.Request` and returns the
breaker name:

```python
import httpx2

from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport

transport = AsyncCircuitBreakerTransport(
    httpx2.AsyncHTTPTransport(),
    name_resolver=lambda request: request.url.host.removesuffix('.query.consul'),
)
```

Returning the same name for several discovery hosts gives them one breaker;
deriving a name from the path can split independent upstreams behind one
gateway host. The result must be a non-empty string containing something other
than whitespace. Invalid results raise `ValueError` with the request URL before
the wrapped transport performs I/O.

The resolved name is used consistently as the registry key, in
`CircuitOpenError`, and in every listener event. Resolve the name here instead
of rewriting listener labels so metrics always identify the breaker whose
state they report. Both synchronous and asynchronous transports accept the
option.

## Share one registry across clients

Pass a caller-owned `Registry` when several clients reach the same dependency.
Requests resolving to the same name then use one breaker instance and one
sliding window, even when they travel through different transports:

```python
import httpx2

from interlock import Config, Registry
from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport, HttpStatusClassifier

registry = Registry(
    config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
    classifier=HttpStatusClassifier(),
)

client_a = httpx2.AsyncClient(
    transport=AsyncCircuitBreakerTransport(httpx2.AsyncHTTPTransport(), registry=registry)
)
client_b = httpx2.AsyncClient(
    transport=AsyncCircuitBreakerTransport(httpx2.AsyncHTTPTransport(), registry=registry)
)
```

`Registry` uses exception-only classification by default. Always configure
`HttpStatusClassifier` as above when you want the integration's normal HTTP
status policy; otherwise a returned `503` counts as a success. The supplied
registry owns `config`, `clock`, `initial_state`, `classifier`, and `listener`,
so none of those options may also be passed to the transport.

Share such a registry with HTTP clients only. `HttpStatusClassifier` reads
`.status_code` off every result it records, so a breaker taken from the same
registry for non-HTTP work — `registry.get('db')` — raises `AttributeError` the
first time that call returns. Keep a separate registry for those.

Closing a client automatically closes its breakers only when the transport
owns the registry. An injected registry remains open while the wrapped
connection pool closes; the application must explicitly call
`await registry.aclose_all()` during async shutdown (or `registry.close_all()`
when all guarded clients are synchronous).

## Reach the wrapped transport

`transport.wrapped` returns the transport being guarded, so a composed object
can be unwrapped without touching private attributes — verifying the pool
limits, TLS context or proxy the inner transport was built with, inspecting it
in a REPL, or walking a chain of wrappers:

```python
import httpx2

from interlock.integrations.httpx2 import AsyncCircuitBreakerTransport

inner = httpx2.AsyncHTTPTransport(limits=httpx2.Limits(max_connections=20))
transport = AsyncCircuitBreakerTransport(inner)

assert transport.wrapped is inner
```

The property is read-only: the wrapped transport is fixed at construction. Both
the synchronous and asynchronous classes expose it.

## What counts as a failure

By default the transport uses `HttpStatusClassifier`:

- a transport exception (connect/read errors) → failure;
- a response with status `429, 500, 502, 503, 504` → failure;
- everything else, including `4xx` client errors like `404`, → success;
- `UnsupportedProtocol` and `LocalProtocolError` → success.

This mirrors the retryable set used by urllib3, AWS and Google clients.
Permanent `5xx` (`501`, `505`) are deliberately excluded — retrying or tripping
the breaker cannot fix a contract or protocol error.

The two excluded exceptions are the caller's own bug — a scheme-less or
unsupported URL, and the local side violating HTTP. They are deterministic and
say nothing about the dependency, so a burst of them must not open the circuit
of a healthy host; they still propagate to the caller unchanged.

`PoolTimeout` is *not* excluded: an exhausted pool is usually the dependency
holding connections open, and shedding load then is the point. Exclude it
explicitly when your pool is sized below your own burst:

```python
import httpx2
from interlock.integrations.httpx2 import HttpStatusClassifier

classifier = HttpStatusClassifier(
    excluded_exceptions=(
        httpx2.LocalProtocolError,
        httpx2.UnsupportedProtocol,
        httpx2.PoolTimeout,
    ),
)
```

`excluded_exceptions` replaces the default set — pass `()` to count every
exception as a failure. An excluded exception is recorded as a *success*: the
sliding window has no third outcome.

## Tuning

Pass any of `config`, `clock`, `classifier`, `listener` to the transport; they
flow to every breaker:

```python
from interlock import Config, LoggingEventListener
from interlock.integrations.httpx2 import CircuitBreakerTransport

transport = CircuitBreakerTransport(
    httpx2.HTTPTransport(),
    config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
    listener=LoggingEventListener(),
)
```

Supply your own `classifier` to change the failure policy — for example to also
fail on `408 Request Timeout`.

---

<!-- source: docs/integrations/httpx.md -->

# httpx

The `interlock-cb[httpx]` extra wraps an
[httpx](https://www.python-httpx.org/) transport so a circuit breaker is
applied **per host** transparently. It supports httpx 0.27.0 and newer.

=== "uv"

    ```bash
    uv add 'interlock-cb[httpx]'
    ```

=== "pip"

    ```bash
    pip install 'interlock-cb[httpx]'
    ```

=== "poetry"

    ```bash
    poetry add 'interlock-cb[httpx]'
    ```

## Synchronous client

```python
import httpx
from interlock.integrations.httpx import CircuitBreakerTransport

transport = CircuitBreakerTransport(httpx.HTTPTransport())
client = httpx.Client(transport=transport)

response = client.get('https://api.example.com/v1/users')
```

## Asynchronous client

```python
import httpx
from interlock.integrations.httpx import AsyncCircuitBreakerTransport

transport = AsyncCircuitBreakerTransport(httpx.AsyncHTTPTransport())
client = httpx.AsyncClient(transport=transport)

response = await client.get('https://api.example.com/v1/users')
```

Use the client as a context manager. Closing it delegates `close()` or
`aclose()` to the wrapped transport and releases both the connection pool and
every breaker created by the transport.

## Safe production rollout

Start in shadow mode when introducing the integration to existing traffic:

```python
import httpx

from interlock import State
from interlock.integrations.httpx import AsyncCircuitBreakerTransport

transport = AsyncCircuitBreakerTransport(
    httpx.AsyncHTTPTransport(),
    initial_state=State.METRICS_ONLY,
    listener=metrics_listener,
)
```

Every host created later starts in `METRICS_ONLY` before its first request: it
records outcomes but never raises `CircuitOpenError`. Use the listener for
production metrics. For local diagnosis,
`transport.registry.get_existing(host)` returns an existing breaker without
creating one; inspect its `state` and `snapshot()`. Hosts are only known at
runtime, so `transport.registry.items()` lists the breakers created so far and
`names()` just the names they were created under. Both are point-in-time copies:
a breaker created afterwards is not in them.

After tuning thresholds, deploy a new transport with the default
`initial_state=State.CLOSED`. Do not reset only the currently known hosts: a
transport configured for shadow mode would still create future hosts in
`METRICS_ONLY`. See the complete [safe-rollout guide](../guides/states.md#safe-rollout).

## Per-host isolation

Each host gets its own lazily created breaker. A failing
`api.a.example.com` trips only that host; traffic to `api.b.example.com`
continues normally. An open breaker raises `CircuitOpenError` before the
wrapped transport performs I/O. A request URL without a host raises
`ValueError` for the same reason: there is no dependency identity to key on.

## Custom breaker keys

Pass `name_resolver` when the request host is transport plumbing rather than
the logical dependency identity. The callback receives the native
`httpx.Request` and returns the breaker name:

```python
import httpx

from interlock.integrations.httpx import AsyncCircuitBreakerTransport

transport = AsyncCircuitBreakerTransport(
    httpx.AsyncHTTPTransport(),
    name_resolver=lambda request: request.url.host.removesuffix('.query.consul'),
)
```

The same callback can split one gateway host into independent breakers, for
example by returning a name derived from the first path segment. It must return
a non-empty string containing something other than whitespace; invalid results
raise `ValueError` with the request URL before the wrapped transport performs
I/O.

The resolved name is the registry key and the name carried by
`CircuitOpenError` and every listener event. Use the resolver, rather than
rewriting labels in a listener, so breaker state and observability labels stay
aligned. Both synchronous and asynchronous transports accept the option.

## Share one registry across clients

Inject one caller-owned `Registry` when several clients should observe the
same dependency health. The transports then resolve the same name to the same
breaker and contribute to one sliding window:

```python
import httpx

from interlock import Config, Registry
from interlock.integrations.httpx import AsyncCircuitBreakerTransport, HttpStatusClassifier

registry = Registry(
    config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
    classifier=HttpStatusClassifier(),
)

client_a = httpx.AsyncClient(
    transport=AsyncCircuitBreakerTransport(httpx.AsyncHTTPTransport(), registry=registry)
)
client_b = httpx.AsyncClient(
    transport=AsyncCircuitBreakerTransport(httpx.AsyncHTTPTransport(), registry=registry)
)
```

The classifier is intentional: a bare `Registry` classifies raised exceptions
but treats returned responses, including `503`, as successes. Configure
`HttpStatusClassifier` to retain the transport's default status policy. A
supplied registry also owns `config`, `clock`, `initial_state`, `classifier`,
and `listener`; combining `registry` with any of those transport options raises
`ValueError` instead of silently ignoring one source of configuration.

Share such a registry with HTTP clients only. `HttpStatusClassifier` reads
`.status_code` off every result it records, so a breaker taken from the same
registry for non-HTTP work — `registry.get('db')` — raises `AttributeError` the
first time that call returns. Keep a separate registry for those.

Closing a client automatically closes its breakers only when the transport
owns the registry. An injected registry remains open while the wrapped
connection pool closes; the application must explicitly call
`await registry.aclose_all()` during async shutdown, or `registry.close_all()`
when every guarded client is synchronous.

## Reach the wrapped transport

`transport.wrapped` returns the transport being guarded, so a composed object
can be unwrapped without touching private attributes — verifying the pool
limits, TLS context or proxy the inner transport was built with, inspecting it
in a REPL, or walking a chain of wrappers:

```python
import httpx

from interlock.integrations.httpx import AsyncCircuitBreakerTransport

inner = httpx.AsyncHTTPTransport(limits=httpx.Limits(max_connections=20))
transport = AsyncCircuitBreakerTransport(inner)

assert transport.wrapped is inner
```

The property is read-only: the wrapped transport is fixed at construction. Both
the synchronous and asynchronous classes expose it.

## What counts as a failure

The default `HttpStatusClassifier` counts these as failures:

- transport exceptions raised before a response is returned;
- response statuses `429, 500, 502, 503, 504`.

Other responses, including caller errors such as `404`, count as successes.
So are the transport exceptions httpx raises for the *caller's* own bug —
`UnsupportedProtocol` (a scheme-less or unsupported URL) and
`LocalProtocolError` (the local side violating HTTP). They are deterministic
and say nothing about the dependency, so a burst of them must not open the
circuit of a healthy host. They still propagate to the caller unchanged.

`PoolTimeout` is *not* excluded: an exhausted pool is usually the dependency
holding connections open, and shedding load then is the point. Exclude it
explicitly when your pool is sized below your own burst:

```python
import httpx
from interlock.integrations.httpx import HttpStatusClassifier

classifier = HttpStatusClassifier(
    excluded_exceptions=(httpx.LocalProtocolError, httpx.UnsupportedProtocol, httpx.PoolTimeout),
)
```

`excluded_exceptions` replaces the default set — pass `()` to count every
exception as a failure. An excluded exception is recorded as a *success*: the
sliding window has no third outcome.

Pass `HttpStatusClassifier(failure_statuses={...})` or another
`FailureClassifier` to change the status side of the policy.

## Streaming responses

The wrapper returns the original `httpx.Response` unchanged, so sync and async
streaming remain lazy and connection cleanup keeps httpx's normal semantics.
Context entry and exit are delegated to the wrapped transport, including for
custom transports that acquire resources in `__enter__` or `__aenter__`.
Because the circuit-breaker call completes when response headers arrive, an
exception raised later while consuming a streaming body is outside that call
and is not recorded by the breaker.

## Tuning

`config`, `clock`, `initial_state`, `classifier`, and `listener` are shared by
every breaker created by the transport:

```python
import httpx

from interlock import Config, LoggingEventListener
from interlock.integrations.httpx import CircuitBreakerTransport

transport = CircuitBreakerTransport(
    httpx.HTTPTransport(),
    config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
    listener=LoggingEventListener(),
)
```

The transport never retries. If another layer owns retries, keep them bounded
and stop retrying when the breaker raises `CircuitOpenError`.

---

<!-- source: docs/integrations/aiohttp.md -->

# aiohttp

The `interlock-cb[aiohttp]` extra guards every request a `ClientSession`
sends with a circuit breaker **per host**, wired in as a client middleware —
no decorators in call sites. Requires aiohttp ≥ 3.12 (client middlewares).

=== "uv"

    ```bash
    uv add 'interlock-cb[aiohttp]'
    ```

=== "pip"

    ```bash
    pip install 'interlock-cb[aiohttp]'
    ```

=== "poetry"

    ```bash
    poetry add 'interlock-cb[aiohttp]'
    ```

## Usage

```python
import aiohttp

from interlock.integrations.aiohttp import CircuitBreakerMiddleware

middleware = CircuitBreakerMiddleware()

async with aiohttp.ClientSession(middlewares=(middleware,)) as session:
    async with session.get('https://api.example.com/orders') as response:
        orders = await response.json()

await middleware.aclose()
```

`ClientSession` does not own middleware resources. Call `middleware.aclose()`
during application shutdown; it releases every per-host breaker and is
idempotent.

Each host gets its own breaker (a failing `api.a` never trips `api.b`),
created lazily and shared across requests. When a host's circuit is open the
request raises [`CircuitOpenError`](../reference.md) *before* a connection is
made.

The breaker observes the time to *response headers*; reading the body happens
outside the guarded call — the same semantics as the
[httpx2 transport](httpx2.md).

## Custom breaker keys

Pass `name_resolver` when host-based isolation does not match the logical
dependencies. The callback receives the native `aiohttp.ClientRequest` and
returns the breaker name:

```python
from interlock.integrations.aiohttp import CircuitBreakerMiddleware

middleware = CircuitBreakerMiddleware(
    name_resolver=lambda request: request.url.host.removesuffix('.query.consul'),
)
```

A resolver can collapse several discovery hosts onto one breaker or derive a
name from the request path to separate upstreams behind a shared gateway. It
must return a non-empty string containing something other than whitespace;
invalid results raise `ValueError` with the request URL before the handler
performs I/O.

The resolved name is used by the registry, `CircuitOpenError`, and every
listener event. Resolve it in the middleware rather than rewriting listener
labels so observed names always match the breaker whose state they describe.

## Share one registry across sessions

Several middleware instances can share one caller-owned registry, so traffic
resolving to the same name contributes to one breaker and one sliding window:

```python
from interlock import Config, Registry
from interlock.integrations.aiohttp import CircuitBreakerMiddleware, HttpStatusClassifier

registry = Registry(
    config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
    classifier=HttpStatusClassifier(),
)

middleware_a = CircuitBreakerMiddleware(registry=registry)
middleware_b = CircuitBreakerMiddleware(registry=registry)
```

Do not omit the classifier when HTTP statuses should affect the breaker. A
bare `Registry` uses exception-only classification, so a returned `503` counts
as a success. The registry owns `config`, `clock`, `initial_state`,
`classifier`, and `listener`; passing any of them to the middleware together
with `registry` raises `ValueError`.

Share such a registry with HTTP sessions only. `HttpStatusClassifier` reads
`.status` off every result it records, so a breaker taken from the same
registry for non-HTTP work — `registry.get('db')` — raises `AttributeError` the
first time that call returns. Keep a separate registry for those.

Calling `aclose()` automatically closes the breakers only when the middleware
owns the registry. An injected registry remains open until the application
explicitly calls `await registry.aclose_all()` during shutdown.

## Safe production rollout

Pass `initial_state=State.METRICS_ONLY` to record real outcomes without
rejecting requests. The state is applied to every lazily created host before
its first request:

```python
from interlock import State

middleware = CircuitBreakerMiddleware(
    initial_state=State.METRICS_ONLY,
    listener=metrics_listener,
)
```

The public `middleware.registry` supports local diagnosis with
`get_existing(host)`, `state` and `snapshot()`, plus `names()` and `items()` for
the breakers created so far — point-in-time copies, without the ones created
afterwards. Use an `EventListener` for
production metrics, then deploy a new middleware with the default `CLOSED`
state to begin enforcement. See [Safe rollout](../guides/states.md#safe-rollout).

## Failure policy

By default a response counts as a failure when its status is in the canonical
retryable set (`429, 500, 502, 503, 504`) and any exception raised while
sending (connect/read errors) is a failure; `4xx` client mistakes like `404`
are successes.

Nothing is excluded by default: aiohttp rejects a malformed or non-HTTP URL
before the middleware chain runs, so no caller-side error of its own reaches
the classifier. Middlewares of your own that sit inside this one are the
exception — an auth middleware refusing to sign a request is your bug, not the
dependency's, so exclude what it raises with
`excluded_exceptions=(MissingCredentials,)`. An excluded exception is recorded
as a *success*, since the sliding window has no third outcome, and still
propagates to the caller.

Change the statuses, or the whole policy:

```python
from interlock import Config
from interlock.integrations.aiohttp import CircuitBreakerMiddleware, HttpStatusClassifier

middleware = CircuitBreakerMiddleware(
    config=Config(failure_rate_threshold=0.3),
    classifier=HttpStatusClassifier(failure_statuses={408, 429, 500, 502, 503, 504}),
)
```

Any custom `FailureClassifier` works too — see
[Failure classification](../guides/failure-classification.md).

## Tuning and observability

The middleware accepts the same collaborators as `CircuitBreaker` — `config`,
`clock`, `initial_state`, `classifier`, `listener`. One middleware instance
holds one registry of resolved breakers; reuse the instance across sessions to
share breaker state, or create separate instances to isolate them. For
application-level retries combine with the [tenacity integration](tenacity.md)
and read [Retries and circuit breakers](../guides/retries.md) first.

---

<!-- source: docs/integrations/requests.md -->

# requests

The `interlock-cb[requests]` extra guards every request a `Session` sends
with a circuit breaker **per host** — mounted once, no decorators in call
sites.

=== "uv"

    ```bash
    uv add 'interlock-cb[requests]'
    ```

=== "pip"

    ```bash
    pip install 'interlock-cb[requests]'
    ```

=== "poetry"

    ```bash
    poetry add 'interlock-cb[requests]'
    ```

## Usage

`CircuitBreakerAdapter` subclasses `requests.adapters.HTTPAdapter` — the
library's native transport extension point — so it mounts like any adapter:

```python
import requests

from interlock.integrations.requests import CircuitBreakerAdapter

session = requests.Session()
adapter = CircuitBreakerAdapter()
session.mount('https://', adapter)
session.mount('http://', adapter)

response = session.get('https://api.example.com/orders')
```

Closing the session closes the adapter's connection pools and every breaker it
created.

Each host gets its own breaker (a failing `api.a` never trips `api.b`),
created lazily and shared across requests. When a host's circuit is open the
request raises [`CircuitOpenError`](../reference.md) *before* a connection is
made.

## Custom breaker keys

Pass `name_resolver` when the request host is not the logical dependency
identity. The callback receives the native `requests.PreparedRequest` and
returns the breaker name. For example, the first path segment can separate
independent upstreams behind one gateway host:

```python
from interlock.integrations.requests import CircuitBreakerAdapter

adapter = CircuitBreakerAdapter(
    name_resolver=lambda request: request.path_url.split('/')[1],
)
```

Returning one name for several discovery hosts instead makes them share a
breaker. The result must be a non-empty string containing something other than
whitespace; invalid results raise `ValueError` with the request URL before the
adapter performs I/O.

The resolved name is used consistently as the registry key, in
`CircuitOpenError`, and in every listener event. Resolve the identity here
instead of rewriting listener labels so metrics remain aligned with breaker
state.

## Share one registry across sessions

Inject a caller-owned `Registry` when independent sessions should use one
breaker and one sliding window for the same resolved name:

```python
import requests

from interlock import Config, Registry
from interlock.integrations.requests import CircuitBreakerAdapter, HttpStatusClassifier

registry = Registry(
    config=Config(failure_rate_threshold=0.25, minimum_number_of_calls=50),
    classifier=HttpStatusClassifier(),
)

session_a = requests.Session()
session_a.mount('https://', CircuitBreakerAdapter(registry=registry))

session_b = requests.Session()
session_b.mount('https://', CircuitBreakerAdapter(registry=registry))
```

The explicit classifier preserves the adapter's normal status policy. Without
it, a bare `Registry` classifies exceptions only and a returned `503` counts
as a success. The registry owns `config`, `clock`, `initial_state`,
`classifier`, and `listener`; combining `registry` with any of those adapter
options raises `ValueError`.

Share such a registry with HTTP sessions only. `HttpStatusClassifier` reads
`.status_code` off every result it records, so a breaker taken from the same
registry for non-HTTP work — `registry.get('db')` — raises `AttributeError` the
first time that call returns. Keep a separate registry for those.

Closing a session automatically closes its breakers only when the adapter owns
the registry. An injected registry remains open while its connection pools
close; the application must explicitly call `registry.close_all()` during
shutdown.

## Safe production rollout

Pass `initial_state=State.METRICS_ONLY` to record real outcomes without
rejecting requests. The state is applied to every lazily created host before
its first request:

```python
from interlock import State

adapter = CircuitBreakerAdapter(
    initial_state=State.METRICS_ONLY,
    listener=metrics_listener,
)
```

The public `adapter.registry` supports local diagnosis with
`get_existing(host)`, `state` and `snapshot()`, plus `names()` and `items()` for
the breakers created so far — point-in-time copies, without the ones created
afterwards. Use an `EventListener` for
production metrics, then deploy a new adapter with the default `CLOSED` state
to begin enforcement. See [Safe rollout](../guides/states.md#safe-rollout).

## Failure policy

By default a response counts as a failure when its status is in the canonical
retryable set (`429, 500, 502, 503, 504`) and a transport exception
(connect/read errors) is a failure; `4xx` client mistakes like `404` are
successes.

`InvalidURL` — raised by the adapter when the request or proxy URL carries no
host, and the parent of `InvalidProxyURL` — is a success too. It is the
caller's own bug: deterministic, and no evidence about the dependency, so a
burst of them must not open the circuit of a healthy host. The exception still
propagates to the caller. Replace that set with `excluded_exceptions=(...)`,
or pass `()` to count every exception as a failure; an excluded exception is
recorded as a *success*, since the sliding window has no third outcome.

Change the statuses, or the whole policy:

```python
from interlock import Config
from interlock.integrations.requests import CircuitBreakerAdapter, HttpStatusClassifier

adapter = CircuitBreakerAdapter(
    config=Config(failure_rate_threshold=0.3),
    classifier=HttpStatusClassifier(failure_statuses={408, 429, 500, 502, 503, 504}),
)
```

Any custom `FailureClassifier` works too — see
[Failure classification](../guides/failure-classification.md).

## Tuning and observability

The adapter accepts the same collaborators as `CircuitBreaker` — `config`,
`clock`, `initial_state`, `classifier`, `listener` — and forwards everything
else
(`pool_connections`, `max_retries`, ...) to `HTTPAdapter`. Note that
`max_retries` is urllib3's connection-level retry; for application-level
retries combine with the [tenacity integration](tenacity.md) and read
[Retries and circuit breakers](../guides/retries.md) first.

---

<!-- source: docs/integrations/llm.md -->

# LLM SDKs (OpenAI, Anthropic) — recipe

LLM APIs fail in exactly the ways circuit breakers exist for: rate limits
(`429`), overloaded backends (`529`/`503`), long hangs. A breaker around your
LLM calls stops a degraded provider from stalling every request thread, and
bounded retries recover from blips without amplifying an outage.

You can protect an SDK at either its call boundary, which enables SDK-specific
classification and slow-call detection, or at the httpx transport,
which applies transparently to every request made by that client.

## Classify SDK errors

Both SDKs raise `APIStatusError` subclasses carrying `status_code`, plus
connection/timeout errors. Not every error should trip the circuit: an
invalid request (`400`) or a missing model (`404`) is your bug, not the
provider's outage.

```python
import anthropic


class LLMFailureClassifier:
    """Trip on provider-side trouble, not on caller mistakes."""

    _FAILURE_STATUSES = frozenset({429, 500, 502, 503, 504, 529})

    def is_failure(self, *, result: object, exception: Exception | None) -> bool:
        if exception is None:
            return False
        if isinstance(exception, anthropic.APIStatusError):
            return exception.status_code in self._FAILURE_STATUSES
        return isinstance(exception, (anthropic.APIConnectionError, anthropic.APITimeoutError))
```

For OpenAI, swap the exception types (`openai.APIStatusError`,
`openai.APIConnectionError`, `openai.APITimeoutError`) — the shape is
identical.

## Guard the calls

```python
import anthropic
from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential_jitter

from interlock import CircuitBreaker, Config
from interlock.integrations.tenacity import retry_unless_open

client = anthropic.AsyncAnthropic()

breaker = CircuitBreaker(
    name='anthropic',
    config=Config(slow_call_duration_threshold=30.0),
    classifier=LLMFailureClassifier(),
)


@breaker
async def complete(prompt: str) -> str:
    message = await client.messages.create(
        model='claude-sonnet-5',
        max_tokens=1024,
        messages=[{'role': 'user', 'content': prompt}],
    )
    return message.content[0].text


retrying = AsyncRetrying(
    retry=retry_unless_open(
        anthropic.APIStatusError,
        anthropic.APIConnectionError,
        anthropic.APITimeoutError,
    ),
    wait=wait_exponential_jitter(initial=1.0, max=30.0),
    stop=stop_after_attempt(4),
    reraise=True,
)

answer = await retrying(complete, 'Summarise this document...')
```

What each layer contributes:

- **Slow-call detection** (`slow_call_duration_threshold`) counts calls
  slower than 30s as failures — a provider that still answers but takes a
  minute per completion trips the breaker too. No other signal catches this.
- **The breaker** stops sending after the failure rate crosses the threshold;
  while open, callers get `CircuitOpenError` in microseconds instead of
  hanging — fail over to a second provider or degrade gracefully.
- **`retry_unless_open`** retries provider blips with jittered backoff but
  stops the moment the circuit opens. The SDK's own retries overlap here —
  either set `max_retries=0` on the client and let tenacity own retries, or
  keep the SDK's and drop the tenacity layer; running both multiplies
  attempts.

## Multiple providers, one pattern

Give each provider its own breaker name (`anthropic`, `openai`, ...) via a
shared `Registry` and check `breaker.state` to route around an open provider.
The [states guide](../guides/states.md) covers manual failover controls.

## Transport-level protection

OpenAI and Anthropic clients accept an httpx client. Install
`interlock-cb[httpx]`, wrap the SDK's underlying transport, and every endpoint
on the provider host shares the same breaker:

```python
import httpx
from openai import DefaultHttpxClient, OpenAI

from interlock.integrations.httpx import CircuitBreakerTransport

transport = CircuitBreakerTransport(httpx.HTTPTransport())

with OpenAI(
    http_client=DefaultHttpxClient(transport=transport),
    max_retries=0,
) as client:
    response = client.responses.create(model='gpt-5.5', input='Summarise this document...')
```

Use `AsyncCircuitBreakerTransport`, `httpx.AsyncHTTPTransport`, and the SDK's
async client for async applications. `max_retries=0` gives the breaker one
observable attempt per SDK call; if retries are required, keep one explicit,
bounded retry owner instead of stacking SDK and application retries.

The transport classifier sees HTTP statuses directly, so no SDK exception
classifier is needed. Choose the call-boundary recipe above when you also need
to classify SDK-specific exceptions, measure the complete SDK operation as a
slow call, or use a breaker name that is not derived from the request host.

---

<!-- source: docs/integrations/tenacity.md -->

# tenacity (retries)

interlock deliberately ships no retry engine of its own:
[tenacity](https://tenacity.readthedocs.io/) already does backoff, jitter,
stop conditions and predicates well. The `interlock-cb[tenacity]` extra adds
the glue where retry × breaker composition goes wrong in practice.

=== "uv"

    ```bash
    uv add 'interlock-cb[tenacity]'
    ```

=== "pip"

    ```bash
    pip install 'interlock-cb[tenacity]'
    ```

=== "poetry"

    ```bash
    poetry add 'interlock-cb[tenacity]'
    ```

Read [Retries and circuit breakers](../guides/retries.md) first if you are
deciding *how* to combine the two patterns; this page documents the helpers.

## Fail fast (recommended default)

`retry_unless_open(*transient)` retries the listed transient exceptions but
stops as soon as the breaker opens. `CircuitOpenError` is not transient: the
breaker rejects instantly, so backing off and retrying it only burns the
attempt budget without ever reaching the dependency.

```python
from tenacity import Retrying, stop_after_attempt, wait_exponential_jitter

from interlock import CircuitBreaker
from interlock.integrations.tenacity import retry_unless_open

breaker = CircuitBreaker(name='payments')


@breaker
def charge(amount: int) -> str:
    return gateway.charge(amount)


retrying = Retrying(
    retry=retry_unless_open(TimeoutError, ConnectionError),
    wait=wait_exponential_jitter(),
    stop=stop_after_attempt(5),
    reraise=True,
)

result = retrying(charge, 100)
```

Called without arguments, `retry_unless_open()` retries any ordinary
`Exception` — still never `CircuitOpenError`.

## Patient mode (wait for the probe)

Background jobs often prefer waiting over failing. `wait_probe(fallback)` is
a wait strategy: when the last attempt was rejected with a `retry_after`
estimate, it sleeps *exactly* until the breaker allows the next probe (plus a
small jitter so concurrent waiters do not storm the single probe slot). Any
other outcome delegates to the `fallback` strategy.

```python
from tenacity import (
    AsyncRetrying,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential_jitter,
)

from interlock import CircuitOpenError
from interlock.integrations.tenacity import wait_probe

retrying = AsyncRetrying(
    retry=retry_if_exception_type((TimeoutError, CircuitOpenError)),
    wait=wait_probe(wait_exponential_jitter()),
    stop=stop_after_attempt(10),
    reraise=True,
)

report = await retrying(nightly_export)
```

Note the retry predicate: patient mode deliberately *does* retry
`CircuitOpenError` — that is what makes `wait_probe` see the rejection and
wait the right amount. Keep a `stop` condition anyway; a dependency can stay
down longer than any job should wait.

`wait_probe(..., jitter=0.5)` widens the random extra wait (seconds) added on
top of `retry_after`; when the rejection carries no estimate (for example
after `force_open()`), the fallback strategy decides.

## `RetryStrategy` — the pipeline layer

The same policies package into a [pipeline](../guides/pipeline.md) strategy:

```python
from interlock import Pipeline
from interlock.integrations.tenacity import RetryStrategy

pipeline = (
    Pipeline.builder()
    .retry(attempts=4)  # this step builds a RetryStrategy
    .circuit_breaker(breaker)
    .timeout(2.0)
    .build()
)
```

Attempts are always capped and the original exception is re-raised when the
budget runs out — no `RetryError` wrapping. The default predicate is
`retry_unless_open()` (fail fast on an open circuit); pass
`wait=wait_probe(...)` and a predicate that includes `CircuitOpenError` for
the patient mode. `name=` and `listener=` make every retry visible through
the `on_retry` hook ([observability](../guides/observability.md)).

## Everything else is plain tenacity

`Retrying`, `AsyncRetrying`, the `@retry` decorator, stop and wait strategies
compose as usual — the helpers are ordinary tenacity predicates and wait
objects, so you can combine them with `|`, `retry_any`, `wait_chain` and
friends.

---

<!-- source: docs/integrations/redis.md -->

# Redis (shared state)

The `interlock-cb[redis]` extra coordinates breaker state across processes and
machines through Redis: when one instance trips, every instance backs off, and
recovery probes are budgeted globally instead of per process.

=== "uv"

    ```bash
    uv add 'interlock-cb[redis]'
    ```

=== "pip"

    ```bash
    pip install 'interlock-cb[redis]'
    ```

=== "poetry"

    ```bash
    poetry add 'interlock-cb[redis]'
    ```

## When to share state — and when not to

Per-instance state is the default for a reason. A local breaker reacts only to
what *this* process observes, cannot be affected by another instance's problem,
and keeps working when Redis does not.

Reach for shared state when all of these hold:

- Many instances call the **same downstream**, and its failure affects all of
  them equally (a shared database, a rate-limited third-party API).
- You want **coordinated back-off**: once the downstream is declared unhealthy,
  no instance should keep hammering it just because its own window has not
  filled yet.
- You want **bounded recovery probing**: N instances should send at most
  `permitted_calls_in_half_open` probes *in total*, not each.

Stay local when instances see genuinely different views of the dependency
(per-AZ endpoints, canary deployments), or when one instance's network problems
must not silence the whole fleet. A shared OPEN gates traffic *everywhere* —
that is the point, and the risk. It is a trade-off you opt into, not a default.

## Usage

Pass a storage to the breaker (or to a `Registry`, which hands it to every
breaker it creates — each coordinates under its own name):

```python
import redis
from interlock import CircuitBreaker, Registry
from interlock.integrations.redis import RedisStorage

storage = RedisStorage(redis.Redis(host='redis.internal'))
breaker = CircuitBreaker(name='payments', storage=storage)

registry = Registry(storage=storage)  # or share one storage across many breakers
```

Async services use the async client and storage:

```python
import redis.asyncio
from interlock import CircuitBreaker
from interlock.integrations.redis import AsyncRedisStorage

storage = AsyncRedisStorage(redis.asyncio.Redis(host='redis.internal'))
breaker = CircuitBreaker(name='payments', storage=storage)
```

A coordinated breaker matches its storage's runtime: a `RedisStorage` serves
only the sync API (`with`, sync `call`), an `AsyncRedisStorage` only the async
one (`async with`, async `call`); mixing the styles raises `InterlockError`
with a clear message. A breaker *without* a storage stays fully dual.

## How coordination works

The local state machine keeps owning the sliding window and trip detection;
Redis owns the shared OPEN/HALF_OPEN state and the global probe budget. All
state for one breaker lives in a single hash (`interlock:cb:<name>` by
default), and every transition runs as a Lua script, so racing instances stay
consistent.

The protected path stays fast:

- **CLOSED / OPEN admission** reads a locally cached view of the shared state —
  zero inline Redis calls. A background poller refreshes the cache every
  `poll_interval` seconds, so a trip on one instance reaches the others within
  roughly one interval.
- **HALF_OPEN admission** is the single inline Redis operation: an atomic probe
  lease that decrements the shared budget, bounding probes across the fleet.
- **Writes** (propagating a local trip, tallying probe outcomes, the final
  close-or-reopen decision) are fire-and-forget on a background worker; they
  never block a protected call.

Time comparisons ("has `wait_duration_in_open` elapsed?") use the *Redis
server's* clock, since instance clocks are not comparable. After the last probe
of a round, the deciding instance applies the same thresholds as the local
state machine and writes the transition guarded by a version check, so a
delayed decision can never overwrite a newer state.

## Coordinated mode contract

`RedisStorage` implements `Storage` / `AsyncStorage`; a third-party backend can
too. Four behaviours are part of that contract, not implementation detail —
skipping any of them breaks the guarantees the coordinator relies on:

1. **Fencing is mandatory.** `trip_open` and `close` accept `expected_version`
   and must apply the write only when the backend's current version matches
   it, otherwise no-op and return the current state. The coordinator relies on
   this for the probe-round decision: a delayed instance computing "probes
   passed" off a stale view must lose to a state another instance already
   wrote, never overwrite it.
2. **A leaked probe slot is bounded only by `state_ttl`.** `lease_probe`
   decrements a shared budget and has no corresponding un-lease operation. A
   `BaseException` — cancellation, process kill — between `lease_probe` and
   `record_probe` never returns that slot; only the key's TTL expiry does.
   With several instances interrupted mid-probe, `HALF_OPEN` can stall for up
   to `state_ttl`. Size it with this in mind — it is not just an
   abandoned-key cleanup knob.
3. **`record_probe` tallies only while `HALF_OPEN`.** Every coordinated write
   is best-effort *and* bounded: dropped while degraded, dropped when the
   write queue is full, superseded by a later decision, reconciled by the next
   poll rather than retried.
4. **Teardown is explicit, not automatic.** `close()` / `aclose()` stop the
   lane deterministically — see [Shutdown](#shutdown) below. A coordinated
   breaker left to be garbage-collected can outlive its owning scope instead.

Also: a breaker's `name` becomes a storage key (`interlock:cb:<name>` for
`RedisStorage`). Never build one from untrusted input — an attacker-controlled
name can collide with, and corrupt, unrelated breaker state.

### Checklist for a custom `Storage` implementation

- [ ] `trip_open` / `close` honor `expected_version`: apply only on a match,
      otherwise no-op and return the current state.
- [ ] Every operation is atomic against concurrent callers (a Lua script, a
      locked transaction, ...) — no read-modify-write races.
- [ ] `ttl` is refreshed on every write so an abandoned key self-expires.
- [ ] `lease_probe` grants only while `HALF_OPEN` and budget remains.
- [ ] `record_probe` tallies only while `HALF_OPEN`; a late or out-of-round
      outcome is dropped, not applied.
- [ ] No method raises into the protected path — the engine treats any
      exception as a signal to degrade to local state, so a backend that
      raises for a routine outcome (e.g. a fencing miss) breaks the contract;
      that case must return the current state instead.
- [ ] Time comparisons (has `wait_duration_in_open` elapsed?) use the
      backend's own clock, not the caller's — instance clocks are not
      comparable across a fleet.

## Manual controls

Manual controls are local to one process and take precedence over Redis while
they are active. `force_open()` rejects every local call; `disable()` and
`metrics_only()` admit local calls without consuming a Redis HALF_OPEN probe.
`reset()` clears the local override and local metrics, but does not reset Redis
for the fleet: the breaker immediately resumes its cached shared `OPEN` or
`HALF_OPEN` state.

## Degradation: Redis down ≠ breaker down

A storage error never reaches your calls. On the first failure the breaker
switches to its local state and keeps protecting the process on its own window;
pending shared writes are dropped, and Redis is left alone before the poller
tries again. That delay starts at `retry_backoff` seconds and, by default,
stays there for as long as the outage lasts — the same fixed cadence every
release before this one used. Set `retry_backoff_multiplier` above `1.0` to
grow it geometrically with each further consecutive failure (capped at
`retry_backoff_max`), plus `retry_jitter` to spread out the retries of a fleet
of instances recovering from the same outage instead of having them all probe
Redis in the same instant. On the first successful operation the shared view
becomes authoritative again — including adopting a shared OPEN that happened
while this instance was cut off — and the failure count resets, so the next
outage starts back at `retry_backoff`.

Both edges are observable through the listener:

```python
class StorageWatch:
    def on_storage_degraded(
        self, *, name: str, error: BaseException
    ) -> None: ...  # alert: running on local state

    def on_storage_recovered(self, *, name: str) -> None: ...  # back to coordinated state
```

`LoggingEventListener` logs degradation at `WARNING` and recovery at `INFO`;
`OTelEventListener` counts both on `interlock.storage.events`. Listeners
written before these hooks existed keep working — the engine calls them only if
present.

## Backpressure: the write queue is bounded

Coordinated writes are fire-and-forget: a local trip and each probe outcome are
queued for the background lane instead of being written on the protected path.
That queue holds at most `write_queue_size` writes (128 by default).

It is not a per-call queue — a healthy lane keeps it near empty, because writes
happen per *transition*, not per call, and probes are capped by
`permitted_calls_in_half_open`. The bound only matters when the lane stops
draining at all: a Redis client blocking without a timeout, or an async lane
whose event loop is gone. Without it, that lane would grow the queue for as
long as the process lives.

When the queue is full the *arriving* write is dropped — never blocked, never
raised into the call that produced it — and reported:

```python
class StorageWatch:
    def on_storage_write_dropped(self, *, name: str) -> None: ...  # shared state falling behind
```

Nothing is retried: the shared state is reconciled by the next successful poll
and, failing that, by `state_ttl` expiring the key. Locally the breaker keeps
protecting the process on its own window exactly as it does while degraded.

The hook is the signal that a lane is wedged — treat a non-zero rate as an
alert, not a tuning hint. `LoggingEventListener` logs it at `WARNING` and
`OTelEventListener` counts it on `interlock.storage.events`.

## Shutdown

A coordinated breaker owns a background lane — a daemon thread for a sync
storage, an asyncio task for an async one. It polls the shared view and drains
fire-and-forget writes. Left alone, it ends only when the breaker is garbage
collected, which is why an async lane can outlive `asyncio.run()` and log
"task was destroyed but it is pending".

`close()` (or `aclose()` for an async storage) ends it deterministically:

```python
breaker = CircuitBreaker(name='payments', storage=storage)
try:
    ...  # serve traffic
finally:
    breaker.close()  # drains queued writes, stops the lane, joins it
```

`Registry` does the whole set at once:

```python
registry = Registry(storage=storage)
...
registry.close_all()  # or: await registry.aclose_all()
```

What it guarantees:

- **Queued writes are drained first.** Ops already on the queue run in order
  before the lane exits, so a trip recorded just before shutdown still reaches
  Redis. Writes that the degraded gate would drop are still dropped.
- **No waiting out `poll_interval`.** A parked lane is woken immediately.
- **The `auto_transition` timer is cancelled**, and no new one is armed.
- **It is idempotent** and safe to call from any thread.

Two consequences worth knowing:

- **Shutdown is terminal.** The lane never restarts. Afterwards the breaker
  keeps protecting calls on its local state, and shared writes are dropped —
  the same behaviour as a degraded storage. `Registry.get()` keeps returning the
  closed instance rather than silently starting a fresh lane.
- **The cached shared view is dropped.** Nothing refreshes it once the lane is
  gone, so keeping it would pin the breaker in whatever a peer last published —
  a shared `OPEN` would never expire. The fallback to local state is reported
  through `on_state_change` like any other.

`close()` is teardown, not a state change: it does **not** close the circuit.
That is `reset()`.

## Tuning

All knobs live on the storage constructor; the core `Config` stays
storage-agnostic:

```python
RedisStorage(
    client,
    key_prefix='interlock:cb:',  # hash key namespace
    state_ttl=300.0,  # key lifetime (s); refreshed on every write
    poll_interval=1.0,  # cache refresh cadence (s)
    retry_backoff=5.0,  # local-only time after a storage failure (s)
    retry_backoff_multiplier=1.0,  # growth per consecutive failure; 1.0 = fixed delay
    retry_backoff_max=None,  # cap on the delay (s), or None for no cap
    retry_jitter=0.0,  # proportional random spread added to the (capped) delay
    write_queue_size=128,  # max pending coordinated writes; further ones are dropped
)
```

- **`state_ttl`** keeps abandoned state from lingering: if every instance
  disappears, the key expires and the breaker starts CLOSED. Keep it well above
  `wait_duration_in_open`.
- **`poll_interval`** is the propagation latency of a coordinated trip. Each
  breaker costs about one Redis read per interval.
- **`retry_backoff`** is the delay before the first retry after a storage
  failure, and the fixed delay for every retry after that while
  `retry_backoff_multiplier` stays at its default of `1.0`.
- **`retry_backoff_multiplier`** grows the delay geometrically with each
  further *consecutive* failure (`retry_backoff * retry_backoff_multiplier **
  attempts`); a recovery resets the attempt count. Must be `>= 1.0`.
- **`retry_backoff_max`** caps the grown delay in seconds. `None` (the
  default) leaves it uncapped.
- **`retry_jitter`** adds up to this fraction of the (capped) delay as random
  spread, so instances that degraded at the same moment do not all retry
  Redis at the same moment too. Deterministic given the same clock reading,
  attempt number and breaker name — it does not depend on process-global
  random state.
- **`write_queue_size`** bounds the pending coordinated writes; see
  [Backpressure](#backpressure-the-write-queue-is-bounded). The default of 128
  is far above what a draining lane ever holds, so raising it does not buy
  throughput — it only lets a wedged lane hold more memory before dropping.

## Compatibility

`RedisStorage` speaks plain commands and `EVAL` — no server-specific features —
so it works against Redis, [Valkey](https://valkey.io), or any RESP-compatible
server. The scripts call `TIME` before writing, which requires effect-based
script replication: **Redis 5.0 or newer**, or any Valkey release. (The
`redis>=5.0.0` dependency pin is the *client* library's version, not the
server's.)

---

<!-- source: docs/integrations/frameworks.md -->

# Flask / Django — recipe

When a route's outgoing dependency trips its breaker, the raised
[`CircuitOpenError`](../reference.md) should become a clean
`503 Service Unavailable` with a `Retry-After` header — the same behaviour
the [FastAPI](fastapi.md) and [Litestar](litestar.md) extras ship as code.
For other frameworks the handler is a few lines; no extra needed.

## Flask

```python
import math

from flask import Flask, jsonify

from interlock import CircuitOpenError

app = Flask(__name__)


@app.errorhandler(CircuitOpenError)
def on_circuit_open(exc: CircuitOpenError):
    response = jsonify({'detail': str(exc)})
    response.status_code = 503

    if exc.retry_after is not None:
        response.headers['Retry-After'] = str(math.ceil(exc.retry_after))

    return response
```

## Django

```python
# middleware.py
import json
import math

from django.http import HttpResponse

from interlock import CircuitOpenError


class CircuitOpenMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        return self.get_response(request)

    def process_exception(self, request, exception):
        if not isinstance(exception, CircuitOpenError):
            return None

        response = HttpResponse(
            json.dumps({'detail': str(exception)}),
            status=503,
            content_type='application/json',
        )

        if exception.retry_after is not None:
            response['Retry-After'] = str(math.ceil(exception.retry_after))

        return response
```

Add it to `MIDDLEWARE` in `settings.py`.

## Where the breakers live

The handler only translates the rejection. The breakers themselves guard your
*outgoing* calls — share one `Registry` across the app and wrap the
dependencies:

```python
from interlock import Registry

registry = Registry()
payments = registry.get('payments-api')


def charge(amount: int) -> str:
    return payments.call(gateway.charge, amount)
```

`Retry-After` is rounded up to whole seconds (per RFC 7231) and omitted when
the breaker cannot estimate the next probe (for example after
`force_open()`).

---

<!-- source: docs/reference.md -->

# API reference

Everything below is importable from the top-level `interlock` package, except
the integration adapters, which live in their own modules to keep the core
dependency-free.

## `CircuitBreaker`

```python
CircuitBreaker(*, name, config=None, clock=None, initial_state=State.CLOSED,
               classifier=None, listener=None, storage=None)
```

A named breaker for sync and async callables.

- **Use as** a decorator (`@breaker`), a sync/async context manager
  (`with` / `async with`), or `breaker.call(fn, *args, **kwargs)`.
- **`call_sync(fn, ...)` / `call_async(fn, ...)`** — the same protection with the
  sync/async dispatch skipped, for callers that already know their own nature
  (the transport integrations use them per request). `call_sync` never awaits: a
  coroutine function passed to it is recorded as an immediate success.
  `call_async` awaits whatever `fn` returns, so it accepts any
  awaitable-returning callable, not only a coroutine function.
- **Properties:** `name: str`, `state: State`.
- **`initial_state`** — one of `CLOSED`, `FORCED_OPEN`, `DISABLED` or
  `METRICS_ONLY`; transitional `OPEN` / `HALF_OPEN` raise `ValueError`.
- **`snapshot() -> WindowSnapshot`** — current, self-consistent window aggregates;
  concurrent call settlement cannot expose a partially updated window.
- **Manual control:** `reset()`, `force_open()`, `disable()`, `metrics_only()`.
- **`close()` / `aclose()`** — release background resources: the coordinator
  lane and the `auto_transition` timer. Teardown, not a state change: neither
  closes the circuit (`reset()` does). Idempotent, and terminal — the lane never
  restarts. See [Shutdown](integrations/redis.md#shutdown).
- **`storage`** — optional shared backend (`Storage` or `AsyncStorage`) for
  coordinated state across instances; see the
  [Redis integration](integrations/redis.md). A coordinated breaker matches its
  storage's runtime (sync storage → sync API, async storage → async API);
  without a storage the breaker stays fully dual.

## `Config`

Frozen dataclass of thresholds, window and timing; validated on construction.
See [Configuration](guides/configuration.md) for every field. Raises
`ValueError` on invalid input.

## `Registry`

```python
Registry(*, config=None, clock=None, initial_state=State.CLOSED,
         classifier=None, listener=None, storage=None)
registry.get(name, *, config=None) -> CircuitBreaker
registry.get_existing(name) -> CircuitBreaker | None
registry.names() -> tuple[str, ...]
registry.items() -> tuple[tuple[str, CircuitBreaker], ...]
registry.close_all() / await registry.aclose_all()
```

Creates and caches named breakers. The same name always returns the same
instance; the per-call `config` override applies only at creation. A cache hit
is served without taking the registry lock, so a registry shared by every
request of a service does not serialise on lookups. A `storage`
is handed to every breaker the registry creates; each coordinates under its own
name. `initial_state` is assigned before a lazy breaker is published;
`get_existing()` inspects the cache without creating a missing name.

`names()` and `items()` enumerate the breakers created so far — the only way to
see the ones the HTTP transports create lazily, one per host. Both take a
point-in-time copy under the registry lock: a breaker created afterwards is not
in it, and the returned tuple never changes.

```python
for name, breaker in registry.items():
    print(name, breaker.state, breaker.snapshot())
```

`close_all()` / `aclose_all()` close every breaker created so far. The cache is
kept, so `get()` keeps returning the same, torn-down instances instead of
silently starting a fresh lane after shutdown.

## Enums

- **`State`** — `CLOSED`, `OPEN`, `HALF_OPEN`, `FORCED_OPEN`, `DISABLED`,
  `METRICS_ONLY`. A `StrEnum`; values are stable lowercase identifiers.
- **`Outcome`** — `SUCCESS`, `FAILURE`, `SLOW_SUCCESS`, `SLOW_FAILURE`, with
  `.is_failure` and `.is_slow` properties.
- **`WindowType`** — `COUNT_BASED`, `TIME_BASED`.

## `WindowSnapshot`

Frozen dataclass: `total_calls`, `failed_calls`, `slow_calls`, plus
`.failure_rate` and `.slow_call_rate` properties (both `0.0` when empty).

## Errors & warnings

- **`InterlockError`** — base of all interlock errors.
- **`CircuitOpenError(breaker_name, *, retry_after=None, last_failure=None)`** —
  raised on rejection; attributes `breaker_name`, `retry_after`, `last_failure`.
- **`CallTimeoutError(timeout)`** — raised by `timeout` and `sync_timeout`;
  attribute `timeout`.
- **`BulkheadFullError(max_concurrent, *, max_wait=0.0)`** — raised by a
  pipeline bulkhead when no concurrency slot frees up in time; attributes
  `max_concurrent`, `max_wait`.
- **`InterlockDeprecationWarning`** — subclasses `UserWarning`, visible by
  default.

## `timeout` / `sync_timeout`

```python
async with timeout(seconds):
    ...  # async block


@sync_timeout(seconds)  # synchronous callable
def work(): ...
```

`timeout` is an async context manager that raises `CallTimeoutError` if the
block exceeds `seconds`. `sync_timeout` is a decorator that runs a synchronous
callable in a daemon worker thread and raises `CallTimeoutError` if it overruns
`seconds`; the worker keeps running after a timeout (Python cannot kill a
thread). See [Timeout](guides/timeout.md).

## Pipeline

Compose strategies around one call, outermost first — see the
[pipeline guide](guides/pipeline.md):

- **`Pipeline(*strategies)`** — the executor; works as a signature-preserving
  decorator and as `pipeline.call(fn, *args, **kwargs)` (detect-dispatching,
  like the breaker's). No context manager by design.
- **`Pipeline.builder()` / `PipelineBuilder`** — step-by-step assembly:
  `.fallback(...)`, `.retry(...)` (lazy `tenacity` extra),
  `.circuit_breaker(breaker)`, `.bulkhead(...)`, `.timeout(seconds)`,
  `.add(custom)`, `.build()`.
- **`Strategy`** — the structural protocol: `execute(call)` /
  `execute_async(call)`; `execute_async` always receives a real coroutine
  function.
- **`CircuitBreakerStrategy(breaker)`** — wraps a standalone breaker unchanged.
- **`TimeoutStrategy(seconds)`** — bounds every attempt via the v1 primitives.
- **`BulkheadStrategy(max_concurrent, *, max_wait=0.0, name='bulkhead',
  listener=None)`** — concurrency cap; raises `BulkheadFullError`.
- **`FallbackStrategy(fallback, *, on=(Exception,), name='fallback',
  listener=None)`** — explicit substitution for selected failures; result
  typed `T | F`.
- **`RetryStrategy(...)`** — lives in `interlock.integrations.tenacity` (see
  below).

## Protocols (extension points)

Implement any of these to swap a core behaviour:

- **`Clock`** — `monotonic() -> float`. Inject a fake for deterministic tests.
- **`SlidingWindow`** — `record(outcome)`, `snapshot() -> WindowSnapshot`.
- **`Storage`** / **`AsyncStorage`** — shared-state backend as atomic *intent*
  operations: `read`, `trip_open`, `begin_half_open_if_elapsed`, `lease_probe`,
  `record_probe`, `close`. `trip_open`/`close` take an optional
  `expected_version` (version-fenced CAS); every write carries a `ttl`.
  Mechanism only — threshold policy stays in the core. `AsyncStorage` is the
  awaitable mirror. See the [Redis integration](integrations/redis.md).
- **`FailureClassifier`** — `is_failure(*, result, exception) -> bool`, where
  `exception` is an `Exception` or `None` — cancellation and shutdown are
  released without being classified. See
  [Failure classification](guides/failure-classification.md).
- **`EventListener`** — `on_state_change`, `on_call`, `on_rejected`, `on_reset`,
  plus `on_storage_degraded` / `on_storage_recovered` /
  `on_storage_write_dropped` for coordinated breakers
  and `on_retry` / `on_bulkhead_rejected` / `on_fallback` for pipeline
  strategies (all optional hooks are dispatched only if present, so older
  listeners keep working). See [Observability](guides/observability.md).

## Shared-state types

- **`SharedState`** — frozen snapshot of one breaker's coordinated state:
  `state`, `opened_at` (backend time), `version` (for fencing), and the
  HALF_OPEN probe accounting (`probes_permitted`, `probes_remaining`,
  `probes_completed`, `probe_failures`, `probe_slows`).
  `SharedState.closed()` is the baseline an absent key implies.
- **`ProbeLease`** — result of `lease_probe`: `granted: bool` plus the
  post-attempt `state: SharedState`.

## Listeners

- **`LoggingEventListener(logger=None)`** — top-level; zero dependencies.
- **`interlock.integrations.otel.OTelEventListener(meter=None)`** — extra `interlock-cb[otel]`.

## httpx2 adapters

Extra `interlock-cb[httpx2]`, module `interlock.integrations.httpx2`:

- **`CircuitBreakerTransport(transport, *, config=None, clock=None,
  initial_state=State.CLOSED, classifier=None, listener=None)`**
- **`AsyncCircuitBreakerTransport(transport, *, ...)`**
- **`HttpStatusClassifier(*, failure_statuses=None, excluded_exceptions=None)`** —
  fails on transport exceptions and statuses `429, 500, 502, 503, 504`
  (override the set via `failure_statuses`). `UnsupportedProtocol` and
  `LocalProtocolError` are caller-side and count as successes; replace that set
  via `excluded_exceptions`.

Both transports expose their per-host `registry` and the guarded transport as a
read-only `wrapped`. `close()` / `aclose()` release the wrapped transport and,
when the transport owns the registry, every breaker in it; a caller-owned
registry stays open and is closed by its owner. See the
[httpx2 integration](integrations/httpx2.md).

## httpx adapters

Extra `interlock-cb[httpx]` (httpx ≥ 0.27.0), module
`interlock.integrations.httpx`:

- **`CircuitBreakerTransport(transport, *, config=None, clock=None,
  initial_state=State.CLOSED, classifier=None, listener=None)`**
- **`AsyncCircuitBreakerTransport(transport, *, ...)`**
- **`HttpStatusClassifier(*, failure_statuses=None, excluded_exceptions=None)`** —
  same policy, including the caller-side `UnsupportedProtocol` /
  `LocalProtocolError` exclusions.

Both transports expose their per-host `registry` and the guarded transport as a
read-only `wrapped`, and preserve streaming responses. `close()` / `aclose()`
release the wrapped transport and, when the transport owns the registry, every
breaker in it; a caller-owned registry stays open and is closed by its owner.
See the [httpx integration](integrations/httpx.md).

## aiohttp adapters

Extra `interlock-cb[aiohttp]` (aiohttp ≥ 3.12), module `interlock.integrations.aiohttp`:

- **`CircuitBreakerMiddleware(*, config=None, clock=None,
  initial_state=State.CLOSED, classifier=None, listener=None)`** —
  client middleware for `ClientSession(middlewares=(...,))`; one breaker per
  request host.
- **`HttpStatusClassifier(*, failure_statuses=None, excluded_exceptions=None)`** —
  same canonical HTTP policy, reading `ClientResponse.status`; nothing is
  excluded by default (aiohttp rejects bad URLs before the middleware runs).

It exposes its `registry`; call `await middleware.aclose()` during application
shutdown. See the [aiohttp integration](integrations/aiohttp.md).

## requests adapters

Extra `interlock-cb[requests]`, module `interlock.integrations.requests`:

- **`CircuitBreakerAdapter(*, config=None, clock=None,
  initial_state=State.CLOSED, classifier=None, listener=None, **adapter_kwargs)`** —
  `HTTPAdapter` subclass for `session.mount(...)`; one breaker per request
  host. Extra kwargs go to `HTTPAdapter`.
- **`HttpStatusClassifier(*, failure_statuses=None, excluded_exceptions=None)`** —
  same policy, reading `Response.status_code`; the caller-side `InvalidURL`
  counts as a success.

It exposes its `registry`; closing the adapter or owning session releases both
the connection pools and breaker resources. See the
[requests integration](integrations/requests.md).

## tenacity helpers

Extra `interlock-cb[tenacity]`, module `interlock.integrations.tenacity`:

- **`retry_unless_open(*transient)`** — tenacity retry predicate: retries the
  listed transient exception types (default: any `Exception`), never
  `CircuitOpenError`.
- **`wait_probe(fallback, *, jitter=0.1)`** — tenacity wait strategy: sleeps
  `CircuitOpenError.retry_after` (+ up to `jitter` seconds) after a
  rejection, delegates to `fallback` otherwise.
- **`RetryStrategy(*, attempts=3, retry=None, wait=None, sleep=None,
  async_sleep=None, before_sleep=None, name='retry', listener=None)`** — a
  bounded retry layer for the pipeline: policy delegated to tenacity,
  attempts always capped, the original exception re-raised when the budget
  runs out, `CircuitOpenError` not retried by default.

See the [tenacity integration](integrations/tenacity.md) and the
[retries guide](guides/retries.md).

## FastAPI adapters

Extra `interlock-cb[fastapi]`, module `interlock.integrations.fastapi`:

- **`breaker_dependency(name, *, registry)`** — returns a `Depends`-compatible
  callable yielding the named breaker from a shared `Registry`.
- **`install_exception_handler(app)`** — registers a handler mapping
  `CircuitOpenError` to `503` with a `Retry-After` header.
- **`circuit_open_handler(request, exc)`** — the handler itself, for custom
  registration.

See the [FastAPI integration](integrations/fastapi.md).

## Litestar adapters

Extra `interlock-cb[litestar]` (Litestar ≥ 2.23), module
`interlock.integrations.litestar`:

- **`breaker_dependency(name, *, registry)`** — returns a `Provide` yielding
  the named breaker from a shared `Registry`; annotate handler parameters
  with `NamedDependency[CircuitBreaker]`.
- **`circuit_open_handler(request, exc)`** — maps `CircuitOpenError` to `503`
  with a `Retry-After` header; pass it in the app's `exception_handlers`.

See the [Litestar integration](integrations/litestar.md).

## Redis adapters

Extra `interlock-cb[redis]`, module `interlock.integrations.redis`:

- **`RedisStorage(client, *, key_prefix='interlock:cb:', state_ttl=300.0, poll_interval=1.0, retry_backoff=5.0, retry_backoff_multiplier=1.0, retry_backoff_max=None, retry_jitter=0.0, write_queue_size=128)`** —
  sync `Storage` over a `redis.Redis` client.
- **`AsyncRedisStorage(client, *, ...)`** — async mirror over
  `redis.asyncio.Redis`.

One Redis hash per breaker; every transition is a Lua script (atomic across
racing instances), elapse checks use the server's `TIME`. Works against Redis
(5.0+), Valkey, or any RESP-compatible server.

See the [Redis integration](integrations/redis.md).
