Add ROS publish transcripts and H1 metal hotpath

Author: AIFlowML

PORT_TO_MLX_TODO.md

@@ -95,6 +95,7 @@ without pausing for replanning after every small success.
 - `DONE` Env/runtime diagnostics now prove the articulated `mac-sim` contract on a real locomotion backend instead of only reporting the high-level backend seam
 - `DONE` ROS/planner software smokes now exercise the real `mac-planners` backend and verify typed round-trip reconstruction of planner world-state and joint trajectories
 - `DONE` Planner/ROS batch helpers now restore batches by `batch_index` and report actual batch envelope counts instead of inferring from message order or `max(index) + 1`
+- `DONE` Planner/ROS batch publish now also emits replayable plain-data transcripts with ordered input envelopes, generated CLI commands, and publish results so process-level failures can be audited after the fact without importing ROS Python bindings
 - `DONE` Stereo/depth smoke now validates raw capture artifacts before processing and writes a machine-checkable JSON summary artifact
 - `DONE` `uv run scripts/bootstrap_uv_mlx.py` now bootstraps the public MLX/mac editable environment in one command
 - `DONE` Upstream-compatible Franka reach/stack/open-drawer controller variants now resolve to the canonical mac-native manipulation slices through the lazy task registry, public MLX wrapper, and installed CLI without inflating the canonical public task surface

README.md

@@ -710,7 +710,7 @@ The current ROS 2 bridge is also intentionally plain. It focuses on message/proc
 - [`source/isaaclab/isaaclab/backends/ros2_compat.py`](source/isaaclab/isaaclab/backends/ros2_compat.py)
 - [`scripts/tools/ros2_bridge_smoke.py`](scripts/tools/ros2_bridge_smoke.py)
 
-Batch publish helpers now reject mixed topic-root batches across command building, typed manifest generation, and CLI publish execution, so planner world-state and joint-trajectory envelopes cannot be silently interleaved and replayed out of contract.
+Batch publish helpers now reject mixed topic-root batches across command building, typed manifest generation, and CLI publish execution, and the replayable batch transcript captures ordered input envelopes, commands, and publish results for deterministic audit, so planner world-state and joint-trajectory envelopes cannot be silently interleaved and replayed out of contract.
 
 Example planner smoke:
 
@@ -733,6 +733,7 @@ This is the current compatibility contract:
 - planner compatibility on macOS means serializable box/sphere/capsule/mesh world updates, attachment metadata, and deterministic timed joint-space plans
 - ROS compatibility on macOS means plain message/process interoperability first, including ROS-friendly world-state and joint-trajectory envelopes plus typed round-trip reconstruction without importing ROS Python bindings
 - batched ROS planner envelopes are reconstructed by `batch_index`, not by input order, so JSONL message reordering cannot silently corrupt planner/world batch recovery
+- batch publish transcripts preserve planned envelopes, generated commands, and publish results as plain data so process-level failures can be audited and replayed without ROS Python bindings
 - CUDA stream transport, NITROS, and GXF remain future follow-on work
 
 ## Kernel Inventory

scripts/tools/mac_planner_smoke.py

@@ -100,6 +100,8 @@ def main() -> int:
     trajectory_batch_envelopes = joint_motion_plan_batch_to_ros_envelopes(batch)
     trajectory_batch_roundtrip = joint_motion_plan_batch_from_ros_envelopes(tuple(reversed(trajectory_batch_envelopes)))
     ros_bridge = Ros2ProcessBridge()
+    planner_batch_publish_transcript = ros_bridge.build_batch_publish_transcript(planner_world_batch_envelopes)
+    trajectory_batch_publish_transcript = ros_bridge.build_batch_publish_transcript(trajectory_batch_envelopes)
 
     payload = {
         "planner": planner.state_dict(),
@@ -115,6 +117,8 @@ def main() -> int:
         == [item.state_dict() for item in batch],
         "planner_ros_batch_pub_commands": ros_bridge.build_topic_pub_batch_commands(planner_world_batch_envelopes),
         "trajectory_ros_batch_pub_commands": ros_bridge.build_topic_pub_batch_commands(trajectory_batch_envelopes),
+        "planner_ros_batch_publish_transcript": planner_batch_publish_transcript,
+        "trajectory_ros_batch_publish_transcript": trajectory_batch_publish_transcript,
     }
     args.output.parent.mkdir(parents=True, exist_ok=True)
     args.output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")

scripts/tools/ros2_bridge_smoke.py

@@ -129,6 +129,8 @@ def main() -> int:
     trajectory_batch_publish_manifest = [
         item.state_dict() for item in bridge.build_batch_publish_manifest(tuple(reversed(planner_plan_batch_envelopes)))
     ]
+    planner_batch_publish_transcript = bridge.build_batch_publish_transcript(tuple(reversed(planner_world_batch_envelopes)))
+    trajectory_batch_publish_transcript = bridge.build_batch_publish_transcript(tuple(reversed(planner_plan_batch_envelopes)))
 
     summary = {
         "cli_available": bridge.cli_available(),
@@ -153,6 +155,8 @@ def main() -> int:
         "trajectory_batch_pub_commands": trajectory_batch_commands,
         "planner_batch_publish_manifest": planner_batch_publish_manifest,
         "trajectory_batch_publish_manifest": trajectory_batch_publish_manifest,
+        "planner_batch_publish_transcript": planner_batch_publish_transcript,
+        "trajectory_batch_publish_transcript": trajectory_batch_publish_transcript,
         "message_summary": message_summary,
     }
     if args.summary_out is not None:

source/isaaclab/isaaclab/backends/mac_sim/hotpath.py

@@ -614,6 +614,77 @@ def _h1_leg_extension_impl(joint_pos: mx.array) -> mx.array:
     return mx.clip(extension, 0.58, 0.98)
 
 
+_h1_leg_extension_compiled = mx.compile(_h1_leg_extension_impl)
+_h1_leg_extension_metal = None
+_h1_leg_extension_hotpath_initialized = False
+H1_LEG_EXTENSION_HOTPATH_BACKEND = HOTPATH_BACKEND
+
+
+def _build_h1_leg_extension_metal_kernel():
+    if not hasattr(mx, "fast") or not hasattr(mx.fast, "metal_kernel"):
+        return None
+    source = r"""
+        uint env_id = thread_position_in_grid.x;
+        uint joint_dim = (uint)params[0];
+        uint base = env_id * joint_dim;
+        for (uint leg_id = 0; leg_id < 2; ++leg_id) {
+            uint joint_base = base + leg_id * 5;
+            float hip_pitch = joint_pos[joint_base + 2];
+            float knee = joint_pos[joint_base + 3];
+            float ankle = joint_pos[joint_base + 4];
+            float extension = 0.40f
+                + 0.20f * metal::cos(hip_pitch + 0.20f)
+                + 0.26f * metal::cos(hip_pitch + knee - 0.10f)
+                + 0.08f * metal::cos(hip_pitch + knee + ankle);
+            extension_out[env_id * 2 + leg_id] = metal::clamp(extension, 0.58f, 0.98f);
+        }
+    """
+    try:
+        return mx.fast.metal_kernel(
+            name="h1_leg_extension",
+            input_names=["joint_pos", "params"],
+            output_names=["extension_out"],
+            source=source,
+        )
+    except Exception:
+        return None
+
+
+def _ensure_h1_leg_extension_hotpath() -> None:
+    global _h1_leg_extension_metal
+    global _h1_leg_extension_hotpath_initialized
+    global H1_LEG_EXTENSION_HOTPATH_BACKEND
+    if _h1_leg_extension_hotpath_initialized:
+        return
+    _h1_leg_extension_hotpath_initialized = True
+    _h1_leg_extension_metal = _build_h1_leg_extension_metal_kernel()
+    if _h1_leg_extension_metal is not None:
+        H1_LEG_EXTENSION_HOTPATH_BACKEND = "mlx-metal-h1-leg-extension"
+
+
+def get_h1_leg_extension_hotpath_backend() -> str:
+    _ensure_h1_leg_extension_hotpath()
+    return H1_LEG_EXTENSION_HOTPATH_BACKEND
+
+
+def h1_leg_extension_hotpath(joint_pos: mx.array) -> mx.array:
+    _ensure_h1_leg_extension_hotpath()
+    joint_pos = mx.array(joint_pos[:, :10], dtype=mx.float32)
+    if _h1_leg_extension_metal is None:
+        return _h1_leg_extension_compiled(joint_pos)
+    if int(joint_pos.shape[0]) == 0:
+        return mx.zeros((0, 2), dtype=mx.float32)
+    params = mx.array([int(joint_pos.shape[1])], dtype=mx.float32)
+    outputs = _h1_leg_extension_metal(
+        inputs=[joint_pos, params],
+        grid=(int(joint_pos.shape[0]), 1, 1),
+        threadgroup=(64, 1, 1),
+        output_shapes=[(int(joint_pos.shape[0]), 2)],
+        output_dtypes=[mx.float32],
+    )
+    return outputs[0]
+
+
 def _h1_body_positions_impl(
     root_pos_w: mx.array,
     joint_pos: mx.array,
@@ -630,7 +701,7 @@ def _h1_body_positions_impl(
     hip_pitch = leg_pos[:, :, 2]
     knee = leg_pos[:, :, 3]
     ankle = leg_pos[:, :, 4]
-    extension = _h1_leg_extension_impl(joint_pos)
+    extension = h1_leg_extension_hotpath(joint_pos)
     command_speed = mx.linalg.norm(commands[:, :2], axis=1, keepdims=True)
     phase = gait_phase.reshape((num_envs, 1)) + gait_phase_offsets.reshape((1, 2))
     swing = mx.maximum(mx.sin(phase), 0.0) * (0.18 + 0.65 * command_speed)
@@ -659,7 +730,6 @@ def _h1_body_positions_impl(
 
 
 anymal_body_positions_hotpath = mx.compile(_anymal_body_positions_impl)
-h1_leg_extension_hotpath = mx.compile(_h1_leg_extension_impl)
 h1_body_positions_hotpath = mx.compile(_h1_body_positions_impl)
 
 

source/isaaclab/isaaclab/backends/ros2_compat.py

@@ -18,6 +18,7 @@
 
 
 ROS2_MESSAGE_ENVELOPE_SCHEMA_VERSION = 1
+ROS2_BATCH_PUBLISH_TRANSCRIPT_SCHEMA_VERSION = 1
 
 
 def _normalize_message_value(value: Any) -> Any:
@@ -399,6 +400,73 @@ def build_batch_publish_manifest(
             for batch_index, envelope in indexed_envelopes
         ]
 
+    def build_batch_publish_transcript(
+        self,
+        envelopes: list[Ros2MessageEnvelope] | tuple[Ros2MessageEnvelope, ...],
+        *,
+        once: bool = True,
+        records: list[Ros2BatchPublishRecord] | tuple[Ros2BatchPublishRecord, ...] | None = None,
+    ) -> dict[str, Any]:
+        """Build a replayable batch publish transcript for audit and deterministic replay."""
+
+        if not envelopes:
+            return {
+                "schema_version": ROS2_BATCH_PUBLISH_TRANSCRIPT_SCHEMA_VERSION,
+                "kind": "ros2_batch_publish_transcript",
+                "publish_state": "planned",
+                "topic_root": None,
+                "message_count": 0,
+                "batch_indices": [],
+                "topic_sequence": [],
+                "msg_type_sequence": [],
+                "command_sequence": [],
+                "input_envelope_sequence": [],
+                "record_sequence": [],
+                "record_count": 0,
+                "observed_record_count": 0,
+                "success": None,
+                "replayable": True,
+            }
+
+        indexed_envelopes = _sorted_batch_envelopes(envelopes, topic_root=_topic_root_for_batch(envelopes[0].topic))
+        planned_records = [
+            Ros2BatchPublishRecord(
+                batch_index=batch_index,
+                topic=envelope.topic,
+                msg_type=envelope.msg_type,
+                command=tuple(self.build_topic_pub_command(envelope, once=once)),
+            )
+            for batch_index, envelope in indexed_envelopes
+        ]
+        transcript_records = planned_records if records is None else list(records)
+        if len(transcript_records) > len(planned_records):
+            raise ValueError("record count cannot exceed planned batch size")
+        publish_state = "planned" if records is None else "published"
+        if records is not None and len(transcript_records) < len(planned_records):
+            publish_state = "failed"
+        elif records is not None and any(record.returncode not in (None, 0) for record in transcript_records):
+            publish_state = "failed"
+        success = None
+        if records is not None:
+            success = publish_state == "published" and len(transcript_records) == len(planned_records)
+        return {
+            "schema_version": ROS2_BATCH_PUBLISH_TRANSCRIPT_SCHEMA_VERSION,
+            "kind": "ros2_batch_publish_transcript",
+            "publish_state": publish_state,
+            "topic_root": _topic_root_for_batch(envelopes[0].topic),
+            "message_count": len(indexed_envelopes),
+            "batch_indices": [batch_index for batch_index, _ in indexed_envelopes],
+            "topic_sequence": [envelope.topic for _, envelope in indexed_envelopes],
+            "msg_type_sequence": [envelope.msg_type for _, envelope in indexed_envelopes],
+            "command_sequence": [list(record.command) for record in planned_records],
+            "input_envelope_sequence": [envelope.state_dict() for _, envelope in indexed_envelopes],
+            "record_sequence": [record.state_dict() for record in transcript_records],
+            "record_count": len(planned_records),
+            "observed_record_count": len(transcript_records),
+            "success": success,
+            "replayable": True,
+        }
+
     def publish_batch_via_cli(
         self,
         envelopes: list[Ros2MessageEnvelope] | tuple[Ros2MessageEnvelope, ...],
@@ -434,6 +502,11 @@ def publish_batch_via_cli(
                     f"returncode={exc.returncode}"
                 )
                 error.batch_publish_manifest = [item.state_dict() for item in results]
+                error.batch_publish_transcript = self.build_batch_publish_transcript(
+                    envelopes,
+                    once=once,
+                    records=results,
+                )
                 raise error from exc
             results.append(
                 replace(

source/isaaclab/test/backends/test_mac_hotpath.py

@@ -36,6 +36,8 @@
     get_franka_stack_rgb_hotpath_backend,
     get_locomotion_hotpath_backend,
     h1_body_positions_hotpath,
+    h1_leg_extension_hotpath,
+    get_h1_leg_extension_hotpath_backend,
     locomotion_root_step_hotpath,
     prime_contact_state,
     quadruped_support_metrics_hotpath,
@@ -186,6 +188,31 @@ def test_h1_body_positions_hotpath_returns_expected_shape_and_base_slot():
     assert np.allclose(np.array(body_pos)[:, 0, :], np.array(root_pos_w))
 
 
+def test_h1_leg_extension_hotpath_matches_reference_math():
+    joint_pos = mx.array(
+        [
+            [0.05, -0.25, 0.70, -0.15, 0.05, 0.12, 0.10, -0.08, 0.18, 0.22, -0.30, 0.04, 0.12, -0.07, 0.20, -0.04, 0.15, 0.08, -0.02, 0.11],
+            [0.00, -0.30, 0.60, 0.08, -0.10, 0.18, -0.05, 0.12, -0.16, 0.06, 0.20, -0.02, 0.14, 0.03, -0.12, 0.10, -0.18, 0.05, 0.09, -0.06],
+        ],
+        dtype=mx.float32,
+    )
+
+    extension = h1_leg_extension_hotpath(joint_pos)
+    mx.eval(extension)
+
+    joint_np = np.array(joint_pos)[:, :10].reshape((2, 2, 5))
+    hip_pitch = joint_np[:, :, 2]
+    knee = joint_np[:, :, 3]
+    ankle = joint_np[:, :, 4]
+    expected = 0.40 + 0.20 * np.cos(hip_pitch + 0.20) + 0.26 * np.cos(hip_pitch + knee - 0.10) + 0.08 * np.cos(
+        hip_pitch + knee + ankle
+    )
+    expected = np.clip(expected, 0.58, 0.98).astype(np.float32)
+
+    assert extension.shape == (2, 2)
+    assert np.allclose(np.array(extension), expected)
+
+
 def test_hotpath_backend_label_is_stable():
     assert HOTPATH_BACKEND == "mlx-compiled"
     assert get_anymal_leg_extension_hotpath_backend() in {"mlx-compiled", "mlx-metal-anymal-leg-extension"}
@@ -196,6 +223,7 @@ def test_hotpath_backend_label_is_stable():
     assert get_franka_stack_hotpath_backend() in {"mlx-compiled", "mlx-metal-franka-stack"}
     assert get_franka_stack_rgb_hotpath_backend() in {"mlx-compiled", "mlx-metal-franka-stack-rgb"}
     assert get_locomotion_hotpath_backend() in {"mlx-compiled", "mlx-metal-root-step"}
+    assert get_h1_leg_extension_hotpath_backend() in {"mlx-compiled", "mlx-metal-h1-leg-extension"}
 
 
 def test_franka_end_effector_hotpath_matches_reference_math():

source/isaaclab/test/backends/test_ros2_bridge.py

@@ -178,6 +178,39 @@ def test_ros2_process_bridge_builds_batch_publish_manifest_in_batch_index_order(
     assert manifest[0].command[4] == "/planner/world_state/0"
 
 
+def test_ros2_process_bridge_builds_batch_publish_transcript_in_batch_index_order():
+    """Batch publish transcripts should capture replayable envelope and command order."""
+    bridge = Ros2ProcessBridge()
+    envelopes = [
+        Ros2MessageEnvelope(
+            topic="/planner/world_state/1",
+            msg_type="robotflow_msgs/msg/PlannerWorldState",
+            payload={"frame_id": "world", "batch_index": 1},
+            batch_index=1,
+        ),
+        Ros2MessageEnvelope(
+            topic="/planner/world_state/0",
+            msg_type="robotflow_msgs/msg/PlannerWorldState",
+            payload={"frame_id": "world", "batch_index": 0},
+            batch_index=0,
+        ),
+    ]
+
+    transcript = bridge.build_batch_publish_transcript(tuple(envelopes))
+
+    assert transcript["schema_version"] == 1
+    assert transcript["kind"] == "ros2_batch_publish_transcript"
+    assert transcript["publish_state"] == "planned"
+    assert transcript["batch_indices"] == [0, 1]
+    assert transcript["topic_sequence"] == ["/planner/world_state/0", "/planner/world_state/1"]
+    assert transcript["msg_type_sequence"] == ["robotflow_msgs/msg/PlannerWorldState", "robotflow_msgs/msg/PlannerWorldState"]
+    assert transcript["record_count"] == 2
+    assert transcript["observed_record_count"] == 2
+    assert transcript["replayable"] is True
+    assert transcript["input_envelope_sequence"][0]["batch_index"] == 0
+    assert transcript["command_sequence"][0][4] == "/planner/world_state/0"
+
+
 def test_ros2_process_bridge_wraps_batch_publish_failures_with_batch_context(monkeypatch):
     """Batch publish failures should report the failing batch index and topic."""
     bridge = Ros2ProcessBridge()
@@ -219,6 +252,10 @@ def _publish_via_cli(envelope, *, once=True, check=True):
     assert manifest[-1]["topic"] == "/planner/world_state/1"
     assert manifest[-1]["returncode"] == 1
     assert manifest[-1]["stderr"] == "boom"
+    transcript = exc_info.value.batch_publish_transcript
+    assert transcript["publish_state"] == "failed"
+    assert transcript["observed_record_count"] == 2
+    assert transcript["record_sequence"][-1]["stderr"] == "boom"
 
 
 def test_ros2_process_bridge_rejects_invalid_batch_envelopes():
@@ -568,3 +605,6 @@ def test_ros2_bridge_smoke_uses_planner_backend_round_trip(tmp_path: Path):
     assert [item["batch_index"] for item in summary["trajectory_batch_publish_manifest"]] == [0, 1]
     assert summary["planner_batch_publish_manifest"][0]["returncode"] is None
     assert summary["trajectory_batch_publish_manifest"][0]["command"][4] == "/planner/joint_trajectory/0"
+    assert summary["planner_batch_publish_transcript"]["publish_state"] == "planned"
+    assert summary["trajectory_batch_publish_transcript"]["publish_state"] == "planned"
+    assert summary["planner_batch_publish_transcript"]["batch_indices"] == [0, 1]