NSPSession: Read Application State and Run Actions - NSP

NSPSession is the high-level interface for interacting with a single attached application. It bundles the process ID and app identifier together so you never have to pass them around manually. Every state read, action call, and streaming subscription goes through a session.

Creating a Session

You create a session by calling one of the attach methods on NSPClient:

# Most common — attach by app name
session = await client.attach("Gmail")

# Attach by PID
session = await client.attach_by_pid(12847)

# Force a fresh probe before the session opens
session = await client.attach("Gmail", fresh=True)

Session Properties

Property	Type	Description
`session.state`	`SubstrateState`	Cached state snapshot. Call `refresh()` to update.
`session.pid`	`int`	OS process ID
`session.app_id`	`str`	Stable semantic app identifier
`session.app_name`	`str`	Human-readable application name
`session.confidence`	`float`	Probe confidence of the cached state (0.0–1.0)
`session.actions`	`dict[str, ActionSchema]`	Available actions keyed by action name
`session.clr`	`CLRSessionProxy`	.NET-specific helpers (CLR apps only)
`session.jvm`	`JVMSessionProxy`	Java-specific helpers (JVM apps only)

Reading State

Typed Accessor Methods

Use the typed getters to read individual values from the cached state. These methods never raise — they return the provided default (or None) when a key is missing:

# Generic — returns Any
value   = session.get("inbox.unread_count")

# Typed — returns the specified type, default on missing key
count   = session.get_int("inbox.unread_count", default=0)
alt     = session.get_float("flight.altitude_m", default=0.0)
is_read = session.get_bool("inbox.emails[0].is_read", default=False)
subject = session.get_str("inbox.emails[0].subject", default="")

Prefix Query

Retrieve all state keys that share a common prefix as a flat dictionary:

# Returns {"inbox.unread_count": 47, "inbox.total_count": 1284, ...}
inbox_data = session.get_keys_with_prefix("inbox.")

for key, value in inbox_data.items():
    print(f"  {key}: {value}")

Accessing the Raw State Object

For operations not covered by the typed helpers, work directly with the SubstrateState object:

state = session.state

# Direct field access via the objects mapping
sender = state.objects.get("inbox.emails[0].from")

# Check whether an action can safely run at current confidence
if state.can_execute("send_reply"):
    print("Safe to send")
else:
    print(f"Confidence {session.confidence:.2f} is too low")

# Look up action schema metadata
schema = state.action("archive")
if schema:
    print(f"Min confidence : {schema.min_confidence_required}")
    print(f"Reversibility  : {schema.reversibility}")
    print(f"Is destructive : {schema.is_destructive}")

Cached State vs. Fresh State

session.state holds the last snapshot received from the daemon. It does not automatically update. You have two ways to get newer data:

session.refresh()

Fetches the latest cached snapshot from the daemon (~1 ms). Use this after an action or when you need a more recent read without triggering a full probe.

session.refresh(fresh_probe=True)

Forces the daemon to run a new probe cycle before returning (50 ms–90 s). Use this only when you need a guaranteed up-to-date reading.

# Fast — fetches daemon's latest cached state
state = await session.refresh()

# Slow but accurate — triggers a fresh probe first
state = await session.refresh(fresh_probe=True)

Executing Actions

`execute()` Method

result = await session.execute(
    action,
    parameters={},
    verify=None,
    verify_expression=None,
    verify_timeout_ms=5000,
    auto_refresh=True,
)

action

str

required

The action name, as it appears in session.actions.

parameters

dict

default:"{}"

Key-value pairs matching the action’s parameter schema.

verify

bool | None

default:"None"

Enable post-action verification. Automatically set to True for IrreversibleWrite actions — you cannot set this to False for irreversible actions without raising NSPIrreversibleNotConfirmedError.

verify_expression

str | None

default:"None"

A JavaScript expression evaluated against the post-action state. If it returns falsy, the action is marked as failed.

verify_timeout_ms

int

default:"5000"

How long (in milliseconds) to poll for state settlement after the action runs. Range: 100–30000.

auto_refresh

bool

default:"True"

When True, refreshes session.state automatically after the action completes.

Usage Examples

# Read-type action — no verification needed
result = await session.execute(
    "get_thread",
    parameters={"thread_id": "msg_18821af3"},
)
print(result.result)   # {"subject": "...", "body": "...", ...}

# Reversible write — verify is optional
result = await session.execute(
    "archive",
    parameters={"email_id": "msg_18821af3"},
)
print(f"Archived: {result.success}  latency={result.latency_ms}ms")

# Irreversible write — verify=True is REQUIRED
result = await session.execute(
    "send_reply",
    parameters={
        "thread_id": "msg_18821af3",
        "body": "Thanks, I will review this shortly.",
    },
    verify=True,
    verify_expression="inbox.sent_count > 0",
    verify_timeout_ms=5000,
)
print(f"Sent: {result.success}  verified: {result.verification.satisfied}")

`ActionResponse` Fields

success

bool

True if the action ran and (when enabled) verification passed.

latency_ms

int

Total round-trip time from dispatch to response, in milliseconds.

result

Any

The raw return value of the action as reported by the runtime. null for actions that return nothing (UI interactions, void methods).

verification

VerificationResult | None

Present when verify=True. Contains satisfied: bool, actual_value, expression, evaluated_at, and latency_ms. See VerificationResult.

error_message

str | None

Human-readable failure description when success is False.

Watch Stream

Open a real-time WebSocket stream of state-change events:

# Basic stream — you manage refreshes manually
async with session.watch() as stream:
    async for event in stream:
        print(f"Changed: {event.changed_keys}")
        print(f"Seq    : {event.sequence}")
        print(f"At     : {event.captured_at}")
        await session.refresh()

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

# Tune queue size and reconnect behavior
async with session.watch(max_queue=64, auto_reconnect=True) as stream:
    async for event in stream:
        ...

See the Streaming page for complete documentation.

Wait Patterns

`wait_for(predicate, *, timeout, poll_interval, fresh_probe)`

Poll the state until a lambda condition is satisfied:

# Wait until the email is marked as read
state = await session.wait_for(
    lambda s: s.get_bool("inbox.emails[0].is_read"),
    timeout=30.0,
    poll_interval=1.0,
)
print("Email is now read")

# Wait for flight mode to become GUIDED, using fresh probes
state = await session.wait_for(
    lambda s: s.get_str("flight.mode") == "GUIDED",
    timeout=60.0,
    poll_interval=0.5,
    fresh_probe=True,
)

`wait_for_key(key, *, expected_value, timeout, poll_interval, fresh_probe)`

Wait for a specific key to reach an exact value:

state = await session.wait_for_key(
    "flight.mode",
    expected_value="GUIDED",
    timeout=30.0,
)
print(f"Mode is now: {state.get_str('flight.mode')}")

Waiting for Change After an Action

Capture a baseline value, execute the action, then wait for the value to change:

before = session.get_int("inbox.sent_count")

await session.execute(
    "send_reply",
    parameters={"thread_id": "msg_18821af3", "body": "On my way."},
    verify=True,
)

await session.wait_for(
    lambda s: s.get_int("inbox.sent_count") > before,
    timeout=10.0,
)
print("Reply confirmed sent")

Complete Agent Example

import asyncio
import nelieo_nsp as nsp

async def email_responder():
    """Read all unread emails and send a canned reply to each."""

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

        unread = session.get_int("inbox.unread_count")
        print(f"Processing {unread} unread emails...")

        for i in range(min(unread, 10)):   # process at most 10
            await session.refresh()

            if session.get_bool(f"inbox.emails[{i}].is_read"):
                continue

            thread_id = session.get_str(f"inbox.emails[{i}].thread_id")
            subject   = session.get_str(f"inbox.emails[{i}].subject")
            sender    = session.get_str(f"inbox.emails[{i}].from")
            print(f"  Replying to '{subject}' from {sender}")

            result = await session.execute(
                "send_reply",
                parameters={
                    "thread_id": thread_id,
                    "body": "Thank you for your email. I will get back to you shortly.",
                },
                verify=True,
                verify_expression="inbox.sent_count > 0",
            )

            if result.success:
                print(f"    ✅ Sent in {result.latency_ms}ms")
            else:
                print(f"    ❌ Failed: {result.error_message}")

asyncio.run(email_responder())

​Creating a Session

​Session Properties

​Reading State

​Typed Accessor Methods

​Prefix Query

​Accessing the Raw State Object

​Cached State vs. Fresh State

session.refresh()

session.refresh(fresh_probe=True)

​Executing Actions

​execute() Method

​Usage Examples

​ActionResponse Fields

​Watch Stream

​Wait Patterns

​wait_for(predicate, *, timeout, poll_interval, fresh_probe)

​wait_for_key(key, *, expected_value, timeout, poll_interval, fresh_probe)

​Waiting for Change After an Action

​Complete Agent Example