feat: generic robot runtime

[maxxgx]

Summary

Replaces the policy-only PolicyRuntime (which hard-wired inference into the loop and exposed ~25 types: a Controller protocol, 4 capability protocols, a lazy Tick, a subclass, and isinstance dispatch in the loop body) with one runtime class + one 3-method protocol. Same fixed-rate loop now drives policy, teleop, and future action sources with no fork and zero isinstance in the loop. Also reorganizes the flat runtime/ package into subpackages.

Scoped to src/physicalai/runtime/ + its CLI/config surface. No changes to inference/robot/capture layers.

ActionSource

class ActionSource(Protocol):
    def connect(self, *, bus, session_id) -> None: ...
    def update(self, robot_state, camera_frames, step) -> np.ndarray: ...
    def disconnect(self) -> None: ...

RobotRuntime owns robot+cameras, timing, resilient IO (retry / stale fallback / circuit breaker), the callback bus, and lifecycle. PolicySource and TeleopSource are the two concrete implementations. run(duration_s) -> int (steps).

Loop semantics (the core behavioral change)

sequenceDiagram
    participant R as RobotRuntime
    participant S as ActionSource
    participant B as CallbackBus
    participant Rb as Robot

    R->>Rb: read robot_state + camera_frames (once/tick, retry+stale)
    R->>S: update(robot_state, camera_frames, step)
    S-->>R: action
    R->>B: on_action_ready(action) [transform chain]
    B-->>R: action'
    R->>Rb: send_action(action') [resilient]
    R->>B: on_action_sent(action') · emit TickEvent
    Note over R: sleep to hold fps

Loading

Key points for review: reads are eager (once per tick, shared by value — no Tick), update() always returns a sendable action (hold/repeat is the source's internal decision, no runtime on_hold), and no isinstance anywhere in the loop. Action-source-specific live metrics (queue depth) ride a dedicated MetricsEvent, keeping the runtime ignorant of policy concepts.

Removed / renamed

Removed: PolicyRuntime, Tick, SupportsBus/Stats/Drain/HoldInfo, RunStats, on_hold, shutdown drain, dual config schema.
Renamed: Controller→ActionSource, PolicyController→PolicySource, TeleopController→TeleopSource, before_send_action→on_action_ready, runtime.py→core.py.

Package structure

Public imports (from physicalai.runtime import X) unchanged; only backing files moved.

runtime/
  core.py  events.py  smoothers.py  _callback_bus.py
  action_sources/   # base, policy, teleop
  execution/        # base, sync, async, rtc, queue, rtc_queue
  callbacks/        # console, jsonl, async_callback, rerun, low_pass
  observer/         # telemetry subscriber (+ relocated _telemetry.py, dormant)

Out of scope

• runtime→rollout naming (deferred). I think rollout is the more established term and it would make sense for us to adopt as well, especially early on before real adoption.
• observer/_telemetry.py relocated but dormant (observer is just scaffolding for now).

Validation

Manually tested teleop via CLI with:

physicalai run --config examples/runtime/teleop_runtime.yaml

teleop_runtime.yaml:

runtime:
  robot:
    class_path: physicalai.robot.SO101
    init_args:
      port: /dev/ttyACM1
      calibration: /home/max/.local/share/physicalai/robots/b5b8ba26-a470-49a2-ad36-25d69323bb33/calibrations/795d23ad-f3f3-4154-a11d-c46c12985a81.json

  action_source:
    class_path: physicalai.runtime.TeleopSource
    init_args:
      leader:
        class_path: physicalai.robot.SO101
        init_args:
          port: /dev/ttyACM0
          calibration: /home/max/.local/share/physicalai/robots/46a715c3-d07c-49d9-ad0d-290ebe3662e4/calibrations/99bf3448-49a5-4bcd-8f0a-37fa244481ab.json
          role: leader

  fps: 100.0

  callbacks:
    - class_path: physicalai.runtime.ConsoleCallback
      init_args:
        throttle_steps: 100   # print once per second at 100fps

run:
  duration_s: 300

Breaking changes

• Config: model/execution now nest under an action_source: block.
• PolicyRuntime(...) → RobotRuntime(robot, action_source=PolicySource(model, execution), ...).
• run() returns int, not RunStats.
• Callbacks: before_send_action→on_action_ready (return required); read event.robot_state/event.camera_frames (not event.tick.*); drop on_hold.

Related issues

• Closes #

[MarkRedeman]

Some comments. Didn't end up fully reviewing the actual design doc yet.

Generally this looks like a nice addition, but I am concerned that this feels overly complex. Keeping a mental model of which components get used where is not easy.
Maybe we can add some diagrams that show how the different runtimes can be used and where each code gets executed (i.e. in a separate thread, process etc?)

[maxxgx]

@MarkRedeman @samet-akcay I've redesigned the runtime architecture, with the priority of simplifying it while retaining the capabilities. Please check the updated PR description for an overview

[Copilot]

Pull request overview

This PR refactors the runtime layer to replace the policy-specific PolicyRuntime with a generic RobotRuntime driven by a minimal ActionSource protocol, enabling policy rollout, teleop, and future action sources to share the same fixed-rate control loop and callback lifecycle. It also reorganizes src/physicalai/runtime/ into subpackages and updates the CLI/config surface, examples, tests, and documentation to match the new API and event model (including MetricsEvent).

Changes:

• Introduce RobotRuntime + ActionSource (with PolicySource and TeleopSource) and remove PolicyRuntime/RunStats/hold & drain concepts.
• Rework callback hooks (on_action_ready) and telemetry/events (TickEvent shape changes + new MetricsEvent).
• Migrate CLI/config schema (action_source: nesting), tests, examples, and docs; reorganize runtime package into subpackages.

Reviewed changes

Copilot reviewed 60 out of 60 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
tests/unit/runtime/test_telemetry.py	Update imports and TickEvent field names; remove queue-depth assertions from tick telemetry.
tests/unit/runtime/test_runtime.py	Migrate runtime tests from PolicyRuntime/RunStats to RobotRuntime/PolicySource and generic ActionSource behavior.
tests/unit/runtime/test_rerun_callback.py	Update RerunCallback expectations for MetricsEvent and camera frames sourced from TickEvent.
tests/unit/runtime/test_fault_tolerance.py	Rebase fault-tolerance tests onto RobotRuntime read/send helpers and stale flag semantics.
tests/unit/runtime/test_execution.py	Update imports and add coverage for “skip inference when queue full” and RTC obs-slot clearing.
tests/unit/runtime/test_action_queue.py	Update imports to new execution.queue export surface.
tests/unit/runtime/conftest.py	Add FakeActionSource for generic ActionSource loop tests.
tests/unit/cli/test_cli.py	Migrate CLI config parsing expectations to runtime.action_source.* schema and RobotRuntime.run() return type.
src/physicalai/runtime/runtime.py	Remove legacy PolicyRuntime implementation.
src/physicalai/runtime/observer/_telemetry.py	Remove queue_remaining from emitted tick payload.
src/physicalai/runtime/observer/_subscriber.py	Update telemetry import path after observer module relocation.
src/physicalai/runtime/observer/main.py	Rename observer description to RobotRuntime sessions.
src/physicalai/runtime/observer/init.py	Update note to reference RobotRuntime rather than PolicyRuntime.
src/physicalai/runtime/execution/sync.py	Add dedicated SyncExecution module under execution/ subpackage.
src/physicalai/runtime/execution/rtc.py	Repoint imports to new execution.base/queue modules and adjust RTC request gating/obs-slot behavior.
src/physicalai/runtime/execution/rtc_queue.py	Minor docstring addition for RTCActionQueue initialization.
src/physicalai/runtime/execution/queue.py	Promote ActionQueue protocol + ChunkedActionQueue under execution package with expanded docstrings.
src/physicalai/runtime/execution/base.py	Introduce shared Execution ABC, NOT_STARTED message, and WorkerDiedError.
src/physicalai/runtime/execution/async_execution.py	Slim async execution module to just AsyncExecution, reusing execution.base types.
src/physicalai/runtime/execution/init.py	New execution package re-exports (Execution, queues, implementations).
src/physicalai/runtime/events.py	Update TickEvent fields and add MetricsEvent.
src/physicalai/runtime/core.py	New RobotRuntime loop owning IO resilience, timing, callback bus, and lifecycle.
src/physicalai/runtime/callbacks/rerun.py	Update RerunCallback to consume TickEvent frames directly and log queue depth via MetricsEvent.
src/physicalai/runtime/callbacks/low_pass.py	Move LowPassFilterCallback into callbacks subpackage with new hook name.
src/physicalai/runtime/callbacks/jsonl.py	Extract JsonlCallback into its own module and update TickEvent field names.
src/physicalai/runtime/callbacks/console.py	Extract ConsoleCallback and adjust output to new event shape.
src/physicalai/runtime/callbacks/async_callback.py	Extract AsyncCallback and update action-hook exclusions and tick frame-copy logic.
src/physicalai/runtime/callbacks/init.py	New callbacks package export surface.
src/physicalai/runtime/action_sources/teleop.py	Add TeleopSource implementation of ActionSource protocol.
src/physicalai/runtime/action_sources/policy.py	Add PolicySource wiring model/execution/queue into ActionSource and emit MetricsEvent.
src/physicalai/runtime/action_sources/base.py	Define the ActionSource protocol seam for RobotRuntime.
src/physicalai/runtime/action_sources/init.py	New action_sources package exports.
src/physicalai/runtime/_callback_bus.py	Rename/reshape action hook chaining and add metrics dispatch.
src/physicalai/runtime/init.py	Preserve public import surface while re-exporting new runtime/action_source/callback/execution APIs.
src/physicalai/cli/run.py	Migrate CLI parser and dispatcher to RobotRuntime and new schema; simplify completion log output.
README.md	Update public docs and examples to RobotRuntime/PolicySource and new config schema.
examples/tutorials/collect_train_deploy.ipynb	Update tutorial code to construct PolicySource + RobotRuntime and print new-style stats.
examples/runtime/sync_inference.py	Migrate example to RobotRuntime + PolicySource and remove RerunCallback camera dependency.
examples/runtime/runtime.yaml	Update example config to nest model/execution under action_source.init_args and move task accordingly.
examples/runtime/run_from_config.py	Migrate config-loading example to RobotRuntime and update summary output.
examples/runtime/rtc_inference.py	Migrate RTC example to RobotRuntime + PolicySource.
examples/runtime/async_inference.py	Migrate async example to RobotRuntime + PolicySource.
docs/reference/runtime-api.md	Update API reference to RobotRuntime/ActionSource/PolicySource/TeleopSource and run() return type.
docs/reference/config-schema.md	Update schema reference to require explicit action_source block.
docs/reference/cli.md	Update CLI doc snippets to RobotRuntime.
docs/index.md	Update high-level flow and examples to RobotRuntime + PolicySource.
docs/how-to/runtime/run-policy-on-robot.md	Migrate how-to to RobotRuntime + PolicySource and updated YAML structure.
docs/how-to/runtime/add-runtime-callbacks.md	Update callback hook name and example construction to new API.
docs/how-to/inference/load-exported-policy.md	Update guidance to recommend RobotRuntime + PolicySource for hardware loops.
docs/how-to/config/write-runtime-config.md	Update runtime config authoring docs to new schema.
docs/how-to/config/instantiate-components.md	Update nested component-spec example to RobotRuntime + PolicySource.
docs/how-to/cli/run.md	Update run command documentation to RobotRuntime.
docs/getting-started/run-a-policy.md	Update getting-started guide to new runtime/action_source split and loop semantics.
docs/getting-started/quickstart.md	Update quickstart guidance to reference RobotRuntime + PolicySource.
docs/explanation/runtime.md	Update conceptual runtime overview, responsibilities, and callback hook names.
docs/explanation/inference.md	Update inference-vs-runtime explanation to reflect PolicySource as the queue/inference owner.
docs/explanation/configuration.md	Update configuration explanation examples to RobotRuntime.
docs/explanation/cli.md	Update CLI explanation to RobotRuntime framing.
docs/explanation/architecture.md	Update architecture doc to include ActionSource and RobotRuntime responsibilities.
docs/design/components/runtime-system/runtime-architecture.md	Add detailed design doc for the runtime redesign and migration rationale.

Comments suppressed due to low confidence (1)

src/physicalai/runtime/execution/async_execution.py:80

• AsyncExecution.warmup() can set _threshold_count to 0 when request_threshold is 0.0 or very small (int truncation). With below_threshold implemented as len < threshold, a threshold of 0 prevents any future inference requests (queue can never be < 0), so the queue will eventually drain and stay empty.

[MarkRedeman]

Why do we ignore the name of the image input if there's only 1? If I were to record with only an overview camera then I think I'd still expect the model to use its name as an input. Mainly so that it would feel more consistent with when retraining with 2 cameras?

(No need to address in this PR.)

[MarkRedeman]

Is the input of each callback the output of the previous callback, or do they all get the output from the action source?

Typically with these things I quite like having a middleware approach:

def my_callback(input, next)
    new_input = self.do_something_with_input(input)
    output = next(new_input)
    new_output = self.do_something_with_output(output)

    return new_output

so then each callback would be able to both tweak the input and output.
I don't know if that's something that would be directly useful here though. The one example I was thinking of is that you could use LowPassFilterCallback and have it make sure the output is always smooth, but even in the middleware case order of your callbacks would still matter.

[MarkRedeman]

Perhaps move this to _callback_bus.py?

[MarkRedeman]

Might be interesting to apply a CircuitBreaker pattern here. We could wrap the robot and camera instances with a small decorator that implements that pattern.

[maxxgx]

I agree, created an issue to track this: #186

[AlexanderBarabanov]

Thanks, comments can be addressed (if required) in a separate PR.

[AlexanderBarabanov]

As _consecutive_error_ticks is used by both _read_robot_resilient() and _resilient_send(), I am thinking about a case when _read_robot_resilient() fails and sets _consecutive_error_ticks += 1 but then _resilient_send() is ok and it sets _consecutive_error_ticks -> 0 (in the same while loop).
In this case, counter can never accumulate toward _max_consecutive_error_ticks, although _read_robot_resilient() always fails.

Is this a real scenario? If real, should we separate the read failure counter from the send failure counter?

[AlexanderBarabanov]

Should we validate provided action for some allowed range or at least with np.isfinite(action)?

[AlexanderBarabanov]

As far as i understand, in case of exception, the runtime continues with a log message without clear runtime signal - should we somehow propagate exception case to the consumer code to let them know that there is an issue?
Perhaps, we could just remove try/except block here to propagate exception.

[AlexanderBarabanov]

Are there any potential issues (like ConnectionError, OSError) that can be originated from ._leader.get_observation? Perhaps we could use try/catch as we do for _read_robot_resilient() or _resilient_send()

[AlexanderBarabanov]

What is the purpose of this _session_id ? Perhaps we may add a small comment to explain that this sesion_id is not for security but for telemetry correlation purposes.

[AlexanderBarabanov]

Should we validate a value against some allowed range or with np.isfinite(action) before return?