feat(evals): TypeSafe (Jev) navigation policy agent

[ruthwikdasyam]

Contribution path

Standalone draft off main. Scaffold for review — not ready to merge.

Problem

We want to evaluate TypeSafe's System One model (Jev) as a navigation policy, not as a grader: feed it a god-view world state, have it emit a direction, and drive the robot in sim until it reports it has arrived. Then grade the episode offline from the recording.

Solution

TypeSafePolicy is an Agent, so it plugs into the existing runner with no framework change. Each tick it assembles a WorldState, asks Jev for one world-frame unit step plus a termination flag, rotates the pick into the body frame, and publishes a single Twist. Grading is entirely post-hoc off the --record recording.

Two contracts are deliberately isolated for iteration:

• Input — WorldState: robot pose, goal, room bounds, and labeled axis-aligned obstacle boxes, all in one global ROS world frame. Obstacles/room/goal come from a scene JSON; pose comes live off /odom.
• Output — STEP_CRITERIA / build_questions(): a Choice over five coordinate-keyed unit steps (0,0) (0,1) (1,0) (0,-1) (-1,0), plus a reached Noul.

Design notes worth reviewing:

• Publishes /nav_cmd_vel, not /cmd_vel. MovementManager owns cmd_vel: Out[Twist] in the go2 stack, so publishing there would collide. This drops into the existing arbitration slot, keeping teleop override and stop_movement working.
• No skill access. The blueprint is ["unitree-go2", "mcp-server"] with no skill container, so MCP comes up with zero tools exposed. mcp-server cannot be dropped entirely — Sim.start() hard-blocks on McpAdapter.wait_for_ready().
• Code owns all geometry. The model picks a world-frame direction; the body-frame rotation stays in Python.
• Raw NNN-request.json / NNN-response.json naming matches what EvalRunner globs for request_attempts.

Scene

apartment_couch.json is generated from the real DimSim manifest (misc/DimSim/scenes/apartment/objects/manifest.json) by dimos.evals.suites.lib.dimsim_scene: 30 floor-standing obstacles with real centres, converted from Three.js Y-up to the ROS world frame using the same axis swap DimSimClient does. Wall-mounted items and tabletop clutter are filtered out (Jev's accuracy falls with irrelevant state).

Footprints are a uniform 0.5 m half-extent — the manifest carries positions but not extents (those live in the GLBs). Real centres, approximate sizes; --half-extent tunes it. A test pins the committed JSON to the generator output so it can't drift from the manifest silently.

The goal is the couch (sectional) with a 2.0 m arrival band, matching the DimSim-native go-to-couch rubric so both harnesses agree on success.

Dependency

New typesafe extra (typesafe-sdk>=0.6.0). The package shipped after the repo's 7-day exclude-newer window, so it gets an exclude-newer-package override — same pattern as the nine packages already listed there. TYPESAFE_API_KEY is added to default.env and the conftest passthrough.

uv sync --extra typesafe --inexact

How to Test

dimos evals run dimos.evals.suites.typesafe_nav \
    --agent dimos.evals.agents.typesafe_policy \
    --set scene_json=dimos/evals/suites/scenes/apartment_couch.json

Offline (no key, no sim):

uv run pytest dimos/evals/agents/test_typesafe_policy.py dimos/evals/suites/lib/test_dimsim_scene.py

Validation so far

• ruff check / ruff format / pre-commit clean.
• 17 offline tests pass — body-frame rotation at 0° and 90° yaw, confidence gate, scene round-trip, observe() merge, preflight rejection, manifest frame swap, clutter filter, room bounds, and committed-JSON-matches-generator.
• Live API verified with a real key: jev-1.13.0 answered a Choice + Noul fan-out and every field the agent reads (model, choice, confidence, probabilities, noul, usage.input_tokens/output_tokens) came back as coded.
• SDK signatures verified against the installed 0.6.0, not just the docs. This caught four bugs in the first commit (fixed in the second): model was never passed to the client, default model is jev-latest not jev-1.13, Usage fields are input_tokens/output_tokens so token metrics were silently zero, and the request timeout was implicit.
• Not run: a live sim eval, mypy (not in the local venv).

Known gaps

• Obstacle footprints are approximate (see Scene). Deriving real extents means parsing the GLBs.
• No baselines yet. A random-direction policy and a greedy-bearing policy are needed before any number here means anything — greedy-bearing is near-optimal in an open room, so the scene must require detours for the eval to discriminate.
• Jev is documented as unreliable at numeric comparison (jev-1.13 jaggedness). Feeding raw coordinates is a deliberate choice; if it underperforms, the fallback is to precompute qualitative facts (bearing words, per-direction clearance) in Python. Early signal: "standing next to the bed" scored reached = 0.27, so the reached instruction and a code-computed goal distance will need to agree on what "arrived" means.
• Call volume: ~180 API calls per case at a 1s tick over a 180s episode.

AI assistance

Claude (Claude Code) wrote the scaffold, generator, tests and PR body under user-directed design review; contracts and architecture decided by the user.

Checklist

• I have read and approved the CLA.

[codecov]

❌ 4 Tests Failed:

Tests completed	Failed	Passed	Skipped
5444	4	5440	217

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

::dimos.evals.agents.test_typesafe_policy

Stack Traces | 0s run time

ImportError while importing test module '.../evals/agents/test_typesafe_policy.py'.
Hint: make sure your test modules/packages have valid Python names.
Traceback:
.../evals/agents/test_typesafe_policy.py:26: in <module>
    from typesafe_sdk import Choice, ChoiceAnswer, Noul
E   ModuleNotFoundError: No module named 'typesafe_sdk'

dimos.evals.test_evals::test_suites_and_agents_importable

Stack Traces | 0.207s run time

def test_suites_and_agents_importable() -> None:
        """Modules construct without data or network (lambdas stay lazy)."""
    
        for module in (examples, go2_smoke, go2_vqa, dimsim_house, dimsim_pointcloud_mapping):
            assert module.SUITE, module.__name__
        agents = list_agents()
>       assert {m.rsplit(".", 1)[1] for m in agents} == {
            "question_answer",
            "blind",
            "mcp_client_adapter",
            "pi",
            "dimcode",
        }
E       AssertionError: assert {'blind', 'di...esafe_policy'} == {'blind', 'di...stion_answer'}
E         
E         Extra items in the left set:
E         #x1B[0m#x1B[33m'#x1B[39;49;00m#x1B[33mtypesafe_policy#x1B[39;49;00m#x1B[33m'#x1B[39;49;00m#x1B[90m#x1B[39;49;00m
E         
E         Full diff:
E         #x1B[0m#x1B[90m #x1B[39;49;00m {#x1B[90m#x1B[39;49;00m
E         #x1B[90m #x1B[39;49;00m     'blind',#x1B[90m#x1B[39;49;00m
E         #x1B[90m #x1B[39;49;00m     'dimcode',#x1B[90m#x1B[39;49;00m
E         #x1B[90m #x1B[39;49;00m     'mcp_client_adapter',#x1B[90m#x1B[39;49;00m
E         #x1B[90m #x1B[39;49;00m     'pi',#x1B[90m#x1B[39;49;00m
E         #x1B[90m #x1B[39;49;00m     'question_answer',#x1B[90m#x1B[39;49;00m
E         #x1B[92m+     'typesafe_policy',#x1B[39;49;00m#x1B[90m#x1B[39;49;00m
E         #x1B[90m #x1B[39;49;00m }#x1B[90m#x1B[39;49;00m

agents     = ['dimos.evals.agents.blind', 'dimos.evals.agents.dimcode', 'dimos.evals.agents.mcp_client_adapter', 'dimos.evals.agents.pi', 'dimos.evals.agents.question_answer', 'dimos.evals.agents.typesafe_policy']
module     = <module 'dimos.evals.suites.dimsim_pointcloud_mapping' from '.../evals/suites/dimsim_pointcloud_mapping.py'>

dimos/evals/test_evals.py:707: AssertionError

dimos.codebase_checks.test_no_sections::test_no_section_markers

Stack Traces | 0.91s run time

def test_no_section_markers():
        """
        Fail if any file contains section-style comment markers.
    
        If a file is too complicated to be understood without sections, then the
        sections should be files. We don't need "subfiles".
        """
        violations = find_section_markers()
        if violations:
            report_lines = [
                f"Found {len(violations)} section marker(s). "
                "If a file is too complicated to be understood without sections, "
                'then the sections should be files. We don\'t need "subfiles".',
                "",
            ]
            for path, lineno, text in violations:
                report_lines.append(f"  {path}:{lineno}: {text.strip()}")
>           raise AssertionError("\n".join(report_lines))
E           AssertionError: Found 8 section marker(s). If a file is too complicated to be understood without sections, then the sections should be files. We don't need "subfiles".
E           
E             .../evals/agents/test_typesafe_policy.py:107: # --- scene loading ---------------------------------------------------------------
E             .../evals/agents/test_typesafe_policy.py:159: # --- controller ------------------------------------------------------------------
E             .../evals/agents/test_typesafe_policy.py:196: # --- preflight -------------------------------------------------------------------
E             .../evals/agents/test_typesafe_policy.py:218: # --- goal from the instruction ---------------------------------------------------
E             .../evals/agents/typesafe_policy.py:50: # --- input contract: what Jev sees each tick ---------------------------------
E             .../evals/agents/typesafe_policy.py:87: # --- output contract: what Jev answers ---------------------------------------
E             .../evals/agents/typesafe_policy.py:129: # --- scene file --------------------------------------------------------------
E             .../evals/agents/typesafe_policy.py:280: # --- the agent ---------------------------------------------------------------

lineno     = 280
path       = '.../evals/agents/typesafe_policy.py'
report_lines = ['Found 8 section marker(s). If a file is too complicated to be understood without sections, then the sections should ...test_typesafe_policy.py:218: # --- goal from the instruction ---------------------------------------------------', ...]
text       = '# --- the agent ---------------------------------------------------------------'
violations = [('.../evals/agents/test_typesafe_policy.py', 107, '# --- scene loading --------------------------------------------...gents/typesafe_policy.py', 87, '# --- output contract: what Jev answers ---------------------------------------'), ...]

dimos/codebase_checks/test_no_sections.py:145: AssertionError

dimos.evals.test_evals::test_sim_launches_base_blueprints_and_agent_modules_in_order

Stack Traces | 4.03s run time

dataset = '.../popen-gw1/test_sim_launches_base_bluepri0/tiny.db'
mocker = <pytest_mock.plugin.MockerFixture object at 0xffb116eb3fe0>

    def test_sim_launches_base_blueprints_and_agent_modules_in_order(
        dataset: str, mocker: MockerFixture
    ) -> None:
        proc = mocker.patch("dimos.evals.environments.sim.DimosCliCall").return_value
        adapter = mocker.patch("dimos.evals.environments.sim.McpAdapter")
        adapter.return_value.wait_for_ready.return_value = True
        sim_client = mocker.patch("dimos.evals.environments.dimsim.DimSimClient")
        setup = mocker.Mock()
        env = _sim(
            scene="empty",
            launch_timeout_s=4.0,
            setup=setup,
            disable=("wavefront-frontier-explorer", "patrolling-module"),
        )
        mocker.patch.object(env, "_wait_recording", return_value=Path(dataset))
    
        try:
>           env.start(("mcp-client", "speak-skill"))

adapter    = <MagicMock name='McpAdapter' id='281136058100064'>
dataset    = '.../popen-gw1/test_sim_launches_base_bluepri0/tiny.db'
env        = <dimos.evals.environments.dimsim.DimSimEnvironment object at 0xffb116e5adb0>
mocker     = <pytest_mock.plugin.MockerFixture object at 0xffb116eb3fe0>
proc       = <MagicMock name='DimosCliCall()' id='281136058096368'>
setup      = <Mock id='281136058446864'>
sim_client = <MagicMock name='DimSimClient' id='281136058442976'>

dimos/evals/test_evals.py:293: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../evals/environments/sim.py:155: in start
    artifacts.update(self.prepare_recording(self._recording, path, deadline))
        SqliteStore = <class 'dimos.memory.store.sqlite.SqliteStore'>
        adapter    = <MagicMock name='McpAdapter()' id='281136052386368'>
        artifacts  = {'recording': PosixPath('.../popen-gw1/test_sim_launches_base_bluepri0/tiny.db')}
        bridge     = []
        deadline   = 552.908788025
        disabled   = ['--disable', 'wavefront-frontier-explorer', '--disable', 'patrolling-module']
        mcp_url    = 'http://localhost:22683/mcp'
        modules    = ('mcp-client', 'speak-skill')
        path       = PosixPath('.../popen-gw1/test_sim_launches_base_bluepri0/tiny.db')
        pid        = <MagicMock name='DimosCliCall().process.pid' id='281136058637376'>
        proc       = <MagicMock name='DimosCliCall()' id='281136058096368'>
        process    = <MagicMock name='DimosCliCall().process' id='281136052381904'>
        self       = <dimos.evals.environments.dimsim.DimSimEnvironment object at 0xffb116e5adb0>
.../evals/environments/dimsim.py:62: in prepare_recording
    self.wait_ready(recording, deadline=deadline)
        deadline   = 552.908788025
        path       = PosixPath('.../popen-gw1/test_sim_launches_base_bluepri0/tiny.db')
        recording  = <dimos.memory.store.sqlite.SqliteStore object at 0xffb11689ef00>
        self       = <dimos.evals.environments.dimsim.DimSimEnvironment object at 0xffb116e5adb0>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.evals.environments.dimsim.DimSimEnvironment object at 0xffb116e5adb0>
recording = <dimos.memory.store.sqlite.SqliteStore object at 0xffb11689ef00>

    def wait_ready(self, recording: Store, *, deadline: float) -> None:
        # The Observation's ts is the recorder's wall clock, which is set even
        # when the sim leaves PoseStamped.ts at zero.
        while time.monotonic() < deadline:
            try:
                if "odom" in recording.streams:
                    latest = recording.streams.odom.last()
                    if time.time() - latest.ts < self.config.fresh_odom_s:
                        return
            except LookupError:
                pass
            time.sleep(0.1)
>       raise TimeoutError("DimSim did not publish fresh odometry before the launch deadline")
E       TimeoutError: DimSim did not publish fresh odometry before the launch deadline

deadline   = 552.908788025
latest     = Observation(id=5, ts=1004.0, data_type=<class 'dimos.msgs.geometry_msgs.PoseStamped.PoseStamped'>, pose_tuple=None, ta...n Backend._make_loader.<locals>.loader at 0xffb1168e7560>, _data_lock=<unlocked _thread.lock object at 0xffb116e4eb00>)
recording  = <dimos.memory.store.sqlite.SqliteStore object at 0xffb11689ef00>
self       = <dimos.evals.environments.dimsim.DimSimEnvironment object at 0xffb116e5adb0>

.../evals/environments/dimsim.py:77: TimeoutError

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