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

# Python SDK - Stream Observer

> Opt-in structured callbacks for SSE lifecycle events in the Python SDK.

Pass a `StreamObserver` via `NexusConfig.stream_observer` to receive structured callbacks for every SSE lifecycle transition. The observer is the **canonical** way to wire SDK state into ops dashboards, custom telemetry, or load-test tooling - far more reliable than scraping `NexusLogger` output by substring match.

The observer is **opt-in**. When `NexusConfig.stream_observer` is `None` the SDK uses `NOOP_STREAM_OBSERVER` at zero overhead.

## The interface

```python theme={null}
from datetime import datetime
from typing import Optional, Protocol, runtime_checkable

@runtime_checkable
class StreamObserver(Protocol):
    def on_connected(self) -> None: ...
    def on_disconnected(self, cause: Optional[BaseException]) -> None: ...
    def on_event(self, name: str) -> None: ...
    def on_reconnect_attempt(self, attempt: int) -> None: ...
    def on_fallback(self, reason: str) -> None: ...
    def on_quarantined(self, reason: str, expires_at: datetime) -> None: ...  # v0.4.0
```

The protocol is structural - any object with the matching method signatures qualifies. Methods are looked up via `getattr`, so consumers may omit any method they don't care about (the SDK handles missing-attribute cases gracefully).

## Hook semantics

| Hook                                            | Fires when                                                                                      | Argument                                                       |
| ----------------------------------------------- | ----------------------------------------------------------------------------------------------- | -------------------------------------------------------------- |
| `on_connected()`                                | `GET /v1/stream` returns `200 OK` and the SDK is ready to read events                           | none                                                           |
| `on_disconnected(cause)`                        | An active stream connection ends - clean disconnect, transport error, or server-initiated close | `None` for clean shutdown, the underlying exception otherwise  |
| `on_event(name)`                                | A whitelisted SSE event arrives, **before** the SDK runs the event-triggered sync               | event name string (e.g. `"config.updated"`)                    |
| `on_reconnect_attempt(n)`                       | The SDK is about to back off and retry after a transport error                                  | 1-based attempt counter (max 3 with current threshold)         |
| `on_fallback(reason)`                           | The SDK enters a degraded state                                                                 | `FALLBACK_REASON_MAX_ERRORS` or `FALLBACK_REASON_RATE_LIMITED` |
| `on_quarantined(reason, expires_at)` *(v0.4.0)* | Server returned `429` with a quarantine body on either the sync or stream path                  | `reason` string; `expires_at` timezone-aware `datetime`        |

### `on_fallback` reasons

```python theme={null}
from westyx_nexus import (
    FALLBACK_REASON_MAX_ERRORS,
    FALLBACK_REASON_RATE_LIMITED,
)
```

| Constant         | When                                                                 | Stream after                                            |
| ---------------- | -------------------------------------------------------------------- | ------------------------------------------------------- |
| `'max-errors'`   | 3 consecutive transport errors - backoff exhausted, the SDK gives up | TTL polling only; reconnect requires `connect_stream()` |
| `'rate-limited'` | server returned `429 Too Many Requests` (tier connection-limit hit)  | depends on the response - see below                     |

**`rate-limited` semantics (v0.3.1+).** This reason now fires in **both** of these situations:

* **Server included a parseable `Retry-After` header.** The SDK fires `on_fallback(FALLBACK_REASON_RATE_LIMITED)`, sleeps the indicated delay clamped to `[5, 300]` seconds, and then reconnects automatically. The 429 does **not** count against `_MAX_STREAM_ERRORS`. Expect a matching `on_connected()` shortly after the sleep window.
* **No parseable `Retry-After`.** Legacy behaviour: the SDK fires `on_fallback(FALLBACK_REASON_RATE_LIMITED)`, closes the stream, and stays on TTL polling. No automatic reconnect - `connect_stream()` is required to recover.

## Threading contract

For the **synchronous** `NexusClient`, observer methods are called **synchronously from the SDK's stream thread**. For the **async** `AsyncNexusClient`, observer methods are called **synchronously from the asyncio task** running the stream - they are NOT awaited even if you define them as coroutines.

<Warning>
  In both cases observer methods MUST return promptly. The SDK does not protect against blocking observers.
</Warning>

If your observer does anything substantial (network I/O, lock contention, disk writes), copy the argument and dispatch the work to your own thread / task before returning:

```python theme={null}
import queue, threading
class SlowObserver:
    def __init__(self) -> None:
        self.q: queue.Queue[str] = queue.Queue(maxsize=1024)
        threading.Thread(target=self._drain, daemon=True).start()

    def on_event(self, name: str) -> None:
        try:
            self.q.put_nowait(name)
        except queue.Full:
            pass  # Drop on overflow - backpressure on the SDK would stall reads.

    def _drain(self) -> None:
        while True:
            event = self.q.get()
            ... # process(event)
```

## Counter pattern

```python theme={null}
from dataclasses import dataclass
from threading import Lock
from westyx_nexus import NexusClient, NexusConfig

@dataclass
class CounterObserver:
    events: int = 0
    reconnects: int = 0
    fallbacks: int = 0
    _lock: Lock = Lock()

    def on_event(self, name: str) -> None:
        with self._lock:
            self.events += 1

    def on_reconnect_attempt(self, attempt: int) -> None:
        with self._lock:
            self.reconnects += 1

    def on_fallback(self, reason: str) -> None:
        with self._lock:
            self.fallbacks += 1

obs = CounterObserver()
client = NexusClient.create(NexusConfig(
    base_url="https://...",
    api_key="wxs_...",
    stream_observer=obs,
))
```

## Active-connection gauge pattern

```python theme={null}
class GaugeObserver:
    def __init__(self) -> None:
        self.connected = False

    def on_connected(self) -> None:
        self.connected = True

    def on_disconnected(self, cause) -> None:
        self.connected = False

    def on_fallback(self, reason: str) -> None:
        self.connected = False
```

`on_fallback` is included because after the 3-strike `FALLBACK_REASON_MAX_ERRORS` case the SDK will not reconnect - the gauge should stay `False` until you call `connect_stream()` again. For `FALLBACK_REASON_RATE_LIMITED` with a `Retry-After` (v0.3.1+) the SDK *does* reconnect on its own; the gauge will flip back to `True` on the next `on_connected()`.