feat(eval): export Pi conversation transcripts

[TomCC7]

Contribution path

• Linked PR: feat(evals): add code-as-policy evaluation runtime #3434
• Stacked on the code-policy evaluation runtime; merge feat(evals): add code-as-policy evaluation runtime #3434 first.

Problem

Pi evaluation runs retain useful conversation history, but raw session JSONL is difficult to review. Generic transcript rendering also hides or truncates the Python REPL calls and results that matter most in code-policy evaluations.

Solution

• retain native Pi session JSONL after successful and failed evaluations
• export a self-contained HTML conversation viewer while the evaluation workspace still exists
• render complete Python calls and results with syntax highlighting and no collapsed result state
• register the JSONL and HTML files as evaluation artifacts
• treat HTML export failures as nonfatal and retain a redacted diagnostic artifact

How to Test

npm --prefix packages/pi-code-policy-extension test
uv run pytest \
  dimos/benchmark/evaluation/test_pi_process.py \
  dimos/benchmark/evaluation/test_policy_runtime.py -q

Validated after restacking: 14 focused Python tests and the Pi extension test pass.

AI assistance

Codex with GPT-5.6 assisted with design discussion, implementation, tests, validation, and PR drafting. The author reviewed the behavior interactively against a real Pi transcript.

Checklist

• I have read and approved the CLA.

[codecov]

❌ 4 Tests Failed:

Tests completed	Failed	Passed	Skipped
4286	4	4282	69

View the top 3 failed test(s) by shortest run time

dimos.e2e_tests.test_dimsim_path_replaning::test_path_replanning

Stack Traces | 109s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76596a7fb0b0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76596a843100>
dim_sim = <dimos.e2e_tests.dim_sim_client.DimSimClient object at 0x76596abc7fb0>
direct_cmd_vel_explorer = <dimos.simulation.mujoco.direct_cmd_vel_explorer.DirectCmdVelExplorer object at 0x765a6e960590>
spawn_wall_on_pose = <function spawn_wall_on_pose.<locals>.spawn at 0x76596a2ec180>

    @pytest.mark.self_hosted_large
    def test_path_replanning(
        lcm_spy, start_blueprint, dim_sim, direct_cmd_vel_explorer, spawn_wall_on_pose
    ) -> None:
        start_blueprint(
            "--dimsim-scene=empty",
            "run",
            "unitree-go2-agentic",
            simulator="dimsim",
        )
        lcm_spy.save_topic(".../McpClient/on_system_modules/res")
        lcm_spy.wait_for_saved_topic(".../McpClient/on_system_modules/res", timeout=1200.0)
    
        # robot spawns at (3, 2)
    
        # side wall
        dim_sim.add_wall(2, -2.5, 12, -2.5)
        # other side wall
        dim_sim.add_wall(2, 3.5, 12, 3.5)
        # back wall (behind robot)
        dim_sim.add_wall(2, -2.5, 2, 3.5)
        # forward wall (far end)
        dim_sim.add_wall(12, -2.5, 12, 3.5)
        # dividing wall at x=7 with doors at y=[-1.5,-0.5] and y=[1.5,2.5]
        dim_sim.add_wall(7, -2.5, 7, -1.5)
        dim_sim.add_wall(7, -0.5, 7, 1.5)
        dim_sim.add_wall(7, 2.5, 7, 3.5)
    
        direct_cmd_vel_explorer.linear_speed = 0.8
>       direct_cmd_vel_explorer.follow_points([(10, 2), (2.5, 2), (3, 2)])

dim_sim    = <dimos.e2e_tests.dim_sim_client.DimSimClient object at 0x76596abc7fb0>
direct_cmd_vel_explorer = <dimos.simulation.mujoco.direct_cmd_vel_explorer.DirectCmdVelExplorer object at 0x765a6e960590>
lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76596a7fb0b0>
spawn_wall_on_pose = <function spawn_wall_on_pose.<locals>.spawn at 0x76596a2ec180>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76596a843100>

dimos/e2e_tests/test_dimsim_path_replaning.py:47: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../simulation/mujoco/direct_cmd_vel_explorer.py:105: in follow_points
    self._wait_for_pose()
        self       = <dimos.simulation.mujoco.direct_cmd_vel_explorer.DirectCmdVelExplorer object at 0x765a6e960590>
        waypoints  = [(10, 2), (2.5, 2), (3, 2)]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.simulation.mujoco.direct_cmd_vel_explorer.DirectCmdVelExplorer object at 0x765a6e960590>

    def _wait_for_pose(self) -> PoseStamped:
        self._new_pose.clear()
        self._new_pose.wait(timeout=5.0)
>       assert self._pose is not None, "No odom received"
E       AssertionError: No odom received

self       = <dimos.simulation.mujoco.direct_cmd_vel_explorer.DirectCmdVelExplorer object at 0x765a6e960590>

.../simulation/mujoco/direct_cmd_vel_explorer.py:65: AssertionError

dimos.e2e_tests.test_manipulation_planning_groups::test_single_arm_plans_and_executes_through_control_coordinator

Stack Traces | 138s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7659b92e6c00>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76596a2ee020>

    def test_single_arm_plans_and_executes_through_control_coordinator(
        lcm_spy: LcmSpy,
        start_blueprint: Callable[..., DimosCliCall],
    ) -> None:
        """Plan with one arm and execute through its trajectory task."""
        _start_openarm_mock_planner(start_blueprint, lcm_spy)
    
        client = RPCClient(None, ManipulationModule)
        coordinator_client = RPCClient(None, ControlCoordinator)
        try:
>           [left] = _wait_for_groups(client, 1)

client     = <dimos.core.rpc_client.RPCClient object at 0x7659d1ecfe30>
coordinator_client = <dimos.core.rpc_client.RPCClient object at 0x7659cf7a5b80>
lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7659b92e6c00>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76596a2ee020>

dimos/e2e_tests/test_manipulation_planning_groups.py:158: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

client = <dimos.core.rpc_client.RPCClient object at 0x7659d1ecfe30>, count = 1

    def _wait_for_groups(
        client: RPCClient,
        count: int,
        *,
        timeout: float = 120.0,
    ) -> tuple[PlanningGroupInfo, ...]:
        deadline = time.time() + timeout
        last_error: BaseException | None = None
        while time.time() < deadline:
            try:
                groups = client.list_planning_groups()
                if len(groups) == count:
                    return groups
            except Exception as exc:
                last_error = exc
            time.sleep(0.5)
>       raise TimeoutError(f"Timed out waiting for {count} planning groups") from last_error
E       TimeoutError: Timed out waiting for 1 planning groups

client     = <dimos.core.rpc_client.RPCClient object at 0x7659d1ecfe30>
count      = 1
deadline   = 1786744796.8268497
groups     = (PlanningGroupInfo('left_arm/manipulator', joints=('left_arm/openarm_left_joint1', 'left_arm/openarm_left_joint2', 'le...right_joint6', 'right_arm/openarm_right_joint7'), base='openarm_body_link0', tip='openarm_right_link7', gripper=False))
last_error = None
timeout    = 120.0

dimos/e2e_tests/test_manipulation_planning_groups.py:64: TimeoutError

dimos.e2e_tests.test_dimsim_walk_forward::test_walk_forward

Stack Traces | 194s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76596bbf5370>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76596a2ecfe0>
human_input = <function human_input.<locals>.send_human_input at 0x76596a2ed080>
dim_sim = <dimos.e2e_tests.dim_sim_client.DimSimClient object at 0x7659b9252cc0>

    @pytest.mark.self_hosted_large
    def test_walk_forward(lcm_spy, start_blueprint, human_input, dim_sim) -> None:
        start_blueprint(
            "run",
            "--disable",
            "spatial-memory",
            "--disable",
            "security-module",
            "unitree-go2-agentic",
            simulator="dimsim",
        )
        lcm_spy.save_topic(".../McpClient/on_system_modules/res")
        lcm_spy.wait_for_saved_topic(".../McpClient/on_system_modules/res", timeout=1200.0)
    
        origin_x, origin_y = 1, 2
        dim_sim.set_agent_position(origin_x, origin_y)
    
        human_input("move forward 3 meter")
    
>       lcm_spy.wait_until_odom_position(origin_x + 3, origin_y, threshold=0.4, timeout=120)

dim_sim    = <dimos.e2e_tests.dim_sim_client.DimSimClient object at 0x7659b9252cc0>
human_input = <function human_input.<locals>.send_human_input at 0x76596a2ed080>
lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76596bbf5370>
origin_x   = 1
origin_y   = 2
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76596a2ecfe0>

dimos/e2e_tests/test_dimsim_walk_forward.py:37: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/lcm_spy.py:167: in wait_until_odom_position
    self.wait_for_message_result(
        predicate  = <function LcmSpy.wait_until_odom_position.<locals>.predicate at 0x76596a2edda0>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76596bbf5370>
        threshold  = 0.4
        timeout    = 120
        x          = 4
        y          = 2
dimos/e2e_tests/lcm_spy.py:153: in wait_for_message_result
    wait_until(
        event      = <threading.Event at 0x76596bdf2d20: unset>
        fail_message = 'Failed to get to position x=4, y=2'
        listener   = <function LcmSpy.wait_for_message_result.<locals>.listener at 0x76596a2ee160>
        predicate  = <function LcmSpy.wait_until_odom_position.<locals>.predicate at 0x76596a2edda0>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x76596bbf5370>
        timeout    = 120
        topic      = '/odom#geometry_msgs.PoseStamped'
        type       = <class 'dimos.msgs.geometry_msgs.PoseStamped.PoseStamped'>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <bound method Event.is_set of <threading.Event at 0x76596bdf2d20: unset>>

    def wait_until(
        predicate: Callable[[], bool],
        *,
        timeout: float,
        interval: float = 0.1,
        message: str | None = None,
    ) -> None:
        """Poll ``predicate`` until it returns truthy or ``timeout`` elapses."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return
            time.sleep(interval)
>       raise TimeoutError(message or f"Timed out after {timeout}s waiting for condition")
E       TimeoutError: Failed to get to position x=4, y=2

deadline   = 3213742.506811044
interval   = 0.1
message    = 'Failed to get to position x=4, y=2'
predicate  = <bound method Event.is_set of <threading.Event at 0x76596bdf2d20: unset>>
timeout    = 120

.../utils/testing/waiting.py:35: TimeoutError

View the full list of 1 ❄️ flaky test(s)

dimos.e2e_tests.test_dimsim_spatial_memory::test_go_to_the_bed

Flake rate in main: 16.11% (Passed 125 times, Failed 24 times)

Stack Traces | 101s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x765a6ea6be30>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76596a2ecb80>
human_input = <function human_input.<locals>.send_human_input at 0x76596a2ecd60>
dim_sim = <dimos.e2e_tests.dim_sim_client.DimSimClient object at 0x76596c312150>
explore_house = <function explore_house.<locals>.explore at 0x76596a2ed440>

    @pytest.mark.self_hosted_large
    def test_go_to_the_bed(lcm_spy, start_blueprint, human_input, dim_sim, explore_house) -> None:
        start_blueprint(
            "run",
            "unitree-go2-agentic",
            simulator="dimsim",
        )
        lcm_spy.save_topic(".../McpClient/on_system_modules/res")
        lcm_spy.wait_for_saved_topic(".../McpClient/on_system_modules/res", timeout=1200.0)
    
>       explore_house()

dim_sim    = <dimos.e2e_tests.dim_sim_client.DimSimClient object at 0x76596c312150>
explore_house = <function explore_house.<locals>.explore at 0x76596a2ed440>
human_input = <function human_input.<locals>.send_human_input at 0x76596a2ecd60>
lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x765a6ea6be30>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x76596a2ecb80>

dimos/e2e_tests/test_dimsim_spatial_memory.py:28: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/conftest.py:249: in explore
    direct_cmd_vel_explorer.follow_points(points)
        direct_cmd_vel_explorer = <dimos.simulation.mujoco.direct_cmd_vel_explorer.DirectCmdVelExplorer object at 0x76596bbf4d70>
        points     = [(3.881, 4.803), (4.16, 1.615), (1.596, 1.505), (1.649, 0.137), (-3.644, -0.064), (-3.759, -2.661), ...]
.../simulation/mujoco/direct_cmd_vel_explorer.py:105: in follow_points
    self._wait_for_pose()
        self       = <dimos.simulation.mujoco.direct_cmd_vel_explorer.DirectCmdVelExplorer object at 0x76596bbf4d70>
        waypoints  = [(3.881, 4.803), (4.16, 1.615), (1.596, 1.505), (1.649, 0.137), (-3.644, -0.064), (-3.759, -2.661), ...]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.simulation.mujoco.direct_cmd_vel_explorer.DirectCmdVelExplorer object at 0x76596bbf4d70>

    def _wait_for_pose(self) -> PoseStamped:
        self._new_pose.clear()
        self._new_pose.wait(timeout=5.0)
>       assert self._pose is not None, "No odom received"
E       AssertionError: No odom received

self       = <dimos.simulation.mujoco.direct_cmd_vel_explorer.DirectCmdVelExplorer object at 0x76596bbf4d70>

.../simulation/mujoco/direct_cmd_vel_explorer.py:65: AssertionError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}