Add Runtime Callbacks#

Use callbacks when you need product-specific behavior around the runtime loop.

The following example records observations and actions.

class RecordingCallback:
    def on_action_ready(self, *, action, step):
        recorder.write_policy_action(step, action)
        return action

    def on_action_sent(self, *, action, step):
        recorder.write_sent_action(step, action)

Attach the callback when you construct the runtime.

runtime = RobotRuntime(
    robot=robot,
    action_source=PolicySource(model=model, execution=execution),
    fps=30,
    callbacks=[RecordingCallback()],
)

The same callback instances receive every run() performed before the runtime is disconnected. Use the start and shutdown lifecycle events to separate per-run state. RobotRuntime.disconnect() or context exit closes callback resources exactly once and permanently disposes that runtime.

As a general rule, keep workflow-specific logic in callbacks unless the same behavior becomes a reusable runtime primitive.