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

# Real-Time State Streaming with WebSocket and StateStream

> Subscribe to live WebSocket state-change events, filter by changed keys, handle reconnects, and build reactive agents with StateStream and session.watch().

The NSP daemon broadcasts a `StateChangeNotification` over WebSocket every time a tracked process's state changes. The Python SDK delivers these events through `StateStream` — an async iterator you consume inside an `async for` loop. This lets you react to application state changes in real time with sub-100 ms latency.

## When to Use Streaming vs. Polling

<CardGroup cols={2}>
  <Card title="Use streaming (session.watch())" icon="radio">
    When you need sub-second reaction to state changes, are building a monitoring loop, or want to trigger an action the moment a condition changes.
  </Card>

  <Card title="Use polling (session.refresh())" icon="clock">
    When you only need periodic reads (e.g. every 5 seconds) or are running a one-shot agent task that does not need to react to changes live.
  </Card>
</CardGroup>

## Opening a Stream

### Via Session (Recommended)

Use `session.watch()` for the most ergonomic experience. It automatically binds the stream to the session's PID:

```python theme={null}
async with session.watch() as stream:
    async for event in stream:
        print(event.changed_keys)   # ["inbox.unread_count", "inbox.emails[0]"]
        print(event.sequence)       # monotonic counter
        print(event.captured_at)    # datetime
        print(event.pid)            # 12847
        print(event.app_id)         # "gmail-chrome-12847"
```

### Via Client (Raw Stream)

Use `client.watch(pid=...)` when you do not have a session object, or when you need a raw stream for a specific PID:

```python theme={null}
async with client.watch(pid=12847) as stream:
    async for event in stream:
        print(event.changed_keys)
```

## `StateChangeNotification` Fields

<ResponseField name="pid" type="int">
  OS process ID of the application that changed.
</ResponseField>

<ResponseField name="app_id" type="str">
  Stable semantic app identifier, e.g. `"gmail-chrome-12847"`.
</ResponseField>

<ResponseField name="sequence" type="int">
  Monotonically increasing counter. A gap between consecutive sequence numbers means one or more events were dropped (queue overflow or reconnection).
</ResponseField>

<ResponseField name="changed_keys" type="list[str]">
  The state keys that changed since the previous notification, e.g. `["inbox.unread_count", "inbox.emails[0].is_read"]`.
</ResponseField>

<ResponseField name="capture_id" type="str">
  Identifier for the state capture that produced this notification.
</ResponseField>

<ResponseField name="prev_capture_id" type="str">
  Identifier for the previous capture, enabling diff lookups.
</ResponseField>

<ResponseField name="captured_at" type="datetime">
  UTC timestamp of the state capture that produced this event.
</ResponseField>

## Refreshing State After Events

When you receive a `StateChangeNotification`, `session.state` is **not automatically updated** unless you pass `auto_refresh_on_event=True`. You control when to refresh:

```python theme={null}
# Manual refresh — call refresh() when you need updated values
async with session.watch() as stream:
    async for event in stream:
        if "inbox.unread_count" in event.changed_keys:
            await session.refresh()
            unread = session.get_int("inbox.unread_count")
            print(f"Unread: {unread}")

# Auto-refresh — session.state is updated before every iteration
async with session.watch(auto_refresh_on_event=True) as stream:
    async for event in stream:
        # session.state is always current here
        unread = session.get_int("inbox.unread_count")
        print(f"[{event.captured_at}] Unread: {unread}")
```

<Tip>
  Do not call `session.refresh()` unconditionally on every event. Check `event.changed_keys` first and only refresh when a key you care about has changed. This reduces unnecessary round-trips to the daemon on high-frequency streams.
</Tip>

## Filtering by Changed Keys

React only to the parts of state you care about by inspecting `changed_keys` before doing any work:

```python theme={null}
async with session.watch() as stream:
    async for event in stream:
        # Skip events that don't touch the inbox
        if not any(k.startswith("inbox.") for k in event.changed_keys):
            continue

        await session.refresh()
        unread = session.get_int("inbox.unread_count")

        if unread > 0:
            await handle_new_emails(session)
```

## Reconnection and Sequence Gaps

By default, `StateStream` reconnects automatically if the WebSocket drops. To detect reconnections, watch for gaps in `event.sequence`:

```python theme={null}
last_seq = 0

async with session.watch(auto_reconnect=True) as stream:
    async for event in stream:
        if event.sequence > last_seq + 1:
            missed = event.sequence - last_seq - 1
            print(f"⚠ Sequence gap: missed {missed} event(s) — reconnected")
        last_seq = event.sequence
        print(f"seq={event.sequence}  changed={event.changed_keys}")
```

Set `auto_reconnect=False` to receive a `NSPConnectionError` instead of a transparent reconnect when the connection drops.

## Queue Configuration

Events are buffered in an internal asyncio queue. If your consumer is slower than the daemon's emit rate, the oldest events are dropped and reflected as sequence gaps:

```python theme={null}
# Default queue size is 128 events
async with session.watch(max_queue=1024) as stream:
    async for event in stream:
        # more headroom for slow consumers
        await slow_processing_function(event)
```

## Timeout and Cancellation

Limit how long you watch using `asyncio.timeout` (Python 3.11+):

```python theme={null}
import asyncio

async def watch_with_timeout():
    try:
        async with asyncio.timeout(60.0):
            async with session.watch() as stream:
                async for event in stream:
                    print(event.changed_keys)
    except asyncio.TimeoutError:
        print("Watch stopped after 60 seconds")
```

Cancel a watching task explicitly:

```python theme={null}
async def watch_task():
    async with session.watch() as stream:
        async for event in stream:
            print(event.changed_keys)

task = asyncio.create_task(watch_task())
await asyncio.sleep(60.0)
task.cancel()
try:
    await task
except asyncio.CancelledError:
    pass
```

## Watching Multiple Apps Simultaneously

Open one stream per session and run them concurrently with `asyncio.gather`:

```python theme={null}
import asyncio
import nelieo_nsp as nsp

async def watch_all():
    async with nsp.NSPClient(api_key="sk_nsp_...") as client:
        apps = await client.list_apps()
        sessions = [await client.session(app.pid) for app in apps]

        async def watch_one(s):
            async with s.watch() as stream:
                async for event in stream:
                    print(f"[{s.app_name}] {event.changed_keys}")

        await asyncio.gather(*[watch_one(s) for s in sessions])

asyncio.run(watch_all())
```

## Low-Level: `open_state_stream(pid, client)`

For advanced use cases, use `open_state_stream()` directly to obtain a `StateStream` without a session:

```python theme={null}
from nelieo_nsp import open_state_stream

async with open_state_stream(pid=12847, client=client) as stream:
    async for event in stream:
        print(event.changed_keys)
```

### WebSocket Wire Format

The stream connects to:

```
ws://localhost:7842/substrate/v1/watch/{pid}?key=sk_nsp_...
```

Each message is a JSON object:

```json theme={null}
{
  "pid": 12847,
  "app_id": "gmail-chrome-12847",
  "sequence": 142,
  "changed_keys": ["inbox.unread_count", "inbox.emails[0].is_read"],
  "capture_id": "cap_9f3a",
  "prev_capture_id": "cap_9f39",
  "captured_at": "2026-07-20T13:00:00.123Z"
}
```

## Complete Example: Reactive Inbox Agent

This agent opens a stream, watches for new emails, and immediately archives any message matching a filter — all in real time:

```python theme={null}
import asyncio
import nelieo_nsp as nsp

ARCHIVE_SENDERS = {"noreply@github.com", "alerts@pagerduty.com"}

async def reactive_inbox_agent():
    async with nsp.NSPClient(api_key="sk_nsp_...") as client:
        session = await client.attach("Gmail")
        print(f"Attached to Gmail  conf={session.confidence:.2f}")

        last_seq = 0

        async with session.watch(auto_reconnect=True) as stream:
            async for event in stream:
                # Detect and report sequence gaps (reconnections / dropped events)
                if event.sequence > last_seq + 1:
                    print(f"⚠ Missed {event.sequence - last_seq - 1} event(s)")
                last_seq = event.sequence

                # Only act when the inbox changed
                if not any(k.startswith("inbox.") for k in event.changed_keys):
                    continue

                # Refresh state only when relevant keys changed
                await session.refresh()

                unread = session.get_int("inbox.unread_count", default=0)
                if unread == 0:
                    continue

                sender  = session.get_str("inbox.emails[0].from", default="")
                subject = session.get_str("inbox.emails[0].subject", default="")
                email_id = session.get_str("inbox.emails[0].id", default="")

                print(f"New email from {sender}: {subject}")

                if sender in ARCHIVE_SENDERS and email_id:
                    result = await session.execute(
                        "archive",
                        parameters={"email_id": email_id},
                    )
                    if result.success:
                        print(f"  ✅ Auto-archived in {result.latency_ms}ms")
                    else:
                        print(f"  ❌ Archive failed: {result.error_message}")

asyncio.run(reactive_inbox_agent())
```