From 8549d5b999c44c5819fc7d5027d932d071a70a8e Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Mon, 20 Jul 2026 16:51:18 +0200 Subject: [PATCH 1/8] add robot serve --- docs/development/robot-serve-plan.md | 462 ++++++++++++++++++ .../robot-zenoh-transport-design.md | 356 ++++++++++++++ docs/explanation/cli.md | 8 +- docs/how-to/runtime/share-a-robot.md | 25 + docs/reference/cli.md | 25 + examples/so101/serve.yaml | 14 + src/physicalai/cli/main.py | 14 +- src/physicalai/cli/robot.py | 184 +++++++ src/physicalai/robot/transport/_owner.py | 28 +- .../robot/transport/_owner_config.py | 24 +- .../robot/transport/_owner_worker.py | 139 ++++-- tests/unit/cli/test_cli.py | 4 +- tests/unit/cli/test_robot_cli.py | 173 +++++++ tests/unit/robot/transport/fake.py | 18 +- .../unit/robot/transport/test_owner_config.py | 25 + .../unit/robot/transport/test_owner_worker.py | 133 +++++ 16 files changed, 1581 insertions(+), 51 deletions(-) create mode 100644 docs/development/robot-serve-plan.md create mode 100644 docs/development/robot-zenoh-transport-design.md create mode 100644 examples/so101/serve.yaml create mode 100644 src/physicalai/cli/robot.py create mode 100644 tests/unit/cli/test_robot_cli.py create mode 100644 tests/unit/robot/transport/test_owner_worker.py diff --git a/docs/development/robot-serve-plan.md b/docs/development/robot-serve-plan.md new file mode 100644 index 00000000..f4ec6cc9 --- /dev/null +++ b/docs/development/robot-serve-plan.md @@ -0,0 +1,462 @@ +# Robot Serve design + +## Status + +Implemented production design. This document replaces the proof-of-concept plan +implemented in commit `2ed63a6ea2a53ea6548a149cc4a82cb1022ce64c`. + +The PoC established that a robot can be exposed through the existing Zenoh +`SharedRobot` transport and consumed from another process or host. The +production implementation must now make foreground ownership, shutdown, +failure reporting, and CLI boundaries explicit. + +## Decision + +Add these operator commands: + +```text +physicalai robot serve +physicalai robot discover +``` + +`physicalai robot serve` runs the hardware owner in the foreground process. It +does not launch a detached persistent child. Foreground serving and the +existing auto-spawn worker call one shared owner runtime that exclusively +constructs, connects, reads, commands, and disconnects the robot. + +The auto-spawn path remains detached and retains its numeric idle timeout. The +explicit CLI path is persistent until interrupted and remains bound to the +lifetime of the foreground process. + +## Goals + +- Provide a convenient YAML- and argument-driven command for serving one robot. +- Reuse the proven `SharedRobot` Zenoh protocol and hardware ownership model. +- Keep exactly one process responsible for each live robot driver. +- Guarantee that normal CLI termination attempts a safe driver disconnect. +- Report startup, loop, forced-termination, and disconnect failures as nonzero. +- Preserve secure local-only networking unless remote access is explicit. +- Keep auto-spawn behavior compatible with existing `SharedRobot` callers. +- Expose deterministic human-readable and machine-readable discovery. + +## Non-goals + +- Authentication, authorization, or encryption in the robot transport. +- Cross-host robot-name arbitration or distributed ownership leases. +- Multiple-controller action arbitration. +- Daemonization, PID files, or a background-service manager in the CLI. +- Zenoh router provisioning. +- Driver-specific CLI arguments. +- Reworking the robot wire protocol. +- Detailed nested shell completion. + +Operating-system service managers such as systemd, Docker, or Kubernetes own +background supervision. The CLI remains a foreground command suitable for +those supervisors. + +## Why the PoC Architecture Must Change + +The PoC reused `RobotOwner`, which always starts its worker with +`start_new_session=True`, and made that detached worker persistent by setting +`idle_timeout=None`. This combines two individually useful behaviors into an +unsafe lifecycle: + +1. The CLI process starts a detached owner. +2. The owner has no idle self-termination. +3. If the CLI is killed or crashes before signaling the owner, the owner can + continue indefinitely while retaining the hardware and ownership locks. + +That violates the operator-facing contract that the robot is served for the +lifetime of `physicalai robot serve`. + +The PoC also made the CLI depend on private transport validators and mixed +explicit-host networking changes into the serve feature. Those concerns should +be separated so each can be reviewed against one contract. + +## Architecture + +### Shared owner runtime + +Extract the hardware-owning behavior from the subprocess entry point into one +internal, in-process runtime: + +```python +class OwnerExitReason(enum.Enum): + SHUTDOWN = "shutdown" + IDLE_TIMEOUT = "idle_timeout" + CONSECUTIVE_READ_FAILURES = "consecutive_read_failures" + LOOP_FAILURE = "loop_failure" + + +@dataclass(frozen=True) +class OwnerResult: + reason: OwnerExitReason + exit_code: int + + +def run_owner( + config: RobotOwnerConfig, + shutdown: threading.Event, + *, + ready: Callable[[], None] | None = None, +) -> OwnerResult: + ... +``` + +The exact names and callback shape are implementation details. The contract is +that this runtime owns the complete lifecycle: + +```text +validate config + -> import and construct driver + -> acquire name and device locks + -> connect driver + -> read initial observation + -> declare Zenoh endpoints + -> report ready + -> run control loop + -> disconnect driver + -> undeclare/close/release best-effort resources + -> return structured result +``` + +No launch adapter duplicates hardware construction, loop, or cleanup logic. + +### Launch adapters + +Two adapters invoke the same runtime with different lifecycle policies: + +| Concern | Auto-spawn worker | `robot serve` | +| -------------------- | -------------------------------- | ---------------------------- | +| Process | Dedicated child | Foreground CLI process | +| Detachment | New session | None | +| Idle timeout | Numeric, default 10 seconds | Disabled (`None`) | +| Readiness | `READY`/`ERROR` IPC | Operator message after ready | +| Shutdown source | SIGTERM or idle timeout | SIGINT/SIGTERM | +| Parent death | Idle timeout eventually stops it | Process itself is gone | +| Exit status consumer | `RobotOwner` parent | Shell/service manager | + +The subprocess module remains a thin adapter around `run_owner` for the +existing auto-spawn handshake. The CLI adapter installs signal handlers, calls +`run_owner` directly, and maps `OwnerResult` to a process exit code. + +### Responsibility boundaries + +`physicalai.robot.transport` owns: + +- Owner configuration and validation. +- Driver import and construction. +- Host-local name and device locks. +- Zenoh endpoints and payloads. +- The control loop and its termination reasons. +- Safe driver disconnect and cleanup ordering. +- Discovery of metadata records. + +`physicalai.cli.robot` owns: + +- Command-line and config-file parsing. +- Direct group and nested command help. +- Signal-to-shutdown-event adaptation for foreground serving. +- Operator-facing output and exit-code presentation. +- Sorting and formatting discovery results. + +The CLI must not import private field validators or reproduce transport +validation. It constructs one supported owner configuration and catches the +documented validation and transport errors at the command boundary. + +## Lifecycle Invariants + +1. Only the owner runtime imports, constructs, connects, reads, commands, and + disconnects a concrete robot driver. +2. Locks are acquired before `driver.connect()` performs hardware access. +3. Readiness is reported only after the driver is connected, an initial valid + observation has been read, and all Zenoh endpoints are declared. +4. A foreground server never leaves a persistent detached worker by design. +5. SIGINT and SIGTERM request graceful shutdown; they do not bypass cleanup. +6. Driver disconnect is attempted after every post-connect exit path. +7. Driver-disconnect failure is safety-critical and forces a nonzero result. +8. Metadata undeclaration, Zenoh session close, and explicit lock release are + best-effort. Their failures are logged and do not replace the primary result. +9. Repeated observation failures and unexpected loop failures are nonzero. +10. Numeric idle timeout and requested shutdown succeed only when safe driver + disconnect also succeeds. +11. Subscriber disconnect never directly stops the robot; owner policy controls + the safe-state transition. + +`SIGKILL`, kernel failure, and power loss cannot run Python cleanup. Drivers and +hardware must fail safely where possible, but that is outside this command's +guarantees. + +## Configuration + +Use the existing flat owner shape rather than introducing another +`class_path`/`init_args` representation: + +```yaml +name: follower-arm +robot_class: physicalai.robot.SO101 +robot_kwargs: + port: /dev/ttyACM0 + calibration: /path/to/calibration.json + role: follower +allow_remote: false +rate_hz: 100.0 +``` + +| Field | Required | Meaning | +| -------------- | -------- | ----------------------------------------------- | +| `name` | Yes | Logical name and Zenoh key segment | +| `robot_class` | Yes | Trusted local dotted path to a driver class | +| `robot_kwargs` | No | JSON-serializable constructor keyword arguments | +| `allow_remote` | No | Expose beyond localhost; default `false` | +| `rate_hz` | No | Positive finite owner loop rate; default 100 Hz | + +`idle_timeout` is not exposed by `robot serve`. Explicit serving is persistent +until interrupted. Auto-spawn continues to supply its existing numeric timeout. + +Validate all fields before constructing hardware: + +- `name` is one nonempty safe Zenoh key segment. +- `robot_class` is a nonempty dotted path. Import occurs only in the owner. +- `robot_kwargs` is JSON-serializable across the worker IPC boundary. +- `rate_hz` is finite and greater than zero; booleans are rejected. + +The class path is trusted local configuration. Network metadata must never be +used as a class path or imported by discovery. + +## CLI Design + +### Serve + +```bash +physicalai robot serve --config examples/so101/serve.yaml +``` + +Direct arguments remain available for scripting: + +```bash +physicalai robot serve \ + --name follower-arm \ + --robot_class physicalai.robot.SO101 \ + --robot_kwargs.port /dev/ttyACM0 \ + --robot_kwargs.calibration /path/to/calibration.json \ + --allow_remote +``` + +`--allow_remote` is a bare `store_true` switch. Local-only operation is the +default. A readiness message goes to stderr so stdout remains available for +future structured output. + +| Condition | Exit | +| ------------------------------------------------ | ------- | +| SIGINT/SIGTERM and successful disconnect | 0 | +| Invalid config | Nonzero | +| Driver construction/connect/initial-read failure | Nonzero | +| Name or device lock contention | Nonzero | +| Repeated read failure | Nonzero | +| Unexpected control-loop failure | Nonzero | +| Driver disconnect failure | Nonzero | + +Errors identify the phase and an operator recovery action where known. Expected +operational errors do not print tracebacks by default. Detailed tracebacks +remain available through debug logging. + +### Discover + +```bash +physicalai robot discover +physicalai robot discover --allow_remote +physicalai robot discover --json +``` + +Discovery delegates to the public transport discovery function, sorts records +by `(name, host)`, and never imports advertised `robot_class` values. + +Human output includes at least name, robot class, host, and joint count. JSON +mode writes exactly one array to stdout. Logs and warnings go to stderr. Empty +discovery is successful and produces `[]` in JSON mode. + +Targeted `--host`/`--name` discovery is not required for the serve redesign. If +retained for unicast-only networks, implement and review it as a separate +transport/API change. User-provided endpoints must use a structured serializer +such as `json.dumps`, not interpolation into JSON5. + +### Help routing + +Register `robot` as one built-in command group. Direct group help can use the +lightweight path: + +```text +physicalai robot --help +``` + +Nested help must reach the real nested parser: + +```text +physicalai robot serve --help +physicalai robot discover --help +``` + +Prefer generic host routing based on direct help versus nested arguments rather +than a robot-specific nested-command registry. Existing plugin help behavior +must remain unchanged and covered by regression tests. + +## Security + +`allow_remote=false` disables remote scouting and binds the owner to loopback. +This is the default for both configuration and CLI arguments. + +`allow_remote=true` exposes an unauthenticated action endpoint. Any reachable +peer can command the physical robot. Documentation and command help must state +that remote serving requires an isolated robot-cell VLAN/firewall or configured +Zenoh ACL/TLS. + +Discovery metadata excludes constructor kwargs, calibration paths, device +paths, and other construction secrets. Physical device identifiers remain +host-local diagnostics and are not advertised remotely. + +Robot names are operator-enforced as unique across the reachable Zenoh network. +Host-local locks do not provide cross-host arbitration. + +## Public API + +Do not add a public `RobotServer` class for this change. There is one concrete +consumer of foreground serving, and the CLI can call a narrow internal owner +runtime. Adding a public lifecycle abstraction now would freeze semantics that +have not demonstrated a second use case. + +Keep `SharedRobot`, `SharedRobotClient`, and `discover_robots` as the user-facing +transport APIs. The owner runtime and owner configuration may remain internal. +Reconsider a public serving API when a second concrete embedding use case +requires it. + +## Migration From the PoC + +Reimplement from the parent of commit +`2ed63a6ea2a53ea6548a149cc4a82cb1022ce64c` rather than repairing that commit in +place. Carry behavior and tests selectively. + +Retain: + +- The `physicalai robot serve` and `physicalai robot discover` experience. +- Flat configuration fields. +- Persistent explicit serve and numeric auto-spawn timeout. +- Structured termination reasons and exit-code requirements. +- Sorted human/JSON discovery behavior. +- Local-only defaults and the remote-network warning. +- Tests that express these contracts. + +Redesign: + +- Run explicit serving in the foreground instead of supervising a detached + persistent worker. +- Extract one owner runtime shared by foreground and subprocess adapters. +- Keep validation within the transport-owned configuration boundary. +- Route nested help generically where possible without plugin regressions. +- Convert operational failures to concise CLI errors. + +Separate or defer: + +- Explicit-host attachment and targeted discovery for unicast-only networks. +- Any unrelated transport refactor. + +Discard: + +- Robot-specific private validator imports in CLI code. +- Persistent detached-child supervision for explicit serve. +- Machine-specific paths and remote-enabled defaults in examples. + +## Implementation Sequence + +### Phase 1: Owner runtime + +1. Restore the pre-PoC transport to a compiling, tested baseline. +2. Extract `run_owner` without changing auto-spawn behavior. +3. Add structured exit reasons and cleanup result mapping. +4. Run focused owner config, worker, handshake, lock, and `SharedRobot` tests. + +### Phase 2: Foreground serve + +1. Add the flat serve parser and config-file support. +2. Call `run_owner` directly in the CLI process. +3. Adapt SIGINT/SIGTERM to the runtime shutdown event. +4. Add concise startup, readiness, contention, and shutdown reporting. +5. Add process-level CLI tests with the fake robot. + +### Phase 3: Discovery + +1. Add the discover parser over existing discovery behavior. +2. Add deterministic human and clean JSON formatting. +3. Verify that network metadata never drives imports. + +### Phase 4: Documentation + +1. Add a portable SO101 example with placeholders and `allow_remote: false`. +2. Update the sharing how-to and CLI reference. +3. Restore the full Zenoh transport design and align its lifecycle invariants + with this design. + +Keep these phases as separate reviewable commits where practical. + +## Test Strategy + +### Unit tests + +- Owner config accepts internal `idle_timeout=None` and round-trips it over IPC. +- Invalid names, rates, and kwargs fail before driver construction. +- Requested shutdown and numeric idle timeout return success after disconnect. +- Repeated reads, loop exceptions, and disconnect failures return nonzero. +- Cleanup continues after each best-effort cleanup failure. +- Auto-spawn remains detached with its 10-second default timeout. +- CLI serving passes `idle_timeout=None` and does not instantiate `RobotOwner`. +- Expected CLI failures do not leak tracebacks. +- Discovery is sorted and JSON stdout is one clean array. +- Group help stays lightweight and nested help reaches the selected parser. +- Plugin help behavior remains unchanged. + +### Process-level tests + +Start `physicalai robot serve` with a fake driver, wait for readiness, attach a +`SharedRobot`, observe state, send a stationary action, terminate the CLI, and +verify: + +- The foreground process exits. +- The fake driver disconnect path ran. +- Name and device locks can be reacquired. +- A clean interruption exits zero. +- An injected disconnect failure exits nonzero. +- No owner process remains after the CLI exits. + +### Manual hardware validation + +On an isolated robot network: + +1. Serve an SO101 on host A. +2. Discover and attach from host B. +3. Exchange one safe stationary action. +4. Disconnect host B and confirm host A remains available. +5. Stop host A and verify safe disconnect and lock release. +6. Restart under a service manager and verify signal-driven shutdown. + +## Acceptance Criteria + +- No tracked source contains the PoC syntax corruption or duplicate API + definitions. +- Explicit serve owns hardware in the foreground and cannot orphan a detached + persistent worker through its normal architecture. +- Auto-spawn remains compatible and exits after its existing idle timeout. +- Every lifecycle invariant in this document has an adjacent test. +- Remote serving remains explicit and documented as unauthenticated. +- The example contains no developer-specific paths and defaults to local-only. +- Design and user documentation agree on lifecycle and exit semantics. +- Focused tests, CLI smoke checks, type checking, and repository hooks pass: + +```bash +uv run pytest tests/unit/robot/transport tests/unit/cli +uv run physicalai --help +uv run physicalai robot --help +uv run physicalai robot serve --help +uv run physicalai robot discover --help +pyrefly check +prek run --all-files +``` diff --git a/docs/development/robot-zenoh-transport-design.md b/docs/development/robot-zenoh-transport-design.md new file mode 100644 index 00000000..af058f0f --- /dev/null +++ b/docs/development/robot-zenoh-transport-design.md @@ -0,0 +1,356 @@ +# Design: Zenoh Transport for Shared Robots + +**Status:** Implemented + +`physicalai.robot.transport` lets one process own a robot's hardware connection while +multiple processes read state and send actions over +[Zenoh](https://zenoh.io/). This is the canonical design for that transport. + +The design follows `SharedCamera`'s probe, spawn-or-attach structure, but uses Zenoh +instead of same-host shared memory. Robot state and action payloads are small and may +cross hosts; camera frames remain in `SharedCamera`. + +## Architecture + +One **owner runtime** constructs and connects the real robot driver. Auto-spawn runs +it in a detached child with a numeric idle timeout; `physicalai robot serve` runs it +in the foreground until interrupted. Both adapters use the same startup, loop, and +cleanup implementation. Each **subscriber** is a `SharedRobot` that pulls the latest +state and publishes absolute joint targets. + +```mermaid +flowchart LR + subgraph Owner["Owner process"] + HW["Robot driver"] <--> LOOP["Single write-first loop"] + end + subgraph Clients["Subscriber processes"] + A["SharedRobot A"] + B["SharedRobot B"] + end + + LOOP -- "publish" --> STATE["/{name}/state"] + ACTION["/{name}/action"] -- "latest action" --> LOOP + LOOP -. "answer queries" .-> META["/{name}/metadata"] + STATE --> A + STATE --> B + A --> ACTION + A -. "probe/validate" .-> META + B -. "probe/validate" .-> META +``` + +The full key prefix is `physicalai/robot`. Each robot has three keys: + +| Key | Pattern | Semantics | +| ---------------------------------- | --------- | -------------------------------------------------- | +| `physicalai/robot/{name}/state` | pub-sub | Owner to subscribers; continuous, latest-wins | +| `physicalai/robot/{name}/action` | pub-sub | Subscribers to owner; fire-and-forget, latest-wins | +| `physicalai/robot/{name}/metadata` | queryable | Discovery, compatibility, and liveness | + +`SharedRobot` satisfies the `Robot` protocol. Camera frames are deliberately excluded; +`SharedRobot.get_observation().images` is always `None`. + +## Public API and identity + +Normal construction means "create this named service if absent, then attach": + +```python +robot = SharedRobot( + name="left-arm", + robot_class=SO101, + robot_kwargs={ + "port": "/dev/ttyACM0", + "calibration": "calibration.json", + }, +) +``` + +Attach-only construction is explicit: + +```python +robot = SharedRobot.attach("left-arm") +``` + +Two identities serve different purposes: + +- `name` is the logical service identity used in Zenoh keys. It is required and must + be one non-empty segment containing only ASCII letters, digits, `_`, or `-`. +- `Robot.device_ids` identifies every physical resource a driver owns exclusively. + It is available before `connect()`, stable across equivalent construction, and used + only for host-local locking. `SharedRobot.device_ids` is empty because a subscriber + owns no hardware. + +Keeping these identities separate avoids guessing physical ownership from constructor +kwargs such as `port` or `ip`, and supports arbitrary and composite robots. + +### Arbitrary robot construction + +`robot_class` may be an importable class or dotted path. Class objects normalize to +`cls.__module__ + "." + cls.__qualname__`; explicit strings are preserved. The owner +subprocess imports the class and calls it with `robot_kwargs`. + +The private `RobotOwnerConfig` carries the normalized path and JSON-serializable +kwargs across the subprocess boundary. Local classes, lambdas, live instances, and +non-serializable kwargs cannot be auto-spawned. Class paths and kwargs are trusted +local application input. A class path received over Zenoh is never imported. + +If an owner already exists, the name is authoritative and the construction recipe is +unused. A caller-supplied class path that differs from metadata produces a warning, +not an error: re-exports, wrappers, and subclasses may preserve the wire contract. + +## Create-or-attach and ownership + +`SharedRobot.connect()` first probes the named `/metadata` key. It attaches when a +compatible owner answers; otherwise create-or-attach instances spawn an owner and +attach-only instances fail. + +```mermaid +sequenceDiagram + participant S as SharedRobot + participant Z as Zenoh metadata + participant O as Owner worker + participant L as Host-local locks + participant H as Hardware + + S->>Z: Query /metadata + alt Owner answers + Z-->>S: Metadata + S->>S: Validate protocol and shape + S->>Z: Subscribe to /state and publish to /action + else No owner + S->>O: Spawn with RobotOwnerConfig + O->>O: Construct driver and read device_ids + O->>L: Acquire name lock, then device locks + O->>H: connect() + O->>Z: Declare endpoints and /metadata + O-->>S: READY + S->>Z: Attach + end +``` + +The worker startup order is an invariant: + +```text +construct driver -> read device_ids -> acquire name lock +-> acquire sorted device locks -> connect -> declare endpoints -> READY +``` + +Construction may validate files or configuration but must not access hardware. +Hardware access starts only after every lock is held. + +### Host-local locking + +POSIX `flock` files in a private runtime directory provide crash-safe arbitration. +Linux uses `$XDG_RUNTIME_DIR/physicalai/robot-locks`; other Unix hosts fall back to +a per-user directory below the platform temporary directory. Lock state is kept out +of cache and durable application-storage directories: + +- one `name:{name}` lock serializes concurrent creation of the same service; +- one `device:{device_id}` lock protects each physical resource; +- lock filenames are SHA-256 hashes of the namespace and identity; +- diagnostic contents record lock kind, raw identity, owner name, and PID; +- device IDs are sorted and deduplicated before acquisition; +- partial acquisitions are released in reverse order; and +- lock files may remain after release because the open descriptor, not file + existence, represents ownership. + +The name lock is necessary because query-then-declare is not atomic and Zenoh permits +multiple entities on one key. A TCP bind failure is not the ownership arbiter. + +An empty `device_ids` tuple is valid only for a virtual robot with no exclusive +physical resource. Composite robots report all constituent resources. + +If two processes concurrently create the same name, the loser re-probes metadata. It +attaches if its resolved device IDs match the winner and raises `RobotNameConflict` if +they differ. A device already locked under another name raises +`RobotDeviceAlreadyOwned` before hardware connection. + +`flock` guarantees exclusivity only within one host. Deployments must designate one +owner host per network robot using topology, firewall policy, or operator +configuration. Distributed leases and hardware fencing are deferred. + +### Local discovery registry + +Local-only sessions cannot use multicast or gossip. Name-lock diagnostic files +therefore also provide candidate names for host-local discovery. Discovery filters +entries by live owner PID, then confirms each candidate through its deterministic +loopback `/metadata` queryable. Stale files are harmless. + +## Transport scope and rendezvous + +Local-only operation is the secure default: + +| `allow_remote` | Reachability | Security contract | +| -------------- | --------------------- | ------------------------------------------------------- | +| `False` | Same-host processes | No Zenoh transport exposed to the LAN | +| `True` | Reachable Zenoh peers | Trusted robot-cell network; deployer provides isolation | + +All sessions use peer mode. With `allow_remote=False`, multicast and gossip scouting +are disabled, the owner listens only on loopback, and subscribers connect only to the +same deterministic loopback endpoint. With `allow_remote=True`, the owner listens on +all interfaces and scouting is enabled. + +The owner and subscribers hash the full robot prefix into an unprivileged port in +`20000-59999`. A bind collision is a structured startup error that reports the exact +derived endpoint and tells the operator to choose another name or configure a local +Zenoh router. The owner does not probe alternate ports because independent subscribers +must derive the same endpoint. + +The caller that spawns an owner fixes its scope for the owner's lifetime. A later +attacher controls only its own session and cannot widen or narrow the running owner. +Remote attachment requires explicit `allow_remote=True`. + +Long-running remote clients can retain one scouting session through +`SharedRobotClient(allow_remote=True)`. The client is attach-only and owns +the session for its context lifetime: its first discovery includes Zenoh +route establishment, and later discovery or attachments reuse that warmed +session. `SharedRobotClient()` also supports same-host reuse with local-only +scope. With no explicit timeout, the first discovery gets a one-second budget +and later discovery gets 0.1 seconds. It disconnects all robots attached +through the client before closing the session. The wildcard discovery timeout +remains a collection window even for a warmed session because Zenoh cannot +signal that every reachable owner has replied. + +## Owner loop and subscriber behavior + +The owner is the only thread that touches the driver. Each tick is: + +1. Pull the newest `/action` sample without blocking and apply it if present. +2. Read the measured observation. +3. Publish `/state`. +4. Update subscriber-presence and idle-shutdown state. +5. Sleep until the next fixed-rate tick without accumulating scheduling debt. + +Write-first ordering minimizes action latency. With no new action, the owner sends +nothing and the servos hold their last target. Malformed or rejected actions are +logged and do not kill the shared owner. Five consecutive observation failures stop +the owner. + +The default rate is 100 Hz. A slower blocking driver naturally runs below that rate; +callers may provide a positive finite `rate_hz` when measurements justify an override. + +The `/state` subscriber uses native `RingChannel(1)` buffering and non-blocking +`try_recv()`. The ring remains active independently of Python callback scheduling, +keeps only the newest state, and avoids stale backlogs. `get_observation()` returns a +cached last-known state when no newer sample is waiting. The payload timestamp exposes +staleness. + +Actions are absolute joint targets, so dropping intermediate actions is safe. This +latest-wins design would not be valid for relative or delta actions. + +A subscriber's `disconnect()` closes only its own Zenoh session. The owner detects +zero state subscribers through publisher matching status, waits `idle_timeout`, calls +the underlying driver's `disconnect()` to satisfy the safe-state contract, and exits. +The explicit foreground command disables idle shutdown and remains available until +SIGINT or SIGTERM requests cleanup. Startup failures after hardware connection also +disconnect the driver before releasing locks. Repeated observation failures, +unexpected loop failures, and driver-disconnect failures produce a nonzero result. +Metadata undeclaration, session close, and explicit lock release remain best-effort +and do not replace the primary result. + +## Wire protocol + +Zenoh carries opaque bytes. State, action, and metadata use msgpack dictionaries; +NumPy arrays use `{dtype, shape, data}` records so dtype and shape round-trip exactly. +Images never enter this protocol. + +`/state` contains: + +```python +{ + "joint_positions": encode_numpy(obs.joint_positions), + "state": encode_numpy(obs.state), + "timestamp": obs.timestamp, + "sensor_data": {key: encode_numpy(value), ...} | None, +} +``` + +The owner ships its computed `obs.state`, not merely joint positions. This is required +because robot state is implementation-specific: SO101 uses joint positions, while +WidowXAI variants concatenate positions and velocities. Recomputing state in the +subscriber would duplicate driver logic and can silently change model input shape. + +`/action` contains the absolute action array, `goal_time`, and a monotonic send +timestamp. + +`/metadata` contains only public transport information: + +```python +{ + "protocol_version": 1, + "name": "left-arm", + "robot_class": "physicalai.robot.SO101", + "device_ids": ["serial:/dev/serial/by-id/usb-FTDI_1234"], + "host": "robot-cell-01", + "joint_names": [...], + "num_joints": 6, + "state_dim": 6, +} +``` + +Metadata excludes constructor kwargs, calibration paths or contents, credentials, +tokens, and arbitrary configuration. `device_ids` may expose serial numbers, paths, +or IP addresses; this is an intentional diagnostic choice within the transport's +trust boundary. + +`ROBOT_TRANSPORT_PROTOCOL_VERSION` versions the wire contract, not robot classes or +package releases. It changes for backward-incompatible payload, required-field, or +semantic changes, but not for additive optional fields or internal refactors. +Subscribers reject unsupported versions before declaring the action publisher. +Metadata validation also requires non-empty unique joint names, +`num_joints == len(joint_names)`, and positive `state_dim`. + +## QoS + +State and action publishers pin Zenoh's low-latency semantics explicitly: + +- `Reliability.BEST_EFFORT`: loss is acceptable under latest-wins semantics; +- `CongestionControl.DROP`: a full queue never blocks the control loop; and +- `express=True`: bypasses batching for small, high-rate messages. + +Functional tests do not detect batching regressions, so p99 action-latency jitter +should be measured at the target loop rate when changing Zenoh configuration. + +## Errors and startup handshake + +The parent sends JSON configuration over stdin. The worker emits exactly one `READY` +or `ERROR:{json}` line on stdout. Structured phases distinguish construction, +connection, lock contention, endpoint collision, timeout, and unexpected startup +failure without parsing human-readable messages. + +The subprocess worker is only a launch adapter over the in-process owner runtime. +`physicalai robot serve` calls that runtime directly, maps SIGINT and SIGTERM to its +shutdown event, and returns its structured exit code to the shell or service manager. + +The public hierarchy stays limited to conditions with different caller recovery: + +```text +RobotError +├── RobotNotConnectedError +└── RobotTransportError + ├── RobotNameConflict + ├── RobotDeviceAlreadyOwned + └── RobotProtocolMismatch +``` + +Other import, construction, connection, handshake, codec, timeout, and endpoint +failures remain `RobotTransportError` with structured phase details. + +## Security boundary + +In remote mode, any peer that can reach `/action` can move the physical robot. The +transport provides no application-level authentication or authorization. It must run +on a trusted robot-cell network isolated by VLAN/firewall policy or Zenoh ACL/TLS. + +Local-only mode avoids that exposure by disabling discovery and binding only to +loopback. Msgpack is used instead of `pickle`, so malformed transport payloads do not +introduce arbitrary code execution. Owner-advertised class paths are diagnostic strings +and are never imported. + +## Deferred work + +The following can be added without breaking the current protocol when a concrete need +appears: + +- action acknowledgement or blocking safety-command channels; +- router-enforced write authorization; +- distributed ownership leases with hardware fencing. 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..98122ce2 100644 --- a/docs/how-to/runtime/share-a-robot.md +++ b/docs/how-to/runtime/share-a-robot.md @@ -45,6 +45,29 @@ 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. 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. 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 +191,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..aa8928ab 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -28,6 +28,31 @@ 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. + +`--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 includes name, class, host, +and joint count. 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..6cae10e1 --- /dev/null +++ b/src/physicalai/cli/robot.py @@ -0,0 +1,184 @@ +# 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 +from typing import TYPE_CHECKING + +from jsonargparse import ActionConfigFile, ArgumentParser + +from physicalai.cli._spec import SubcommandSpec # noqa: PLC2701 +from physicalai.robot.errors import RobotTransportError +from physicalai.robot.transport import discover_robots +from physicalai.robot.transport._owner_config import DEFAULT_RATE_HZ, RobotOwnerConfig # noqa: PLC2701 +from physicalai.robot.transport._owner_worker import run_owner # noqa: PLC2701 + +if TYPE_CHECKING: + 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." +_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={}, + 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.") + 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 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. + """ + 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: + sys.stderr.write(f"Invalid robot configuration: {exc}\n") + return 1 + + shutdown = threading.Event() + + def _handle_signal(_signum: int, _frame: object) -> None: + shutdown.set() + + def _ready() -> None: + scope = "remote" if config.allow_remote else "local-only" + sys.stderr.write(f"Serving robot {config.name!r} ({scope}).\n") + sys.stderr.flush() + + previous_sigterm = signal.signal(signal.SIGTERM, _handle_signal) + previous_sigint = signal.signal(signal.SIGINT, _handle_signal) + try: + try: + return run_owner(config, shutdown, ready=_ready).exit_code + except RobotTransportError as exc: + phase = f" during {exc.phase}" if exc.phase else "" + sys.stderr.write(f"Failed to start robot owner{phase}: {exc}\n") + return 1 + 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 + + 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("No robots discovered.") # noqa: T201 + return 0 + for robot in robots: + print( # noqa: T201 + f"{robot.get('name', '?')}\t{robot.get('robot_class', '?')}\t" + f"host={robot.get('host', '?')}\tjoints={robot.get('num_joints', '?')}", + ) + 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/_owner.py b/src/physicalai/robot/transport/_owner.py index 5888c85c..024c3c1d 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,8 +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: - self._process = None + if proc is None: + return + + if proc.poll() is not None: + self._exit_code = proc.returncode return proc.terminate() @@ -141,7 +146,24 @@ def stop(self, timeout: float = 5.0) -> None: proc.kill() proc.wait(timeout=1) - self._process = None + self._exit_code = proc.returncode + + 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: diff --git a/src/physicalai/robot/transport/_owner_config.py b/src/physicalai/robot/transport/_owner_config.py index 66e0d863..c277116a 100644 --- a/src/physicalai/robot/transport/_owner_config.py +++ b/src/physicalai/robot/transport/_owner_config.py @@ -24,6 +24,7 @@ from __future__ import annotations import json +import math from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any @@ -95,7 +96,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 +105,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..443961a2 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,6 +54,7 @@ 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 @@ -59,6 +62,23 @@ 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 + + def sigterm_handler(_signum: int, _frame: FrameType | None) -> None: shutdown.set() @@ -171,9 +191,10 @@ def _run_loop( action_sub: Any, # noqa: ANN401 *, rate_hz: float, - idle_timeout: float, + idle_timeout: float | None, name: str, -) -> None: + shutdown_event: threading.Event, +) -> OwnerExitReason: """Single-threaded write-first owner loop. Ordering per tick: apply newest action (minimizes action latency), read @@ -189,13 +210,17 @@ 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. + + Returns: + The reason the loop stopped. """ period = 1.0 / rate_hz idle_since: float | None = None consecutive_failures = 0 next_tick = time.monotonic() - while not shutdown.is_set(): + while not shutdown_event.is_set(): sample = action_sub.try_recv() if sample is not None: try: @@ -212,7 +237,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 @@ -232,9 +257,9 @@ def _run_loop( 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 next_tick += period sleep_time = next_tick - time.monotonic() @@ -244,6 +269,8 @@ def _run_loop( # Fell behind (slow bus / blocking read); don't accumulate debt. next_tick = time.monotonic() + return OwnerExitReason.SHUTDOWN + @dataclass class _Endpoints: @@ -409,63 +436,101 @@ 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, +) -> 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. - 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, ) + 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}") + exit_code = 1 with contextlib.suppress(Exception): endpoints.metadata_queryable.undeclare() with contextlib.suppress(Exception): endpoints.session.close() - endpoints.locks.release_all() + with contextlib.suppress(Exception): + endpoints.locks.release_all() + + 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 + + saved_stdout_fd = suppress_stdout() + stdout_restored = False - return 0 + 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..bc04c6ee --- /dev/null +++ b/tests/unit/cli/test_robot_cli.py @@ -0,0 +1,173 @@ +# 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 physicalai.cli import robot as robot_module +from physicalai.robot.errors import RobotTransportError +from physicalai.robot.transport._lock import acquire_locks +from physicalai.robot.transport._owner_worker import OwnerExitReason, OwnerResult + +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, + } + 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) -> OwnerResult: + captured["config"] = config + captured["shutdown"] = shutdown + assert callable(ready) + 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] + + +def test_signal_requests_runtime_shutdown() -> None: + def _run_owner(_config: object, shutdown: threading.Event, *, ready: object) -> OwnerResult: # noqa: ARG001 + 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 + + +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_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) \ No newline at end of file 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..426d60d5 --- /dev/null +++ b/tests/unit/robot/transport/test_owner_worker.py @@ -0,0 +1,133 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import threading +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 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_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 \ No newline at end of file From 296a6f63d56265df2d3edd081625c6dd1356e5cf Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Mon, 20 Jul 2026 16:51:49 +0200 Subject: [PATCH 2/8] remove design artifact --- .../robot-zenoh-transport-design.md | 356 ------------------ 1 file changed, 356 deletions(-) delete mode 100644 docs/development/robot-zenoh-transport-design.md diff --git a/docs/development/robot-zenoh-transport-design.md b/docs/development/robot-zenoh-transport-design.md deleted file mode 100644 index af058f0f..00000000 --- a/docs/development/robot-zenoh-transport-design.md +++ /dev/null @@ -1,356 +0,0 @@ -# Design: Zenoh Transport for Shared Robots - -**Status:** Implemented - -`physicalai.robot.transport` lets one process own a robot's hardware connection while -multiple processes read state and send actions over -[Zenoh](https://zenoh.io/). This is the canonical design for that transport. - -The design follows `SharedCamera`'s probe, spawn-or-attach structure, but uses Zenoh -instead of same-host shared memory. Robot state and action payloads are small and may -cross hosts; camera frames remain in `SharedCamera`. - -## Architecture - -One **owner runtime** constructs and connects the real robot driver. Auto-spawn runs -it in a detached child with a numeric idle timeout; `physicalai robot serve` runs it -in the foreground until interrupted. Both adapters use the same startup, loop, and -cleanup implementation. Each **subscriber** is a `SharedRobot` that pulls the latest -state and publishes absolute joint targets. - -```mermaid -flowchart LR - subgraph Owner["Owner process"] - HW["Robot driver"] <--> LOOP["Single write-first loop"] - end - subgraph Clients["Subscriber processes"] - A["SharedRobot A"] - B["SharedRobot B"] - end - - LOOP -- "publish" --> STATE["/{name}/state"] - ACTION["/{name}/action"] -- "latest action" --> LOOP - LOOP -. "answer queries" .-> META["/{name}/metadata"] - STATE --> A - STATE --> B - A --> ACTION - A -. "probe/validate" .-> META - B -. "probe/validate" .-> META -``` - -The full key prefix is `physicalai/robot`. Each robot has three keys: - -| Key | Pattern | Semantics | -| ---------------------------------- | --------- | -------------------------------------------------- | -| `physicalai/robot/{name}/state` | pub-sub | Owner to subscribers; continuous, latest-wins | -| `physicalai/robot/{name}/action` | pub-sub | Subscribers to owner; fire-and-forget, latest-wins | -| `physicalai/robot/{name}/metadata` | queryable | Discovery, compatibility, and liveness | - -`SharedRobot` satisfies the `Robot` protocol. Camera frames are deliberately excluded; -`SharedRobot.get_observation().images` is always `None`. - -## Public API and identity - -Normal construction means "create this named service if absent, then attach": - -```python -robot = SharedRobot( - name="left-arm", - robot_class=SO101, - robot_kwargs={ - "port": "/dev/ttyACM0", - "calibration": "calibration.json", - }, -) -``` - -Attach-only construction is explicit: - -```python -robot = SharedRobot.attach("left-arm") -``` - -Two identities serve different purposes: - -- `name` is the logical service identity used in Zenoh keys. It is required and must - be one non-empty segment containing only ASCII letters, digits, `_`, or `-`. -- `Robot.device_ids` identifies every physical resource a driver owns exclusively. - It is available before `connect()`, stable across equivalent construction, and used - only for host-local locking. `SharedRobot.device_ids` is empty because a subscriber - owns no hardware. - -Keeping these identities separate avoids guessing physical ownership from constructor -kwargs such as `port` or `ip`, and supports arbitrary and composite robots. - -### Arbitrary robot construction - -`robot_class` may be an importable class or dotted path. Class objects normalize to -`cls.__module__ + "." + cls.__qualname__`; explicit strings are preserved. The owner -subprocess imports the class and calls it with `robot_kwargs`. - -The private `RobotOwnerConfig` carries the normalized path and JSON-serializable -kwargs across the subprocess boundary. Local classes, lambdas, live instances, and -non-serializable kwargs cannot be auto-spawned. Class paths and kwargs are trusted -local application input. A class path received over Zenoh is never imported. - -If an owner already exists, the name is authoritative and the construction recipe is -unused. A caller-supplied class path that differs from metadata produces a warning, -not an error: re-exports, wrappers, and subclasses may preserve the wire contract. - -## Create-or-attach and ownership - -`SharedRobot.connect()` first probes the named `/metadata` key. It attaches when a -compatible owner answers; otherwise create-or-attach instances spawn an owner and -attach-only instances fail. - -```mermaid -sequenceDiagram - participant S as SharedRobot - participant Z as Zenoh metadata - participant O as Owner worker - participant L as Host-local locks - participant H as Hardware - - S->>Z: Query /metadata - alt Owner answers - Z-->>S: Metadata - S->>S: Validate protocol and shape - S->>Z: Subscribe to /state and publish to /action - else No owner - S->>O: Spawn with RobotOwnerConfig - O->>O: Construct driver and read device_ids - O->>L: Acquire name lock, then device locks - O->>H: connect() - O->>Z: Declare endpoints and /metadata - O-->>S: READY - S->>Z: Attach - end -``` - -The worker startup order is an invariant: - -```text -construct driver -> read device_ids -> acquire name lock --> acquire sorted device locks -> connect -> declare endpoints -> READY -``` - -Construction may validate files or configuration but must not access hardware. -Hardware access starts only after every lock is held. - -### Host-local locking - -POSIX `flock` files in a private runtime directory provide crash-safe arbitration. -Linux uses `$XDG_RUNTIME_DIR/physicalai/robot-locks`; other Unix hosts fall back to -a per-user directory below the platform temporary directory. Lock state is kept out -of cache and durable application-storage directories: - -- one `name:{name}` lock serializes concurrent creation of the same service; -- one `device:{device_id}` lock protects each physical resource; -- lock filenames are SHA-256 hashes of the namespace and identity; -- diagnostic contents record lock kind, raw identity, owner name, and PID; -- device IDs are sorted and deduplicated before acquisition; -- partial acquisitions are released in reverse order; and -- lock files may remain after release because the open descriptor, not file - existence, represents ownership. - -The name lock is necessary because query-then-declare is not atomic and Zenoh permits -multiple entities on one key. A TCP bind failure is not the ownership arbiter. - -An empty `device_ids` tuple is valid only for a virtual robot with no exclusive -physical resource. Composite robots report all constituent resources. - -If two processes concurrently create the same name, the loser re-probes metadata. It -attaches if its resolved device IDs match the winner and raises `RobotNameConflict` if -they differ. A device already locked under another name raises -`RobotDeviceAlreadyOwned` before hardware connection. - -`flock` guarantees exclusivity only within one host. Deployments must designate one -owner host per network robot using topology, firewall policy, or operator -configuration. Distributed leases and hardware fencing are deferred. - -### Local discovery registry - -Local-only sessions cannot use multicast or gossip. Name-lock diagnostic files -therefore also provide candidate names for host-local discovery. Discovery filters -entries by live owner PID, then confirms each candidate through its deterministic -loopback `/metadata` queryable. Stale files are harmless. - -## Transport scope and rendezvous - -Local-only operation is the secure default: - -| `allow_remote` | Reachability | Security contract | -| -------------- | --------------------- | ------------------------------------------------------- | -| `False` | Same-host processes | No Zenoh transport exposed to the LAN | -| `True` | Reachable Zenoh peers | Trusted robot-cell network; deployer provides isolation | - -All sessions use peer mode. With `allow_remote=False`, multicast and gossip scouting -are disabled, the owner listens only on loopback, and subscribers connect only to the -same deterministic loopback endpoint. With `allow_remote=True`, the owner listens on -all interfaces and scouting is enabled. - -The owner and subscribers hash the full robot prefix into an unprivileged port in -`20000-59999`. A bind collision is a structured startup error that reports the exact -derived endpoint and tells the operator to choose another name or configure a local -Zenoh router. The owner does not probe alternate ports because independent subscribers -must derive the same endpoint. - -The caller that spawns an owner fixes its scope for the owner's lifetime. A later -attacher controls only its own session and cannot widen or narrow the running owner. -Remote attachment requires explicit `allow_remote=True`. - -Long-running remote clients can retain one scouting session through -`SharedRobotClient(allow_remote=True)`. The client is attach-only and owns -the session for its context lifetime: its first discovery includes Zenoh -route establishment, and later discovery or attachments reuse that warmed -session. `SharedRobotClient()` also supports same-host reuse with local-only -scope. With no explicit timeout, the first discovery gets a one-second budget -and later discovery gets 0.1 seconds. It disconnects all robots attached -through the client before closing the session. The wildcard discovery timeout -remains a collection window even for a warmed session because Zenoh cannot -signal that every reachable owner has replied. - -## Owner loop and subscriber behavior - -The owner is the only thread that touches the driver. Each tick is: - -1. Pull the newest `/action` sample without blocking and apply it if present. -2. Read the measured observation. -3. Publish `/state`. -4. Update subscriber-presence and idle-shutdown state. -5. Sleep until the next fixed-rate tick without accumulating scheduling debt. - -Write-first ordering minimizes action latency. With no new action, the owner sends -nothing and the servos hold their last target. Malformed or rejected actions are -logged and do not kill the shared owner. Five consecutive observation failures stop -the owner. - -The default rate is 100 Hz. A slower blocking driver naturally runs below that rate; -callers may provide a positive finite `rate_hz` when measurements justify an override. - -The `/state` subscriber uses native `RingChannel(1)` buffering and non-blocking -`try_recv()`. The ring remains active independently of Python callback scheduling, -keeps only the newest state, and avoids stale backlogs. `get_observation()` returns a -cached last-known state when no newer sample is waiting. The payload timestamp exposes -staleness. - -Actions are absolute joint targets, so dropping intermediate actions is safe. This -latest-wins design would not be valid for relative or delta actions. - -A subscriber's `disconnect()` closes only its own Zenoh session. The owner detects -zero state subscribers through publisher matching status, waits `idle_timeout`, calls -the underlying driver's `disconnect()` to satisfy the safe-state contract, and exits. -The explicit foreground command disables idle shutdown and remains available until -SIGINT or SIGTERM requests cleanup. Startup failures after hardware connection also -disconnect the driver before releasing locks. Repeated observation failures, -unexpected loop failures, and driver-disconnect failures produce a nonzero result. -Metadata undeclaration, session close, and explicit lock release remain best-effort -and do not replace the primary result. - -## Wire protocol - -Zenoh carries opaque bytes. State, action, and metadata use msgpack dictionaries; -NumPy arrays use `{dtype, shape, data}` records so dtype and shape round-trip exactly. -Images never enter this protocol. - -`/state` contains: - -```python -{ - "joint_positions": encode_numpy(obs.joint_positions), - "state": encode_numpy(obs.state), - "timestamp": obs.timestamp, - "sensor_data": {key: encode_numpy(value), ...} | None, -} -``` - -The owner ships its computed `obs.state`, not merely joint positions. This is required -because robot state is implementation-specific: SO101 uses joint positions, while -WidowXAI variants concatenate positions and velocities. Recomputing state in the -subscriber would duplicate driver logic and can silently change model input shape. - -`/action` contains the absolute action array, `goal_time`, and a monotonic send -timestamp. - -`/metadata` contains only public transport information: - -```python -{ - "protocol_version": 1, - "name": "left-arm", - "robot_class": "physicalai.robot.SO101", - "device_ids": ["serial:/dev/serial/by-id/usb-FTDI_1234"], - "host": "robot-cell-01", - "joint_names": [...], - "num_joints": 6, - "state_dim": 6, -} -``` - -Metadata excludes constructor kwargs, calibration paths or contents, credentials, -tokens, and arbitrary configuration. `device_ids` may expose serial numbers, paths, -or IP addresses; this is an intentional diagnostic choice within the transport's -trust boundary. - -`ROBOT_TRANSPORT_PROTOCOL_VERSION` versions the wire contract, not robot classes or -package releases. It changes for backward-incompatible payload, required-field, or -semantic changes, but not for additive optional fields or internal refactors. -Subscribers reject unsupported versions before declaring the action publisher. -Metadata validation also requires non-empty unique joint names, -`num_joints == len(joint_names)`, and positive `state_dim`. - -## QoS - -State and action publishers pin Zenoh's low-latency semantics explicitly: - -- `Reliability.BEST_EFFORT`: loss is acceptable under latest-wins semantics; -- `CongestionControl.DROP`: a full queue never blocks the control loop; and -- `express=True`: bypasses batching for small, high-rate messages. - -Functional tests do not detect batching regressions, so p99 action-latency jitter -should be measured at the target loop rate when changing Zenoh configuration. - -## Errors and startup handshake - -The parent sends JSON configuration over stdin. The worker emits exactly one `READY` -or `ERROR:{json}` line on stdout. Structured phases distinguish construction, -connection, lock contention, endpoint collision, timeout, and unexpected startup -failure without parsing human-readable messages. - -The subprocess worker is only a launch adapter over the in-process owner runtime. -`physicalai robot serve` calls that runtime directly, maps SIGINT and SIGTERM to its -shutdown event, and returns its structured exit code to the shell or service manager. - -The public hierarchy stays limited to conditions with different caller recovery: - -```text -RobotError -├── RobotNotConnectedError -└── RobotTransportError - ├── RobotNameConflict - ├── RobotDeviceAlreadyOwned - └── RobotProtocolMismatch -``` - -Other import, construction, connection, handshake, codec, timeout, and endpoint -failures remain `RobotTransportError` with structured phase details. - -## Security boundary - -In remote mode, any peer that can reach `/action` can move the physical robot. The -transport provides no application-level authentication or authorization. It must run -on a trusted robot-cell network isolated by VLAN/firewall policy or Zenoh ACL/TLS. - -Local-only mode avoids that exposure by disabling discovery and binding only to -loopback. Msgpack is used instead of `pickle`, so malformed transport payloads do not -introduce arbitrary code execution. Owner-advertised class paths are diagnostic strings -and are never imported. - -## Deferred work - -The following can be added without breaking the current protocol when a concrete need -appears: - -- action acknowledgement or blocking safety-command channels; -- router-enforced write authorization; -- distributed ownership leases with hardware fencing. From 32803663f785daa73368dad36427b52a873597ac Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Mon, 20 Jul 2026 22:44:48 +0200 Subject: [PATCH 3/8] improve stdout/logs --- docs/how-to/runtime/share-a-robot.md | 11 +- docs/reference/cli.md | 10 +- src/physicalai/cli/robot.py | 120 +++++++++++++++--- src/physicalai/robot/transport/__init__.py | 14 +- .../robot/transport/_owner_worker.py | 71 +++++++++-- tests/unit/cli/test_robot_cli.py | 81 +++++++++++- .../unit/robot/transport/test_owner_worker.py | 38 +++++- 7 files changed, 306 insertions(+), 39 deletions(-) diff --git a/docs/how-to/runtime/share-a-robot.md b/docs/how-to/runtime/share-a-robot.md index 98122ce2..fa731fb8 100644 --- a/docs/how-to/runtime/share-a-robot.md +++ b/docs/how-to/runtime/share-a-robot.md @@ -54,8 +54,10 @@ robot lifecycle: physicalai robot serve --config examples/so101/serve.yaml ``` -The command constructs and connects the driver in its own foreground process. The command does not daemonize; use your service manager for -background supervision. +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: @@ -65,8 +67,9 @@ physicalai robot discover --json physicalai robot discover --allow_remote ``` -Discovery is local-only unless `--allow_remote` is explicit. JSON mode writes one -sorted array to stdout, including `[]` when no robot answers. +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: diff --git a/docs/reference/cli.md b/docs/reference/cli.md index aa8928ab..65af4c6c 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -40,6 +40,9 @@ Direct CLI arguments use the same names, including nested constructor values suc `--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. @@ -49,9 +52,10 @@ an isolated robot-cell VLAN/firewall or with Zenoh ACL/TLS. physicalai robot discover [--allow_remote] [--timeout 2] [--json] ``` -Results are sorted by robot name and host. Human output includes name, class, host, -and joint count. JSON mode writes exactly one array to stdout; empty discovery is -successful and writes `[]`. +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 diff --git a/src/physicalai/cli/robot.py b/src/physicalai/cli/robot.py index 6cae10e1..702934f6 100644 --- a/src/physicalai/cli/robot.py +++ b/src/physicalai/cli/robot.py @@ -10,17 +10,25 @@ 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 discover_robots -from physicalai.robot.transport._owner_config import DEFAULT_RATE_HZ, RobotOwnerConfig # noqa: PLC2701 -from physicalai.robot.transport._owner_worker import run_owner # noqa: PLC2701 +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." @@ -59,6 +67,7 @@ def _build_serve_parser() -> ArgumentParser: ), ) 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 @@ -95,12 +104,55 @@ def print_help(prog: str) -> None: ) +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 _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. """ + _configure_serve_logging(verbose=cfg.verbose) try: config = RobotOwnerConfig( name=cfg.name, @@ -111,28 +163,55 @@ def serve(cfg: Namespace) -> int: idle_timeout=None, ) except (TypeError, ValueError) as exc: - sys.stderr.write(f"Invalid robot configuration: {exc}\n") + 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: + def _handle_signal(signum: int, _frame: object) -> None: + nonlocal received_signum + received_signum = signum shutdown.set() def _ready() -> None: - scope = "remote" if config.allow_remote else "local-only" - sys.stderr.write(f"Serving robot {config.name!r} ({scope}).\n") - sys.stderr.flush() - + 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}") + + logger.info(f"Starting robot {config.name!r} using {config.robot_class}") previous_sigterm = signal.signal(signal.SIGTERM, _handle_signal) previous_sigint = signal.signal(signal.SIGINT, _handle_signal) try: try: - return run_owner(config, shutdown, ready=_ready).exit_code + result = run_owner(config, shutdown, ready=_ready, on_event=_on_event) except RobotTransportError as exc: phase = f" during {exc.phase}" if exc.phase else "" - sys.stderr.write(f"Failed to start robot owner{phase}: {exc}\n") + 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) @@ -148,6 +227,7 @@ def discover(cfg: Namespace) -> int: 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", ""))), @@ -156,13 +236,21 @@ def discover(cfg: Namespace) -> int: sys.stdout.write(json.dumps(robots) + "\n") return 0 if not robots: - print("No robots discovered.") # noqa: T201 + print(f"No robots discovered in {time.monotonic() - started_at:.1f}s.") # noqa: T201 return 0 - for robot in robots: - print( # noqa: T201 - f"{robot.get('name', '?')}\t{robot.get('robot_class', '?')}\t" - f"host={robot.get('host', '?')}\tjoints={robot.get('num_joints', '?')}", + 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 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/_owner_worker.py b/src/physicalai/robot/transport/_owner_worker.py index 443961a2..f98cec15 100644 --- a/src/physicalai/robot/transport/_owner_worker.py +++ b/src/physicalai/robot/transport/_owner_worker.py @@ -58,6 +58,7 @@ from types import FrameType _MAX_CONSECUTIVE_FAILURES = 5 +_HEARTBEAT_INTERVAL_S = 30.0 shutdown = threading.Event() @@ -79,6 +80,14 @@ class OwnerResult: 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() @@ -185,6 +194,18 @@ 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}", exc_info=True) + + def _run_loop( driver: Robot, state_pub: Any, # noqa: ANN401 @@ -194,6 +215,8 @@ def _run_loop( idle_timeout: float | None, name: str, 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. @@ -211,6 +234,8 @@ def _run_loop( 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. @@ -219,17 +244,19 @@ def _run_loop( idle_since: float | None = None consecutive_failures = 0 next_tick = time.monotonic() + next_heartbeat = next_tick + heartbeat_interval_s + subscribers_present = False + + 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(): - 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) + _apply_pending_action(driver, action_sub, name) try: obs = driver.get_observation() @@ -253,7 +280,15 @@ 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 @@ -261,6 +296,10 @@ def _run_loop( logger.info(f"No subscribers for {idle_timeout}s -- shutting down owner {name}") 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() if sleep_time > 0: @@ -299,12 +338,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) @@ -356,6 +397,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): @@ -394,6 +436,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}" @@ -404,6 +447,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) @@ -441,6 +485,7 @@ def run_owner( 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. @@ -448,6 +493,7 @@ def run_owner( 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. Returns: Structured termination reason and process-compatible exit code. @@ -472,6 +518,7 @@ def run_owner( 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 @@ -480,15 +527,19 @@ def run_owner( shutdown_event.set() try: endpoints.driver.disconnect() + logger.trace(f"Disconnected robot driver for {config.name!r}") except Exception: # noqa: BLE001 logger.exception(f"driver disconnect failed for {config.name}") 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() + 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) diff --git a/tests/unit/cli/test_robot_cli.py b/tests/unit/cli/test_robot_cli.py index bc04c6ee..ebcceb73 100644 --- a/tests/unit/cli/test_robot_cli.py +++ b/tests/unit/cli/test_robot_cli.py @@ -15,10 +15,12 @@ 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 physicalai.robot.transport._owner_worker import OwnerExitReason, OwnerResult from tests.unit.robot.transport.conftest import requires_zenoh @@ -30,6 +32,7 @@ def _serve_cfg(**overrides: object) -> SimpleNamespace: "robot_kwargs": {"port": "/dev/fake0"}, "allow_remote": False, "rate_hz": 100.0, + "verbose": False, } values.update(overrides) return SimpleNamespace(**values) @@ -38,10 +41,11 @@ def _serve_cfg(**overrides: object) -> SimpleNamespace: 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) -> OwnerResult: + 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) @@ -55,14 +59,21 @@ def _run_owner(config: object, shutdown: threading.Event, *, ready: object) -> O assert config.name == "left-arm" # type: ignore[attr-defined] -def test_signal_requests_runtime_shutdown() -> None: - def _run_owner(_config: object, shutdown: threading.Event, *, ready: object) -> OwnerResult: # noqa: ARG001 +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: @@ -96,6 +107,68 @@ def test_discovery_json_is_sorted_and_clean(capsys: object) -> None: 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=[]): diff --git a/tests/unit/robot/transport/test_owner_worker.py b/tests/unit/robot/transport/test_owner_worker.py index 426d60d5..95bdf09d 100644 --- a/tests/unit/robot/transport/test_owner_worker.py +++ b/tests/unit/robot/transport/test_owner_worker.py @@ -4,13 +4,14 @@ 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 OwnerExitReason, _run_loop, run_owner +from physicalai.robot.transport._owner_worker import OwnerEvent, OwnerExitReason, _run_loop, run_owner from .fake import FakeRobot @@ -88,6 +89,41 @@ def test_repeated_reads_return_failure() -> None: 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() From 3baf7ef6470c1aebf12490ae437980d18d4325fc Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Mon, 20 Jul 2026 22:51:20 +0200 Subject: [PATCH 4/8] remove plan --- docs/development/robot-serve-plan.md | 462 --------------------------- 1 file changed, 462 deletions(-) delete mode 100644 docs/development/robot-serve-plan.md diff --git a/docs/development/robot-serve-plan.md b/docs/development/robot-serve-plan.md deleted file mode 100644 index f4ec6cc9..00000000 --- a/docs/development/robot-serve-plan.md +++ /dev/null @@ -1,462 +0,0 @@ -# Robot Serve design - -## Status - -Implemented production design. This document replaces the proof-of-concept plan -implemented in commit `2ed63a6ea2a53ea6548a149cc4a82cb1022ce64c`. - -The PoC established that a robot can be exposed through the existing Zenoh -`SharedRobot` transport and consumed from another process or host. The -production implementation must now make foreground ownership, shutdown, -failure reporting, and CLI boundaries explicit. - -## Decision - -Add these operator commands: - -```text -physicalai robot serve -physicalai robot discover -``` - -`physicalai robot serve` runs the hardware owner in the foreground process. It -does not launch a detached persistent child. Foreground serving and the -existing auto-spawn worker call one shared owner runtime that exclusively -constructs, connects, reads, commands, and disconnects the robot. - -The auto-spawn path remains detached and retains its numeric idle timeout. The -explicit CLI path is persistent until interrupted and remains bound to the -lifetime of the foreground process. - -## Goals - -- Provide a convenient YAML- and argument-driven command for serving one robot. -- Reuse the proven `SharedRobot` Zenoh protocol and hardware ownership model. -- Keep exactly one process responsible for each live robot driver. -- Guarantee that normal CLI termination attempts a safe driver disconnect. -- Report startup, loop, forced-termination, and disconnect failures as nonzero. -- Preserve secure local-only networking unless remote access is explicit. -- Keep auto-spawn behavior compatible with existing `SharedRobot` callers. -- Expose deterministic human-readable and machine-readable discovery. - -## Non-goals - -- Authentication, authorization, or encryption in the robot transport. -- Cross-host robot-name arbitration or distributed ownership leases. -- Multiple-controller action arbitration. -- Daemonization, PID files, or a background-service manager in the CLI. -- Zenoh router provisioning. -- Driver-specific CLI arguments. -- Reworking the robot wire protocol. -- Detailed nested shell completion. - -Operating-system service managers such as systemd, Docker, or Kubernetes own -background supervision. The CLI remains a foreground command suitable for -those supervisors. - -## Why the PoC Architecture Must Change - -The PoC reused `RobotOwner`, which always starts its worker with -`start_new_session=True`, and made that detached worker persistent by setting -`idle_timeout=None`. This combines two individually useful behaviors into an -unsafe lifecycle: - -1. The CLI process starts a detached owner. -2. The owner has no idle self-termination. -3. If the CLI is killed or crashes before signaling the owner, the owner can - continue indefinitely while retaining the hardware and ownership locks. - -That violates the operator-facing contract that the robot is served for the -lifetime of `physicalai robot serve`. - -The PoC also made the CLI depend on private transport validators and mixed -explicit-host networking changes into the serve feature. Those concerns should -be separated so each can be reviewed against one contract. - -## Architecture - -### Shared owner runtime - -Extract the hardware-owning behavior from the subprocess entry point into one -internal, in-process runtime: - -```python -class OwnerExitReason(enum.Enum): - SHUTDOWN = "shutdown" - IDLE_TIMEOUT = "idle_timeout" - CONSECUTIVE_READ_FAILURES = "consecutive_read_failures" - LOOP_FAILURE = "loop_failure" - - -@dataclass(frozen=True) -class OwnerResult: - reason: OwnerExitReason - exit_code: int - - -def run_owner( - config: RobotOwnerConfig, - shutdown: threading.Event, - *, - ready: Callable[[], None] | None = None, -) -> OwnerResult: - ... -``` - -The exact names and callback shape are implementation details. The contract is -that this runtime owns the complete lifecycle: - -```text -validate config - -> import and construct driver - -> acquire name and device locks - -> connect driver - -> read initial observation - -> declare Zenoh endpoints - -> report ready - -> run control loop - -> disconnect driver - -> undeclare/close/release best-effort resources - -> return structured result -``` - -No launch adapter duplicates hardware construction, loop, or cleanup logic. - -### Launch adapters - -Two adapters invoke the same runtime with different lifecycle policies: - -| Concern | Auto-spawn worker | `robot serve` | -| -------------------- | -------------------------------- | ---------------------------- | -| Process | Dedicated child | Foreground CLI process | -| Detachment | New session | None | -| Idle timeout | Numeric, default 10 seconds | Disabled (`None`) | -| Readiness | `READY`/`ERROR` IPC | Operator message after ready | -| Shutdown source | SIGTERM or idle timeout | SIGINT/SIGTERM | -| Parent death | Idle timeout eventually stops it | Process itself is gone | -| Exit status consumer | `RobotOwner` parent | Shell/service manager | - -The subprocess module remains a thin adapter around `run_owner` for the -existing auto-spawn handshake. The CLI adapter installs signal handlers, calls -`run_owner` directly, and maps `OwnerResult` to a process exit code. - -### Responsibility boundaries - -`physicalai.robot.transport` owns: - -- Owner configuration and validation. -- Driver import and construction. -- Host-local name and device locks. -- Zenoh endpoints and payloads. -- The control loop and its termination reasons. -- Safe driver disconnect and cleanup ordering. -- Discovery of metadata records. - -`physicalai.cli.robot` owns: - -- Command-line and config-file parsing. -- Direct group and nested command help. -- Signal-to-shutdown-event adaptation for foreground serving. -- Operator-facing output and exit-code presentation. -- Sorting and formatting discovery results. - -The CLI must not import private field validators or reproduce transport -validation. It constructs one supported owner configuration and catches the -documented validation and transport errors at the command boundary. - -## Lifecycle Invariants - -1. Only the owner runtime imports, constructs, connects, reads, commands, and - disconnects a concrete robot driver. -2. Locks are acquired before `driver.connect()` performs hardware access. -3. Readiness is reported only after the driver is connected, an initial valid - observation has been read, and all Zenoh endpoints are declared. -4. A foreground server never leaves a persistent detached worker by design. -5. SIGINT and SIGTERM request graceful shutdown; they do not bypass cleanup. -6. Driver disconnect is attempted after every post-connect exit path. -7. Driver-disconnect failure is safety-critical and forces a nonzero result. -8. Metadata undeclaration, Zenoh session close, and explicit lock release are - best-effort. Their failures are logged and do not replace the primary result. -9. Repeated observation failures and unexpected loop failures are nonzero. -10. Numeric idle timeout and requested shutdown succeed only when safe driver - disconnect also succeeds. -11. Subscriber disconnect never directly stops the robot; owner policy controls - the safe-state transition. - -`SIGKILL`, kernel failure, and power loss cannot run Python cleanup. Drivers and -hardware must fail safely where possible, but that is outside this command's -guarantees. - -## Configuration - -Use the existing flat owner shape rather than introducing another -`class_path`/`init_args` representation: - -```yaml -name: follower-arm -robot_class: physicalai.robot.SO101 -robot_kwargs: - port: /dev/ttyACM0 - calibration: /path/to/calibration.json - role: follower -allow_remote: false -rate_hz: 100.0 -``` - -| Field | Required | Meaning | -| -------------- | -------- | ----------------------------------------------- | -| `name` | Yes | Logical name and Zenoh key segment | -| `robot_class` | Yes | Trusted local dotted path to a driver class | -| `robot_kwargs` | No | JSON-serializable constructor keyword arguments | -| `allow_remote` | No | Expose beyond localhost; default `false` | -| `rate_hz` | No | Positive finite owner loop rate; default 100 Hz | - -`idle_timeout` is not exposed by `robot serve`. Explicit serving is persistent -until interrupted. Auto-spawn continues to supply its existing numeric timeout. - -Validate all fields before constructing hardware: - -- `name` is one nonempty safe Zenoh key segment. -- `robot_class` is a nonempty dotted path. Import occurs only in the owner. -- `robot_kwargs` is JSON-serializable across the worker IPC boundary. -- `rate_hz` is finite and greater than zero; booleans are rejected. - -The class path is trusted local configuration. Network metadata must never be -used as a class path or imported by discovery. - -## CLI Design - -### Serve - -```bash -physicalai robot serve --config examples/so101/serve.yaml -``` - -Direct arguments remain available for scripting: - -```bash -physicalai robot serve \ - --name follower-arm \ - --robot_class physicalai.robot.SO101 \ - --robot_kwargs.port /dev/ttyACM0 \ - --robot_kwargs.calibration /path/to/calibration.json \ - --allow_remote -``` - -`--allow_remote` is a bare `store_true` switch. Local-only operation is the -default. A readiness message goes to stderr so stdout remains available for -future structured output. - -| Condition | Exit | -| ------------------------------------------------ | ------- | -| SIGINT/SIGTERM and successful disconnect | 0 | -| Invalid config | Nonzero | -| Driver construction/connect/initial-read failure | Nonzero | -| Name or device lock contention | Nonzero | -| Repeated read failure | Nonzero | -| Unexpected control-loop failure | Nonzero | -| Driver disconnect failure | Nonzero | - -Errors identify the phase and an operator recovery action where known. Expected -operational errors do not print tracebacks by default. Detailed tracebacks -remain available through debug logging. - -### Discover - -```bash -physicalai robot discover -physicalai robot discover --allow_remote -physicalai robot discover --json -``` - -Discovery delegates to the public transport discovery function, sorts records -by `(name, host)`, and never imports advertised `robot_class` values. - -Human output includes at least name, robot class, host, and joint count. JSON -mode writes exactly one array to stdout. Logs and warnings go to stderr. Empty -discovery is successful and produces `[]` in JSON mode. - -Targeted `--host`/`--name` discovery is not required for the serve redesign. If -retained for unicast-only networks, implement and review it as a separate -transport/API change. User-provided endpoints must use a structured serializer -such as `json.dumps`, not interpolation into JSON5. - -### Help routing - -Register `robot` as one built-in command group. Direct group help can use the -lightweight path: - -```text -physicalai robot --help -``` - -Nested help must reach the real nested parser: - -```text -physicalai robot serve --help -physicalai robot discover --help -``` - -Prefer generic host routing based on direct help versus nested arguments rather -than a robot-specific nested-command registry. Existing plugin help behavior -must remain unchanged and covered by regression tests. - -## Security - -`allow_remote=false` disables remote scouting and binds the owner to loopback. -This is the default for both configuration and CLI arguments. - -`allow_remote=true` exposes an unauthenticated action endpoint. Any reachable -peer can command the physical robot. Documentation and command help must state -that remote serving requires an isolated robot-cell VLAN/firewall or configured -Zenoh ACL/TLS. - -Discovery metadata excludes constructor kwargs, calibration paths, device -paths, and other construction secrets. Physical device identifiers remain -host-local diagnostics and are not advertised remotely. - -Robot names are operator-enforced as unique across the reachable Zenoh network. -Host-local locks do not provide cross-host arbitration. - -## Public API - -Do not add a public `RobotServer` class for this change. There is one concrete -consumer of foreground serving, and the CLI can call a narrow internal owner -runtime. Adding a public lifecycle abstraction now would freeze semantics that -have not demonstrated a second use case. - -Keep `SharedRobot`, `SharedRobotClient`, and `discover_robots` as the user-facing -transport APIs. The owner runtime and owner configuration may remain internal. -Reconsider a public serving API when a second concrete embedding use case -requires it. - -## Migration From the PoC - -Reimplement from the parent of commit -`2ed63a6ea2a53ea6548a149cc4a82cb1022ce64c` rather than repairing that commit in -place. Carry behavior and tests selectively. - -Retain: - -- The `physicalai robot serve` and `physicalai robot discover` experience. -- Flat configuration fields. -- Persistent explicit serve and numeric auto-spawn timeout. -- Structured termination reasons and exit-code requirements. -- Sorted human/JSON discovery behavior. -- Local-only defaults and the remote-network warning. -- Tests that express these contracts. - -Redesign: - -- Run explicit serving in the foreground instead of supervising a detached - persistent worker. -- Extract one owner runtime shared by foreground and subprocess adapters. -- Keep validation within the transport-owned configuration boundary. -- Route nested help generically where possible without plugin regressions. -- Convert operational failures to concise CLI errors. - -Separate or defer: - -- Explicit-host attachment and targeted discovery for unicast-only networks. -- Any unrelated transport refactor. - -Discard: - -- Robot-specific private validator imports in CLI code. -- Persistent detached-child supervision for explicit serve. -- Machine-specific paths and remote-enabled defaults in examples. - -## Implementation Sequence - -### Phase 1: Owner runtime - -1. Restore the pre-PoC transport to a compiling, tested baseline. -2. Extract `run_owner` without changing auto-spawn behavior. -3. Add structured exit reasons and cleanup result mapping. -4. Run focused owner config, worker, handshake, lock, and `SharedRobot` tests. - -### Phase 2: Foreground serve - -1. Add the flat serve parser and config-file support. -2. Call `run_owner` directly in the CLI process. -3. Adapt SIGINT/SIGTERM to the runtime shutdown event. -4. Add concise startup, readiness, contention, and shutdown reporting. -5. Add process-level CLI tests with the fake robot. - -### Phase 3: Discovery - -1. Add the discover parser over existing discovery behavior. -2. Add deterministic human and clean JSON formatting. -3. Verify that network metadata never drives imports. - -### Phase 4: Documentation - -1. Add a portable SO101 example with placeholders and `allow_remote: false`. -2. Update the sharing how-to and CLI reference. -3. Restore the full Zenoh transport design and align its lifecycle invariants - with this design. - -Keep these phases as separate reviewable commits where practical. - -## Test Strategy - -### Unit tests - -- Owner config accepts internal `idle_timeout=None` and round-trips it over IPC. -- Invalid names, rates, and kwargs fail before driver construction. -- Requested shutdown and numeric idle timeout return success after disconnect. -- Repeated reads, loop exceptions, and disconnect failures return nonzero. -- Cleanup continues after each best-effort cleanup failure. -- Auto-spawn remains detached with its 10-second default timeout. -- CLI serving passes `idle_timeout=None` and does not instantiate `RobotOwner`. -- Expected CLI failures do not leak tracebacks. -- Discovery is sorted and JSON stdout is one clean array. -- Group help stays lightweight and nested help reaches the selected parser. -- Plugin help behavior remains unchanged. - -### Process-level tests - -Start `physicalai robot serve` with a fake driver, wait for readiness, attach a -`SharedRobot`, observe state, send a stationary action, terminate the CLI, and -verify: - -- The foreground process exits. -- The fake driver disconnect path ran. -- Name and device locks can be reacquired. -- A clean interruption exits zero. -- An injected disconnect failure exits nonzero. -- No owner process remains after the CLI exits. - -### Manual hardware validation - -On an isolated robot network: - -1. Serve an SO101 on host A. -2. Discover and attach from host B. -3. Exchange one safe stationary action. -4. Disconnect host B and confirm host A remains available. -5. Stop host A and verify safe disconnect and lock release. -6. Restart under a service manager and verify signal-driven shutdown. - -## Acceptance Criteria - -- No tracked source contains the PoC syntax corruption or duplicate API - definitions. -- Explicit serve owns hardware in the foreground and cannot orphan a detached - persistent worker through its normal architecture. -- Auto-spawn remains compatible and exits after its existing idle timeout. -- Every lifecycle invariant in this document has an adjacent test. -- Remote serving remains explicit and documented as unauthenticated. -- The example contains no developer-specific paths and defaults to local-only. -- Design and user documentation agree on lifecycle and exit semantics. -- Focused tests, CLI smoke checks, type checking, and repository hooks pass: - -```bash -uv run pytest tests/unit/robot/transport tests/unit/cli -uv run physicalai --help -uv run physicalai robot --help -uv run physicalai robot serve --help -uv run physicalai robot discover --help -pyrefly check -prek run --all-files -``` From 56de22fcbcbac5670f84268184a04947eac37d56 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 20 Jul 2026 21:45:32 +0000 Subject: [PATCH 5/8] fix: verify flock is still held in registered_owner_names to prevent stale PID false positives --- src/physicalai/robot/transport/_lock.py | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/src/physicalai/robot/transport/_lock.py b/src/physicalai/robot/transport/_lock.py index 1b252e9b..e783f6cf 100644 --- a/src/physicalai/robot/transport/_lock.py +++ b/src/physicalai/robot/transport/_lock.py @@ -107,7 +107,26 @@ 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 + 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) + return None + except OSError: + # Could not acquire — some process holds the lock; this is a live owner. + return diagnostics + finally: + os.close(fd) def _active_owner_name(path: Path) -> str | None: From c2d52b579071d3ddd415f79cffa524c46c06c8f4 Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Tue, 21 Jul 2026 11:13:41 +0200 Subject: [PATCH 6/8] address copilot comments --- src/physicalai/cli/robot.py | 2 +- src/physicalai/robot/transport/_owner.py | 2 ++ src/physicalai/robot/transport/_owner_worker.py | 5 +++-- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/src/physicalai/cli/robot.py b/src/physicalai/cli/robot.py index 702934f6..df964042 100644 --- a/src/physicalai/cli/robot.py +++ b/src/physicalai/cli/robot.py @@ -54,7 +54,7 @@ def _build_serve_parser() -> ArgumentParser: parser.add_argument( "--robot_kwargs", type=dict, - default={}, + default=None, help="JSON-serializable driver constructor arguments.", ) parser.add_argument( diff --git a/src/physicalai/robot/transport/_owner.py b/src/physicalai/robot/transport/_owner.py index 024c3c1d..a708d352 100644 --- a/src/physicalai/robot/transport/_owner.py +++ b/src/physicalai/robot/transport/_owner.py @@ -136,6 +136,7 @@ def stop(self, timeout: float = 5.0) -> None: if proc.poll() is not None: self._exit_code = proc.returncode + self._process = None return proc.terminate() @@ -147,6 +148,7 @@ def stop(self, timeout: float = 5.0) -> None: 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. diff --git a/src/physicalai/robot/transport/_owner_worker.py b/src/physicalai/robot/transport/_owner_worker.py index f98cec15..05ac0439 100644 --- a/src/physicalai/robot/transport/_owner_worker.py +++ b/src/physicalai/robot/transport/_owner_worker.py @@ -528,8 +528,9 @@ def run_owner( try: endpoints.driver.disconnect() logger.trace(f"Disconnected robot driver for {config.name!r}") - except Exception: # noqa: BLE001 - logger.exception(f"driver disconnect failed for {config.name}") + 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() From d3bee0d02e662ca471528f9d8dbffbdbfd8de887 Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Tue, 21 Jul 2026 11:42:19 +0200 Subject: [PATCH 7/8] fix style --- src/physicalai/robot/transport/_lock.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/physicalai/robot/transport/_lock.py b/src/physicalai/robot/transport/_lock.py index e783f6cf..b9493985 100644 --- a/src/physicalai/robot/transport/_lock.py +++ b/src/physicalai/robot/transport/_lock.py @@ -116,17 +116,18 @@ def _read_live_name_diagnostics(path: Path) -> dict[str, object] | None: 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) - return None except OSError: # Could not acquire — some process holds the lock; this is a live owner. - return diagnostics + held = True finally: os.close(fd) + return diagnostics if held else None def _active_owner_name(path: Path) -> str | None: From 5e8dac974d079d0de24b92d816f2f5a417cb5082 Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Wed, 22 Jul 2026 10:47:32 +0200 Subject: [PATCH 8/8] address comments --- src/physicalai/cli/robot.py | 22 ++++++++++++++-- .../robot/transport/_owner_config.py | 25 ++++++++++++------ .../robot/transport/_owner_worker.py | 3 ++- tests/unit/cli/test_robot_cli.py | 26 ++++++++++++++++++- .../unit/robot/transport/test_owner_worker.py | 2 +- 5 files changed, 65 insertions(+), 13 deletions(-) diff --git a/src/physicalai/cli/robot.py b/src/physicalai/cli/robot.py index df964042..b07ba970 100644 --- a/src/physicalai/cli/robot.py +++ b/src/physicalai/cli/robot.py @@ -34,6 +34,11 @@ 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} @@ -115,6 +120,19 @@ def _configure_serve_logging(*, verbose: bool) -> None: ) +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``. @@ -152,7 +170,7 @@ def serve(cfg: Namespace) -> int: Returns: The owner runtime exit code, or 1 for an expected startup failure. """ - _configure_serve_logging(verbose=cfg.verbose) + _prepare_serve_logging(allow_remote=cfg.allow_remote, verbose=cfg.verbose) try: config = RobotOwnerConfig( name=cfg.name, @@ -194,7 +212,7 @@ def _on_event(event: OwnerEvent) -> None: status = "subscriber(s) connected" if subscribers_present else "no subscribers" logger.info(f"Healthy · uptime {_format_duration(uptime_s)} · {status}") - logger.info(f"Starting robot {config.name!r} using {config.robot_class}") + _log_serve_start(config) previous_sigterm = signal.signal(signal.SIGTERM, _handle_signal) previous_sigint = signal.signal(signal.SIGINT, _handle_signal) try: diff --git a/src/physicalai/robot/transport/_owner_config.py b/src/physicalai/robot/transport/_owner_config.py index c277116a..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 @@ -78,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). @@ -87,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. """ diff --git a/src/physicalai/robot/transport/_owner_worker.py b/src/physicalai/robot/transport/_owner_worker.py index 05ac0439..fed690cb 100644 --- a/src/physicalai/robot/transport/_owner_worker.py +++ b/src/physicalai/robot/transport/_owner_worker.py @@ -203,7 +203,8 @@ def _apply_pending_action(driver: Robot, action_sub: Any, name: str) -> None: # 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}", exc_info=True) + logger.warning(f"Failed to apply action for {name}") + logger.opt(exception=True).trace("action apply traceback") def _run_loop( diff --git a/tests/unit/cli/test_robot_cli.py b/tests/unit/cli/test_robot_cli.py index ebcceb73..ab19564a 100644 --- a/tests/unit/cli/test_robot_cli.py +++ b/tests/unit/cli/test_robot_cli.py @@ -57,6 +57,30 @@ def _run_owner(config: object, shutdown: threading.Event, *, ready: object, on_e 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: @@ -243,4 +267,4 @@ def test_serve_process_sigterm_disconnects_and_releases_locks(tmp_path: Path) -> finally: if process.poll() is None: process.kill() - process.wait(timeout=5.0) \ No newline at end of file + process.wait(timeout=5.0) diff --git a/tests/unit/robot/transport/test_owner_worker.py b/tests/unit/robot/transport/test_owner_worker.py index 95bdf09d..b22fbf81 100644 --- a/tests/unit/robot/transport/test_owner_worker.py +++ b/tests/unit/robot/transport/test_owner_worker.py @@ -166,4 +166,4 @@ def _fail_ready() -> None: assert result.reason is OwnerExitReason.LOOP_FAILURE assert result.exit_code == 1 - assert driver.disconnect_called \ No newline at end of file + assert driver.disconnect_called