xr_ai_launcher#

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

Uses the standard library and the lightweight xr-ai-gpu utility so samples do not pull in the dependency chain of the processes they manage.

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.

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.
`has_ownership_marker`(→ bool)	Return whether pid was started with `<prefix>_MANAGED=1` and `<prefix>_PORT=<port>`.
`pid_on_port_checked`(→ tuple[int \| None, bool, bool])	Return the listening PID, inspection status, and listener presence.
`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 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.

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.

Raises:: GPUInventoryError: Inventory fails or no profile matches the topology.

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.

xr_ai_launcher.has_ownership_marker(pid: int, port: int, prefix: str) → bool#: Return whether pid was started with <prefix>_MANAGED=1 and <prefix>_PORT=<port>.

xr_ai_launcher.pid_on_port_checked(port: int) → tuple[int | None, bool, bool]#

Return the listening PID, inspection status, and listener presence.

Tries ss first (always present on modern Linux), falls back to lsof. A listener without a visible PID is still reported so callers fail closed instead of mistaking an uninspectable listener for an unused port.

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. A member that signals readiness remains monitored until the whole group is ready. An unexpected exit during that interval stops owned processes and persistent processes that have not signaled readiness. A successful persist process may exit after readiness only when run_stack(..., exit_after_ready=True) permits bootstrap wrappers to outlive the launcher.

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 once it has signaled readiness, including when another process fails during startup. 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, health_port: int | None = None, ) → 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.

Pass health_port as an integer TCP port (1–65535) to expose GET http://127.0.0.1:<port>/healthz for this foreground stack. It returns 503 during startup and cleanup, or when any spawned process has exited; 200 means startup completed and every spawned process is still running. The endpoint is disabled with health_port=None (the default) and for exit_after_ready=True. Invalid ports and bind failures abort before any process is launched. This is process health, not an inference probe, and does not check services declared with launch_mode="reuse". The JSON body reports status, spawned processes (running or exited), and unchecked_reuse process names. HEAD is also supported.

After all processes are ready the launcher monitors them: if any exits, the remaining foreground processes are terminated. The launcher preserves a positive child status, maps a signal to 128 + signal, and maps an unexpected zero exit to 1. Startup failure or cancellation stops owned processes and persistent processes that have not signaled readiness. Already-ready persistent processes and reused services are left running. SIGINT or SIGTERM during startup exits with 128 + signal. After readiness, signals follow normal shutdown. Spawn errors, readiness failures, and unexpected monitored exits print a concise summary with the constructed command and an existing configured run-log directory when available. 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)