xr_ai_launcher#

xr-ai-launcher — process management for the xr-ai stack.

Intentionally stdlib-only so it can be added to any sample without pulling in the dependency chain of the processes it manages.

Typical usage:

from xr_ai_launcher import Parallel, Process, run_stack

_BASE = Path(__file__).resolve().parent

PROCESSES = [
    Process("hub",    "../../services/device-io-hub", "device_io_hub"),
    Parallel([
        Process("stt", "../../services/stt-server", "stt_server"),
        Process("tts", "../../services/pocket-tts",  "pocket_tts_server"),
    ]),
    Process("worker", "worker", "my_agent_worker"),
]

def run() -> None:
    run_stack(PROCESSES, _BASE)

Attributes#

`NATIVE_DEVICE_PROFILES`	CloudXR device profiles that use direct native transport instead of WSS.
`XR_RUNTIME_VAR`	OpenXR runtime selector written by the CloudXR runtime service.

Exceptions#

GPUInventoryError

Raised when visible GPUs cannot be inventoried or safely classified.

Classes#

`ModelDeployment`	Launcher-facing process ownership and credentials for model endpoints.
`Parallel`	A group of processes that start simultaneously.
`Process`	Declare one process in a launcher-managed stack.

Functions#

`is_native_profile`(→ bool)	True if profile names a native-transport CloudXR device profile.
`load_cloudxr_env`(→ None)	Parse an `export KEY=VALUE` env file and merge into `os.environ`.
`read_device_profile`(→ str)	Return NV_DEVICE_PROFILE from the environment, or from yaml_path when unset.
`read_config_scalar`(→ str)	Read one scalar from a top-level YAML worker configuration.
`ensure_credentials`(→ None)	Ensure each named token is available in os.environ.
`load_credentials`(→ None)	Inject saved credentials into os.environ.
`require_credentials`(→ None)	Exit non-zero unless each named token is available; NEVER prompts.
`warn_if_missing`(→ None)	Non-interactively surface missing optional tokens — NEVER prompts.
`detect_gpu_config`(→ str)	Select an existing config only when the complete topology is supported.
`load_deployment_profile`(→ ModelDeployment)	Load a JSON deployment profile directly (no worker YAML indirection).
`load_model_deployment`(→ ModelDeployment)	Load the JSON profile selected by a YAML worker configuration.
`ManagedProcess`(name, cmd[, cwd, env])	Run cmd as a subprocess and terminate it cleanly when the context exits.
`run_stack`(→ None)	Start processes in declaration order, waiting for each item to signal

Package Contents#

xr_ai_launcher.NATIVE_DEVICE_PROFILES#: CloudXR device profiles that use direct native transport instead of WSS.

xr_ai_launcher.XR_RUNTIME_VAR = 'XR_RUNTIME_JSON'#: OpenXR runtime selector written by the CloudXR runtime service.

xr_ai_launcher.is_native_profile(profile: str) → bool#: True if profile names a native-transport CloudXR device profile.

xr_ai_launcher.load_cloudxr_env(path: pathlib.Path) → None#: Parse an export KEY=VALUE env file and merge into os.environ.

xr_ai_launcher.read_device_profile(yaml_path) → str#: Return NV_DEVICE_PROFILE from the environment, or from yaml_path when unset.

xr_ai_launcher.read_config_scalar( path: pathlib.Path, key: str, default: str = '', ) → str#: Read one scalar from a top-level YAML worker configuration.

xr_ai_launcher.ensure_credentials(*names: str) → None#

Ensure each named token is available in os.environ.

Priority (highest first):

Already set in os.environ
Saved in ~/.config/xr-ai/credentials.json
Stored in ~/.cache/huggingface/token (HF_TOKEN only)
Prompted interactively and then saved to both locations

Any token entered interactively is saved so future runs are non-interactive. Pressing Enter without typing skips the token (left unset, not saved).

xr_ai_launcher.load_credentials() → None#

Inject saved credentials into os.environ.

Values already set in os.environ are left unchanged. For HF_TOKEN, also checks ~/.cache/huggingface/token (written by huggingface-cli login) as a fallback so existing HuggingFace logins are picked up automatically.

xr_ai_launcher.require_credentials(*names: str, allow_missing: bool = False) → None#

Exit non-zero unless each named token is available; NEVER prompts.

allow_missing (wired to the orchestrators’ --allow-anonymous flag) downgrades to warn_if_missing behavior.

xr_ai_launcher.warn_if_missing(*names: str) → None#

Non-interactively surface missing optional tokens — NEVER prompts.

Loads saved / huggingface-cli login / env tokens, then prints an actionable notice per still-missing token and continues, so a missing token is visible early without blocking startup on an interactive prompt.

exception xr_ai_launcher.GPUInventoryError#: Raised when visible GPUs cannot be inventoried or safely classified.

xr_ai_launcher.detect_gpu_config() → str#

Select an existing config only when the complete topology is supported.

This compatibility bridge remains until capability-based service planning replaces named config selection. It deliberately has no fallback.

class xr_ai_launcher.ModelDeployment#

Launcher-facing process ownership and credentials for model endpoints.

profile_path: pathlib.Path#: Resolved path of the selected model profile.

services: dict[str, Literal['own', 'reuse']]#: Launcher ownership mode keyed by managed service name.

required_credentials: tuple[str, Ellipsis]#: Environment-variable names required by configured model endpoints.

launch_mode(service: str) → Literal['own', 'reuse'] | None#: Return the configured ownership mode for service, if managed.

xr_ai_launcher.load_deployment_profile(profile_path: pathlib.Path) → ModelDeployment#: Load a JSON deployment profile directly (no worker YAML indirection).

xr_ai_launcher.load_model_deployment(worker_config: pathlib.Path) → ModelDeployment#: Load the JSON profile selected by a YAML worker configuration.

async xr_ai_launcher.ManagedProcess( name: str, cmd: list[str], cwd: pathlib.Path | None = None, env: dict[str, str] | None = None, )#

Run cmd as a subprocess and terminate it cleanly when the context exits.

Output routing: when the parent has called xr_ai_logging.setup_logging() (per-run env vars stamped), stdout and stderr are appended line-by-line to <log_dir>/<name>.log with HH:MM:SS.mmm timestamps. Lines emitted via a custom lovr.log callback (LOVR_LOG\t<level>\t<tag>\t<msg>) are parsed for authoritative severity; warn/error records are also mirrored to the parent’s stdout so they reach the terminal. Native bytes that don’t carry the marker fall back to a regex match on LOVR’s level vocabulary (case-insensitive error|warn|warning|fatal|panic) before being mirrored.

When the env vars are unset (standalone use), output is not lost: lines fall back to prefixed [name] … prints on the parent’s stdout.

Sends SIGTERM on exit; escalates to SIGKILL after _STOP_TIMEOUT seconds. env, if given, replaces the child’s environment entirely; otherwise the child inherits the parent’s.

class xr_ai_launcher.Parallel(processes: Sequence[Process])#

A group of processes that start simultaneously.

The launcher spawns every member at once and waits for all of them to signal readiness before advancing to the next item in the stack sequence. If any member exits before signaling ready the launcher shuts everything down, just as it would for a serial process.

Example:

Parallel([
    Process("stt", "../../services/stt-server", "stt_server"),
    Process("tts", "../../services/pocket-tts",  "pocket_tts_server"),
])

processes: tuple[Process, Ellipsis]#: Processes started together before the launcher advances.

class xr_ai_launcher.Process#

Declare one process in a launcher-managed stack.

name: str#: Label used in launcher output and readiness diagnostics.

project: str | pathlib.Path#: uv project path, relative to the sample root or absolute.

command: str#: Entry-point command to run inside the project’s environment.

config: str | pathlib.Path | None = None#

Optional YAML path, relative to the sample root or absolute.

The launcher passes the resolved path to the process as --config.

gpu: str | None = None#: Optional CUDA_VISIBLE_DEVICES value, such as "0" or "0,1".

launch_mode: str = 'own'#

Spawn and shutdown behavior.

own starts the process and stops it with the stack. persist starts it but leaves it running at shutdown. reuse assumes the process is already running and neither starts nor stops it.

port: int | None = None#

Optional service port metadata.

The launcher stores but does not inspect this value or stop persistent services by port.

quiet_native_output: bool = False#

Route unformatted native output to DEBUG instead of the terminal.

Use this for processes that interleave native C/C++ output with Python loguru records. All output remains available in the per-process log file.

xr_ai_launcher.run_stack( processes: Sequence[Process | Parallel], base: pathlib.Path, *, exit_after_ready: bool = False, ) → None#

Start processes in declaration order, waiting for each item to signal readiness before advancing to the next.

Each item is either a single Process (started and awaited alone) or a Parallel group (all members started at once; the launcher waits for every member before moving on).

Each process receives --ready-file <path>. When fully initialized it must Path(ready_file).touch() to signal the launcher. Progress is printed every five seconds so model-loading or CloudXR start-up remains visible.

After all processes are ready the launcher monitors them: if any exits, all others are terminated and the launcher exits. Pass exit_after_ready=True to return immediately once everything is ready instead — useful for launchers whose processes are all launch_mode="persist" and should outlive the orchestrator (e.g. model-servers).

base is the sample root — all relative paths in Process.project and Process.config are resolved against it:

_BASE = Path(__file__).resolve().parent

PROCESSES = [
    Process("hub",    "../../services/device-io-hub", "device_io_hub",
            config="yaml/device_io_hub.yaml"),
    Parallel([
        Process("stt", "../../services/stt-server", "stt_server"),
        Process("tts", "../../services/pocket-tts",  "pocket_tts_server"),
    ]),
    Process("worker", "worker", "my_worker",
            config="yaml/my_worker.yaml"),
]

def run() -> None:
    run_stack(PROCESSES, _BASE)