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="<green>{time:HH:mm:ss}</green>  <level>{level: <8}</level> {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