Skip to content

feat(agents): TypeSafeAgent: world state in, joystick cmd_vel out - #4212

Draft
spomichter wants to merge 3 commits into
mainfrom
feat/typesafe-agent-v2
Draft

spomichter wants to merge 3 commits into
mainfrom
feat/typesafe-agent-v2

Conversation

@spomichter

@spomichter spomichter commented Sep 18, 2026

Copy link
Copy Markdown
Contributor

Problem

DimOS agents are LLM chat loops. There is no way to run a fast, calibrated, text-only decision model as a reactive controller: perception in, typed decisions out, at a fixed rate, with confidence the code can gate on.

Solution

TypeSafeAgent (dimos/agents/typesafe/: world_state.py in, drive.py adapter out, agent.py module) drives a robot from TypeSafe System One decisions. Rewrite of #4210 from scratch on main, applying the review there.

  • State: world_state.build_world_state() folds odom (PoseStamped or Odometry), 3D or 2D detections and the scan into one JSON state. New typed to_json() on PoseStamped, Detection2DArray, Detection3DArray, PointCloud2 (8-sector clearance in the robot frame). Numbers carry word buckets (bearing, distance, sector state).
  • Questions: drive.questions() builds one 3-way Choice per joystick axis (x, y, yaw), a stop Noul and a target Choice over the visible labels, in the documented API shapes via client.choice() / client.noul().
  • Request: one requests.post in agent._post to POST /v1/systemone with {state, model, questions}. No client class, no retries: a failed tick zeroes the target and the next tick asks again.
  • Drive: drive.decode() gates each axis on confidence; agent._steer() scales turn rate by bearing error and speed by distance to the chosen target, declares arrival under 0.5 m, ramps the Twist at 10 Hz from its own thread, zeroes on a deadman of max(1 s, 2.5 / rate_hz). Holds (no request) without odom or detections; gives up after 5 s without motion.
  • Keeps the human_input / agent / agent_idle contract, so humancli and the cockpit chat attach unchanged.

Blueprints:

  • habitat-typesafe: habitat_teleop + HabitatConnection(publish_objects=True). HM3D annotations carry no object boxes, so the native renders the semantic sensor and boxes each visible instance from its depth pixels. Generated LCM constructors share nested defaults; Detection3D is built fresh at every level.
  • unitree-go2-typesafe, with a DimSim e2e that publishes a chair and drives to it.

Spec: dimensionalOS/agents engineering/typesafe-agent/spec.md.

Testing

  • 54 unit tests (message to_json, world-state buckets, drive decoding, agent loop with a fake client: hold, arrival, odometry input, failure, deadman; habitat server round trips incl. two-object boxes). mypy clean on all changed files.
  • Live habitat run on this branch (habitat-typesafe, headless, traced): chair 4.5 m ahead at spawn, target=chair, drove 3.8 m and stopped at the chair.
  • DimSim e2e (test_dimsim_typesafe_agent.py) passes on this branch: 86 s, target confidence 1.0.

Run: TYPESAFE_API_KEY=... uv run dimos run habitat-typesafe, then uv run humancli --transport zenoh and go to the chair.

Supersedes #4210.

A reactive text-only agent on TypeSafe System One. Each tick folds odom,
3D (or 2D) detections and the scan into one JSON state, asks one request
(a 3-way Choice per joystick axis, a stop Noul, a target Choice over the
visible labels) and drives: the model picks directions, code sets
magnitudes from the target's bearing and distance, ramps the Twist at
10 Hz and zeroes it on a deadman. Holds without odom or detections.

`to_json` on PoseStamped, Detection2DArray, Detection3DArray and
PointCloud2 (8-sector clearance), all TypedDict-typed.

`habitat-typesafe`: HabitatConnection(publish_objects=True) renders the
semantic sensor and boxes each visible annotated instance from its depth
pixels (HM3D annotations carry no object boxes); generated LCM
constructors share nested defaults, so Detection3D is built fresh at
every level. `unitree-go2-typesafe` plus a DimSim e2e that publishes a
chair and drives to it.
@codecov

codecov Bot commented Sep 18, 2026

Copy link
Copy Markdown

❌ 2 Tests Failed:

Tests completed Failed Passed Skipped
5536 2 5534 217
View the top 2 failed test(s) by shortest run time
dimos.evals.environments.test_habitat::test_explicit_spawn_uses_real_constructor[False]
Stack Traces | 0.006s run time
mocker = <pytest_mock.plugin.MockerFixture object at 0xff76d9fc8350>
navigable = False

    @pytest.mark.parametrize("navigable", [True, False])
    def test_explicit_spawn_uses_real_constructor(mocker, navigable):
        from dimos.simulation.habitat.server import HabitatHost
    
        hs = Mock()
        mocker.patch.dict("sys.modules", {"habitat_sim": hs})
        sim = hs.Simulator.return_value
        sim.pathfinder.is_navigable.return_value = navigable
        config = dict(
            scene_id="example",
            scene_dataset_config="default",
            seed=4,
            start_position_ros=(1, 2, 3),
            width=640,
            height=360,
            hfov_deg=90,
            camera_height_m=0.45,
        )
        if not navigable:
            with pytest.raises(ValueError, match="not navigable"):
>               HabitatHost(config)

HabitatHost = <class 'dimos.simulation.habitat.server.HabitatHost'>
config     = {'camera_height_m': 0.45, 'height': 360, 'hfov_deg': 90, 'scene_dataset_config': 'default', ...}
hs         = <Mock id='280885927684832'>
mocker     = <pytest_mock.plugin.MockerFixture object at 0xff76d9fc8350>
navigable  = False
sim        = <Mock name='mock.Simulator()' id='280885928427744'>

.../evals/environments/test_habitat.py:182: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../simulation/habitat/server.py:313: in __init__
    self._build(cfg["scene_id"])
        cfg        = {'camera_height_m': 0.45, 'height': 360, 'hfov_deg': 90, 'scene_dataset_config': 'default', ...}
        habitat_sim = <Mock id='280885927684832'>
        self       = <dimos.simulation.habitat.server.HabitatHost object at 0xff76d9fc8410>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.simulation.habitat.server.HabitatHost object at 0xff76d9fc8410>
scene_id = 'example'

    def _build(self, scene_id: str) -> None:
        hs = self.hs
        backend = hs.SimulatorConfiguration()
        backend.scene_dataset_config_file = self.cfg["scene_dataset_config"]
        backend.scene_id = scene_id
        backend.enable_physics = False
        backend.random_seed = int(self.cfg.get("seed", 0))
    
        def cam(uuid: str, sensor_type: Any) -> Any:
            spec = hs.CameraSensorSpec()
            spec.uuid = uuid
            spec.sensor_type = sensor_type
            spec.resolution = [self.height, self.width]
            spec.position = [0.0, self.camera_height, 0.0]
            spec.hfov = self.hfov
            return spec
    
        specs = [cam("rgb", hs.SensorType.COLOR), cam("depth", hs.SensorType.DEPTH)]
        if self.publish_semantic:
            specs.append(cam("semantic", hs.SensorType.SEMANTIC))
    
        agent_cfg = hs.agent.AgentConfiguration()
        agent_cfg.sensor_specifications = specs
        # Actions are unused: motion is applied as a twist against the navmesh.
        agent_cfg.action_space = {}
    
        if self._sim is not None:
            self._sim.close()
        self._sim = hs.Simulator(hs.Configuration(backend, [agent_cfg]))
        self._agent = self._sim.initialize_agent(0)
>       self.labels = [
            o.category.name() if o.category is not None else ""
            for o in self._sim.semantic_scene.objects
        ]
E       TypeError: 'Mock' object is not iterable

agent_cfg  = <Mock name='mock.agent.AgentConfiguration()' id='280885928428944'>
backend    = <Mock name='mock.SimulatorConfiguration()' id='280885939636848'>
cam        = <function HabitatHost._build.<locals>.cam at 0xff76d9fb0900>
hs         = <Mock id='280885927684832'>
scene_id   = 'example'
self       = <dimos.simulation.habitat.server.HabitatHost object at 0xff76d9fc8410>
specs      = [<Mock name='mock.CameraSensorSpec()' id='280885928428800'>, <Mock name='mock.CameraSensorSpec()' id='280885928428800'>]

.../simulation/habitat/server.py:345: TypeError
dimos.evals.environments.test_habitat::test_explicit_spawn_uses_real_constructor[True]
Stack Traces | 0.008s run time
mocker = <pytest_mock.plugin.MockerFixture object at 0xff76d9f12c90>
navigable = True

    @pytest.mark.parametrize("navigable", [True, False])
    def test_explicit_spawn_uses_real_constructor(mocker, navigable):
        from dimos.simulation.habitat.server import HabitatHost
    
        hs = Mock()
        mocker.patch.dict("sys.modules", {"habitat_sim": hs})
        sim = hs.Simulator.return_value
        sim.pathfinder.is_navigable.return_value = navigable
        config = dict(
            scene_id="example",
            scene_dataset_config="default",
            seed=4,
            start_position_ros=(1, 2, 3),
            width=640,
            height=360,
            hfov_deg=90,
            camera_height_m=0.45,
        )
        if not navigable:
            with pytest.raises(ValueError, match="not navigable"):
                HabitatHost(config)
        else:
>           host = HabitatHost(config)

HabitatHost = <class 'dimos.simulation.habitat.server.HabitatHost'>
config     = {'camera_height_m': 0.45, 'height': 360, 'hfov_deg': 90, 'scene_dataset_config': 'default', ...}
hs         = <Mock id='280885927684640'>
mocker     = <pytest_mock.plugin.MockerFixture object at 0xff76d9f12c90>
navigable  = True
sim        = <Mock name='mock.Simulator()' id='280885927685792'>

.../evals/environments/test_habitat.py:184: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../simulation/habitat/server.py:313: in __init__
    self._build(cfg["scene_id"])
        cfg        = {'camera_height_m': 0.45, 'height': 360, 'hfov_deg': 90, 'scene_dataset_config': 'default', ...}
        habitat_sim = <Mock id='280885927684640'>
        self       = <dimos.simulation.habitat.server.HabitatHost object at 0xff76da9c87a0>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.simulation.habitat.server.HabitatHost object at 0xff76da9c87a0>
scene_id = 'example'

    def _build(self, scene_id: str) -> None:
        hs = self.hs
        backend = hs.SimulatorConfiguration()
        backend.scene_dataset_config_file = self.cfg["scene_dataset_config"]
        backend.scene_id = scene_id
        backend.enable_physics = False
        backend.random_seed = int(self.cfg.get("seed", 0))
    
        def cam(uuid: str, sensor_type: Any) -> Any:
            spec = hs.CameraSensorSpec()
            spec.uuid = uuid
            spec.sensor_type = sensor_type
            spec.resolution = [self.height, self.width]
            spec.position = [0.0, self.camera_height, 0.0]
            spec.hfov = self.hfov
            return spec
    
        specs = [cam("rgb", hs.SensorType.COLOR), cam("depth", hs.SensorType.DEPTH)]
        if self.publish_semantic:
            specs.append(cam("semantic", hs.SensorType.SEMANTIC))
    
        agent_cfg = hs.agent.AgentConfiguration()
        agent_cfg.sensor_specifications = specs
        # Actions are unused: motion is applied as a twist against the navmesh.
        agent_cfg.action_space = {}
    
        if self._sim is not None:
            self._sim.close()
        self._sim = hs.Simulator(hs.Configuration(backend, [agent_cfg]))
        self._agent = self._sim.initialize_agent(0)
>       self.labels = [
            o.category.name() if o.category is not None else ""
            for o in self._sim.semantic_scene.objects
        ]
E       TypeError: 'Mock' object is not iterable

agent_cfg  = <Mock name='mock.agent.AgentConfiguration()' id='280885927684688'>
backend    = <Mock name='mock.SimulatorConfiguration()' id='280885938915856'>
cam        = <function HabitatHost._build.<locals>.cam at 0xff76dac9eca0>
hs         = <Mock id='280885927684640'>
scene_id   = 'example'
self       = <dimos.simulation.habitat.server.HabitatHost object at 0xff76da9c87a0>
specs      = [<Mock name='mock.CameraSensorSpec()' id='280885939644720'>, <Mock name='mock.CameraSensorSpec()' id='280885939644720'>]

.../simulation/habitat/server.py:345: TypeError

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

… publishes goal: PointStamped and steers to it out of view
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant