IPC Client Integration Guide¶

This guide explains how to connect an external process to a running Lightfall instance over NATS. No knowledge of Lightfall internals is required.

Prerequisites¶

• A running NATS server reachable from your client (ask your beamline controls group for the URL and port).

• The server’s TLS CA certificate, or a certificate signed by a trusted CA.

• Python 3.10+ with nats-py installed:

  pip install nats-py
  

• A topic prefix matching the one configured in Lightfall (default: als.7011).

• The connection-capability helper. On bcgnats every connection must present a capability token (see the NATS Use Model). Import it from lightfall_utils.ipc.capability if your client can depend on lightfall-utils; otherwise copy lightfall_utils/ipc/capability.py byte-for-byte into your project as nats_capability.py (stdlib only) and add a pinning test, as described in the use model.

Connecting¶

Lightfall’s NATS server requires TLS. Pass an ssl.SSLContext to nats.connect, together with a fresh connection capability:

import asyncio
import ssl
import nats
from lightfall_utils.ipc.capability import new_capability  # or your vendored nats_capability

NATS_URL = "nats://bcgnats.als.private.lbl.gov:4222"
TOPIC_PREFIX = "als.7011"

async def main():
    tls_ctx = ssl.create_default_context()
    # If using a private CA:
    # tls_ctx.load_verify_locations("/path/to/ca.crt")

    cap = new_capability()
    nc = await nats.connect(NATS_URL, tls=tls_ctx, **cap.connect_kwargs())
    print("Connected; private root", cap.priv_root)
    # ... use nc ...
    await nc.drain()

asyncio.run(main())

connect_kwargs() is {"token": <secret>, "inbox_prefix": "_INBOX.<p>"}. The bcgnats auth callout rejects a connection without a valid token, and the inbox prefix is what lets your client receive replies: on bcgnats a connection may subscribe only to its own _INBOX.<p>.> and _PRIV.<p>.>, plus the public subject roots. Never log or publish the secret. A local nats-server without the callout ignores both options.

Authentication¶

Before sending commands, you must authenticate with Lightfall. This is a request/reply handshake on the instance’s private auth.request subject: first ask {prefix}.meta.actions for the instance’s priv_root (_PRIV.<p>), then send the request to {priv_root}.auth.request. Only that Lightfall instance can receive the request or answer it. Lightfall will show a trust dialog the first time; subsequent requests from the same app_name are approved or denied automatically.

private_auth_request() does both steps. It sends the auth.request leg with private_request() (see “The capability channel” below), so the reply comes back on an unguessable subject that no other connection can forge:

from lightfall_utils.ipc.capability import PrivateRootUnavailable, private_auth_request

async def authenticate(nc, app_name: str, app_version: str = "") -> dict:
    # Timeout >60 s to allow the user to respond to the dialog
    try:
        response = await private_auth_request(
            nc, TOPIC_PREFIX, {"app_name": app_name, "app_version": app_version}, timeout=70
        )
    except PrivateRootUnavailable as exc:
        raise RuntimeError(f"{TOPIC_PREFIX} predates the private handshake: {exc}") from exc

    if response.get("status") == "approved":
        print("Authenticated. Tiled token:", response.get("tiled_token"))
        print("Tiled URL:", response.get("tiled_url"))
        return response
    else:
        reason = response.get("reason", "denied")
        raise PermissionError(f"Lightfall denied the connection request: {reason}")

A successful response has this shape:

{
  "status": "approved",
  "session_token": "<url-safe-random-token>",
  "session_subject": "_PRIV.<p>.session.<url-safe-random-token>",
  "tiled_token": "<api_key_secret>",
  "tiled_url": "https://tiled.als.lbl.gov",
  "session_id": "<keycloak-sub-or-null>",
  "contract_version": 1
}

On bcgnats a responder may send one reply per received request, within 10 minutes, so the trust dialog and slow commands (e.g. commands.device.put with wait=true) are fine. Longer-running work should reply immediately with an accepted id and report completion via events.

session_token (remote-control contract v1) identifies your capability channel, and session_subject is the channel’s full subject root under the instance’s private root; every commands.* call goes to {session_subject}.<suffix> — see the “The capability channel” section below. PrivateRootUnavailable means the instance didn’t advertise a well-formed priv_root: it predates the private handshake (or the answer wasn’t Lightfall’s). contract_version is 1; send it back on every request (see “Structured Errors” below).

Trust is per login session, not per process. The trust decision and the capability channel both live only as long as the current Lightfall login session. When the logged-in user logs out of Lightfall, every capability channel is torn down and every app’s trust decision is forgotten, even if the Lightfall process keeps running. Your client will not receive a notification — the channel simply stops answering. Detect this by timing out on a request and re-running auth.request from scratch (a fresh handshake mints a new session_token; the old one is dead).

Auth v2 (since 2026-05): The tiled_token field name is preserved for wire-format compatibility, but the value is now a Tiled API key (not a Keycloak JWT bearer). Consume it via the Tiled client’s api_key= parameter:

from tiled.client import from_uri
client = from_uri(tiled_url, api_key=tiled_token)

The key has a TTL (~1 week by default) configured server-side; clients should handle 401 responses by re-requesting via the IPC auth.request flow.

A denial looks like:

{"status": "denied", "reason": "timeout"}

Denial reason values include:

Token Refresh¶

If a Tiled request comes back 401, re-run the authentication handshake. Under auth-v2, tiled_token is a server-issued Tiled API key with a TTL (typically 1 week) — it may outlive the IPC requester’s local session, and conversely a new Lightfall session (restart, or a logout/login cycle) will invalidate old keys. On a 401 from Tiled, re-run auth.request to obtain a fresh key (this also mints a fresh session_token).

Discovering Available Actions and Events¶

Before hard-coding subject names, you can ask Lightfall what it supports. These discovery subjects are public, and each reply also carries the instance’s priv_root. Discovery answers are not authenticated: on a shared broker another process could answer them (see the NATS Use Model).

import json

async def discover(nc):
    # List all request/reply actions
    msg = await nc.request(f"{TOPIC_PREFIX}.meta.actions", b"{}", timeout=5)
    actions = json.loads(msg.data)["actions"]
    for a in actions:
        print(f"  action: {a['subject']} — {a['description']}")

    # List all outbound events
    msg = await nc.request(f"{TOPIC_PREFIX}.meta.events", b"{}", timeout=5)
    events = json.loads(msg.data)["events"]
    for e in events:
        print(f"  event:  {e['subject']} — {e['description']}")

The capability channel¶

NATS core messages carry no sender identity, so Lightfall cannot tell “who” sent a message just from the subject or payload. Instead, every commands.* action lives behind a capability channel: a private subject built from the session_token you got back from auth.request.

• Subject shape: {session_subject}.<command-suffix>, i.e. _PRIV.<p>.session.{session_token}.<command-suffix> — e.g. _PRIV.<p>.session.<token>.commands.plan.list. Use the session_subject from the auth.request reply rather than building it yourself. There is no bare commands.plan.list you can call directly; sending a request to the un-channeled subject gets you back a structured denied error instead of a real reply.

• Every request should include contract_version: 1 in its JSON body. If it’s missing, the server currently assumes 1 for backward tolerance, but you should send it explicitly — a mismatched version (once the contract bumps past 1) gets a version_mismatch error instead of being silently misinterpreted.

• The broadcast events are the exception: runs.new, runs.complete, and state.engine are not behind the capability channel — they’re plain {prefix}.runs.new etc., meant to be subscribed by multiple listeners at once and carry no secrets.

• The channel dies on logout (see the Authentication section above) — treat request timeouts as a signal to re-authenticate rather than retrying forever.

• Send every channel request with private_request(), never nc.request(). Inbox isolation stops other connections from reading your replies, not from writing into your inbox: any connection can reflect a forged message into it through allow_responses if it can guess the reply subject, and the reply subjects nc.request() generates are predictable to anyone who has seen one of your public requests. private_request() waits on a fresh _INBOX.<p>.<random token> that only Lightfall sees, and returns the decoded JSON reply. Replies to public requests (such as meta.actions above) can still be forged.

from lightfall_utils.ipc.capability import private_request

async def call(nc, session_subject: str, suffix: str, payload: dict | None = None, timeout: float = 5.0) -> dict:
    subject = f"{session_subject}.{suffix}"
    body = dict(payload or {})
    body.setdefault("contract_version", 1)
    return await private_request(nc, subject, body, timeout=timeout)

Sending Commands¶

All request payloads are JSON objects sent on the capability channel (see above). Replies are also JSON and always carry contract_version: 1. Send {} (plus contract_version) when no other payload is required. This is the full v1 verb table.

async def run_plan(nc, session_subject: str, plan_name: str, params: dict) -> dict:
    reply = await call(nc, session_subject, "commands.plan.run", {
        "plan_name": plan_name,
        "params": params,
    }, timeout=30)
    return reply

# Example
result = await run_plan(nc, session_subject, "count", {"detectors": ["det1"], "num": 5})
# {"status": "submitted", "plan_name": "count", "item_id": "...", "run_uid": "...", "contract_version": 1}
# run_uid may be null if the start document hadn't arrived yet — listen for runs.new.

async def abort_run(nc, session_subject: str, reason: str = "", item_id: str | None = None, run_uid: str | None = None) -> dict:
    payload = {"reason": reason}
    if item_id is not None:
        payload["item_id"] = item_id
    if run_uid is not None:
        payload["run_uid"] = run_uid
    return await call(nc, session_subject, "commands.plan.abort", payload, timeout=10)

# {"status": "abort_requested", "contract_version": 1}
# Selector that doesn't match anything current/queued -> {"status": "not_aborted", "message": "...", "contract_version": 1}
# Both item_id and run_uid given -> {"status": "error", "code": "bad_request", "message": "...", "contract_version": 1}

async def add_logbook_entry(nc, session_subject: str, title: str, content: str = "", tags: list[str] | None = None) -> dict:
    return await call(nc, session_subject, "commands.logbook.add", {
        "title": title,
        "content": content,
        "tags": tags or [],
    }, timeout=10)

# {"status": "created", "entry_id": "...", "contract_version": 1}

async def send_agent_message(nc, session_subject: str, message: str) -> dict:
    return await call(nc, session_subject, "commands.agent.message", {"message": message}, timeout=10)

# {"status": "sent", "contract_version": 1}

Subscribing to Events¶

Lightfall publishes run lifecycle and engine state changes as NATS core messages on plain, un-channeled subjects (they carry no secrets and are meant for multiple simultaneous listeners). Subscribe before starting a plan so you don’t miss early events.

Breaking rename: the runs.new/runs.complete field previously named run_id is now run_uid. runs.new also gained item_id (was implicitly procedure_id in the old commands.plan.run reply; the two concepts are now the same field name across the whole contract).

Run Start and Completion¶

import json

async def watch_runs(nc):
    async def on_run_new(msg):
        data = json.loads(msg.data)
        print(f"Run started: {data['run_uid']} (item {data['item_id']}, plan {data['plan_name']})")

    async def on_run_complete(msg):
        data = json.loads(msg.data)
        print(f"Run finished: {data['run_uid']} — {data['exit_status']}")

    await nc.subscribe(f"{TOPIC_PREFIX}.runs.new", cb=on_run_new)
    await nc.subscribe(f"{TOPIC_PREFIX}.runs.complete", cb=on_run_complete)

runs.new payload: {"item_id": str, "run_uid": str, "plan_name": str}. runs.complete payload: {"run_uid": str, "exit_status": "success" | "abort" | "error"}.

Engine State Changes¶

import json

async def watch_engine_state(nc):
    async def on_state(msg):
        data = json.loads(msg.data)
        print(f"Engine state: {data['state']}")

    await nc.subscribe(f"{TOPIC_PREFIX}.state.engine", cb=on_state)

Structured Errors¶

Any reply with "status": "error" carries a code and a human-readable message, alongside the usual contract_version:

{"status": "error", "code": "busy", "message": "Engine is busy and behavior is 'reject'", "contract_version": 1}

code is one of:

This supersedes the older, unstructured {"error": true, "message": ...} shape — check reply.get("status") == "error" rather than reply.get("error").

Closed-loop¶

External services participate in closed experimental loops by combining the NATS bus (for notifications and suggestions) with Tiled (for the measured data). The canonical loop, as implemented by Lightfall’s built-in adaptive_experiment plan with Tsuchinoko:

1. At run start, the plan publishes a bind message carrying the run UID, the Tiled URL, and a Tiled API key, so the external engine can read the run’s data directly from the catalog.

2. After each measurement (or batch), the plan publishes a notification on {prefix}.adaptive.measured.

3. The autonomous engine reads the new points from Tiled, updates its surrogate model, and publishes the next measurement targets on its own subject (tsuchinoko.targets).

4. The plan, polling that subject between plan messages via NATSPlanBridge (lightfall.acquire.nats_bridge), moves the motors to each target and measures — and the loop closes.

No participant in this loop requires modifications to Lightfall’s core: notifications and suggestions travel over the same bus described in this guide, and the data travels through the same Tiled catalog every other client uses. The script below is a minimal implementation of a simpler participant that uses only the generic action and event subjects.

Complete Example: Reference Client¶

tests/integration/remote_client.py in the Lightfall repo is the canonical reference client for this contract — raw nats-py plus the stdlib-only private_auth_request and private_request helpers, and the starting point for building a language-specific client of your own. The trimmed flow below (handshake → capability call → event subscribe) mirrors it; the Tsuchinoko-style single-file example that used to live here has been replaced by this and by the reference client itself.

#!/usr/bin/env python3
"""Example: connect to Lightfall, submit a plan, wait for it to finish."""

import asyncio
import json
import ssl
import nats
from lightfall_utils.ipc.capability import new_capability, private_auth_request, private_request

NATS_URL = "nats://bcgnats.als.private.lbl.gov:4222"
TOPIC_PREFIX = "als.7011"
APP_NAME = "my-client"
APP_VERSION = "1.0.0"
CONTRACT_VERSION = 1


async def main():
    tls_ctx = ssl.create_default_context()
    nc = await nats.connect(NATS_URL, tls=tls_ctx, **new_capability().connect_kwargs())
    print("Connected to NATS")

    # 1. Handshake — discover priv_root via meta.actions, then {priv_root}.auth.request
    auth = await private_auth_request(
        nc, TOPIC_PREFIX, {"app_name": APP_NAME, "app_version": APP_VERSION}, timeout=90
    )
    if auth.get("status") != "approved":
        raise PermissionError(f"Authentication denied: {auth.get('reason', 'unknown')}")
    session_subject = auth["session_subject"]
    print(f"Authenticated. Tiled URL: {auth.get('tiled_url')}")

    async def call(suffix: str, payload: dict | None = None, timeout: float = 5.0) -> dict:
        subject = f"{session_subject}.{suffix}"
        body = dict(payload or {})
        body.setdefault("contract_version", CONTRACT_VERSION)
        reply = await private_request(nc, subject, body, timeout=timeout)  # never nc.request
        if reply.get("status") == "error":
            raise RuntimeError(f"{reply['code']}: {reply['message']}")
        return reply

    # 2. Subscribe to run events (public subjects, not behind the capability channel)
    run_done = asyncio.Event()
    last_run_uid = None

    async def on_run_new(msg):
        nonlocal last_run_uid
        data = json.loads(msg.data)
        last_run_uid = data["run_uid"]
        print(f"Run started: {last_run_uid} ({data['plan_name']})")

    async def on_run_complete(msg):
        data = json.loads(msg.data)
        print(f"Run complete: {data['run_uid']} — {data['exit_status']}")
        if data["run_uid"] == last_run_uid:
            run_done.set()

    await nc.subscribe(f"{TOPIC_PREFIX}.runs.new", cb=on_run_new)
    await nc.subscribe(f"{TOPIC_PREFIX}.runs.complete", cb=on_run_complete)

    # 3. Capability call — submit a plan
    plan_reply = await call("commands.plan.run", {
        "plan_name": "count",
        "params": {"detectors": ["det1"], "num": 3},
    }, timeout=30)
    print(f"Plan submitted (item_id: {plan_reply['item_id']}, run_uid: {plan_reply.get('run_uid')})")

    # Wait for the run to complete (with a generous timeout)
    await asyncio.wait_for(run_done.wait(), timeout=300)
    print("Done")

    await nc.drain()


asyncio.run(main())

Message Format Reference¶

All messages use JSON encoding (UTF-8). The following fields appear in replies:

procedure_id and run_id from the pre-v1 contract are gone — see the breaking-rename callout under “Subscribing to Events”; procedure_id is now item_id throughout.

Topic Hierarchy Reference¶

All subjects below are prefixed with the configured topic_prefix (default: als.7011). The full NATS subject is {prefix}.{suffix}, except for auth.request, which lives on the instance’s private root ({priv_root}.auth.request), and commands.* actions, which go to the per-session capability channel {session_subject}.{suffix} = {priv_root}.session.{session_token}.{suffix} (see “The capability channel”).

Bare (non-channeled) commands.* subjects still exist as NATS subscriptions internally, but every request sent to one gets a structured denied reply rather than executing the action — treat them as unreachable from a client’s perspective.