diff --git a/docs/explanation/cli.md b/docs/explanation/cli.md
index 822de564..6de7f34c 100644
--- a/docs/explanation/cli.md
+++ b/docs/explanation/cli.md
@@ -30,9 +30,11 @@ source <(pai completion zsh)
## Runtime Commands
-| Command | Purpose |
-| ---------------- | --------------------------------------------------------------- |
-| `physicalai run` | Runs a trained policy (or any action source) on robot hardware. |
+| Command | Purpose |
+| --------------------------- | --------------------------------------------------------------- |
+| `physicalai run` | Runs a trained policy (or any action source) on robot hardware. |
+| `physicalai robot serve` | Serves one shared robot from a foreground owner process. |
+| `physicalai robot discover` | Lists reachable shared-robot owners. |
## Training Commands
diff --git a/docs/how-to/runtime/share-a-robot.md b/docs/how-to/runtime/share-a-robot.md
index 6be2c857..fa731fb8 100644
--- a/docs/how-to/runtime/share-a-robot.md
+++ b/docs/how-to/runtime/share-a-robot.md
@@ -45,6 +45,32 @@ or the path itself, e.g. `robot_class="physicalai.robot.so101.SO101"` — any
importable class works, including third-party plugin robots, with no
registry to update.
+## Serve a robot in the foreground
+
+Use the operator command when a shell, systemd, Docker, or Kubernetes should own the
+robot lifecycle:
+
+```bash
+physicalai robot serve --config examples/so101/serve.yaml
+```
+
+The command constructs and connects the driver in its own foreground process. Normal
+output reports readiness, state-subscriber presence changes, a health summary every
+30 seconds, and clean shutdown. Add `--verbose` for startup and cleanup details. The
+command does not daemonize; use your service manager for background supervision.
+
+List reachable owners without importing their advertised driver class:
+
+```bash
+physicalai robot discover
+physicalai robot discover --json
+physicalai robot discover --allow_remote
+```
+
+Discovery is local-only unless `--allow_remote` is explicit. Human output is a sorted
+ASCII table. JSON mode writes one sorted array to stdout, including `[]` when no robot
+answers.
+
Notes:
- `get_observation()` returns the newest owner-published state; if no new
@@ -168,6 +194,8 @@ wire contract.
- When the last subscriber disconnects (cleanly or by crashing), the owner
waits `idle_timeout` seconds, then calls the driver's `disconnect()` —
honoring the safe-state contract (hold/home) — and exits.
+- An explicit `physicalai robot serve` owner has no idle timeout. It remains in the
+ foreground until interrupted or until the owner loop fails.
- A subscriber's `disconnect()` never stops the robot's motors; the owner
owns safe-state.
- Subscribers reject an owner advertising an unsupported transport
diff --git a/docs/reference/cli.md b/docs/reference/cli.md
index f165f13b..65af4c6c 100644
--- a/docs/reference/cli.md
+++ b/docs/reference/cli.md
@@ -28,6 +28,35 @@ with runtime:
runtime.run(duration_s=60)
```
+## `physicalai robot serve`
+
+```bash
+physicalai robot serve --config examples/so101/serve.yaml
+```
+
+The flat YAML fields are `name`, `robot_class`, optional `robot_kwargs`, optional
+`allow_remote` (default `false`), and optional positive `rate_hz` (default 100 Hz).
+Direct CLI arguments use the same names, including nested constructor values such as
+`--robot_kwargs.port /dev/ttyACM0`. Serving stays in the foreground until SIGINT or
+SIGTERM and returns nonzero for startup, loop, repeated-read, or disconnect failures.
+
+Use `--verbose` to include driver construction, lock acquisition, initial observation,
+endpoint declaration, and cleanup details.
+
+`--allow_remote` exposes an unauthenticated physical action endpoint. Use it only on
+an isolated robot-cell VLAN/firewall or with Zenoh ACL/TLS.
+
+## `physicalai robot discover`
+
+```bash
+physicalai robot discover [--allow_remote] [--timeout 2] [--json]
+```
+
+Results are sorted by robot name and host. Human output is an ASCII table containing
+name, class, host, and joint count, followed by the result count and elapsed discovery
+time. JSON mode writes exactly one array to stdout; empty discovery is successful and
+writes `[]`.
+
## Shell Completion
Shell completion scripts can be printed directly from the CLI and sourced in
diff --git a/examples/so101/serve.yaml b/examples/so101/serve.yaml
new file mode 100644
index 00000000..31100d97
--- /dev/null
+++ b/examples/so101/serve.yaml
@@ -0,0 +1,14 @@
+# Example config for `physicalai robot serve --config examples/so101/serve.yaml`.
+#
+# Replace `port` and `calibration` with the values for your machine before
+# running. See docs/how-to/runtime/share-a-robot.md for the full host A /
+# host B walkthrough (serve on the host with the physical robot attached,
+# attach from anywhere else with a `SharedRobot` runtime config).
+name: follower-arm
+robot_class: physicalai.robot.SO101
+robot_kwargs:
+ port: /dev/ttyACM0
+ calibration: /path/to/so101-calibration.json
+ role: follower
+allow_remote: false
+rate_hz: 100.0
diff --git a/src/physicalai/cli/main.py b/src/physicalai/cli/main.py
index 0359efcd..576bc7a6 100644
--- a/src/physicalai/cli/main.py
+++ b/src/physicalai/cli/main.py
@@ -29,6 +29,7 @@
from jsonargparse import ArgumentParser
+from physicalai.cli import robot as robot_cmd
from physicalai.cli import run as run_cmd
from physicalai.cli._discovery import discover_subcommands # noqa: PLC2701
from physicalai.cli._spec import SubcommandSpec # noqa: PLC2701
@@ -45,10 +46,12 @@
# ``--help`` listing never has to build (or import) a parser.
_BUILTINS: dict[str, Callable[[], SubcommandSpec]] = {
"run": run_cmd.register,
+ "robot": robot_cmd.register,
}
_BUILTIN_HELP: dict[str, str] = {
"completion": "Print a shell completion script.",
"run": run_cmd.HELP,
+ "robot": robot_cmd.HELP,
}
_COMPLETION_SHELLS = frozenset({"bash", "zsh", "fish"})
_HELP_FLAGS = frozenset({"-h", "--help"})
@@ -97,8 +100,8 @@ def _load_subcommand_module(name: str, entry_points: dict[str, EntryPoint]) -> M
Returns:
Imported subcommand module when available, otherwise ``None``.
"""
- if name == "run":
- return run_cmd
+ if name in _BUILTINS:
+ return sys.modules.get(_BUILTINS[name].__module__)
try:
register = entry_points[name].load()
@@ -128,6 +131,11 @@ def _is_help_request(argv: Sequence[str]) -> bool:
return any(token in _HELP_FLAGS for token in argv)
+def _is_direct_help_request(argv: Sequence[str]) -> bool:
+ """Return whether arguments request help for the selected command itself."""
+ return bool(argv) and all(token in _HELP_FLAGS for token in argv)
+
+
def _ep_help(ep: EntryPoint) -> str:
"""Per-subcommand help for an entry point, falling back to distribution metadata.
@@ -284,7 +292,7 @@ def main(argv: Sequence[str] | None = None) -> int:
if selected_name == "completion":
return _print_completion(sub_argv, entry_points, prog)
- if _is_help_request(sub_argv) and _print_fast_help(selected_name, entry_points, prog):
+ if _is_direct_help_request(sub_argv) and _print_fast_help(selected_name, entry_points, prog):
return 0
spec = _load_spec(selected_name, entry_points)
diff --git a/src/physicalai/cli/robot.py b/src/physicalai/cli/robot.py
new file mode 100644
index 00000000..b07ba970
--- /dev/null
+++ b/src/physicalai/cli/robot.py
@@ -0,0 +1,290 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""Serve and discover shared robots over Zenoh."""
+
+from __future__ import annotations
+
+import json
+import math
+import signal
+import sys
+import threading
+import time
+from typing import TYPE_CHECKING
+
+from jsonargparse import ActionConfigFile, ArgumentParser
+from loguru import logger
+
+from physicalai.cli._spec import SubcommandSpec # noqa: PLC2701
+from physicalai.robot.errors import RobotTransportError
+from physicalai.robot.transport import (
+ DEFAULT_RATE_HZ,
+ OwnerEvent,
+ RobotOwnerConfig,
+ discover_robots,
+ run_owner,
+)
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from jsonargparse import Namespace
+
+HELP = "Serve or discover shared robots over Zenoh."
+_SERVE_HELP = "Serve one robot in the foreground as a persistent shared-robot owner."
+_DISCOVER_HELP = "Enumerate reachable shared robots."
+_ALLOW_REMOTE_WARNING = (
+ "WARNING: The action endpoint is unauthenticated and reachable from all "
+ "network interfaces on the derived port. Use only on an isolated robot-cell "
+ "network or with Zenoh ACL/TLS.\n"
+)
+_HELP_TEMPLATE = """usage: {prog} {{serve,discover}} ...
+
+{description}
+
+subcommands:
+ serve {serve_help}
+ discover {discover_help}
+
+Run '{prog} serve --help' or '{prog} discover --help' for subcommand options.
+"""
+
+
+def _build_serve_parser() -> ArgumentParser:
+ parser = ArgumentParser(description=_SERVE_HELP)
+ parser.add_argument("--config", action=ActionConfigFile, help="YAML/JSON config file.")
+ parser.add_argument("--name", type=str, required=True, help="Robot logical name.")
+ parser.add_argument("--robot_class", type=str, required=True, help="Trusted dotted path to the driver class.")
+ parser.add_argument(
+ "--robot_kwargs",
+ type=dict,
+ default=None,
+ help="JSON-serializable driver constructor arguments.",
+ )
+ parser.add_argument(
+ "--allow_remote",
+ action="store_true",
+ default=False,
+ help=(
+ "Expose the unauthenticated action endpoint beyond localhost. "
+ "Use only on an isolated robot-cell network or with Zenoh ACL/TLS."
+ ),
+ )
+ parser.add_argument("--rate_hz", type=float, default=DEFAULT_RATE_HZ, help="Owner loop rate in Hz.")
+ parser.add_argument("--verbose", action="store_true", default=False, help="Show startup and cleanup details.")
+ return parser
+
+
+def _build_discover_parser() -> ArgumentParser:
+ parser = ArgumentParser(description=_DISCOVER_HELP)
+ parser.add_argument(
+ "--allow_remote",
+ action="store_true",
+ default=False,
+ help="Discover owners beyond localhost.",
+ )
+ parser.add_argument("--timeout", type=float, default=2.0, help="Discovery timeout in seconds.")
+ parser.add_argument("--json", action="store_true", default=False, help="Write one JSON array to stdout.")
+ return parser
+
+
+def build_parser() -> ArgumentParser:
+ """Build the nested robot command parser.
+
+ Returns:
+ Parser for ``physicalai robot``.
+ """
+ parser = ArgumentParser(prog="physicalai robot", description=HELP)
+ subcommands = parser.add_subcommands(required=True)
+ subcommands.add_subcommand("serve", _build_serve_parser(), help=_SERVE_HELP)
+ subcommands.add_subcommand("discover", _build_discover_parser(), help=_DISCOVER_HELP)
+ return parser
+
+
+def print_help(prog: str) -> None:
+ """Print group help without building nested parsers."""
+ print( # noqa: T201
+ _HELP_TEMPLATE.format(prog=prog, description=HELP, serve_help=_SERVE_HELP, discover_help=_DISCOVER_HELP),
+ )
+
+
+def _configure_serve_logging(*, verbose: bool) -> None:
+ """Configure concise process-wide Loguru output for foreground serving."""
+ logger.remove()
+ logger.add(
+ sys.stderr,
+ level="TRACE" if verbose else "INFO",
+ format="{time:HH:mm:ss} {level: <8} {message}",
+ colorize=sys.stderr.isatty(),
+ )
+
+
+def _prepare_serve_logging(*, allow_remote: bool, verbose: bool) -> None:
+ """Emit the remote-exposure banner (if needed), then configure Loguru."""
+ if allow_remote:
+ sys.stderr.write(_ALLOW_REMOTE_WARNING)
+ _configure_serve_logging(verbose=verbose)
+
+
+def _log_serve_start(config: RobotOwnerConfig) -> None:
+ """Log the audited serve start line, including local vs remote mode."""
+ mode_tag = "remote" if config.allow_remote else "local-only"
+ logger.info(f"Starting robot {config.name!r} using {config.robot_class} [{mode_tag}]")
+
+
+def _format_duration(seconds: float) -> str:
+ """Format elapsed seconds as ``HH:MM:SS``.
+
+ Returns:
+ The formatted duration.
+ """
+ total_seconds = max(0, int(seconds))
+ hours, remainder = divmod(total_seconds, 3600)
+ minutes, seconds = divmod(remainder, 60)
+ return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
+
+
+def _format_table(headers: tuple[str, ...], rows: Sequence[tuple[str, ...]], *, right_align: set[int]) -> str:
+ """Render a compact dependency-free ASCII table.
+
+ Returns:
+ The rendered table.
+ """
+ widths = [max(len(header), *(len(row[index]) for row in rows)) for index, header in enumerate(headers)]
+
+ def _row(values: tuple[str, ...]) -> str:
+ cells = [
+ value.rjust(widths[index]) if index in right_align else value.ljust(widths[index])
+ for index, value in enumerate(values)
+ ]
+ return " ".join(cells).rstrip()
+
+ separator = " ".join("-" * width for width in widths)
+ return "\n".join([_row(headers), separator, *(_row(row) for row in rows)])
+
+
+def serve(cfg: Namespace) -> int:
+ """Serve one robot in the current foreground process.
+
+ Returns:
+ The owner runtime exit code, or 1 for an expected startup failure.
+ """
+ _prepare_serve_logging(allow_remote=cfg.allow_remote, verbose=cfg.verbose)
+ try:
+ config = RobotOwnerConfig(
+ name=cfg.name,
+ robot_class=cfg.robot_class,
+ robot_kwargs=dict(cfg.robot_kwargs or {}),
+ allow_remote=cfg.allow_remote,
+ rate_hz=cfg.rate_hz,
+ idle_timeout=None,
+ )
+ except (TypeError, ValueError) as exc:
+ logger.error(f"Invalid robot configuration: {exc}")
+ return 1
+
+ shutdown = threading.Event()
+ ready_at: float | None = None
+ subscribers_present = False
+ received_signum: int | None = None
+
+ def _handle_signal(signum: int, _frame: object) -> None:
+ nonlocal received_signum
+ received_signum = signum
+ shutdown.set()
+
+ def _ready() -> None:
+ nonlocal ready_at
+ ready_at = time.monotonic()
+ logger.success(f"Serving robot {config.name!r} · {config.rate_hz:g} Hz")
+
+ def _on_event(event: OwnerEvent) -> None:
+ nonlocal subscribers_present
+ if event is OwnerEvent.SUBSCRIBERS_PRESENT:
+ subscribers_present = True
+ logger.info("Subscriber(s) connected")
+ elif event is OwnerEvent.NO_SUBSCRIBERS:
+ subscribers_present = False
+ logger.info("No subscribers remain")
+ elif event is OwnerEvent.HEARTBEAT:
+ uptime_s = 0.0 if ready_at is None else time.monotonic() - ready_at
+ status = "subscriber(s) connected" if subscribers_present else "no subscribers"
+ logger.info(f"Healthy · uptime {_format_duration(uptime_s)} · {status}")
+
+ _log_serve_start(config)
+ previous_sigterm = signal.signal(signal.SIGTERM, _handle_signal)
+ previous_sigint = signal.signal(signal.SIGINT, _handle_signal)
+ try:
+ try:
+ result = run_owner(config, shutdown, ready=_ready, on_event=_on_event)
+ except RobotTransportError as exc:
+ phase = f" during {exc.phase}" if exc.phase else ""
+ logger.error(f"Failed to start robot owner{phase}: {exc}")
+ return 1
+ else:
+ if received_signum is not None:
+ logger.info(f"Shutdown requested by {signal.Signals(received_signum).name}")
+ if result.exit_code == 0:
+ logger.success(f"Robot {config.name!r} disconnected safely")
+ else:
+ logger.error(f"Robot {config.name!r} stopped with reason {result.reason.value}")
+ return result.exit_code
+ finally:
+ signal.signal(signal.SIGTERM, previous_sigterm)
+ signal.signal(signal.SIGINT, previous_sigint)
+
+
+def discover(cfg: Namespace) -> int:
+ """Discover robots without importing advertised driver classes.
+
+ Returns:
+ Zero on successful discovery, including an empty result; otherwise 1.
+ """
+ if isinstance(cfg.timeout, bool) or not math.isfinite(cfg.timeout) or cfg.timeout <= 0:
+ sys.stderr.write(f"timeout must be finite and greater than zero, got {cfg.timeout!r}\n")
+ return 1
+
+ started_at = time.monotonic()
+ robots = sorted(
+ discover_robots(timeout=cfg.timeout, allow_remote=cfg.allow_remote),
+ key=lambda robot: (str(robot.get("name", "")), str(robot.get("host", ""))),
+ )
+ if cfg.json:
+ sys.stdout.write(json.dumps(robots) + "\n")
+ return 0
+ if not robots:
+ print(f"No robots discovered in {time.monotonic() - started_at:.1f}s.") # noqa: T201
+ return 0
+ headers = ("NAME", "ROBOT CLASS", "HOST", "JOINTS")
+ rows = [
+ (
+ str(robot.get("name", "?")),
+ str(robot.get("robot_class", "?")),
+ str(robot.get("host", "?")),
+ str(robot.get("num_joints", "?")),
+ )
+ for robot in robots
+ ]
+ print(_format_table(headers, rows, right_align={3})) # noqa: T201
+ noun = "robot" if len(robots) == 1 else "robots"
+ print(f"\n{len(robots)} {noun} found in {time.monotonic() - started_at:.1f}s") # noqa: T201
+ return 0
+
+
+def _dispatch(parser: ArgumentParser, cfg: Namespace) -> int: # noqa: ARG001
+ if cfg.subcommand == "serve":
+ return serve(cfg.serve)
+ if cfg.subcommand == "discover":
+ return discover(cfg.discover)
+ msg = f"unknown robot subcommand: {cfg.subcommand!r}"
+ raise AssertionError(msg)
+
+
+def register() -> SubcommandSpec:
+ """Register the robot command group with the CLI host.
+
+ Returns:
+ The robot subcommand specification.
+ """
+ return SubcommandSpec(name="robot", parser=build_parser(), dispatch=_dispatch, help=HELP)
diff --git a/src/physicalai/robot/transport/__init__.py b/src/physicalai/robot/transport/__init__.py
index de19246f..38d89d8d 100644
--- a/src/physicalai/robot/transport/__init__.py
+++ b/src/physicalai/robot/transport/__init__.py
@@ -24,6 +24,18 @@
from __future__ import annotations
from ._client import SharedRobotClient
+from ._owner_config import DEFAULT_RATE_HZ, RobotOwnerConfig
+from ._owner_worker import OwnerEvent, OwnerExitReason, OwnerResult, run_owner
from ._shared_robot import SharedRobot, discover_robots
-__all__ = ["SharedRobot", "SharedRobotClient", "discover_robots"]
+__all__ = [
+ "DEFAULT_RATE_HZ",
+ "OwnerEvent",
+ "OwnerExitReason",
+ "OwnerResult",
+ "RobotOwnerConfig",
+ "SharedRobot",
+ "SharedRobotClient",
+ "discover_robots",
+ "run_owner",
+]
diff --git a/src/physicalai/robot/transport/_lock.py b/src/physicalai/robot/transport/_lock.py
index 1b252e9b..b9493985 100644
--- a/src/physicalai/robot/transport/_lock.py
+++ b/src/physicalai/robot/transport/_lock.py
@@ -107,7 +107,27 @@ def _read_live_name_diagnostics(path: Path) -> dict[str, object] | None:
os.kill(pid, 0)
except OSError:
return None
- return diagnostics
+
+ # A PID that is alive but has already released the lock (e.g. after a clean
+ # shutdown or after a caller used acquire_locks/release_all for verification)
+ # is not an active owner. Confirm by attempting a non-blocking exclusive
+ # trylock: success means nobody holds it; EWOULDBLOCK means someone does.
+ try:
+ fd = os.open(path, os.O_RDWR)
+ except OSError:
+ return None
+ held = False
+ try:
+ fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ # Acquired the lock — no process is currently holding it.
+ with contextlib.suppress(OSError):
+ fcntl.flock(fd, fcntl.LOCK_UN)
+ except OSError:
+ # Could not acquire — some process holds the lock; this is a live owner.
+ held = True
+ finally:
+ os.close(fd)
+ return diagnostics if held else None
def _active_owner_name(path: Path) -> str | None:
diff --git a/src/physicalai/robot/transport/_owner.py b/src/physicalai/robot/transport/_owner.py
index 5888c85c..a708d352 100644
--- a/src/physicalai/robot/transport/_owner.py
+++ b/src/physicalai/robot/transport/_owner.py
@@ -38,6 +38,7 @@ class RobotOwner:
def __init__(self, config: RobotOwnerConfig) -> None:
self._config = config
self._process: subprocess.Popen[bytes] | None = None
+ self._exit_code: int | None = None
def start(self, timeout: float = _DEFAULT_START_TIMEOUT) -> None:
"""Start the owner subprocess and wait for the READY handshake.
@@ -60,6 +61,7 @@ def start(self, timeout: float = _DEFAULT_START_TIMEOUT) -> None:
"""
if self.is_alive:
return
+ self._exit_code = None
# B603 suppressed: the argv list is static — sys.executable (the
# active interpreter) plus a hardcoded internal module path.
@@ -129,7 +131,11 @@ def stop(self, timeout: float = 5.0) -> None:
timeout: Maximum seconds to wait for graceful shutdown.
"""
proc = self._process
- if proc is None or proc.poll() is not None:
+ if proc is None:
+ return
+
+ if proc.poll() is not None:
+ self._exit_code = proc.returncode
self._process = None
return
@@ -141,8 +147,26 @@ def stop(self, timeout: float = 5.0) -> None:
proc.kill()
proc.wait(timeout=1)
+ self._exit_code = proc.returncode
self._process = None
+ def wait(self) -> int:
+ """Wait for the owner to exit and return its retained exit code.
+
+ Returns:
+ The worker process exit code.
+
+ Raises:
+ RuntimeError: If the owner has never been started.
+ """
+ if self._process is None:
+ if self._exit_code is None:
+ msg = "cannot wait for robot owner before start"
+ raise RuntimeError(msg)
+ return self._exit_code
+ self._exit_code = self._process.wait()
+ return self._exit_code
+
@property
def is_alive(self) -> bool:
"""Whether the owner subprocess is running."""
diff --git a/src/physicalai/robot/transport/_owner_config.py b/src/physicalai/robot/transport/_owner_config.py
index 66e0d863..88643d11 100644
--- a/src/physicalai/robot/transport/_owner_config.py
+++ b/src/physicalai/robot/transport/_owner_config.py
@@ -1,17 +1,17 @@
# Copyright (C) 2026 Intel Corporation
# SPDX-License-Identifier: Apache-2.0
-"""Serializable owner-subprocess construction config for shared robots.
+"""Serializable owner construction config for shared robots.
Named ``RobotOwnerConfig`` (not ``RobotSpec``) to avoid colliding with the
unrelated ``physicalai.inference.manifest.RobotSpec`` (a manifest-schema
-pydantic model). This is private, process-internal IPC — not a user-facing
-configuration object.
+pydantic model). Used by the foreground serve path and by the owner
+subprocess stdin handshake.
-The owner subprocess must construct the robot driver itself: a live
-serial/socket handle cannot cross a process boundary (D15). Only an
-importable ``robot_class`` plus JSON-serializable ``robot_kwargs`` survive
-that boundary — arbitrary robot types, including third-party plugins, work
+The owner must construct the robot driver itself: a live serial/socket
+handle cannot cross a process boundary (D15). Only an importable
+``robot_class`` plus JSON-serializable ``robot_kwargs`` survive that
+boundary — arbitrary robot types, including third-party plugins, work
without any registry lookup here.
Security: *robot_class* is trusted local application/config input, exactly
@@ -24,6 +24,7 @@
from __future__ import annotations
import json
+import math
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
@@ -77,7 +78,13 @@ class happened to be imported by the caller.
@dataclass(frozen=True)
class RobotOwnerConfig:
- """Everything the owner subprocess needs to construct and run a robot.
+ """Everything the owner needs to construct and run a robot.
+
+ Warning:
+ ``allow_remote=True`` exposes an unauthenticated physical ``/action``
+ endpoint beyond localhost. Any peer that can reach the owner's Zenoh
+ session can move the robot. Use only on an isolated robot-cell
+ network (VLAN/firewall) or with Zenoh ACL/TLS.
Attributes:
name: The robot's logical name (keys the Zenoh topics).
@@ -86,6 +93,9 @@ class RobotOwnerConfig:
driver constructor (e.g. ``calibration`` as a file path).
allow_remote: Whether the owner's Zenoh session is reachable beyond
localhost. Fixed for the owner's lifetime once spawned.
+ ``True`` exposes an unauthenticated physical ``/action`` endpoint
+ — use only on an isolated robot-cell network or with Zenoh
+ ACL/TLS. Default ``False`` keeps the owner unreachable off-host.
rate_hz: Owner loop rate.
idle_timeout: Seconds with zero subscribers before self-exit.
"""
@@ -95,7 +105,7 @@ class RobotOwnerConfig:
robot_kwargs: dict[str, Any] = field(default_factory=dict)
allow_remote: bool = False
rate_hz: float = DEFAULT_RATE_HZ
- idle_timeout: float = 10.0
+ idle_timeout: float | None = 10.0
def __post_init__(self) -> None:
"""Validate ``rate_hz`` and that ``robot_kwargs`` is JSON-serializable.
@@ -104,9 +114,28 @@ def __post_init__(self) -> None:
ValueError: If ``rate_hz`` is not finite and positive, or
``robot_kwargs`` contains non-JSON-serializable values.
"""
- if not (self.rate_hz > 0 and self.rate_hz < float("inf")):
+ if (
+ isinstance(self.rate_hz, bool)
+ or not isinstance(self.rate_hz, (int, float))
+ or not math.isfinite(self.rate_hz)
+ or self.rate_hz <= 0
+ ):
msg = f"rate_hz must be finite and greater than zero, got {self.rate_hz!r}"
raise ValueError(msg)
+ if self.idle_timeout is not None and (
+ isinstance(self.idle_timeout, bool)
+ or not isinstance(self.idle_timeout, (int, float))
+ or not math.isfinite(self.idle_timeout)
+ or self.idle_timeout <= 0
+ ):
+ msg = f"idle_timeout must be finite and greater than zero, got {self.idle_timeout!r}"
+ raise ValueError(msg)
+ from ._ids import validate_name # noqa: PLC0415
+
+ validate_name(self.name)
+ if not isinstance(self.robot_class, str) or not self.robot_class.strip() or "." not in self.robot_class:
+ msg = f"robot_class must be a nonempty dotted path, got {self.robot_class!r}"
+ raise ValueError(msg)
try:
json.dumps(self.robot_kwargs)
except TypeError as exc:
diff --git a/src/physicalai/robot/transport/_owner_worker.py b/src/physicalai/robot/transport/_owner_worker.py
index c77a0100..fed690cb 100644
--- a/src/physicalai/robot/transport/_owner_worker.py
+++ b/src/physicalai/robot/transport/_owner_worker.py
@@ -22,6 +22,7 @@
from __future__ import annotations
import contextlib
+import enum
import json
import signal
import sys
@@ -33,6 +34,7 @@
from loguru import logger
+from physicalai.robot.errors import RobotTransportError
from physicalai.robot.interface import Robot
from physicalai.robot.transport._codec import ( # noqa: PLC2701
ROBOT_TRANSPORT_PROTOCOL_VERSION,
@@ -52,13 +54,40 @@
from physicalai.robot.transport._session import open_session # noqa: PLC2701
if TYPE_CHECKING:
+ from collections.abc import Callable
from types import FrameType
_MAX_CONSECUTIVE_FAILURES = 5
+_HEARTBEAT_INTERVAL_S = 30.0
shutdown = threading.Event()
+class OwnerExitReason(enum.Enum):
+ """Reason the shared owner runtime stopped."""
+
+ SHUTDOWN = "shutdown"
+ IDLE_TIMEOUT = "idle_timeout"
+ CONSECUTIVE_READ_FAILURES = "consecutive_read_failures"
+ LOOP_FAILURE = "loop_failure"
+
+
+@dataclass(frozen=True)
+class OwnerResult:
+ """Structured result returned by the shared owner runtime."""
+
+ reason: OwnerExitReason
+ exit_code: int
+
+
+class OwnerEvent(enum.Enum):
+ """Operator-relevant pulse emitted by the owner loop."""
+
+ SUBSCRIBERS_PRESENT = "subscribers_present"
+ NO_SUBSCRIBERS = "no_subscribers"
+ HEARTBEAT = "heartbeat"
+
+
def sigterm_handler(_signum: int, _frame: FrameType | None) -> None:
shutdown.set()
@@ -165,15 +194,31 @@ def _build_metadata(
return metadata
+def _apply_pending_action(driver: Robot, action_sub: Any, name: str) -> None: # noqa: ANN401
+ """Apply the newest pending action without letting invalid input stop the owner."""
+ sample = action_sub.try_recv()
+ if sample is None:
+ return
+ try:
+ action, goal_time, _send_ts = decode_action(sample.payload.to_bytes())
+ driver.send_action(action, goal_time=goal_time)
+ except Exception: # noqa: BLE001
+ logger.warning(f"Failed to apply action for {name}")
+ logger.opt(exception=True).trace("action apply traceback")
+
+
def _run_loop(
driver: Robot,
state_pub: Any, # noqa: ANN401
action_sub: Any, # noqa: ANN401
*,
rate_hz: float,
- idle_timeout: float,
+ idle_timeout: float | None,
name: str,
-) -> None:
+ shutdown_event: threading.Event,
+ on_event: Callable[[OwnerEvent], None] | None = None,
+ heartbeat_interval_s: float = _HEARTBEAT_INTERVAL_S,
+) -> OwnerExitReason:
"""Single-threaded write-first owner loop.
Ordering per tick: apply newest action (minimizes action latency), read
@@ -189,22 +234,30 @@ def _run_loop(
rate_hz: Fixed loop rate.
idle_timeout: Seconds with zero subscribers before self-exit.
name: For logging.
+ shutdown_event: Event requesting graceful loop termination.
+ on_event: Optional callback for subscriber transitions and heartbeat telemetry.
+ heartbeat_interval_s: Seconds between heartbeat events.
+
+ Returns:
+ The reason the loop stopped.
"""
period = 1.0 / rate_hz
idle_since: float | None = None
consecutive_failures = 0
next_tick = time.monotonic()
+ next_heartbeat = next_tick + heartbeat_interval_s
+ subscribers_present = False
- while not shutdown.is_set():
- sample = action_sub.try_recv()
- if sample is not None:
- try:
- action, goal_time, _send_ts = decode_action(sample.payload.to_bytes())
- driver.send_action(action, goal_time=goal_time)
- except Exception: # noqa: BLE001
- # A malformed or out-of-range action from one subscriber must
- # not kill the owner shared by everyone else.
- logger.warning(f"Failed to apply action for {name}", exc_info=True)
+ def _emit(event: OwnerEvent) -> None:
+ if on_event is None:
+ return
+ try:
+ on_event(event)
+ except Exception: # noqa: BLE001
+ logger.warning(f"Owner event callback failed for {name}", exc_info=True)
+
+ while not shutdown_event.is_set():
+ _apply_pending_action(driver, action_sub, name)
try:
obs = driver.get_observation()
@@ -212,7 +265,7 @@ def _run_loop(
consecutive_failures += 1
if consecutive_failures >= _MAX_CONSECUTIVE_FAILURES:
logger.error(f"{consecutive_failures} consecutive read failures -- shutting down owner {name}")
- break
+ return OwnerExitReason.CONSECUTIVE_READ_FAILURES
continue
consecutive_failures = 0
@@ -228,13 +281,25 @@ def _run_loop(
now = time.monotonic()
# Runtime returns a MatchingStatus object (always truthy); the bool
# lives on its .matching attribute — the type stub says `-> bool`.
- if state_pub.matching_status.matching:
+ matching = state_pub.matching_status.matching
+ if matching and not subscribers_present:
+ subscribers_present = True
+ _emit(OwnerEvent.SUBSCRIBERS_PRESENT)
+ elif not matching and subscribers_present:
+ subscribers_present = False
+ _emit(OwnerEvent.NO_SUBSCRIBERS)
+
+ if matching:
idle_since = None
elif idle_since is None:
idle_since = now
- elif now - idle_since > idle_timeout:
+ elif idle_timeout is not None and now - idle_since > idle_timeout:
logger.info(f"No subscribers for {idle_timeout}s -- shutting down owner {name}")
- break
+ return OwnerExitReason.IDLE_TIMEOUT
+
+ if now >= next_heartbeat:
+ _emit(OwnerEvent.HEARTBEAT)
+ next_heartbeat = now + heartbeat_interval_s
next_tick += period
sleep_time = next_tick - time.monotonic()
@@ -244,6 +309,8 @@ def _run_loop(
# Fell behind (slow bus / blocking read); don't accumulate debt.
next_tick = time.monotonic()
+ return OwnerExitReason.SHUTDOWN
+
@dataclass
class _Endpoints:
@@ -272,12 +339,14 @@ def _connect_and_build_metadata(
raises.
"""
try:
+ logger.trace(f"Connecting robot driver for {config.name!r}")
driver.connect()
except Exception as exc:
msg = f"driver.connect() failed: {exc}"
raise _StartupError(msg, phase="connection_failed") from exc
first_obs = driver.get_observation()
+ logger.trace(f"Received initial observation for {config.name!r}")
metadata = _build_metadata(config, driver, device_ids, state_dim=int(first_obs.state.shape[0]))
return encode_metadata(metadata)
@@ -329,6 +398,7 @@ def _answer_metadata(query: Any) -> None: # noqa: ANN401
)
action_sub = session.declare_subscriber(action_key(config.name), zenoh.handlers.RingChannel(1))
metadata_queryable = session.declare_queryable(metadata_key_expr, _answer_metadata)
+ logger.trace(f"Declared state, action, and metadata endpoints for {config.name!r}")
except Exception as exc:
if session is not None:
with contextlib.suppress(Exception):
@@ -367,6 +437,7 @@ def _startup(config: RobotOwnerConfig) -> _Endpoints:
via :func:`signal_error`.
"""
try:
+ logger.trace(f"Constructing robot driver {config.robot_class!r}")
driver = config.build()
except Exception as exc:
msg = f"failed to construct {config.robot_class!r}: {exc}"
@@ -377,6 +448,7 @@ def _startup(config: RobotOwnerConfig) -> _Endpoints:
raise _StartupError(msg, phase="construction_failed")
device_ids = tuple(sorted(set(driver.device_ids)))
+ logger.trace(f"Acquiring ownership locks for {config.name!r}")
try:
locks = acquire_locks(config.name, device_ids)
@@ -409,63 +481,109 @@ def _startup(config: RobotOwnerConfig) -> _Endpoints:
)
-def main() -> int:
- """Entry point for the owner worker process.
+def run_owner(
+ config: RobotOwnerConfig,
+ shutdown_event: threading.Event,
+ *,
+ ready: Callable[[], None] | None = None,
+ on_event: Callable[[OwnerEvent], None] | None = None,
+) -> OwnerResult:
+ """Own a robot driver and its transport endpoints in the current process.
- Returns:
- Exit code: 0 on success, 1 on startup failure.
- """
- signal.signal(signal.SIGTERM, sigterm_handler)
+ Args:
+ config: Validated owner configuration.
+ shutdown_event: Event requesting graceful shutdown.
+ ready: Optional callback invoked after startup is complete.
+ on_event: Optional callback for operator-facing runtime telemetry.
- raw = sys.stdin.read()
- sys.stdin.close()
- try:
- config = RobotOwnerConfig.from_json_dict(json.loads(raw))
- except (json.JSONDecodeError, ValueError, KeyError, TypeError) as exc:
- signal_error(f"invalid worker config: {exc}", phase="invalid_config")
- return 1
+ Returns:
+ Structured termination reason and process-compatible exit code.
- saved_stdout_fd = suppress_stdout()
+ Raises:
+ RobotTransportError: If construction, locking, connection, or endpoint setup fails.
+ """
try:
endpoints = _startup(config)
except _StartupError as exc:
- restore_stdout(saved_stdout_fd)
- signal_error(str(exc), tb=traceback.format_exc(), phase=exc.phase, device_ids=exc.device_ids)
- return 1
- except Exception as exc: # noqa: BLE001
- restore_stdout(saved_stdout_fd)
- signal_error(f"{type(exc).__name__}: {exc}", tb=traceback.format_exc(), phase="unexpected_startup_failure")
- return 1
- restore_stdout(saved_stdout_fd)
-
- signal_ready()
-
+ raise RobotTransportError(str(exc), phase=exc.phase, device_ids=exc.device_ids) from exc
+ reason = OwnerExitReason.LOOP_FAILURE
+ exit_code = 1
try:
- _run_loop(
+ if ready is not None:
+ ready()
+ reason = _run_loop(
endpoints.driver,
endpoints.state_pub,
endpoints.action_sub,
rate_hz=config.rate_hz,
idle_timeout=config.idle_timeout,
name=config.name,
+ shutdown_event=shutdown_event,
+ on_event=on_event,
)
+ exit_code = 0 if reason in {OwnerExitReason.SHUTDOWN, OwnerExitReason.IDLE_TIMEOUT} else 1
except Exception: # noqa: BLE001
logger.exception(f"owner loop failed for {config.name}")
finally:
- shutdown.set()
- # Safe-state contract: the owner (not subscribers) stops/homes the
- # robot on exit, whether idle-timeout, SIGTERM, or loop failure.
+ shutdown_event.set()
try:
endpoints.driver.disconnect()
- except Exception: # noqa: BLE001
- logger.exception(f"driver disconnect failed for {config.name}")
+ logger.trace(f"Disconnected robot driver for {config.name!r}")
+ except Exception as exc: # noqa: BLE001
+ logger.error(f"driver disconnect failed for {config.name}: {exc}")
+ logger.opt(exception=True).trace("driver disconnect traceback")
+ exit_code = 1
with contextlib.suppress(Exception):
endpoints.metadata_queryable.undeclare()
+ logger.trace(f"Undeclared metadata endpoint for {config.name!r}")
with contextlib.suppress(Exception):
endpoints.session.close()
- endpoints.locks.release_all()
+ logger.trace(f"Closed Zenoh session for {config.name!r}")
+ with contextlib.suppress(Exception):
+ endpoints.locks.release_all()
+ logger.trace(f"Released ownership locks for {config.name!r}")
+
+ return OwnerResult(reason=reason, exit_code=exit_code)
+
+
+def main() -> int:
+ """Entry point for the owner worker process.
+
+ Returns:
+ Exit code: 0 on success, 1 on startup failure.
+ """
+ signal.signal(signal.SIGTERM, sigterm_handler)
+
+ raw = sys.stdin.read()
+ sys.stdin.close()
+ try:
+ config = RobotOwnerConfig.from_json_dict(json.loads(raw))
+ except (json.JSONDecodeError, ValueError, KeyError, TypeError) as exc:
+ signal_error(f"invalid worker config: {exc}", phase="invalid_config")
+ return 1
- return 0
+ saved_stdout_fd = suppress_stdout()
+ stdout_restored = False
+
+ def _ready() -> None:
+ nonlocal stdout_restored
+ restore_stdout(saved_stdout_fd)
+ stdout_restored = True
+ signal_ready()
+
+ try:
+ result = run_owner(config, shutdown, ready=_ready)
+ except RobotTransportError as exc:
+ if not stdout_restored:
+ restore_stdout(saved_stdout_fd)
+ signal_error(str(exc), tb=traceback.format_exc(), phase=exc.phase, device_ids=exc.device_ids)
+ return 1
+ except Exception as exc: # noqa: BLE001
+ if not stdout_restored:
+ restore_stdout(saved_stdout_fd)
+ signal_error(f"{type(exc).__name__}: {exc}", tb=traceback.format_exc(), phase="unexpected_startup_failure")
+ return 1
+ return result.exit_code
if __name__ == "__main__":
diff --git a/tests/unit/cli/test_cli.py b/tests/unit/cli/test_cli.py
index 865f0d06..66568461 100644
--- a/tests/unit/cli/test_cli.py
+++ b/tests/unit/cli/test_cli.py
@@ -433,8 +433,8 @@ def register() -> SubcommandSpec:
assert exit_code == 0
assert "fast help for pytest fit" in capsys.readouterr().out
- def test_builtins_contain_run_only(self) -> None:
- assert list(main_module._BUILTINS) == ["run"] # noqa: SLF001
+ def test_builtins_contain_run_and_robot_only(self) -> None:
+ assert list(main_module._BUILTINS) == ["run", "robot"] # noqa: SLF001
def test_unknown_subcommand_errors(self) -> None:
with pytest.raises(SystemExit) as exc:
diff --git a/tests/unit/cli/test_robot_cli.py b/tests/unit/cli/test_robot_cli.py
new file mode 100644
index 00000000..ab19564a
--- /dev/null
+++ b/tests/unit/cli/test_robot_cli.py
@@ -0,0 +1,270 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import json
+import select
+import signal
+import subprocess
+import sys
+import threading
+import time
+import uuid
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from loguru import logger
+
+from physicalai.cli import robot as robot_module
+from physicalai.robot.errors import RobotTransportError
+from physicalai.robot.transport import OwnerEvent, OwnerExitReason, OwnerResult
+from physicalai.robot.transport._lock import acquire_locks
+
+from tests.unit.robot.transport.conftest import requires_zenoh
+
+
+def _serve_cfg(**overrides: object) -> SimpleNamespace:
+ values: dict[str, object] = {
+ "name": "left-arm",
+ "robot_class": "tests.unit.robot.transport.fake.FakeRobot",
+ "robot_kwargs": {"port": "/dev/fake0"},
+ "allow_remote": False,
+ "rate_hz": 100.0,
+ "verbose": False,
+ }
+ values.update(overrides)
+ return SimpleNamespace(**values)
+
+
+def test_serve_runs_owner_in_foreground_with_persistent_timeout(capsys: object) -> None:
+ captured: dict[str, object] = {}
+
+ def _run_owner(config: object, shutdown: threading.Event, *, ready: object, on_event: object) -> OwnerResult:
+ captured["config"] = config
+ captured["shutdown"] = shutdown
+ assert callable(ready)
+ assert callable(on_event)
+ ready()
+ shutdown.set()
+ return OwnerResult(OwnerExitReason.SHUTDOWN, 0)
+
+ with patch.object(robot_module, "run_owner", side_effect=_run_owner) as run:
+ assert robot_module.serve(_serve_cfg()) == 0
+
+ run.assert_called_once()
+ config = captured["config"]
+ assert config.idle_timeout is None # type: ignore[attr-defined]
+ assert config.name == "left-arm" # type: ignore[attr-defined]
+ stderr = capsys.readouterr().err # type: ignore[attr-defined]
+ assert "unauthenticated" not in stderr
+ assert "[local-only]" in stderr
+
+
+def test_serve_allow_remote_warns_and_tags_mode(capsys: object) -> None:
+ def _run_owner( # noqa: ARG001
+ _config: object,
+ _shutdown: threading.Event,
+ *,
+ ready: object,
+ on_event: object,
+ ) -> OwnerResult:
+ return OwnerResult(OwnerExitReason.SHUTDOWN, 0)
+
+ with patch.object(robot_module, "run_owner", side_effect=_run_owner):
+ assert robot_module.serve(_serve_cfg(allow_remote=True)) == 0
+
+ stderr = capsys.readouterr().err # type: ignore[attr-defined]
+ assert robot_module._ALLOW_REMOTE_WARNING.strip() in stderr
+ assert "[remote]" in stderr
+ warning_at = stderr.index("WARNING: The action endpoint is unauthenticated")
+ starting_at = stderr.index("Starting robot")
+ assert warning_at < starting_at
+
+
+def test_signal_requests_runtime_shutdown(capsys: object) -> None:
+ def _run_owner( # noqa: ARG001
+ _config: object,
+ shutdown: threading.Event,
+ *,
+ ready: object,
+ on_event: object,
+ ) -> OwnerResult:
+ signal.raise_signal(signal.SIGTERM)
+ assert shutdown.is_set()
+ return OwnerResult(OwnerExitReason.SHUTDOWN, 0)
+
+ with patch.object(robot_module, "run_owner", side_effect=_run_owner):
+ assert robot_module.serve(_serve_cfg()) == 0
+ assert "Shutdown requested by SIGTERM" in capsys.readouterr().err # type: ignore[attr-defined]
+
+
+def test_expected_startup_error_is_concise(capsys: object) -> None:
+ error = RobotTransportError("name is already owned", phase="name_lock_contention")
+ with patch.object(robot_module, "run_owner", side_effect=error):
+ assert robot_module.serve(_serve_cfg()) == 1
+
+ stderr = capsys.readouterr().err # type: ignore[attr-defined]
+ assert "name_lock_contention" in stderr
+ assert "Traceback" not in stderr
+
+
+def test_invalid_config_fails_before_runtime(capsys: object) -> None:
+ with patch.object(robot_module, "run_owner") as run:
+ assert robot_module.serve(_serve_cfg(name="left/arm")) == 1
+ run.assert_not_called()
+ assert "Invalid robot configuration" in capsys.readouterr().err # type: ignore[attr-defined]
+
+
+def test_discovery_json_is_sorted_and_clean(capsys: object) -> None:
+ records = [
+ {"name": "z-arm", "host": "b", "robot_class": "untrusted.Z", "num_joints": 7},
+ {"name": "a-arm", "host": "a", "robot_class": "untrusted.A", "num_joints": 6},
+ ]
+ cfg = SimpleNamespace(timeout=1.0, allow_remote=False, json=True)
+ with patch.object(robot_module, "discover_robots", return_value=records):
+ assert robot_module.discover(cfg) == 0
+
+ output = capsys.readouterr() # type: ignore[attr-defined]
+ assert output.err == ""
+ assert [record["name"] for record in json.loads(output.out)] == ["a-arm", "z-arm"]
+
+
+def test_discovery_human_output_is_table(capsys: object) -> None:
+ records = [
+ {"name": "z-arm", "host": "b", "robot_class": "untrusted.Z", "num_joints": 7},
+ {"name": "a-arm", "host": "a", "robot_class": "untrusted.A", "num_joints": 6},
+ ]
+ cfg = SimpleNamespace(timeout=1.0, allow_remote=False, json=False)
+ with patch.object(robot_module, "discover_robots", return_value=records):
+ assert robot_module.discover(cfg) == 0
+
+ output = capsys.readouterr().out # type: ignore[attr-defined]
+ assert "NAME" in output
+ assert "ROBOT CLASS" in output
+ assert output.index("a-arm") < output.index("z-arm")
+ assert "2 robots found" in output
+
+
+def test_owner_events_are_logged(capsys: object) -> None:
+ def _run_owner( # noqa: ARG001
+ _config: object,
+ _shutdown: threading.Event,
+ *,
+ ready: object,
+ on_event: object,
+ ) -> OwnerResult:
+ assert callable(ready)
+ assert callable(on_event)
+ ready()
+ on_event(OwnerEvent.SUBSCRIBERS_PRESENT)
+ on_event(OwnerEvent.HEARTBEAT)
+ on_event(OwnerEvent.NO_SUBSCRIBERS)
+ return OwnerResult(OwnerExitReason.SHUTDOWN, 0)
+
+ with patch.object(robot_module, "run_owner", side_effect=_run_owner):
+ assert robot_module.serve(_serve_cfg()) == 0
+
+ stderr = capsys.readouterr().err # type: ignore[attr-defined]
+ assert "Subscriber(s) connected" in stderr
+ assert "Healthy" in stderr
+ assert "Subscriber(s) connected" in stderr
+ assert "No subscribers remain" in stderr
+
+
+def test_verbose_controls_trace_details(capsys: object) -> None:
+ def _run_owner( # noqa: ARG001
+ _config: object,
+ _shutdown: threading.Event,
+ *,
+ ready: object,
+ on_event: object,
+ ) -> OwnerResult:
+ logger.trace("startup phase detail")
+ return OwnerResult(OwnerExitReason.SHUTDOWN, 0)
+
+ with patch.object(robot_module, "run_owner", side_effect=_run_owner):
+ assert robot_module.serve(_serve_cfg(verbose=False)) == 0
+ assert "startup phase detail" not in capsys.readouterr().err # type: ignore[attr-defined]
+
+ with patch.object(robot_module, "run_owner", side_effect=_run_owner):
+ assert robot_module.serve(_serve_cfg(verbose=True)) == 0
+ assert "startup phase detail" in capsys.readouterr().err # type: ignore[attr-defined]
+
+
+def test_empty_discovery_json_is_array(capsys: object) -> None:
+ cfg = SimpleNamespace(timeout=1.0, allow_remote=False, json=True)
+ with patch.object(robot_module, "discover_robots", return_value=[]):
+ assert robot_module.discover(cfg) == 0
+ assert capsys.readouterr().out == "[]\n" # type: ignore[attr-defined]
+
+
+def test_parser_does_not_expose_idle_timeout() -> None:
+ parser = robot_module.build_parser()
+ cfg = parser.parse_args(
+ ["serve", "--name", "left-arm", "--robot_class", "pkg.mod.Robot"],
+ )
+ assert not hasattr(cfg.serve, "idle_timeout")
+
+
+def test_discover_rejects_invalid_timeout(capsys: object) -> None:
+ cfg = SimpleNamespace(timeout=float("nan"), allow_remote=False, json=False)
+ with patch.object(robot_module, "discover_robots") as discover:
+ assert robot_module.discover(cfg) == 1
+ discover.assert_not_called()
+ assert "timeout must be finite" in capsys.readouterr().err # type: ignore[attr-defined]
+
+
+@requires_zenoh
+def test_serve_process_sigterm_disconnects_and_releases_locks(tmp_path: Path) -> None:
+ name = f"cli-{uuid.uuid4().hex[:8]}"
+ port = f"/dev/{name}"
+ device_id = f"fake:{port}"
+ marker = tmp_path / "disconnected"
+ process = subprocess.Popen( # noqa: S603
+ [
+ sys.executable,
+ "-m",
+ "physicalai.cli.main",
+ "robot",
+ "serve",
+ "--name",
+ name,
+ "--robot_class",
+ "tests.unit.robot.transport.fake.FakeRobot",
+ "--robot_kwargs.port",
+ port,
+ "--robot_kwargs.disconnect_marker",
+ str(marker),
+ ],
+ stderr=subprocess.PIPE,
+ text=True,
+ )
+ assert process.stderr is not None
+ deadline = time.monotonic() + 20.0
+ ready = False
+ stderr_lines: list[str] = []
+ try:
+ while time.monotonic() < deadline:
+ readable, _, _ = select.select([process.stderr], [], [], 0.5)
+ if readable:
+ line = process.stderr.readline()
+ stderr_lines.append(line)
+ if "Serving robot" in line:
+ ready = True
+ break
+ if process.poll() is not None:
+ break
+ assert ready, "".join(stderr_lines)
+
+ process.terminate()
+ assert process.wait(timeout=10.0) == 0
+ assert marker.exists()
+
+ locks = acquire_locks(name, [device_id])
+ locks.release_all()
+ finally:
+ if process.poll() is None:
+ process.kill()
+ process.wait(timeout=5.0)
diff --git a/tests/unit/robot/transport/fake.py b/tests/unit/robot/transport/fake.py
index 9f9ff16c..ecda1a81 100644
--- a/tests/unit/robot/transport/fake.py
+++ b/tests/unit/robot/transport/fake.py
@@ -13,6 +13,7 @@
import time
from dataclasses import dataclass
+from pathlib import Path
import numpy as np
@@ -53,12 +54,19 @@ def __init__(
device_ids: tuple[str, ...] | None = None,
fail_connect: bool = False,
fail_observation: bool = False,
+ fail_observation_after: int | None = None,
+ fail_disconnect: bool = False,
+ disconnect_marker: str | None = None,
**_ignored: object,
) -> None:
self._port = port
self._device_ids = device_ids if device_ids is not None else (f"fake:{port}",)
self._fail_connect = fail_connect
self._fail_observation = fail_observation
+ self._fail_observation_after = fail_observation_after
+ self._fail_disconnect = fail_disconnect
+ self._disconnect_marker = disconnect_marker
+ self._observation_calls = 0
self._connected = False
self._last_action = np.zeros(NUM_JOINTS, dtype=np.float32)
self.disconnect_called = False
@@ -80,12 +88,20 @@ def connect(self) -> None:
def disconnect(self) -> None:
self.disconnect_called = True
self._connected = False
+ if self._disconnect_marker is not None:
+ Path(self._disconnect_marker).touch()
+ if self._fail_disconnect:
+ msg = f"fake disconnect failure on {self._port}"
+ raise RuntimeError(msg)
def get_observation(self) -> FakeObservation:
if not self._connected:
msg = "Robot is not connected. Call connect() first."
raise ConnectionError(msg)
- if self._fail_observation:
+ self._observation_calls += 1
+ if self._fail_observation or (
+ self._fail_observation_after is not None and self._observation_calls > self._fail_observation_after
+ ):
msg = f"fake observation failure on {self._port}"
raise RuntimeError(msg)
# Echo the last commanded action as measured position so tests can
diff --git a/tests/unit/robot/transport/test_owner_config.py b/tests/unit/robot/transport/test_owner_config.py
index 73935b11..08730208 100644
--- a/tests/unit/robot/transport/test_owner_config.py
+++ b/tests/unit/robot/transport/test_owner_config.py
@@ -6,6 +6,7 @@
import json
import pickle
import sys
+from typing import Any, cast
from unittest.mock import MagicMock, patch
import pytest
@@ -65,6 +66,30 @@ def test_defaults(self) -> None:
assert config.rate_hz == 100.0
assert config.idle_timeout == 10.0
+ def test_persistent_idle_timeout_json_roundtrip(self) -> None:
+ config = RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Cls", idle_timeout=None)
+ restored = RobotOwnerConfig.from_json_dict(json.loads(json.dumps(config.to_json_dict())))
+ assert restored == config
+ assert restored.idle_timeout is None
+
+ def test_invalid_name_raises(self) -> None:
+ with pytest.raises(ValueError, match="invalid robot name"):
+ RobotOwnerConfig(name="left/arm", robot_class="pkg.mod.Cls")
+
+ @pytest.mark.parametrize("rate_hz", [float("nan"), True, "100"])
+ def test_invalid_rate_types_raise(self, rate_hz: object) -> None:
+ with pytest.raises(ValueError, match="rate_hz must be finite"):
+ RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Cls", rate_hz=rate_hz) # type: ignore[arg-type]
+
+ @pytest.mark.parametrize("idle_timeout", [0.0, -1.0, float("inf"), float("nan"), True, "10"])
+ def test_invalid_idle_timeout_raises(self, idle_timeout: object) -> None:
+ with pytest.raises(ValueError, match="idle_timeout must be finite"):
+ RobotOwnerConfig(
+ name="left-arm",
+ robot_class="pkg.mod.Cls",
+ idle_timeout=cast(Any, idle_timeout),
+ )
+
def test_zero_rate_raises(self) -> None:
with pytest.raises(ValueError, match="rate_hz must be finite"):
RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Cls", rate_hz=0.0)
diff --git a/tests/unit/robot/transport/test_owner_worker.py b/tests/unit/robot/transport/test_owner_worker.py
new file mode 100644
index 00000000..b22fbf81
--- /dev/null
+++ b/tests/unit/robot/transport/test_owner_worker.py
@@ -0,0 +1,169 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import threading
+import time
+from dataclasses import dataclass, field
+from types import SimpleNamespace
+from typing import Any
+
+from physicalai.robot.transport import _owner_worker
+from physicalai.robot.transport._owner_config import RobotOwnerConfig
+from physicalai.robot.transport._owner_worker import OwnerEvent, OwnerExitReason, _run_loop, run_owner
+
+from .fake import FakeRobot
+
+
+@dataclass
+class _MatchingStatus:
+ matching: bool = True
+
+
+@dataclass
+class _StatePublisher:
+ matching: bool = True
+ fail_put: bool = False
+ puts: list[bytes] = field(default_factory=list)
+
+ @property
+ def matching_status(self) -> _MatchingStatus:
+ return _MatchingStatus(self.matching)
+
+ def put(self, payload: bytes) -> None:
+ if self.fail_put:
+ raise RuntimeError("fake publish failure")
+ self.puts.append(payload)
+
+
+class _ActionSubscriber:
+ def try_recv(self) -> Any | None:
+ return None
+
+
+def _driver(**kwargs: object) -> FakeRobot:
+ driver = FakeRobot(**kwargs)
+ driver.connect()
+ return driver
+
+
+def test_shutdown_returns_shutdown() -> None:
+ shutdown = threading.Event()
+ shutdown.set()
+ reason = _run_loop(
+ _driver(),
+ _StatePublisher(),
+ _ActionSubscriber(),
+ rate_hz=1000.0,
+ idle_timeout=None,
+ name="test",
+ shutdown_event=shutdown,
+ )
+ assert reason is OwnerExitReason.SHUTDOWN
+
+
+def test_idle_timeout_returns_idle_timeout() -> None:
+ reason = _run_loop(
+ _driver(),
+ _StatePublisher(matching=False),
+ _ActionSubscriber(),
+ rate_hz=1000.0,
+ idle_timeout=0.01,
+ name="test",
+ shutdown_event=threading.Event(),
+ )
+ assert reason is OwnerExitReason.IDLE_TIMEOUT
+
+
+def test_repeated_reads_return_failure() -> None:
+ reason = _run_loop(
+ _driver(fail_observation=True),
+ _StatePublisher(),
+ _ActionSubscriber(),
+ rate_hz=1000.0,
+ idle_timeout=10.0,
+ name="test",
+ shutdown_event=threading.Event(),
+ )
+ assert reason is OwnerExitReason.CONSECUTIVE_READ_FAILURES
+
+
+def test_subscriber_transitions_and_heartbeat_emit_events() -> None:
+ events: list[OwnerEvent] = []
+ publisher = _StatePublisher(matching=False)
+ shutdown = threading.Event()
+
+ def _change_matching() -> None:
+ publisher.matching = True
+ time.sleep(0.02)
+ publisher.matching = False
+ time.sleep(0.02)
+ shutdown.set()
+
+ thread = threading.Thread(target=_change_matching)
+ thread.start()
+ try:
+ reason = _run_loop(
+ _driver(),
+ publisher,
+ _ActionSubscriber(),
+ rate_hz=1000.0,
+ idle_timeout=None,
+ name="test",
+ shutdown_event=shutdown,
+ on_event=events.append,
+ heartbeat_interval_s=0.01,
+ )
+ finally:
+ thread.join()
+
+ assert reason is OwnerExitReason.SHUTDOWN
+ assert OwnerEvent.SUBSCRIBERS_PRESENT in events
+ assert OwnerEvent.NO_SUBSCRIBERS in events
+ assert OwnerEvent.HEARTBEAT in events
+
+
+def test_disconnect_failure_upgrades_clean_shutdown(monkeypatch: Any) -> None: # noqa: ANN401
+ driver = _driver(fail_disconnect=True)
+ shutdown = threading.Event()
+ shutdown.set()
+ endpoints = SimpleNamespace(
+ driver=driver,
+ state_pub=_StatePublisher(),
+ action_sub=_ActionSubscriber(),
+ metadata_queryable=SimpleNamespace(undeclare=lambda: None),
+ session=SimpleNamespace(close=lambda: None),
+ locks=SimpleNamespace(release_all=lambda: None),
+ )
+ monkeypatch.setattr(_owner_worker, "_startup", lambda _config: endpoints)
+ config = RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Robot", idle_timeout=None)
+
+ result = run_owner(config, shutdown)
+
+ assert result.reason is OwnerExitReason.SHUTDOWN
+ assert result.exit_code == 1
+ assert driver.disconnect_called
+
+
+def test_ready_failure_still_disconnects(monkeypatch: Any) -> None: # noqa: ANN401
+ driver = _driver()
+ endpoints = SimpleNamespace(
+ driver=driver,
+ state_pub=_StatePublisher(),
+ action_sub=_ActionSubscriber(),
+ metadata_queryable=SimpleNamespace(undeclare=lambda: None),
+ session=SimpleNamespace(close=lambda: None),
+ locks=SimpleNamespace(release_all=lambda: None),
+ )
+ monkeypatch.setattr(_owner_worker, "_startup", lambda _config: endpoints)
+ config = RobotOwnerConfig(name="left-arm", robot_class="pkg.mod.Robot", idle_timeout=None)
+
+ def _fail_ready() -> None:
+ raise RuntimeError("readiness output failed")
+
+ result = run_owner(config, threading.Event(), ready=_fail_ready)
+
+ assert result.reason is OwnerExitReason.LOOP_FAILURE
+ assert result.exit_code == 1
+ assert driver.disconnect_called