Stream Real-Time App State Changes with NSP WebSocket

Instead of polling the state endpoint on a timer, you can subscribe to the NSP daemon’s WebSocket watch endpoint and receive a push notification every time a tracked application’s state changes. The daemon fires a lightweight JSON message on each probe cycle that detects at least one changed key — typically every two seconds. Your code wakes up only when something actually changed, processes only the keys that changed, and then fetches fresh state if it needs the updated values. This model is far more efficient than high-frequency polling and does not count against your action rate limit.

WebSocket Endpoint

Connect to the watch endpoint using the tracked process’s PID:

ws://127.0.0.1:7842/substrate/v1/watch/{pid}?key=<api_key>

Because WebSocket clients don’t uniformly support custom HTTP headers during the upgrade handshake, pass your API key as a query parameter rather than a header:

ws://localhost:7842/substrate/v1/watch/12847?key=nsp_live_sk_a1b2c3d4...

Find the PID for your target app by calling GET /substrate/v1/apps first, or by reading session.state.pid from the Python SDK after calling client.attach().

Message Format

The server sends one JSON message per probe cycle in which it detects state changes. Messages are never batched — each message represents one detection event:

{
  "pid": 12847,
  "app_id": "gmail-chrome-12847",
  "sequence": 142,
  "changed_keys": [
    "inbox.unread_count",
    "inbox.emails[0].is_read"
  ],
  "captured_at": "2026-07-20T13:00:00.123Z"
}

Field	Type	Description
`pid`	`integer`	OS process ID of the tracked application
`app_id`	`string`	Stable application identifier — survives process restarts
`sequence`	`integer`	Monotonically increasing counter per connection. Use this to detect missed events.
`changed_keys`	`string[]`	Dot-notation keys whose values changed since the previous notification
`captured_at`	`datetime`	ISO 8601 timestamp of the state capture that produced this event

Behavior Notes

Understanding how the watch stream behaves helps you write robust reconnection and event-handling logic:

Cadence: The server fires one event per successful probe cycle that detects changes. With the default poll_interval_secs = 2, you should expect at most one event every two seconds when the app is active.
Process exit: If the tracked process exits, the server closes the WebSocket connection with close code 1000 (Normal Closure). Treat this as a clean shutdown, not an error.
Daemon restart: If the NSP daemon itself restarts, your connection is dropped without a clean close frame. Your client should detect this and reconnect.
No heartbeat required: The server does not send ping frames and does not expect pong frames. The stream is event-driven — silence means no changes were detected.
Sequence gaps: If event.sequence jumps by more than 1 between consecutive messages (for example, from 142 to 145), it means the daemon detected state changes in intermediate cycles but your client did not receive them — possibly due to network buffering or a brief reconnect. Treat a gap as a signal that you may have missed intermediate updates and re-fetch full state if you need to be certain.

Python SDK

The Python SDK wraps the raw WebSocket connection in a typed async stream. Use session.watch() as an async context manager and iterate over the events it yields.

Attach to the target app

Call client.attach() with the app’s name. The SDK resolves the PID automatically and returns a session with session.state already populated:

session = await client.attach("Gmail")
print(f"Attached to Gmail (pid={session.state.pid})")

Open the watch stream

Use session.watch() as an async context manager. The stream opens a WebSocket connection to the /watch/{pid} endpoint and keeps it alive for the duration of the async with block:

async with session.watch() as stream:
    # stream is now active

Iterate over events

Use async for to receive each event as it arrives. Each event object exposes event.changed_keys (list of strings) and event.sequence (integer):

async for event in stream:
    print(f"[seq={event.sequence}] Changed: {event.changed_keys}")

Call session.refresh() to get updated values

The event tells you which keys changed, but not their new values. Call session.refresh() inside your handler to pull the latest state snapshot from the daemon, then use the typed session.get_*() helpers to read specific keys:

if "inbox.unread_count" in event.changed_keys:
    await session.refresh()
    unread = session.get_int("inbox.unread_count")
    print(f"  Unread count is now: {unread}")

Full Python Example

import asyncio
import nelieo_nsp as nsp

async def watch_gmail():
    async with nsp.NSPClient() as client:
        session = await client.attach("Gmail")
        print(f"Watching Gmail (pid={session.state.pid})...")

        async with session.watch() as stream:
            async for event in stream:
                print(f"[seq={event.sequence}] Changed: {event.changed_keys}")

                if "inbox.unread_count" in event.changed_keys:
                    await session.refresh()
                    unread = session.get_int("inbox.unread_count")
                    print(f"  Unread count: {unread}")

asyncio.run(watch_gmail())

Sample output:

Watching Gmail (pid=12847)...
[seq=141] Changed: ['context.url']
[seq=142] Changed: ['inbox.unread_count', 'inbox.emails[0].is_read']
  Unread count: 46
[seq=143] Changed: ['inbox.emails[0].subject', 'inbox.emails[1].from']

JavaScript Example

Use the browser’s native WebSocket API (or Node’s ws package) to connect to the stream. Parse each incoming message as JSON and react to the changed keys:

const pid = 12847;
const apiKey = "nsp_live_sk_a1b2c3d4...";

const ws = new WebSocket(
  `ws://localhost:7842/substrate/v1/watch/${pid}?key=${apiKey}`
);

ws.onopen = () => {
  console.log("Watch stream connected");
};

ws.onmessage = (msg) => {
  const event = JSON.parse(msg.data);
  console.log(`[seq=${event.sequence}] Changed:`, event.changed_keys);

  if (event.changed_keys.includes("inbox.unread_count")) {
    // Fetch updated state via REST and update your UI
    fetchUpdatedState(event.pid);
  }
};

ws.onclose = (e) => {
  console.log(`Stream closed (code=${e.code}):`, e.reason);
  if (e.code !== 1000) {
    // Unexpected close — reconnect
    scheduleReconnect();
  }
};

ws.onerror = (err) => {
  console.error("WebSocket error:", err);
};

Reconnection Strategy

The watch connection can drop if the daemon restarts, the machine sleeps and wakes, or there is a brief network hiccup. Implement an exponential backoff reconnection loop to handle this automatically:

import asyncio
import random
import nelieo_nsp as nsp

async def watch_with_reconnect(app_name: str):
    attempt = 0
    max_backoff = 60.0

    while True:
        try:
            async with nsp.NSPClient() as client:
                session = await client.attach(app_name)
                print(f"Connected (attempt {attempt + 1}), watching {app_name}...")
                attempt = 0  # reset on successful connection

                async with session.watch() as stream:
                    async for event in stream:
                        await handle_event(session, event)

        except nsp.NSPConnectionError as e:
            attempt += 1
            backoff = min(2 ** attempt + random.uniform(0, 1), max_backoff)
            print(f"Connection lost ({e}). Reconnecting in {backoff:.1f}s...")
            await asyncio.sleep(backoff)

        except nsp.NSPAppNotFoundError:
            # App is not running yet — wait and retry
            await asyncio.sleep(5)


async def handle_event(session, event):
    if "inbox.unread_count" in event.changed_keys:
        await session.refresh()
        unread = session.get_int("inbox.unread_count")
        print(f"[seq={event.sequence}] Unread: {unread}")

asyncio.run(watch_with_reconnect("Gmail"))

Key points in this pattern:

Reset the attempt counter after a successful connection so the backoff stays short after a clean reconnect.
Cap the backoff at a sensible ceiling (60 seconds here) so you do not wait forever if the daemon is down for maintenance.
Distinguish between a connection error (retry) and an app-not-found error (app not running yet — also retry, but with a fixed short delay).

Handling Sequence Gaps

Track the sequence number across events and alert when a gap appears:

last_sequence = None

async for event in stream:
    if last_sequence is not None:
        gap = event.sequence - last_sequence - 1
        if gap > 0:
            print(f"Warning: missed {gap} event(s) between seq {last_sequence} and {event.sequence}")
            # Re-fetch full state to catch any changes you missed
            await session.refresh()

    last_sequence = event.sequence
    await handle_event(session, event)

A gap of one or two events is usually harmless for display purposes. If your agent makes decisions based on accumulated state (for example, a counter that it increments), a gap means you should refresh the full state before acting rather than relying on what you saw in the last event.

Performance Notes

Use watch instead of polling

The watch stream wakes your code only when something changes. Polling GET /state/{pid} every 2 seconds makes 1 800 requests per hour even when nothing is happening. Watch events arrive in the same timeframe with zero wasted requests.

No action rate limit impact

Watch events are delivered over a persistent WebSocket connection and are counted separately from REST requests. They do not consume any portion of your 10 actions/second or 100 state reads/second rate limit budgets.

Filter event.changed_keys before calling session.refresh(). If the only changed key is context.url and your agent only cares about inbox.unread_count, skip the refresh entirely. Every unnecessary refresh is a round-trip to the daemon that adds latency and burns state read quota.

HTTP Upgrade Error Reference

If the WebSocket upgrade request fails, the server returns an HTTP error before the connection is established:

Status	Description
`401`	Missing or invalid API key — check the `key` query parameter
`404`	PID is not currently tracked — call `GET /substrate/v1/apps` to confirm the process is running
`426`	Client did not send a valid WebSocket upgrade request — ensure you are using a WebSocket client, not a plain HTTP client

Get Started

The Daemon

Runtime Probes

Security

Guides

WebSocket Endpoint

Message Format

Behavior Notes

Python SDK

Full Python Example

JavaScript Example

Reconnection Strategy

Handling Sequence Gaps

Performance Notes

Use watch instead of polling

No action rate limit impact

HTTP Upgrade Error Reference

​WebSocket Endpoint

​Message Format

​Behavior Notes

​Python SDK

​Full Python Example

​JavaScript Example

​Reconnection Strategy

​Handling Sequence Gaps

​Performance Notes

Use watch instead of polling

No action rate limit impact

​HTTP Upgrade Error Reference

WebSocket Endpoint

Message Format

Behavior Notes

Python SDK

Full Python Example

JavaScript Example

Reconnection Strategy

Handling Sequence Gaps

Performance Notes

HTTP Upgrade Error Reference