run_policy: optional per-session rosbag (--task.debug.record-rosbag)

Spawns 'ros2 bag record --all' in its own process group for the lifetime of a
run_policy run and SIGINTs it for a clean MCAP close on exit (Ctrl-C, normal
completion, or error), then prints the bag's absolute path. Each run gets its
own short, self-contained bag, so debugging one run no longer means hunting for
a timestamp inside a long fleet-telemetry recording.

Wrapped around policy.run() via a SessionRecorder context manager; no-op when
the flag is off, and any failure to start recording is logged and swallowed so
it can never take down the policy run. Output dir configurable via
--task.debug.rosbag-dir (default ~/run_policy_bags).

Author: tomasz-lewicki

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

@@ -25,6 +25,16 @@ class DebugConfig:
     force_zero_action: bool = False
     """Zero out the scaled policy action (robot holds default pose)."""
 
+    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
+    stops it on exit (Ctrl-C, normal completion, or error), then prints the bag's
+    absolute path. Each run gets its own short, self-contained bag — no hunting
+    for a timestamp inside a long fleet-telemetry recording."""
+
+    rosbag_dir: str = "~/run_policy_sessions"
+    """Directory for per-session rosbags recorded via ``record_rosbag``."""
+
 
 @dataclass(frozen=True)
 class Ros2DepthConsumerConfig:

src/holosoma_inference/holosoma_inference/run_policy.py

@@ -22,6 +22,7 @@
 from holosoma_inference.policies.dual_mode import DualModePolicy, _select_policy_class
 from holosoma_inference.utils.config_registry import parse_config
 from holosoma_inference.utils.misc import restore_terminal_settings
+from holosoma_inference.utils.session_recorder import SessionRecorder
 
 
 def _print_control_guide(policy_class, use_joystick: bool, dual_mode: bool = False):
@@ -120,7 +121,10 @@ def run_policy(config: InferenceConfig):
         logger.info("✅ Policy initialized successfully!")
         use_joystick = bool({"joystick", "interface"} & {config.task.velocity_input, config.task.state_input})
         _print_control_guide(policy_class, use_joystick, dual_mode=dual_mode)
-        policy.run()
+        # Optional per-session rosbag: records ros2 bag record --all for this run
+        # only, stopping (and printing the bag path) on Ctrl-C / completion / error.
+        with SessionRecorder.from_debug_config(config.task.debug):
+            policy.run()
         logger.info("✅ Policy execution completed!")
 
     except Exception as e:

src/holosoma_inference/holosoma_inference/utils/session_recorder.py

@@ -0,0 +1,86 @@
+"""Per-session rosbag recorder for run_policy.
+
+Spawns ``ros2 bag record --all`` in its own process group for the lifetime of a
+policy run, then SIGINTs it for a clean MCAP close on exit. Each run_policy
+invocation gets its own short, self-contained bag, so debugging a single run no
+longer means hunting for a timestamp inside a long fleet-telemetry recording.
+
+Usage::
+
+    with SessionRecorder.from_debug_config(cfg.task.debug):
+        policy.run()
+    # bag path is logged on exit (and available as the recorder's .bag_path)
+"""
+
+from __future__ import annotations
+
+import os
+import signal
+import subprocess
+import time
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from loguru import logger
+from typing_extensions import Self  # typing.Self is 3.11+; robot runs 3.10
+
+if TYPE_CHECKING:
+    from holosoma_inference.config.config_types.task import DebugConfig
+
+
+class SessionRecorder:
+    """Context manager that records a rosbag for the duration of the block.
+
+    No-op when ``enabled`` is False, so callers can wrap unconditionally. Failure
+    to start the recorder is logged and swallowed — a missing ``ros2`` or a bag
+    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"):
+        self.enabled = enabled
+        self.out_dir = Path(out_dir).expanduser()
+        self.storage = storage
+        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)
+
+    def __enter__(self) -> Self:
+        if not self.enabled:
+            return self
+        try:
+            self.out_dir.mkdir(parents=True, exist_ok=True)
+            self.bag_path = self.out_dir / time.strftime("run_policy_session_%Y%m%d_%H%M%S")
+            # fmt: off
+            cmd = [
+                "ros2", "bag", "record", "--all",
+                "--storage", self.storage,
+                "--disable-keyboard-controls",
+                "-o", str(self.bag_path),
+            ]
+            # fmt: on
+            # 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)")
+        except Exception as exc:  # never let recording break the run
+            logger.warning(f"Could not start session rosbag recorder: {exc}")
+            self._proc = None
+            self.bag_path = None
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        if self._proc is None:
+            return
+        # SIGINT for a clean MCAP close; SIGKILL if it won't exit.
+        try:
+            os.killpg(os.getpgid(self._proc.pid), signal.SIGINT)
+            self._proc.wait(timeout=15)
+        except subprocess.TimeoutExpired:
+            os.killpg(os.getpgid(self._proc.pid), signal.SIGKILL)
+        except OSError:
+            pass  # recorder already exited
+        if self.bag_path and self.bag_path.exists():
+            logger.info(f"💾 Session rosbag saved: {self.bag_path.resolve()}")