run_policy: dry-run modes, synthetic interface, ORT pin, recorder cpu pin

Robot-local work captured from fmd-rccu-1:
- DebugConfig.dry_run / dryer_run: run the loop without sending commands
  (dry_run) or without any bridge/hardware via a fabricated-state
  SyntheticInterface (dryer_run).
- base.py: wire the synthetic interface for dryer_run; pin ONNX Runtime to a
  single intra-op thread so it stays off the isolated RT cores.
- SessionRecorder: pin ros2 bag record to its own core (DebugConfig.rosbag_cpu,
  default 2) so it stops stealing cycles from the policy core.

Author: tomasz-lewicki

src/holosoma_inference/holosoma_inference/config/config_types/task.py

@@ -25,6 +25,20 @@ class DebugConfig:
     force_zero_action: bool = False
     """Zero out the scaled policy action (robot holds default pose)."""
 
+    dry_run: bool = False
+    """Run the full policy loop (obs, inference, postprocess) but do NOT send
+    commands to the robot. Safe for bring-up / on-hardware testing: everything
+    computes and logs as normal, but send_low_command is skipped so no torque
+    is applied. Pair with print_observations to inspect what would be sent."""
+
+    dryer_run: bool = False
+    """Like ``dry_run``, but ALSO fabricate the robot state instead of reading
+    it from a bridge/driver. Swaps in a synthetic interface that returns a
+    plausible standing pose (default_dof_angles, upright IMU, zero velocities)
+    and never opens a DDS connection, so the loop runs with no sim/hardware
+    present at all. Implies ``dry_run`` (never sends commands). Use to smoke-test
+    config loading, ONNX inference, and the obs pipeline off-robot."""
+
     record_rosbag: bool = False
     """Record a per-session rosbag for the lifetime of this run_policy process.
     Spawns ``ros2 bag record --all`` in the background when the policy starts and
@@ -35,6 +49,13 @@ class DebugConfig:
     rosbag_dir: str = "~/run_policy_sessions"
     """Directory for per-session rosbags recorded via ``record_rosbag``."""
 
+    rosbag_cpu: int = 2
+    """CPU core to pin the ``ros2 bag record`` process to (via ``taskset``). The
+    recorder otherwise inherits the policy's affinity (e.g. ``taskset -c 8``) and
+    contends with inference on that core, adding policy-loop jitter. Pin it to a
+    general-purpose core off both the policy core and the isolated RT cores.
+    Set to a negative value to disable pinning (inherit the parent's affinity)."""
+
 
 @dataclass(frozen=True)
 class Ros2DepthConsumerConfig:

src/holosoma_inference/holosoma_inference/policies/base.py

@@ -149,6 +149,18 @@ def _init_communication_components(self):
         if hasattr(self, "_shared_hardware_source"):
             self.interface = self._shared_hardware_source.interface
             return
+        # Dryer run: fabricate state, never open a DDS connection. Bypasses the
+        # SDK backend entirely so the loop runs with no bridge/driver present.
+        if self.config.task.debug.dryer_run:
+            from holosoma_inference.sdk.synthetic_interface import SyntheticInterface
+
+            self.interface = SyntheticInterface(
+                self.robot_config,
+                self.config.task.domain_id,
+                self.config.task.interface,
+                False,
+            )
+            return
         # The SDK's own wireless-controller path is only needed when an
         # "interface" channel is selected.  The "joystick" channel is read
         # via host-side evdev (UsbJoystickInput) and does not touch the SDK.
@@ -384,9 +396,36 @@ def _setup_keyboard_listener(self):
     # Policy Methods
     # ============================================================================
 
+    @staticmethod
+    def _make_ort_session_options():
+        """ONNX Runtime options tuned for the real-time control loop.
+
+        By default ORT sizes its intra-op thread pool to the CPU count AND pins
+        one worker to each core via sched_setaffinity. That per-thread pin
+        bypasses the kernel's ``isolcpus`` reservation, so the workers land on
+        the isolated RT cores (10-13 here) and contend with the 1 kHz EtherCAT
+        actuator threads. The policy is a small MLP run at 50 Hz (~1 ms
+        single-threaded), so a multi-core pool buys nothing and the pinning is
+        pure harm.
+
+        Setting ``intra_op_num_threads = 1`` (with sequential exec) means ORT
+        builds no worker pool at all — inference runs on the calling thread,
+        which stays within the process's inherited (0-9) affinity mask. So
+        there is nothing left to per-core-pin; we deliberately do NOT set
+        ``session.intra_op_thread_affinities`` (ORT rejects an empty string, and
+        a non-empty one would re-introduce explicit pinning).
+        """
+        opts = onnxruntime.SessionOptions()
+        opts.intra_op_num_threads = 1
+        opts.inter_op_num_threads = 1
+        opts.execution_mode = onnxruntime.ExecutionMode.ORT_SEQUENTIAL
+        return opts
+
     def setup_policy(self, model_path):
         """Setup ONNX policy model and extract metadata."""
-        self.onnx_policy_session = onnxruntime.InferenceSession(model_path)
+        self.onnx_policy_session = onnxruntime.InferenceSession(
+            model_path, sess_options=self._make_ort_session_options()
+        )
         input_names = [inp.name for inp in self.onnx_policy_session.get_inputs()]
         output_names = [out.name for out in self.onnx_policy_session.get_outputs()]
 
@@ -700,14 +739,31 @@ def policy_action(self):
 
         # Stage 5: Action Pub
         with self.latency_tracker.measure("action_pub"):
-            self.interface.send_low_command(
-                self.cmd_q,
-                self.cmd_dq,
-                self.cmd_tau,
-                robot_state_data[0, 7 : 7 + self.num_dofs],
-                kp_override=kp_override,
-                kd_override=kd_override,
-            )
+            # dry_run: the interface is the REAL bridge/driver, so we must NOT
+            # call send_low_command at all. dryer_run: the interface is the
+            # SyntheticInterface, whose send_low_command only publishes to a
+            # synthetic measurement topic (nothing reaches the robot), so we DO
+            # call it — that's what produces the measurable command stream.
+            if self.config.task.debug.dry_run and not self.config.task.debug.dryer_run:
+                # Log once so it's unmistakable no torque is being applied.
+                if not getattr(self, "_dry_run_logged", False):
+                    logger.warning(
+                        colored(
+                            "DRY RUN: policy loop active but send_low_command is SKIPPED "
+                            "(no commands sent to the robot).",
+                            "yellow",
+                        )
+                    )
+                    self._dry_run_logged = True
+            else:
+                self.interface.send_low_command(
+                    self.cmd_q,
+                    self.cmd_dq,
+                    self.cmd_tau,
+                    robot_state_data[0, 7 : 7 + self.num_dofs],
+                    kp_override=kp_override,
+                    kd_override=kd_override,
+                )
 
         # Telemetry hook: fires every control tick in every state (policy,
         # stiff-hold, init ramp) with the final executed command. Default

src/holosoma_inference/holosoma_inference/sdk/synthetic_interface.py

@@ -0,0 +1,176 @@
+"""Synthetic (offline) interface for ``--task.debug.dryer-run``.
+
+Fabricates a plausible standing robot state instead of reading it from a sim
+bridge or hardware driver, so the policy loop runs with nothing else present.
+Used to smoke-test config loading, ONNX inference, and the observation pipeline
+off-robot.
+
+For measurement, ``send_low_command`` publishes the commands the policy *would*
+send to a **separate synthetic topic** (``/dryer_run/cmd``, ``sensor_msgs/
+JointState``) that no real driver subscribes to — so you can ``ros2 topic hz /
+dryer_run/cmd`` / ``echo`` / ``bag record`` the loop's output while nothing is
+ever sent to the actual robot. If ROS2 is unavailable (pure offline run) it
+degrades to a no-op and stays fully offline.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+from loguru import logger
+
+from holosoma_inference.config.config_types import RobotConfig
+from holosoma_inference.sdk.base.base_interface import BaseInterface
+
+# Deliberately NOT a real driver topic (e.g. /alpha/joints/cmd). Nothing on the
+# robot subscribes here, so publishing is safe — it's for measurement only.
+_SYNTHETIC_CMD_TOPIC = "/dryer_run/cmd"
+
+
+class SyntheticInterface(BaseInterface):
+    """Returns a fixed, upright standing state; publishes cmds to a fake topic.
+
+    State layout matches :meth:`BaseInterface.get_low_state`:
+    ``[base_pos(3), quat(4), dof_pos(N), lin_vel(3), ang_vel(3), dof_vel(N)]``
+    followed by an appended ``projected_gravity(3)`` (upright).
+    """
+
+    def __init__(
+        self,
+        robot_config: RobotConfig,
+        domain_id: int = 0,
+        interface_str: str | None = None,
+        use_joystick: bool = False,
+        publish_cmd: bool = True,
+        cmd_topic: str = _SYNTHETIC_CMD_TOPIC,
+    ):
+        super().__init__(robot_config, domain_id, interface_str, use_joystick)
+
+        default_angles = getattr(robot_config, "default_dof_angles", None)
+        if default_angles is not None and len(default_angles) > 0:
+            self._dof_pos = np.asarray(default_angles, dtype=float)
+        else:
+            n = len(getattr(robot_config, "dof_names", []) or [])
+            self._dof_pos = np.zeros(n if n > 0 else 1)
+        self._num_dof = self._dof_pos.shape[0]
+
+        dof_names = list(getattr(robot_config, "dof_names", []) or [])
+        if len(dof_names) != self._num_dof:
+            dof_names = [f"joint_{i}" for i in range(self._num_dof)]
+        self._dof_names = dof_names
+
+        # Upright: identity quaternion (w,x,y,z), gravity straight down in the
+        # base frame, zero linear/angular velocity, zero joint velocity.
+        self._quat = np.array([1.0, 0.0, 0.0, 0.0])
+        self._projected_gravity = np.array([0.0, 0.0, -1.0])
+
+        self._kp_level = 1.0
+        self._kd_level = 1.0
+
+        logger.warning(
+            "DRYER RUN: using SyntheticInterface — state is fabricated (upright, "
+            f"default pose, zero velocities across {self._num_dof} DOF) and NO "
+            "connection to any bridge/driver is opened."
+        )
+
+        # Optional command publishing to a synthetic topic for measurement.
+        self._cmd_topic = cmd_topic
+        self._cmd_pub = None
+        self._node = None
+        self._JointState = None
+        if publish_cmd:
+            self._setup_cmd_publisher()
+
+    def _setup_cmd_publisher(self) -> None:
+        try:
+            import rclpy
+            from rclpy.node import Node
+            from rclpy.qos import QoSProfile, ReliabilityPolicy
+            from sensor_msgs.msg import JointState
+        except Exception as exc:  # ROS2 not sourced → stay fully offline.
+            logger.warning(
+                f"DRYER RUN: command publishing requested but ROS2 is unavailable "
+                f"({exc}); running fully offline — no synthetic cmd topic."
+            )
+            return
+
+        if not rclpy.ok():
+            rclpy.init()
+        self._node = Node("dryer_run_synthetic")
+        qos = QoSProfile(depth=10, reliability=ReliabilityPolicy.RELIABLE)
+        self._cmd_pub = self._node.create_publisher(JointState, self._cmd_topic, qos)
+        self._JointState = JointState
+        logger.warning(
+            f"DRYER RUN: publishing synthetic commands on {self._cmd_topic} "
+            "(sensor_msgs/JointState: name/position/velocity/effort) for "
+            "measurement. No real driver subscribes to this topic — nothing is "
+            "sent to the robot."
+        )
+
+    def get_low_state(self) -> np.ndarray:
+        return np.concatenate(
+            [
+                np.zeros(3),  # base_pos
+                self._quat,  # quat (w,x,y,z)
+                self._dof_pos,  # dof_pos
+                np.zeros(3),  # base lin_vel
+                np.zeros(3),  # base ang_vel
+                np.zeros(self._num_dof),  # dof_vel
+                self._projected_gravity,  # projected_gravity (upright)
+            ]
+        ).reshape(1, -1)
+
+    def send_low_command(
+        self,
+        cmd_q: np.ndarray,
+        cmd_dq: np.ndarray,
+        cmd_tau: np.ndarray,
+        dof_pos_latest: np.ndarray = None,
+        kp_override: np.ndarray = None,
+        kd_override: np.ndarray = None,
+    ) -> None:
+        # Only ever publishes to the synthetic measurement topic — never to a
+        # real driver. If publishing is disabled/unavailable, this is a no-op.
+        if self._cmd_pub is None:
+            return
+        # On shutdown (Ctrl-C / SIGTERM) the rclpy context can be torn down
+        # while a final publish is in flight — guard so exit stays clean.
+        try:
+            import rclpy
+
+            if not rclpy.ok():
+                return
+        except Exception:
+            return
+        msg = self._JointState()
+        msg.header.stamp = self._node.get_clock().now().to_msg()
+        msg.name = self._dof_names
+        msg.position = np.asarray(cmd_q, dtype=float).reshape(-1).tolist()
+        msg.velocity = np.asarray(cmd_dq, dtype=float).reshape(-1).tolist()
+        msg.effort = np.asarray(cmd_tau, dtype=float).reshape(-1).tolist()
+        try:
+            self._cmd_pub.publish(msg)
+        except Exception:
+            # Context invalidated mid-publish during shutdown — ignore.
+            return
+
+    def get_joystick_msg(self):
+        return None
+
+    def get_joystick_key(self, wc_msg=None):
+        return None
+
+    @property
+    def kp_level(self):
+        return self._kp_level
+
+    @kp_level.setter
+    def kp_level(self, value):
+        self._kp_level = float(value)
+
+    @property
+    def kd_level(self):
+        return self._kd_level
+
+    @kd_level.setter
+    def kd_level(self, value):
+        self._kd_level = float(value)

src/holosoma_inference/holosoma_inference/utils/session_recorder.py

@@ -36,17 +36,24 @@ class SessionRecorder:
     that won't start must never take down the policy run.
     """
 
-    def __init__(self, enabled: bool, out_dir: str = "~/run_policy_sessions", storage: str = "mcap"):
+    def __init__(
+        self,
+        enabled: bool,
+        out_dir: str = "~/run_policy_sessions",
+        storage: str = "mcap",
+        cpu: int = 2,
+    ):
         self.enabled = enabled
         self.out_dir = Path(out_dir).expanduser()
         self.storage = storage
+        self.cpu = cpu
         self.bag_path: Path | None = None
         self._proc: subprocess.Popen | None = None
 
     @classmethod
     def from_debug_config(cls, debug: DebugConfig) -> SessionRecorder:
-        """Build from a task ``DebugConfig`` (reads ``record_rosbag`` / ``rosbag_dir``)."""
-        return cls(enabled=debug.record_rosbag, out_dir=debug.rosbag_dir)
+        """Build from a task ``DebugConfig`` (reads ``record_rosbag`` / ``rosbag_dir`` / ``rosbag_cpu``)."""
+        return cls(enabled=debug.record_rosbag, out_dir=debug.rosbag_dir, cpu=debug.rosbag_cpu)
 
     def __enter__(self) -> Self:
         if not self.enabled:
@@ -62,9 +69,18 @@ def __enter__(self) -> Self:
                 "-o", str(self.bag_path),
             ]
             # fmt: on
+            # Pin the recorder off the policy core: as a child of the (often
+            # taskset-pinned) policy process it would otherwise inherit the
+            # policy's CPU affinity and steal cycles from inference, adding
+            # control-loop jitter. taskset -c <cpu> moves it to its own core.
+            if self.cpu >= 0:
+                cmd = ["taskset", "-c", str(self.cpu), *cmd]
             # setsid: own process group so we can SIGINT only the recorder on exit.
             self._proc = subprocess.Popen(cmd, preexec_fn=os.setsid)  # noqa: PLW1509
-            logger.info(f"💾 Recording session rosbag → {self.bag_path} (ros2 bag record --all)")
+            logger.info(
+                f"💾 Recording session rosbag → {self.bag_path} "
+                f"(ros2 bag record --all{f', taskset -c {self.cpu}' if self.cpu >= 0 else ''})"
+            )
         except Exception as exc:  # never let recording break the run
             logger.warning(f"Could not start session rosbag recorder: {exc}")
             self._proc = None