Real-Time State Streaming with WebSocket and StateStream - NSP

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

Use streaming (session.watch())

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.

Use polling (session.refresh())

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.

Opening a Stream

Via Session (Recommended)

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

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:

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

`StateChangeNotification` Fields

pid

int

OS process ID of the application that changed.

app_id

str

Stable semantic app identifier, e.g. "gmail-chrome-12847".

sequence

int

Monotonically increasing counter. A gap between consecutive sequence numbers means one or more events were dropped (queue overflow or reconnection).

changed_keys

list[str]

The state keys that changed since the previous notification, e.g. ["inbox.unread_count", "inbox.emails[0].is_read"].

capture_id

str

Identifier for the state capture that produced this notification.

prev_capture_id

str

Identifier for the previous capture, enabling diff lookups.

captured_at

datetime

UTC timestamp of the state capture that produced this event.

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:

# 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}")

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.

Filtering by Changed Keys

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

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:

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:

# 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+):

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:

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:

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:

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:

{
  "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:

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())

​When to Use Streaming vs. Polling

Use streaming (session.watch())

Use polling (session.refresh())

​Opening a Stream

​Via Session (Recommended)

​Via Client (Raw Stream)

​StateChangeNotification Fields

​Refreshing State After Events

​Filtering by Changed Keys

​Reconnection and Sequence Gaps

​Queue Configuration

​Timeout and Cancellation

​Watching Multiple Apps Simultaneously

​Low-Level: open_state_stream(pid, client)

​WebSocket Wire Format

​Complete Example: Reactive Inbox Agent

When to Use Streaming vs. Polling

Opening a Stream

Via Session (Recommended)

Via Client (Raw Stream)

`StateChangeNotification` Fields

Refreshing State After Events

Filtering by Changed Keys

Reconnection and Sequence Gaps

Queue Configuration

Timeout and Cancellation

Watching Multiple Apps Simultaneously

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

WebSocket Wire Format

Complete Example: Reactive Inbox Agent