feat: robot serve CLI command

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
 

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

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

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

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)