diff --git a/.basedpyright/baseline.json b/.basedpyright/baseline.json
index 01c7315f3..531aa5996 100644
--- a/.basedpyright/baseline.json
+++ b/.basedpyright/baseline.json
@@ -1163,6 +1163,14 @@
             }
         ],
         "./positronic/cfg/policy.py": [
+            {
+                "code": "reportMissingImports",
+                "range": {
+                    "startColumn": 9,
+                    "endColumn": 42,
+                    "lineCount": 1
+                }
+            },
             {
                 "code": "reportArgumentType",
                 "range": {
@@ -8215,54 +8223,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportCallIssue",
-                "range": {
-                    "startColumn": 4,
-                    "endColumn": 73,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 31,
-                    "endColumn": 57,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportCallIssue",
-                "range": {
-                    "startColumn": 4,
-                    "endColumn": 67,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 31,
-                    "endColumn": 51,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportCallIssue",
-                "range": {
-                    "startColumn": 4,
-                    "endColumn": 83,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 31,
-                    "endColumn": 52,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOperatorIssue",
                 "range": {
@@ -8537,14 +8497,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 55,
-                    "endColumn": 63,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportAttributeAccessIssue",
                 "range": {
@@ -10541,6 +10493,14 @@
             }
         ],
         "./positronic/simulator/robolab/env.py": [
+            {
+                "code": "reportMissingImports",
+                "range": {
+                    "startColumn": 7,
+                    "endColumn": 12,
+                    "lineCount": 1
+                }
+            },
             {
                 "code": "reportMissingImports",
                 "range": {
@@ -10891,6 +10851,14 @@
             }
         ],
         "./positronic/simulator/robolab/validate.py": [
+            {
+                "code": "reportMissingImports",
+                "range": {
+                    "startColumn": 7,
+                    "endColumn": 12,
+                    "lineCount": 1
+                }
+            },
             {
                 "code": "reportMissingImports",
                 "range": {
@@ -10997,22 +10965,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportCallIssue",
-                "range": {
-                    "startColumn": 11,
-                    "endColumn": 26,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 11,
-                    "endColumn": 26,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportAttributeAccessIssue",
                 "range": {
@@ -11109,94 +11061,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportAttributeAccessIssue",
-                "range": {
-                    "startColumn": 36,
-                    "endColumn": 40,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportCallIssue",
-                "range": {
-                    "startColumn": 14,
-                    "endColumn": 33,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 14,
-                    "endColumn": 33,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportCallIssue",
-                "range": {
-                    "startColumn": 15,
-                    "endColumn": 25,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 15,
-                    "endColumn": 25,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportCallIssue",
-                "range": {
-                    "startColumn": 9,
-                    "endColumn": 41,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 9,
-                    "endColumn": 41,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportCallIssue",
-                "range": {
-                    "startColumn": 15,
-                    "endColumn": 39,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 15,
-                    "endColumn": 39,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportCallIssue",
-                "range": {
-                    "startColumn": 35,
-                    "endColumn": 43,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 35,
-                    "endColumn": 43,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportArgumentType",
                 "range": {
diff --git a/positronic/cfg/analysis.py b/positronic/cfg/analysis.py
index 26e59a247..f5a6a215c 100644
--- a/positronic/cfg/analysis.py
+++ b/positronic/cfg/analysis.py
@@ -7,6 +7,7 @@
 import pos3
 
 import positronic.cfg.ds as base_cfg
+from positronic import keys
 from positronic.cfg.ds import internal
 from positronic.cfg.eval.real import tasks
 from positronic.dataset.episode import Episode
@@ -20,7 +21,7 @@
 def task_code(ep: Episode) -> str:
     if 'eval.object' in ep:
         return ep['eval.object']
-    match ep['task']:
+    match ep[keys.TASK]:
         case tasks.TOWELS_TASK:
             return 'Towels'
         case tasks.SPOONS_TASK:
@@ -340,9 +341,9 @@ def box_distance_progress(episode: Episode) -> float | None:
 
 
 def ee_pose_movement(episode: Episode) -> float | None:
-    if 'robot_state.ee_pose' not in episode:
+    if keys.EE_POSE not in episode:
         return None
-    signal_values = episode['robot_state.ee_pose'].values()
+    signal_values = episode[keys.EE_POSE].values()
     result = 0.0
     prev_translation = signal_values[0][:3]
     for ee_pose in signal_values[1:]:
@@ -486,20 +487,20 @@ def group_fn(episodes: list[Episode]):
 
 def calculate_units(episode: Episode) -> int:
     """Estimates the number of pick-and-place operations. Vibe-coded heuristic."""
-    if episode['task'] in FIXED_ITEM_COUNTS:
-        return FIXED_ITEM_COUNTS[episode['task']]
+    if episode[keys.TASK] in FIXED_ITEM_COUNTS:
+        return FIXED_ITEM_COUNTS[episode[keys.TASK]]
 
     if 'target_grip' in episode.signals:
         grip_sig = episode.signals['target_grip']
-    elif 'grip' in episode.signals:
-        grip_sig = episode.signals['grip']
+    elif keys.GRIP in episode.signals:
+        grip_sig = episode.signals[keys.GRIP]
     else:
         return 0
 
-    if 'robot_state.ee_pose' not in episode.signals:
+    if keys.EE_POSE not in episode.signals:
         return 0
 
-    pose_sig = episode.signals['robot_state.ee_pose']
+    pose_sig = episode.signals[keys.EE_POSE]
 
     # Sample signals at 10Hz to reduce noise and computation
     times = np.arange(episode.start_ts, episode.last_ts, int(1e8))
diff --git a/positronic/cfg/codecs.py b/positronic/cfg/codecs.py
index 9d84bff18..98d0a7647 100644
--- a/positronic/cfg/codecs.py
+++ b/positronic/cfg/codecs.py
@@ -2,7 +2,7 @@
 
 import configuronic as cfn
 
-from positronic import geom
+from positronic import geom, keys
 from positronic.policy.observation import ObservationCodec
 
 RotRep = geom.Rotation.Representation
@@ -19,27 +19,27 @@ def general_obs(
 
 
 eepose_grip_obs = general_obs.override(
-    state_name='observation.state', state_features={'robot_state.ee_pose': 7, 'grip': 1}, image_size=(224, 224)
+    state_name='observation.state', state_features={keys.EE_POSE: 7, keys.GRIP: 1}, image_size=(224, 224)
 )
 
 joints_grip_obs = general_obs.override(
-    state_name='observation.state', state_features={'robot_state.q': 7, 'grip': 1}, image_size=(224, 224)
+    state_name='observation.state', state_features={keys.JOINTS: 7, keys.GRIP: 1}, image_size=(224, 224)
 )
 
 eepose_grip_joints_obs = general_obs.override(
     state_name='observation.state',
-    state_features={'robot_state.ee_pose': 7, 'grip': 1, 'robot_state.q': 7},
+    state_features={keys.EE_POSE: 7, keys.GRIP: 1, keys.JOINTS: 7},
     image_size=(224, 224),
 )
 
 eepose_obs = eepose_grip_obs.override(
-    image_mappings={'observation.images.left': 'image.wrist', 'observation.images.side': 'image.exterior'}
+    image_mappings={'observation.images.left': keys.WRIST_IMAGE, 'observation.images.side': keys.EXTERIOR_IMAGE}
 )
 joints_obs = joints_grip_obs.override(
-    image_mappings={'observation.images.left': 'image.wrist', 'observation.images.side': 'image.exterior'}
+    image_mappings={'observation.images.left': keys.WRIST_IMAGE, 'observation.images.side': keys.EXTERIOR_IMAGE}
 )
 eepose_joints_obs = eepose_grip_joints_obs.override(
-    image_mappings={'observation.images.left': 'image.wrist', 'observation.images.side': 'image.exterior'}
+    image_mappings={'observation.images.left': keys.WRIST_IMAGE, 'observation.images.side': keys.EXTERIOR_IMAGE}
 )
 
 
@@ -96,14 +96,14 @@ def joint_delta_action(num_joints: int):
     return JointDeltaAction(num_joints=num_joints)
 
 
-traj_ee_action = absolute_pos_action.override(tgt_ee_pose_key='robot_state.ee_pose', tgt_grip_key='grip')
+traj_ee_action = absolute_pos_action.override(tgt_ee_pose_key=keys.EE_POSE, tgt_grip_key=keys.GRIP)
 
 
 @cfn.config(
     solver='dls_limits',
     tgt_ee_pose_key='robot_command.pose',
     tgt_grip_key='target_grip',
-    current_q_key='robot_state.q',
+    current_q_key=keys.JOINTS,
     num_joints=7,
 )
 def ik_joints_action(solver, tgt_ee_pose_key, tgt_grip_key, current_q_key, num_joints):
diff --git a/positronic/cfg/ds/internal.py b/positronic/cfg/ds/internal.py
index 93fd47d75..7fc9152d8 100644
--- a/positronic/cfg/ds/internal.py
+++ b/positronic/cfg/ds/internal.py
@@ -13,6 +13,7 @@
 import numpy as np
 import pos3
 
+from positronic import keys
 from positronic.cfg.eval.real.tasks import BATTERIES_TASK, SCISSORS_TASK, SPOONS_TASK, TOWELS_TASK
 from positronic.dataset.dataset import ConcatDataset, FilterDataset
 from positronic.dataset.local_dataset import load_all_datasets
@@ -35,7 +36,7 @@
 # transform, so the transform always supplies them — overriding any stale value a recording baked
 # into static (pre-rename recordings carry `robot_commands.pose` in their `pose_signals`).
 _ROBOT_SIGNAL_POINTERS = Derive(
-    joint_signal=FromValue('robot_state.q'), pose_signals=FromValue(['robot_state.ee_pose', 'robot_command.pose'])
+    joint_signal=FromValue(keys.JOINTS), pose_signals=FromValue([keys.EE_POSE, 'robot_command.pose'])
 )
 
 # The bundled model (URDF + collision meshes + joint names + control frame + gripper) for the real
@@ -116,16 +117,16 @@ def _flip_grip(key: str):
 old_to_new = Group(
     Derive(**{
         'robot_command.pose': Concat('target_robot_position_translation', 'target_robot_position_quaternion'),
-        'robot_state.ee_pose': Concat('robot_position_translation', 'robot_position_quaternion'),
-        'task': FromValue('Pick up the green cube and place it on the red cube.'),
-        'grip': _flip_grip('grip'),
+        keys.EE_POSE: Concat('robot_position_translation', 'robot_position_quaternion'),
+        keys.TASK: FromValue('Pick up the green cube and place it on the red cube.'),
+        keys.GRIP: _flip_grip(keys.GRIP),
         'target_grip': _flip_grip('target_grip'),
     }),
     Rename(**{
-        'robot_state.q': 'robot_joints',
-        'robot_state.dq': 'robot_joints_velocity',
-        'image.wrist': 'image.handcam_left',
-        'image.exterior': 'image.back_view',
+        keys.JOINTS: 'robot_joints',
+        keys.JOINT_VEL: 'robot_joints_velocity',
+        keys.WRIST_IMAGE: 'image.handcam_left',
+        keys.EXTERIOR_IMAGE: 'image.back_view',
     }),
     Identity(select=['mjSTATE_FULLPHYSICS', 'mjSTATE_INTEGRATION', 'mjSTATE_WARMSTART']),
 )
@@ -140,10 +141,10 @@ def _flip_grip(key: str):
         Group(
             Derive(
                 task=FromValue('Pick up objects from the red tote and place them in the green tote.'),
-                grip=_flip_grip('grip'),
+                grip=_flip_grip(keys.GRIP),
                 target_grip=_flip_grip('target_grip'),
             ),
-            Rename(**{'image.exterior': 'image.back_view'}),
+            Rename(**{keys.EXTERIOR_IMAGE: 'image.back_view'}),
             Identity(),
         ),
         SIM_ROBOT_TRANSFORM,
@@ -191,7 +192,7 @@ def droid_clean(dataset):
 # (combined with --share/--seed in the LeRobot conversion to produce 100% / 50% / 25% variants).
 @cfn.config(dataset=droid_clean)
 def droid_spoons(dataset):
-    return FilterDataset(dataset, lambda ep: ep.static.get('task') == SPOONS_TASK)
+    return FilterDataset(dataset, lambda ep: ep.static.get(keys.TASK) == SPOONS_TASK)
 
 
 droid_recovery = droid_clean.override(
diff --git a/positronic/cfg/embodiment.py b/positronic/cfg/embodiment.py
index d2680f49e..7bc708eec 100644
--- a/positronic/cfg/embodiment.py
+++ b/positronic/cfg/embodiment.py
@@ -4,6 +4,7 @@
 import positronic.cfg.hardware.camera
 import positronic.cfg.hardware.gripper
 import positronic.cfg.hardware.roboarm
+from positronic import keys
 from positronic.dataset.serializers import Serializers
 from positronic.drivers.roboarm import command as roboarm_command
 from positronic.eval import ROBOT_STATIC_META, Command, Embodiment, Observation
@@ -13,10 +14,10 @@
     robot_arm=positronic.cfg.hardware.roboarm.franka_droid,
     gripper=positronic.cfg.hardware.gripper.robotiq,
     cameras={
-        'image.wrist': positronic.cfg.hardware.camera.zed_m.override(
+        keys.WRIST_IMAGE: positronic.cfg.hardware.camera.zed_m.override(
             view='left', resolution='hd720', fps=30, image_enhancement=True
         ),
-        'image.exterior': positronic.cfg.hardware.camera.zed_2i.override(
+        keys.EXTERIOR_IMAGE: positronic.cfg.hardware.camera.zed_2i.override(
             view='left', resolution='hd720', fps=30, image_enhancement=True
         ),
     },
@@ -25,7 +26,7 @@ def droid(robot_arm, gripper, cameras):
     """Real single-arm Franka (DROID) + Robotiq gripper + ZED cameras."""
     observations = {
         'robot_state': Observation(robot_arm.state, Serializers.robot_state),
-        'grip': Observation(gripper.grip, None),
+        keys.GRIP: Observation(gripper.grip, None),
         **{name: Observation(cam.frame, Serializers.camera_images) for name, cam in cameras.items()},
     }
     commands = {
@@ -52,7 +53,7 @@ def mujoco_franka(sim, camera_dict):
     """
     observations = {
         'robot_state': Observation(sim.state, Serializers.robot_state),
-        'grip': Observation(sim.grip, None),
+        keys.GRIP: Observation(sim.grip, None),
         **{name: Observation(sim.cameras[orig], Serializers.camera_images) for name, orig in camera_dict.items()},
     }
     # Home to the scene's initial pose, not `Reset()`: in MujocoSim `Reset()` rebuilds the whole scene, wiping the
diff --git a/positronic/cfg/eval/sim/libero.py b/positronic/cfg/eval/sim/libero.py
index a3d50aa8f..a8ee97cf1 100644
--- a/positronic/cfg/eval/sim/libero.py
+++ b/positronic/cfg/eval/sim/libero.py
@@ -1,5 +1,6 @@
 import configuronic as cfn
 
+from positronic import keys
 from positronic.cfg.eval import build_trials
 from positronic.drivers.roboarm.models import bundled_panda_model
 from positronic.eval import Eval, Observation, Task
@@ -13,7 +14,7 @@
 
 
 @cfn.config(
-    camera_dict={'image.agentview': 'agentview_image', 'image.wrist': 'eye_in_hand_image'},
+    camera_dict={'image.agentview': 'agentview_image', keys.WRIST_IMAGE: 'eye_in_hand_image'},
     camera_resolution=256,
     control_mode='ee',
     timeout=20.0,
diff --git a/positronic/cfg/eval/sim/positronic.py b/positronic/cfg/eval/sim/positronic.py
index 6b22517c3..90479f7d5 100644
--- a/positronic/cfg/eval/sim/positronic.py
+++ b/positronic/cfg/eval/sim/positronic.py
@@ -1,6 +1,7 @@
 import configuronic as cfn
 
 import positronic.cfg.simulator
+from positronic import keys
 from positronic.cfg.embodiment import mujoco_franka
 from positronic.cfg.eval import build_trials
 from positronic.eval import Eval, Observation, Task
@@ -12,7 +13,11 @@
     mujoco_model_path=package_assets_path('assets/mujoco/franka_table.xml'),
     loaders=positronic.cfg.simulator.stack_cubes_loaders,
     camera_fps=15,
-    camera_dict={'image.wrist': 'handcam_left_ph', 'image.exterior': 'back_view_ph', 'image.agent_view': 'agentview'},
+    camera_dict={
+        keys.WRIST_IMAGE: 'handcam_left_ph',
+        keys.EXTERIOR_IMAGE: 'back_view_ph',
+        'image.agent_view': 'agentview',
+    },
     timeout=15,
     seed=None,
     trial_count=1,
diff --git a/positronic/cfg/eval/sim/robolab.py b/positronic/cfg/eval/sim/robolab.py
index c64fba196..4fcc41bda 100644
--- a/positronic/cfg/eval/sim/robolab.py
+++ b/positronic/cfg/eval/sim/robolab.py
@@ -1,5 +1,6 @@
 import configuronic as cfn
 
+from positronic import keys
 from positronic.eval import Eval, Observation, Task
 from positronic.simulator.env_server.proxy import RemoteEnvControlSystem, remote_franka_embodiment
 from positronic.simulator.robolab.adapter import RobolabAdapter
@@ -147,7 +148,7 @@ def _resolve_tasks(task) -> list[str]:
 
 
 @cfn.config(
-    camera_dict={'image.exterior': 'over_shoulder_left_camera', 'image.wrist': 'wrist_cam'},
+    camera_dict={keys.EXTERIOR_IMAGE: 'over_shoulder_left_camera', keys.WRIST_IMAGE: 'wrist_cam'},
     instruction_type='default',
     trial_count=1,
     timeout=None,
diff --git a/positronic/cfg/phail/v1_0.py b/positronic/cfg/phail/v1_0.py
index bf2bebaff..9f1c82078 100644
--- a/positronic/cfg/phail/v1_0.py
+++ b/positronic/cfg/phail/v1_0.py
@@ -13,6 +13,7 @@
 import configuronic as cfn
 import pos3
 
+from positronic import keys
 from positronic.cfg.ds import group, local_all, transform
 from positronic.cfg.ds.internal import REAL_ROBOT_TRANSFORM
 from positronic.cfg.eval.real.tasks import UNIFIED_TASK
@@ -61,7 +62,7 @@ def episodes_table():
 def group_by_task():
     def group_fn(episodes: list[Episode]):
         duration = sum(ep.duration_ns / 1e9 / 3600 for ep in episodes)
-        return {'task': episodes[0]['task'], 'duration': duration, 'count': len(episodes)}
+        return {'task': episodes[0][keys.TASK], 'duration': duration, 'count': len(episodes)}
 
     format_table = {
         'task': C(label='Task'),
diff --git a/positronic/cfg/policy.py b/positronic/cfg/policy.py
index 1e3391b98..4c612e0f4 100644
--- a/positronic/cfg/policy.py
+++ b/positronic/cfg/policy.py
@@ -1,6 +1,7 @@
 import configuronic as cfn
 import pos3
 
+from positronic import keys
 from positronic.cfg import codecs
 from positronic.policy import Codec, Policy, RemotePolicy, SampledPolicy
 from positronic.policy.sampler import Sampler
@@ -104,5 +105,5 @@ def phail_single(hostname, w_openpi=1.0, w_groot=1.0, w_act=1.0):
 phail_multiple = production.override(
     endpoints={'smolvla': 'notebook:8000', 'act': 'notebook:8001', 'groot': 'desktop:8000', 'openpi': 'vm-openpi:8000'},
     sampler=balanced,
-    group_fields=['task', 'eval.object', 'eval.tote_placement', 'eval.external_camera'],
+    group_fields=[keys.TASK, 'eval.object', 'eval.tote_placement', 'eval.external_camera'],
 )
diff --git a/positronic/cfg/server.py b/positronic/cfg/server.py
index 0a468b88c..b4f6c27ba 100644
--- a/positronic/cfg/server.py
+++ b/positronic/cfg/server.py
@@ -5,6 +5,7 @@
 import configuronic as cfn
 import pos3
 
+from positronic import keys
 from positronic.dataset import Episode
 from positronic.dataset.transforms.episode import Derive, FromValue, Group, Identity, Rename
 from positronic.server.positronic_server import ColumnConfig as C
@@ -87,7 +88,7 @@ def group_fn(episodes: list[Episode]):
             duration += ep.duration_ns / 1e9 / 3600
             units += ep['units']
 
-        result = {'task': episodes[0]['task']}
+        result = {'task': episodes[0][keys.TASK]}
         result.update({'duration': duration, 'count': len(episodes), 'uph': units / duration})
         return result
 
diff --git a/positronic/cfg/wrappers.py b/positronic/cfg/wrappers.py
index fa7e2ca90..d8c2300e4 100644
--- a/positronic/cfg/wrappers.py
+++ b/positronic/cfg/wrappers.py
@@ -1,5 +1,6 @@
 import configuronic as cfn
 
+from positronic import keys as obs_keys
 from positronic.policy.wrappers import ChunkedSchedule, TemporalStack
 
 chunked_schedule = cfn.Config(ChunkedSchedule)
@@ -24,7 +25,9 @@ def _frame_offsets_sec(history_frames: int, stride: int, fps: float) -> tuple[fl
     return tuple(-f / fps for f in reversed(frames))
 
 
-@cfn.config(keys=('image.wrist', 'image.exterior', 'robot_state.ee_pose', 'grip'), fps=15.0, pad_start=True)
+@cfn.config(
+    keys=(obs_keys.WRIST_IMAGE, obs_keys.EXTERIOR_IMAGE, obs_keys.EE_POSE, obs_keys.GRIP), fps=15.0, pad_start=True
+)
 def video_context_wrappers(history_frames: int, stride: int, keys: tuple[str, ...], fps: float, pad_start: bool):
     """The pipeline's local half for video-conditioned policies: strided temporal context, scheduling.
 
diff --git a/positronic/data_collection.py b/positronic/data_collection.py
index 2d02864bf..e1a1bb59f 100644
--- a/positronic/data_collection.py
+++ b/positronic/data_collection.py
@@ -15,7 +15,7 @@
 import positronic.cfg.simulator
 import positronic.cfg.sound
 import positronic.cfg.webxr
-from positronic import geom, utils, wire
+from positronic import geom, keys, utils, wire
 from positronic.dataset.ds_writer_agent import DsWriterAgent, DsWriterCommand, TimeMode
 from positronic.dataset.local_dataset import LocalDatasetWriter
 from positronic.dataset.serializers import Serializers
@@ -238,7 +238,7 @@ def main(
     camera_emitters = {name: cam.frame for name, cam in camera_instances.items()}
     static_meta = {}
     if task is not None:
-        static_meta['task'] = task
+        static_meta[keys.TASK] = task
     if robot_arm is not None:
         static_meta.update(wire.ROBOT_STATIC_META)
     data_collection = DataCollectionController(operator_position.value, static_meta=static_meta)
@@ -269,8 +269,8 @@ def main(
     mujoco_model_path=package_assets_path('assets/mujoco/franka_table.xml'),
     webxr=positronic.cfg.webxr.oculus,
     cameras={
-        'image.wrist': 'handcam_left_ph',
-        'image.exterior': 'back_view_ph',
+        keys.WRIST_IMAGE: 'handcam_left_ph',
+        keys.EXTERIOR_IMAGE: 'back_view_ph',
         'image.handcam_right': 'handcam_right_ph',
         'image.wrist_2': 'wrist_cam_ph',
     },
@@ -297,7 +297,7 @@ def main_sim(
 
     static_meta = dict(wire.ROBOT_STATIC_META)
     if task is not None:
-        static_meta['task'] = task
+        static_meta[keys.TASK] = task
 
     data_collection = DataCollectionController(
         operator_position.value,
@@ -365,8 +365,8 @@ def so101cfg(robot_arm, **kwargs):
     webxr=positronic.cfg.webxr.oculus,
     sound=positronic.cfg.sound.sound,
     cameras={
-        'image.wrist': positronic.cfg.hardware.camera.zed_m.override(view='left', resolution='hd720', fps=30),
-        'image.exterior': positronic.cfg.hardware.camera.zed_2i.override(view='left', resolution='hd720', fps=30),
+        keys.WRIST_IMAGE: positronic.cfg.hardware.camera.zed_m.override(view='left', resolution='hd720', fps=30),
+        keys.EXTERIOR_IMAGE: positronic.cfg.hardware.camera.zed_2i.override(view='left', resolution='hd720', fps=30),
     },
     operator_position=OperatorPosition.BACK,
 )
@@ -378,7 +378,9 @@ def so101cfg(robot_arm, **kwargs):
     gripper=None,
     webxr=positronic.cfg.webxr.oculus,
     sound=positronic.cfg.sound.sound,
-    cameras={'image.exterior': positronic.cfg.hardware.camera.zed_2i.override(view='left', resolution='hd720', fps=30)},
+    cameras={
+        keys.EXTERIOR_IMAGE: positronic.cfg.hardware.camera.zed_2i.override(view='left', resolution='hd720', fps=30)
+    },
     operator_position=OperatorPosition.BACK,
 )
 
diff --git a/positronic/dataset/tests/test_ds_writer_agent.py b/positronic/dataset/tests/test_ds_writer_agent.py
index adc1d3674..40049c411 100644
--- a/positronic/dataset/tests/test_ds_writer_agent.py
+++ b/positronic/dataset/tests/test_ds_writer_agent.py
@@ -6,7 +6,7 @@
 import pytest
 
 import pimm
-from positronic import geom
+from positronic import geom, keys
 from positronic.dataset import DatasetWriter, EpisodeWriter
 from positronic.dataset.ds_writer_agent import (
     DsWriterAgent,
@@ -400,10 +400,10 @@ def test_robot_state_serializer_drops_reset_and_emits_components(world):
     w = ds.created[-1]
     items = {name: val for (name, val, _, _) in w.appends}
     # Should not contain any data from RESETTING
-    assert set(items.keys()) == {'robot_state.q', 'robot_state.dq', 'robot_state.ee_pose'}
-    np.testing.assert_allclose(items['robot_state.q'], q)
-    np.testing.assert_allclose(items['robot_state.dq'], dq)
-    np.testing.assert_allclose(items['robot_state.ee_pose'], np.concatenate([t, geom.Rotation.identity.as_quat]))
+    assert set(items.keys()) == {keys.JOINTS, keys.JOINT_VEL, keys.EE_POSE}
+    np.testing.assert_allclose(items[keys.JOINTS], q)
+    np.testing.assert_allclose(items[keys.JOINT_VEL], dq)
+    np.testing.assert_allclose(items[keys.EE_POSE], np.concatenate([t, geom.Rotation.identity.as_quat]))
 
 
 def test_robot_command_serializer_variants(world):
diff --git a/positronic/dataset/transforms/quality.py b/positronic/dataset/transforms/quality.py
index a16383ffc..e91d233e4 100644
--- a/positronic/dataset/transforms/quality.py
+++ b/positronic/dataset/transforms/quality.py
@@ -6,13 +6,15 @@
 
 import numpy as np
 
+from positronic import keys
+
 from .signals import Elementwise, Join, diff, norm, view
 
 _TRANSLATION = slice(0, 3)
 _DT_SEC = 1 / 15
 
 
-def idle_mask(episode, signal='robot_state.q', velocity_threshold=0.015, dt_sec=_DT_SEC):
+def idle_mask(episode, signal=keys.JOINTS, velocity_threshold=0.015, dt_sec=_DT_SEC):
     """Per-frame bool: True where joint speed < threshold (rad/s)."""
     speed = norm(diff(episode.signals[signal], dt_sec))
 
@@ -22,12 +24,12 @@ def fn(vals):
     return Elementwise(speed, fn)
 
 
-def jerk(episode, signal='robot_state.q', dt_sec=_DT_SEC):
+def jerk(episode, signal=keys.JOINTS, dt_sec=_DT_SEC):
     """Per-frame joint acceleration magnitude (rad/s^2)."""
     return norm(diff(episode.signals[signal], dt_sec, order=2))
 
 
-def cmd_lag(episode, cmd_signal='robot_command.pose', state_signal='robot_state.ee_pose', components=_TRANSLATION):
+def cmd_lag(episode, cmd_signal='robot_command.pose', state_signal=keys.EE_POSE, components=_TRANSLATION):
     """Per-frame distance between commanded and actual pose (meters)."""
     cmd = episode.signals[cmd_signal]
     ee = episode.signals[state_signal]
diff --git a/positronic/drivers/roboarm/models.py b/positronic/drivers/roboarm/models.py
index 7f087ab27..173357318 100644
--- a/positronic/drivers/roboarm/models.py
+++ b/positronic/drivers/roboarm/models.py
@@ -6,6 +6,8 @@
 from functools import lru_cache
 from pathlib import Path
 
+from positronic import keys
+
 _FLANGE_LINK = 'link8'
 # Seat the gripper on the flange, rotated about the approach axis to match the real 2F-85 coupler
 # (a +45deg Z, i.e. 90deg off the franka ``end_effector`` frame).
@@ -110,7 +112,7 @@ def _bundled_robotiq_2f85() -> dict:
     return {
         'subtree': subtree,
         'meshes': {f.name: f.read_bytes() for f in sorted(mesh_dir.glob('*.stl'))},
-        'gripper': {'signal': 'grip', 'joints': _ROBOTIQ_2F85_JOINTS, 'travel': 0.8},
+        'gripper': {'signal': keys.GRIP, 'joints': _ROBOTIQ_2F85_JOINTS, 'travel': 0.8},
     }
 
 
@@ -165,5 +167,5 @@ def bundled_panda_model() -> dict:
         'joint_names': [f'joint{i}' for i in range(1, 8)],
         'control_frame': 'end_effector',
         # ``grip`` is recorded in [0, 1] (open→closed); each finger slides 0..0.04 m along its axis.
-        'gripper': {'signal': 'grip', 'joints': ['finger_joint1', 'finger_joint2'], 'travel': 0.04},
+        'gripper': {'signal': keys.GRIP, 'joints': ['finger_joint1', 'finger_joint2'], 'travel': 0.04},
     }
diff --git a/positronic/drivers/roboarm/tests/test_ik.py b/positronic/drivers/roboarm/tests/test_ik.py
index 23ebc6b2f..b37c9a90e 100644
--- a/positronic/drivers/roboarm/tests/test_ik.py
+++ b/positronic/drivers/roboarm/tests/test_ik.py
@@ -5,6 +5,7 @@
 import numpy as np
 import pytest
 
+from positronic import keys
 from positronic.dataset.episode import EpisodeContainer
 from positronic.dataset.tests.utils import DummySignal
 from positronic.drivers.roboarm.ik import DLSIKSolver, DLSIKSolverWithLimits, LMIKSolver, ik_joints_from_episode
@@ -92,14 +93,14 @@ def test_ik_joints_from_episode():
 
     episode = EpisodeContainer(
         data={
-            'robot_state.q': DummySignal(ts, q_traj),
+            keys.JOINTS: DummySignal(ts, q_traj),
             'robot_command.pose': DummySignal(ts, ee_poses),
             'urdf': URDF,
             'joint_names': JOINT_NAMES,
             'control_frame': CONTROL_FRAME,
         }
     )
-    result = ik_joints_from_episode(episode, DLSIKSolverWithLimits, 'robot_command.pose', 'robot_state.q')
+    result = ik_joints_from_episode(episode, DLSIKSolverWithLimits, 'robot_command.pose', keys.JOINTS)
 
     assert len(result) == n_steps
     for i in range(n_steps):
diff --git a/positronic/eval.py b/positronic/eval.py
index 722ca36d5..a8fb68d34 100644
--- a/positronic/eval.py
+++ b/positronic/eval.py
@@ -3,10 +3,11 @@
 from typing import Any
 
 import pimm
+from positronic import keys
 from positronic.dataset.serializers import Serializer
 
 # Embodiment-level static meta: how recorded signals map to the canonical robot fields.
-ROBOT_STATIC_META = {'joint_signal': 'robot_state.q', 'pose_signals': ['robot_state.ee_pose', 'robot_command.pose']}
+ROBOT_STATIC_META = {'joint_signal': keys.JOINTS, 'pose_signals': [keys.EE_POSE, 'robot_command.pose']}
 
 
 @dataclass
diff --git a/positronic/gui/eval.py b/positronic/gui/eval.py
index 6318ffa8d..8fe6c8fec 100644
--- a/positronic/gui/eval.py
+++ b/positronic/gui/eval.py
@@ -9,6 +9,7 @@
 import numpy as np
 
 import pimm
+from positronic import keys
 from positronic.cfg.eval.real.tasks import SCISSORS_TASK, SPOONS_TASK, TOWELS_TASK, UNIFIED_TASK
 from positronic.dataset.edits import EditedDataset
 from positronic.dataset.local_dataset import LocalDataset
@@ -402,7 +403,7 @@ def start(self, sender=None, app_data=None):
 
         task_value = dpg.get_value('task_radio')
         task_name = dpg.get_value('custom_input') if task_value == 'Other' else task_value
-        context = {'task': task_name}
+        context = {keys.TASK: task_name}
 
         if task_value in TASK_TO_OBJECT or task_value == UNIFIED_TASK:
             obj_value = dpg.get_value('object_radio')
@@ -463,7 +464,7 @@ def _select(self, idx: int):
             self._sel = max(0, min(idx, self._count - 1))
             ep = self._edited.overlay(self._base[self._sel])
             static = ep.static
-            dpg.set_value('ed_task', static.get('task', ''))
+            dpg.set_value('ed_task', static.get(keys.TASK, ''))
             dpg.set_value('ed_outcome', static.get('eval.outcome', OUTCOMES[0]))
             dpg.set_value('ed_total', static.get('eval.total_items', 1))
             dpg.set_value('ed_success', static.get('eval.successful_items', 0))
diff --git a/positronic/keys.py b/positronic/keys.py
new file mode 100644
index 000000000..15ed00548
--- /dev/null
+++ b/positronic/keys.py
@@ -0,0 +1,15 @@
+"""Canonical raw observation-signal keys of the positronic embodiment/inference wire.
+
+Every sim adapter and embodiment produces these keys and every vendor codec consumes them. They
+are defined here once, in a leaf module with no positronic imports, so a rename is a single-site
+change the type checker propagates instead of a string literal duplicated across codecs, evals,
+configs, adapters and datasets.
+"""
+
+JOINTS = 'robot_state.q'
+JOINT_VEL = 'robot_state.dq'
+EE_POSE = 'robot_state.ee_pose'
+GRIP = 'grip'
+TASK = 'task'
+WRIST_IMAGE = 'image.wrist'
+EXTERIOR_IMAGE = 'image.exterior'
diff --git a/positronic/offboard/tests/test_offboard.py b/positronic/offboard/tests/test_offboard.py
index 1c4534a46..a41c0560a 100644
--- a/positronic/offboard/tests/test_offboard.py
+++ b/positronic/offboard/tests/test_offboard.py
@@ -2,6 +2,7 @@
 
 import numpy as np
 
+from positronic import keys
 from positronic.drivers.roboarm.command import CartesianPosition, JointDelta, JointPosition, Reset
 from positronic.geom import Rotation, Transform3D
 from positronic.offboard.client import InferenceClient
@@ -91,14 +92,14 @@ def test_wire_serialisation_accepts_mappingproxy():
 def test_jpeg_round_trips_single_image_and_stack():
     """An ``encode_jpeg`` marker survives the wire and decodes back to the original shape and order."""
     single = np.full((16, 24, 3), 90, dtype=np.uint8)
-    restored_single = deserialise(serialise({'image.wrist': encode_jpeg(single)}))['image.wrist']
+    restored_single = deserialise(serialise({keys.WRIST_IMAGE: encode_jpeg(single)}))[keys.WRIST_IMAGE]
     assert isinstance(restored_single, np.ndarray)
     assert restored_single.shape == (16, 24, 3)
     assert restored_single.dtype == np.uint8
     np.testing.assert_allclose(restored_single, single, atol=4)
 
     stack = np.stack([np.full((16, 24, 3), (t + 1) * 60, dtype=np.uint8) for t in range(3)])
-    restored_stack = deserialise(serialise({'image.wrist': encode_jpeg(stack)}))['image.wrist']
+    restored_stack = deserialise(serialise({keys.WRIST_IMAGE: encode_jpeg(stack)}))[keys.WRIST_IMAGE]
     assert restored_stack.shape == (3, 16, 24, 3)
     # q90 JPEG on solid colors is near-lossless; this also verifies per-frame order is preserved.
     np.testing.assert_allclose(restored_stack, stack, atol=4)
diff --git a/positronic/offboard/tests/test_remote_policy.py b/positronic/offboard/tests/test_remote_policy.py
index 5495d5370..55eb7c0c0 100644
--- a/positronic/offboard/tests/test_remote_policy.py
+++ b/positronic/offboard/tests/test_remote_policy.py
@@ -4,6 +4,7 @@
 import numpy as np
 import pytest
 
+from positronic import keys
 from positronic.offboard.client import DEFAULT_INFER_TIMEOUT, InferenceClient
 from positronic.policy import RemotePolicy
 from positronic.policy.codec import ActionHorizon
@@ -50,12 +51,12 @@ def test_compression_reaches_nested_images(self):
             'cam': _make_image(48, 64),
             'video': {'wrist': _make_image(48, 64)},
             'state': np.array([1.0, 2.0]),
-            'task': 'pick cube',
+            keys.TASK: 'pick cube',
         })
         assert isinstance(result['cam'], dict)
         assert isinstance(result['video']['wrist'], dict)
         np.testing.assert_array_equal(result['state'], np.array([1.0, 2.0]))
-        assert result['task'] == 'pick cube'
+        assert result[keys.TASK] == 'pick cube'
 
 
 class TestInferenceClientHeaders:
diff --git a/positronic/policy/action.py b/positronic/policy/action.py
index df6fb0f5d..cf3781dca 100644
--- a/positronic/policy/action.py
+++ b/positronic/policy/action.py
@@ -2,7 +2,7 @@
 
 import numpy as np
 
-from positronic import geom
+from positronic import geom, keys
 from positronic.dataset import transforms
 from positronic.dataset.episode import Episode
 from positronic.dataset.signal import Signal
@@ -114,7 +114,7 @@ def __init__(
         solver_cls,
         *,
         tgt_ee_pose_key='robot_command.pose',
-        current_q_key='robot_state.q',
+        current_q_key=keys.JOINTS,
         tgt_joints_key='robot_command.joints',
     ):
         self.solver_cls = solver_cls
@@ -140,7 +140,7 @@ class RelativePositionAction(Codec):
     def __init__(
         self,
         rotation_rep: RotRep | str = RotRep.QUAT,
-        robot_pose_key: str = 'robot_state.ee_pose',
+        robot_pose_key: str = keys.EE_POSE,
         target_pose_key: str = 'robot_command.pose',
         target_grip_key: str = 'target_grip',
     ):
@@ -161,7 +161,7 @@ def _decode_single(self, data: dict, context: dict | None) -> dict:
         q_diff = geom.Rotation.create_from(rotation, self.rot_rep)
         tr_diff = action_vector[self.rot_rep.size : self.rot_rep.size + 3]
 
-        robot_pose = context['robot_state.ee_pose']
+        robot_pose = context[keys.EE_POSE]
 
         rot_mul = geom.Rotation.from_quat(robot_pose[3:7]) * q_diff
         tr_add = robot_pose[0:3] + tr_diff
diff --git a/positronic/policy/codec.py b/positronic/policy/codec.py
index b0904df0b..4482df09a 100644
--- a/positronic/policy/codec.py
+++ b/positronic/policy/codec.py
@@ -15,6 +15,7 @@
 import numpy as np
 from PIL import Image as PilImage
 
+from positronic import keys as obs_keys
 from positronic.dataset.transforms import Elementwise
 from positronic.dataset.transforms.episode import Derive, EpisodeTransform, Group, Identity
 from positronic.policy.base import PAR, SEQ, DelegatingSession, PolicyWrapper, Session, _ComposedWrapper
@@ -379,8 +380,8 @@ class FlipGrip(Codec):
 
     def encode(self, data):
         # Copy: the original dict is also the decode ``context`` and the raw recording tap's input.
-        if 'grip' in data:
-            data = {**data, 'grip': 1.0 - data['grip']}
+        if obs_keys.GRIP in data:
+            data = {**data, obs_keys.GRIP: 1.0 - data[obs_keys.GRIP]}
         return data
 
     @property
diff --git a/positronic/policy/harness.py b/positronic/policy/harness.py
index 249a16eb0..62dd01974 100644
--- a/positronic/policy/harness.py
+++ b/positronic/policy/harness.py
@@ -5,6 +5,7 @@
 from typing import Any
 
 import pimm
+from positronic import keys
 from positronic.dataset.ds_writer_agent import DsWriterCommand
 from positronic.dataset.serializers import expand_suffixed
 from positronic.eval import Embodiment, Task
@@ -219,7 +220,7 @@ def _begin_episode(self, context: dict[str, Any], clock: pimm.Clock) -> None:
         if self._task is not None and self._task.reset is not None:
             self._task.reset(self.context)
         if self._task is not None:
-            self.context = {**self.context, 'task': self._task.instruction}
+            self.context = {**self.context, keys.TASK: self._task.instruction}
         self._session = self.policy.new_session(self.context, clock.now)
         self._running = True
         self._deadline = clock.now() + self._task.timeout if self._task is not None else None
diff --git a/positronic/policy/observation.py b/positronic/policy/observation.py
index d9c4789bc..d958a7586 100644
--- a/positronic/policy/observation.py
+++ b/positronic/policy/observation.py
@@ -4,6 +4,7 @@
 import numpy as np
 from PIL import Image as PilImage
 
+from positronic import keys
 from positronic.dataset import Signal, transforms
 from positronic.dataset.episode import Episode
 from positronic.dataset.transforms import image
@@ -29,7 +30,7 @@ def __init__(
 
         self._derive_transforms = {k: partial(self._derive_state, k) for k in state.keys()}
         self._derive_transforms.update({k: partial(self._derive_image, k) for k in images.keys()})
-        self._derive_transforms['task'] = Get('task', '')
+        self._derive_transforms['task'] = Get(keys.TASK, '')
 
         lerobot_features: dict[str, Any] = {}
         for name, features in state.items():
@@ -53,8 +54,8 @@ def _decode_single(self, data: dict, context: dict | None) -> dict:
     def encode(self, inputs: dict[str, Any]) -> dict[str, Any]:
         obs: dict[str, Any] = {}
 
-        if 'task' in inputs:
-            obs[self._task_field] = inputs['task']
+        if keys.TASK in inputs:
+            obs[self._task_field] = inputs[keys.TASK]
 
         for out_name, (input_key, (width, height)) in self._image_configs.items():
             if input_key not in inputs:
diff --git a/positronic/policy/recording.py b/positronic/policy/recording.py
index 2c698ab2e..8caf45bb3 100644
--- a/positronic/policy/recording.py
+++ b/positronic/policy/recording.py
@@ -54,6 +54,7 @@
 import rerun.blueprint as rrb
 
 from positronic import geom
+from positronic import keys as obs_keys
 from positronic.drivers.roboarm import command as roboarm_command
 from positronic.policy.base import DelegatingSession, PolicyWrapper, Session
 from positronic.policy.codec import is_action
@@ -388,8 +389,8 @@ def _log_action_chunk(self, prefix: str, actions: list[dict], obs: dict) -> None
         tv = self._rec._timeline_values
         base_ns = int(tv.get('obs_time', tv.get('wall_time', next(iter(tv.values()), 0))))
         grip = _stack_numeric([a['target_grip'] for a in actions]) if all('target_grip' in a for a in actions) else None
-        actual_pos = obs.get('robot_state.ee_pose') if isinstance(obs, Mapping) else None
-        actual_grip = obs.get('grip') if isinstance(obs, Mapping) else None
+        actual_pos = obs.get(obs_keys.EE_POSE) if isinstance(obs, Mapping) else None
+        actual_grip = obs.get(obs_keys.GRIP) if isinstance(obs, Mapping) else None
         # Under TemporalStack these arrive as (T, 7) / (T,) stacks; the overlay draws the current pose,
         # which is the last frame (offsets end at 0 = now), mirroring the image collapse in `_as_image`.
         if actual_pos is not None:
diff --git a/positronic/policy/tests/test_golden_pipeline.py b/positronic/policy/tests/test_golden_pipeline.py
index 8e7dbd908..0d8e60c2d 100644
--- a/positronic/policy/tests/test_golden_pipeline.py
+++ b/positronic/policy/tests/test_golden_pipeline.py
@@ -32,7 +32,7 @@
 import pytest
 
 import pimm
-from positronic import wire
+from positronic import keys, wire
 from positronic.dataset.ds_writer_agent import TimeMode
 from positronic.dataset.local_dataset import LocalDataset, LocalDatasetWriter
 from positronic.dataset.serializers import Serializers
@@ -60,12 +60,12 @@
 CONTROL_PERIOD_S = 0.005  # fake robot/gripper sampling cadence (200 Hz)
 
 # State signals captured at the DsWriterAgent output and locked by the golden.
-CAPTURED_SIGNALS = ('robot_state.ee_pose', 'robot_state.q', 'grip')
+CAPTURED_SIGNALS = (keys.EE_POSE, keys.JOINTS, keys.GRIP)
 
 
 class _ScriptedSession(Session):
     def __call__(self, obs):
-        current = np.asarray(obs['robot_state.ee_pose'][:3], dtype=np.float32)
+        current = np.asarray(obs[keys.EE_POSE][:3], dtype=np.float32)
         delta = TARGET_POS - current
         chunk = []
         for i in range(10):
@@ -194,7 +194,7 @@ def _run_pipeline(tmp_path: Path) -> dict:
             descriptor='',
             observations={
                 'robot_state': Observation(robot.state, Serializers.robot_state),
-                'grip': Observation(gripper.grip, None),
+                keys.GRIP: Observation(gripper.grip, None),
             },
             commands={
                 'robot_command': Command(robot.commands, Reset(), Serializers.robot_command),
diff --git a/positronic/policy/tests/test_harness.py b/positronic/policy/tests/test_harness.py
index 84614f729..5916080a3 100644
--- a/positronic/policy/tests/test_harness.py
+++ b/positronic/policy/tests/test_harness.py
@@ -1,10 +1,11 @@
 from functools import partial
+from typing import Any
 
 import numpy as np
 import pytest
 
 import pimm
-from positronic import wire
+from positronic import keys, wire
 from positronic.dataset.ds_writer_agent import DsWriterCommand, DsWriterCommandType
 from positronic.dataset.serializers import Serializers
 from positronic.drivers import roboarm
@@ -39,7 +40,7 @@ def make_embodiment(descriptor: str = '', cameras=('image.cam',)) -> Embodiment:
     """
     observations = {
         'robot_state': Observation(pimm.NoOpEmitter(), Serializers.robot_state),
-        'grip': Observation(pimm.NoOpEmitter(), None),
+        keys.GRIP: Observation(pimm.NoOpEmitter(), None),
     }
     for cam in cameras:
         observations[cam] = Observation(pimm.NoOpEmitter(), Serializers.camera_images)
@@ -66,7 +67,7 @@ def __init__(self, command: roboarm.command.CommandType | None = None, target_gr
             command = CartesianPosition(pose=pose)
         self.command = command
         self.target_grip = float(target_grip)
-        self.last_obs: dict[str, object] | None = None
+        self.last_obs: dict[str, Any] | None = None
         self.reset_calls: int = 0
         self.last_reset_context = None
 
@@ -105,7 +106,7 @@ def __init__(
             command = CartesianPosition(pose=pose)
         self.command = command
         self.target_grip = float(target_grip)
-        self.last_obs: dict[str, object] | None = None
+        self.last_obs: dict[str, Any] | None = None
         self.observations: list[dict[str, object]] = []
         self.reset_calls = 0
         self.last_reset_context = None
@@ -186,7 +187,7 @@ def _pair_all(world, harness):
     return {
         'frame_em': world.pair(harness.observations['image.cam']),
         'robot_em': world.pair(harness.observations['robot_state']),
-        'grip_em': world.pair(harness.observations['grip']),
+        'grip_em': world.pair(harness.observations[keys.GRIP]),
         'directive_em': world.pair(harness.directive),
         'command_rx': world.pair(harness.commands['robot_command']),
         'grip_rx': world.pair(harness.commands['target_grip']),
@@ -252,7 +253,7 @@ def test_harness_emits_cartesian_move(world):
 
     frame_em = world.pair(harness.observations['image.cam'])
     robot_em = world.pair(harness.observations['robot_state'])
-    grip_em = world.pair(harness.observations['grip'])
+    grip_em = world.pair(harness.observations[keys.GRIP])
     directive_em = world.pair(harness.directive)
 
     robot_state = make_robot_state([0.1, 0.2, 0.3], [0.4, 0.5, 0.6])
@@ -270,21 +271,21 @@ def test_harness_emits_cartesian_move(world):
     obs = policy.last_obs
     assert 'image.cam' in obs
     expected_pose = np.concatenate([robot_state.ee_pose.translation, robot_state.ee_pose.rotation.as_quat])
-    np.testing.assert_allclose(obs['robot_state.ee_pose'], expected_pose)
-    np.testing.assert_allclose(obs['robot_state.q'], robot_state.q)
-    np.testing.assert_allclose(obs['robot_state.dq'], np.zeros_like(robot_state.q))
-    assert obs['grip'] == pytest.approx(0.25)
-    assert obs['task'] == 'stack-blocks'
+    np.testing.assert_allclose(obs[keys.EE_POSE], expected_pose)
+    np.testing.assert_allclose(obs[keys.JOINTS], robot_state.q)
+    np.testing.assert_allclose(obs[keys.JOINT_VEL], np.zeros_like(robot_state.q))
+    assert obs[keys.GRIP] == pytest.approx(0.25)
+    assert obs[keys.TASK] == 'stack-blocks'
     assert obs['descriptor'] == ''  # no descriptor passed -> empty string reaches the policy
     # Recording == canonical policy I/O: the policy sees the same ``robot_state`` serializer
     # the dataset records. wall/obs timestamps carry volatile values, so lock the stable key set.
     assert set(obs) - {'wall_time_ns', 'obs_time_ns'} == {
         'image.cam',
-        'robot_state.q',
-        'robot_state.dq',
-        'robot_state.ee_pose',
-        'grip',
-        'task',
+        keys.JOINTS,
+        keys.JOINT_VEL,
+        keys.EE_POSE,
+        keys.GRIP,
+        keys.TASK,
         'descriptor',
     }
 
@@ -311,7 +312,7 @@ def test_harness_passes_descriptor_to_policy(world):
 
     frame_em = world.pair(harness.observations['image.cam'])
     robot_em = world.pair(harness.observations['robot_state'])
-    grip_em = world.pair(harness.observations['grip'])
+    grip_em = world.pair(harness.observations[keys.GRIP])
     directive_em = world.pair(harness.directive)
 
     robot_state = make_robot_state([0.1, 0.2, 0.3], [0.4, 0.5, 0.6])
@@ -341,7 +342,7 @@ def test_harness_waits_for_complete_inputs(world):
 
     frame_em = world.pair(harness.observations['image.cam'])
     robot_em = world.pair(harness.observations['robot_state'])
-    grip_em = world.pair(harness.observations['grip'])
+    grip_em = world.pair(harness.observations[keys.GRIP])
     directive_em = world.pair(harness.directive)
 
     assert 'image.cam' in harness.observations
@@ -380,7 +381,7 @@ def assert_no_inference():
 @pytest.mark.timeout(3.0)
 def test_episode_meta_stamped_at_finalize(world):
     policy = StubPolicy(meta={'type': 'stub', 'checkpoint': 'v1'})
-    harness = Harness(policy, make_embodiment(), static_meta={'joint_signal': 'robot_state.q'})
+    harness = Harness(policy, make_embodiment(), static_meta={'joint_signal': keys.JOINTS})
     p = _pair_all(world, harness)
 
     driver = ManualDriver([
@@ -396,12 +397,12 @@ def test_episode_meta_stamped_at_finalize(world):
     stops = [c for c in _ds_commands(p) if c.type == DsWriterCommandType.STOP_EPISODE]
     assert len(stops) == 1
     meta = stops[0].static_data
-    assert meta['joint_signal'] == 'robot_state.q'
+    assert meta['joint_signal'] == keys.JOINTS
     assert meta['urdf'] == '<robot/>'
     assert meta['joint_names'] == ['j1']
     assert meta['inference.policy.type'] == 'stub'
     assert meta['inference.policy.checkpoint'] == 'v1'
-    assert meta['task'] == 'test'
+    assert meta[keys.TASK] == 'test'
 
 
 @pytest.mark.timeout(3.0)
@@ -757,7 +758,7 @@ def test_trial_plan_self_drives(world):
 
     stops = [c for c in _ds_commands(p) if c.type == DsWriterCommandType.STOP_EPISODE]
     assert [s.static_data['eval.trial_index'] for s in stops] == [0, 1]
-    assert all(s.static_data['task'] == 'stack' for s in stops)
+    assert all(s.static_data[keys.TASK] == 'stack' for s in stops)
     assert len(stops) == 2
     assert all(s.static_data['eval.terminated'] is False for s in stops)
     assert policy.reset_calls == 2
@@ -780,7 +781,7 @@ def test_timeout_crossed_during_latency_sleep_drops_chunk(world):
 
     frame_em = world.pair(harness.observations['image.cam'])
     robot_em = world.pair(harness.observations['robot_state'])
-    grip_em = world.pair(harness.observations['grip'])
+    grip_em = world.pair(harness.observations[keys.GRIP])
 
     robot_state = make_robot_state([0.1, 0.2, 0.3], [0.4, 0.5, 0.6])
 
@@ -858,7 +859,7 @@ def test_run_calls_policy_reset_with_context(world):
     drive_scheduler(scheduler, steps=5)
 
     assert policy.reset_calls == 1
-    assert policy.last_reset_context == {'task': 'test-task'}
+    assert policy.last_reset_context == {keys.TASK: 'test-task'}
 
 
 @pytest.mark.timeout(3.0)
@@ -878,7 +879,7 @@ def reset(_context):
     scheduler = world.start([harness])
     drive_scheduler(scheduler, steps=200)
 
-    assert policy.last_reset_context['task'] == 'resolved-on-reset'
+    assert policy.last_reset_context[keys.TASK] == 'resolved-on-reset'
 
 
 @pytest.mark.timeout(3.0)
@@ -909,7 +910,7 @@ def emit(self, data, ts: int = -1):
 
     frame_em = world.pair(harness.observations['image.cam'])
     robot_em = world.pair(harness.observations['robot_state'])
-    grip_em = world.pair(harness.observations['grip'])
+    grip_em = world.pair(harness.observations[keys.GRIP])
     directive_em = world.pair(harness.directive)
 
     robot_state = make_robot_state([0.1, 0.2, 0.3], [0.4, 0.5, 0.6])
@@ -963,7 +964,7 @@ def new_session(self, context=None, now=None):
 
     frame_em = world.pair(harness.observations['image.cam'])
     robot_em = world.pair(harness.observations['robot_state'])
-    grip_em = world.pair(harness.observations['grip'])
+    grip_em = world.pair(harness.observations[keys.GRIP])
     directive_em = world.pair(harness.directive)
 
     robot_state = make_robot_state([0.1, 0.2, 0.3], [0.4, 0.5, 0.6])
@@ -1074,7 +1075,7 @@ def test_harness_skips_inference_on_error(world):
 
 
 def test_directive_preserves_payload():
-    assert Directive.RUN(task='test').payload == {'task': 'test'}
+    assert Directive.RUN(task='test').payload == {keys.TASK: 'test'}
     assert Directive.FINISH(outcome='Success').payload == {'outcome': 'Success'}
     assert Directive.FINISH().payload == {}
     assert Directive.ABORT().payload is None
@@ -1201,7 +1202,7 @@ def emit(self, data, ts: int = -1):
 
     frame_em = world.pair(harness.observations['image.cam'])
     robot_em = world.pair(harness.observations['robot_state'])
-    grip_em = world.pair(harness.observations['grip'])
+    grip_em = world.pair(harness.observations[keys.GRIP])
     directive_em = world.pair(harness.directive)
 
     robot_state = make_robot_state([0.1, 0.2, 0.3], [0.4, 0.5, 0.6])
diff --git a/positronic/policy/tests/test_policy_io.py b/positronic/policy/tests/test_policy_io.py
index 6288c627a..7719e7053 100644
--- a/positronic/policy/tests/test_policy_io.py
+++ b/positronic/policy/tests/test_policy_io.py
@@ -2,6 +2,7 @@
 import pytest
 
 import positronic.drivers.roboarm.command as cmd_module
+from positronic import keys as obs_keys
 from positronic.cfg.codecs import compose
 from positronic.dataset.episode import EpisodeContainer
 from positronic.dataset.tests.utils import DummySignal
@@ -60,11 +61,11 @@ def test_observation_encode_missing_state_inputs_raise():
 
 def test_observation_encode_task():
     enc = ObservationCodec(state={'observation.state': ['a']}, images={})
-    obs = enc.encode({'a': 1.0, 'task': 'test_task'})
-    assert obs['task'] == 'test_task'
+    obs = enc.encode({'a': 1.0, obs_keys.TASK: 'test_task'})
+    assert obs[obs_keys.TASK] == 'test_task'
 
     obs_no_task = enc.encode({'a': 1.0})
-    assert 'task' not in obs_no_task
+    assert obs_keys.TASK not in obs_no_task
 
 
 def test_absolute_position_action_encode_decode_quat():
@@ -106,7 +107,7 @@ def test_relative_target_position_action_encode_decode_quat():
     tgt_pose = [np.concatenate([t_tgt[0], q_tgt[0].as_quat]).astype(np.float32)]
 
     ep = EpisodeContainer({
-        'robot_state.ee_pose': DummySignal(ts, cur_pose),
+        obs_keys.EE_POSE: DummySignal(ts, cur_pose),
         'robot_command.pose': DummySignal(ts, tgt_pose),
         'target_grip': DummySignal(ts, g_tgt),
     })
@@ -120,7 +121,7 @@ def test_relative_target_position_action_encode_decode_quat():
     assert np.isclose(vec[7], g_tgt[0])
 
     decoded = act._decode_single(
-        {'action': vec}, context={'robot_state.ee_pose': np.concatenate([t_cur[0], q_cur[0].as_quat])}
+        {'action': vec}, context={obs_keys.EE_POSE: np.concatenate([t_cur[0], q_cur[0].as_quat])}
     )
     command = decoded['robot_command']
     target_grip = decoded['target_grip']
@@ -355,18 +356,18 @@ def test_composed_training_encoder_uses_parallel():
     img = [np.zeros((4, 4, 3), dtype=np.uint8) for _ in ts]
 
     ep = EpisodeContainer({
-        'robot_state.q': DummySignal(ts, joints),
-        'grip': DummySignal(ts, grip),
+        obs_keys.JOINTS: DummySignal(ts, joints),
+        obs_keys.GRIP: DummySignal(ts, grip),
         'robot_command.joints': DummySignal(ts, joints),
         'target_grip': DummySignal(ts, grip),
-        'image.wrist': DummySignal(ts, img),
-        'image.exterior': DummySignal(ts, img),
-        'task': 'test',
+        obs_keys.WRIST_IMAGE: DummySignal(ts, img),
+        obs_keys.EXTERIOR_IMAGE: DummySignal(ts, img),
+        obs_keys.TASK: 'test',
     })
 
     obs = ObservationCodec(
-        state={'observation.state': {'robot_state.q': 7, 'grip': 1}},
-        images={'observation.images.left': ('image.wrist', (4, 4))},
+        state={'observation.state': {obs_keys.JOINTS: 7, obs_keys.GRIP: 1}},
+        images={'observation.images.left': (obs_keys.WRIST_IMAGE, (4, 4))},
     )
     action = AbsoluteJointsAction('robot_command.joints', 'target_grip', num_joints=7)
     timing = ActionTiming(fps=15.0)
@@ -407,11 +408,11 @@ def test_binarize_grip_inference():
 
 def test_binarize_grip_training():
     ts = [1000, 2000]
-    ep = EpisodeContainer({'grip': DummySignal(ts, [0.3, 0.8]), 'target_grip': DummySignal(ts, [0.7, 0.2])})
+    ep = EpisodeContainer({obs_keys.GRIP: DummySignal(ts, [0.3, 0.8]), 'target_grip': DummySignal(ts, [0.7, 0.2])})
 
-    binarize = BinarizeGripTraining(('grip', 'target_grip'))
+    binarize = BinarizeGripTraining((obs_keys.GRIP, 'target_grip'))
     result = binarize.training_encoder(ep)
-    grip_vals = [v for v, _ in result['grip']]
+    grip_vals = [v for v, _ in result[obs_keys.GRIP]]
     tgt_vals = [v for v, _ in result['target_grip']]
     np.testing.assert_array_equal(grip_vals, [0.0, 1.0])
     np.testing.assert_array_equal(tgt_vals, [1.0, 0.0])
@@ -419,16 +420,16 @@ def test_binarize_grip_training():
 
 def test_binarize_grip_training_respects_threshold():
     ts = [1000]
-    ep = EpisodeContainer({'grip': DummySignal(ts, [0.4]), 'target_grip': DummySignal(ts, [0.4])})
+    ep = EpisodeContainer({obs_keys.GRIP: DummySignal(ts, [0.4]), 'target_grip': DummySignal(ts, [0.4])})
 
-    keys = ('grip', 'target_grip')
+    keys = (obs_keys.GRIP, 'target_grip')
     default = BinarizeGripTraining(keys)
     result = default.training_encoder(ep)
-    assert list(result['grip'])[0][0] == pytest.approx(0.0)
+    assert list(result[obs_keys.GRIP])[0][0] == pytest.approx(0.0)
 
     low = BinarizeGripTraining(keys, threshold=0.3)
     result = low.training_encoder(ep)
-    assert list(result['grip'])[0][0] == pytest.approx(1.0)
+    assert list(result[obs_keys.GRIP])[0][0] == pytest.approx(1.0)
 
 
 def test_binarize_grip_training_composed_with_action_codec():
@@ -437,7 +438,7 @@ def test_binarize_grip_training_composed_with_action_codec():
 
     ep = EpisodeContainer({'robot_command.joints': DummySignal(ts, joints), 'target_grip': DummySignal(ts, [0.7])})
 
-    binarize = BinarizeGripTraining(('grip', 'target_grip'))
+    binarize = BinarizeGripTraining((obs_keys.GRIP, 'target_grip'))
     action = AbsoluteJointsAction('robot_command.joints', 'target_grip', num_joints=7)
     composed = binarize | action
 
@@ -449,9 +450,9 @@ def test_binarize_grip_training_composed_with_action_codec():
 def test_flip_grip():
     flip = FlipGrip()
 
-    obs = {'grip': 0.2, 'other': 1.0}
-    assert flip.encode(obs) == {'grip': pytest.approx(0.8), 'other': 1.0}
-    assert obs['grip'] == 0.2  # the original obs dict doubles as decode context and must stay intact
+    obs = {obs_keys.GRIP: 0.2, 'other': 1.0}
+    assert flip.encode(obs) == {obs_keys.GRIP: pytest.approx(0.8), 'other': 1.0}
+    assert obs[obs_keys.GRIP] == 0.2  # the original obs dict doubles as decode context and must stay intact
     assert flip.encode({'other': 1.0}) == {'other': 1.0}
 
     assert flip._decode_single({'target_grip': 0.9}, None) == {'target_grip': pytest.approx(0.1)}
@@ -463,11 +464,11 @@ def test_flip_grip():
 
 
 def test_flip_grip_composed_with_obs_and_action():
-    obs = ObservationCodec(state={'observation.state': ['grip']}, images={})
+    obs = ObservationCodec(state={'observation.state': {obs_keys.GRIP: 1}}, images={})
     action = AbsolutePositionAction('robot_command.pose', 'target_grip', Rotation.Representation.QUAT)
     composed = FlipGrip() | (obs & action)
 
-    encoded = composed.encode({'grip': 0.2})
+    encoded = composed.encode({obs_keys.GRIP: 0.2})
     np.testing.assert_allclose(encoded['observation.state'], [0.8])
 
     vec = np.concatenate([[0.1, -0.2, 0.3], Rotation.identity.as_quat, [0.9]]).astype(np.float32)
@@ -509,20 +510,20 @@ def test_sequential_into_parallel_training():
     joints = [np.array([0.1, -0.2, 0.3, 0.4, -0.5, 0.6, 0.7], dtype=np.float32)]
 
     ep = EpisodeContainer({
-        'robot_state.q': DummySignal(ts, joints),
-        'grip': DummySignal(ts, [0.7]),
+        obs_keys.JOINTS: DummySignal(ts, joints),
+        obs_keys.GRIP: DummySignal(ts, [0.7]),
         'robot_command.joints': DummySignal(ts, joints),
         'target_grip': DummySignal(ts, [0.3]),
-        'image.wrist': DummySignal(ts, [np.zeros((4, 4, 3), dtype=np.uint8)]),
-        'image.exterior': DummySignal(ts, [np.zeros((4, 4, 3), dtype=np.uint8)]),
+        obs_keys.WRIST_IMAGE: DummySignal(ts, [np.zeros((4, 4, 3), dtype=np.uint8)]),
+        obs_keys.EXTERIOR_IMAGE: DummySignal(ts, [np.zeros((4, 4, 3), dtype=np.uint8)]),
     })
 
     obs = ObservationCodec(
-        state={'observation.state': {'robot_state.q': 7, 'grip': 1}},
-        images={'observation.images.left': ('image.wrist', (4, 4))},
+        state={'observation.state': {obs_keys.JOINTS: 7, obs_keys.GRIP: 1}},
+        images={'observation.images.left': (obs_keys.WRIST_IMAGE, (4, 4))},
     )
     action = AbsoluteJointsAction('robot_command.joints', 'target_grip', num_joints=7)
-    binarize = BinarizeGripTraining(('grip', 'target_grip'))
+    binarize = BinarizeGripTraining((obs_keys.GRIP, 'target_grip'))
     composed = binarize | (obs & action)
 
     result = composed.training_encoder(ep)
@@ -545,19 +546,19 @@ def test_compose_training_encoder_produces_only_derived_keys():
     img = [np.zeros((4, 4, 3), dtype=np.uint8) for _ in ts]
 
     ep = EpisodeContainer({
-        'robot_state.q': DummySignal(ts, joints),
-        'grip': DummySignal(ts, grip),
+        obs_keys.JOINTS: DummySignal(ts, joints),
+        obs_keys.GRIP: DummySignal(ts, grip),
         'robot_command.joints': DummySignal(ts, joints),
         'target_grip': DummySignal(ts, grip),
-        'image.wrist': DummySignal(ts, img),
-        'image.exterior': DummySignal(ts, img),
-        'task': 'test',
+        obs_keys.WRIST_IMAGE: DummySignal(ts, img),
+        obs_keys.EXTERIOR_IMAGE: DummySignal(ts, img),
+        obs_keys.TASK: 'test',
     })
 
     codec = compose(
         obs=ObservationCodec(
-            state={'observation.state': {'robot_state.q': 7, 'grip': 1}},
-            images={'observation.images.left': ('image.wrist', (4, 4))},
+            state={'observation.state': {obs_keys.JOINTS: 7, obs_keys.GRIP: 1}},
+            images={'observation.images.left': (obs_keys.WRIST_IMAGE, (4, 4))},
         ),
         action=AbsoluteJointsAction('robot_command.joints', 'target_grip', num_joints=7),
     )
@@ -571,8 +572,8 @@ def test_compose_training_encoder_produces_only_derived_keys():
     # Original episode keys must NOT leak through — this fails if compose uses | instead of &
     assert 'target_grip' not in result
     assert 'robot_command.joints' not in result
-    assert 'robot_state.q' not in result
-    assert 'grip' not in result
+    assert obs_keys.JOINTS not in result
+    assert obs_keys.GRIP not in result
 
 
 def test_operator_precedence():
diff --git a/positronic/policy/tests/test_recording.py b/positronic/policy/tests/test_recording.py
index d7b79fe0f..3717407d9 100644
--- a/positronic/policy/tests/test_recording.py
+++ b/positronic/policy/tests/test_recording.py
@@ -1,5 +1,6 @@
 import numpy as np
 
+from positronic import keys
 from positronic.drivers.roboarm.command import CartesianPosition
 from positronic.geom import Rotation, Transform3D
 from positronic.policy.base import Policy, Session
@@ -109,11 +110,11 @@ def test_obs_log_filtering_uses_pure_tap_names(tmp_path):
     session = rec.tap('cam').wrap(_TrackingPolicy([{'v': 1.0, 'timestamp': 0.0}])).new_session()
     session({
         'wall_time_ns': 1_000_000,
-        'task': 'pick up the cube',
+        keys.TASK: 'pick up the cube',
         'camera': np.zeros((4, 4, 3), dtype=np.uint8),
         'joint_pos': np.array([1.0, 2.0], dtype=np.float32),
         'joints_list': [0.1, 0.2, 0.3],
-        'grip': 0.5,
+        keys.GRIP: 0.5,
     })
 
     assert 'cam/camera' in rec._image_paths
diff --git a/positronic/policy/tests/test_sampled_e2e.py b/positronic/policy/tests/test_sampled_e2e.py
index a2da718f3..8990b0b8b 100644
--- a/positronic/policy/tests/test_sampled_e2e.py
+++ b/positronic/policy/tests/test_sampled_e2e.py
@@ -15,6 +15,7 @@
 import pytest
 
 import pimm
+from positronic import keys
 from positronic.dataset.ds_writer_agent import DsWriterCommandType
 from positronic.drivers import roboarm
 from positronic.drivers.roboarm.command import CartesianPosition
@@ -33,7 +34,7 @@ def __init__(self, target, meta):
         self._meta = meta
 
     def __call__(self, obs):
-        current_pos = np.asarray(obs['robot_state.ee_pose'][:3], dtype=np.float32)
+        current_pos = np.asarray(obs[keys.EE_POSE][:3], dtype=np.float32)
         delta = self._target - current_pos
         actions = []
         for i in range(5):
@@ -89,7 +90,7 @@ def _pair_all(world, harness):
     return {
         'frame_em': world.pair(harness.observations['image.cam']),
         'robot_em': world.pair(harness.observations['robot_state']),
-        'grip_em': world.pair(harness.observations['grip']),
+        'grip_em': world.pair(harness.observations[keys.GRIP]),
         'directive_em': world.pair(harness.directive),
         'meta_em': world.pair(harness.robot_meta_in),
         'ds_recorder': ds_recorder,
diff --git a/positronic/policy/tests/test_wrappers.py b/positronic/policy/tests/test_wrappers.py
index 2c662f335..6fc251a80 100644
--- a/positronic/policy/tests/test_wrappers.py
+++ b/positronic/policy/tests/test_wrappers.py
@@ -5,6 +5,7 @@
 import numpy as np
 import pytest
 
+from positronic import keys
 from positronic.policy import spec
 from positronic.policy.action import (
     AbsoluteJointsAction,
@@ -268,7 +269,9 @@ def test_spec_round_trip(self):
         assert rebuilt is not None and rebuilt.to_spec() == stack.to_spec()
 
     def test_codec_spec_round_trip(self):
-        obs = ObservationCodec(state={'observation.state': {'grip': 1}}, images={'left': ('image.wrist', (224, 224))})
+        obs = ObservationCodec(
+            state={'observation.state': {'grip': 1}}, images={'left': (keys.WRIST_IMAGE, (224, 224))}
+        )
         local = ChunkedSchedule() | ActionTimestamp(fps=10.0) | (obs & AbsolutePositionAction('pose', 'grip'))
         rebuilt = spec.from_spec(local.to_spec())
         assert rebuilt is not None and rebuilt.to_spec() == local.to_spec()
diff --git a/positronic/probe.py b/positronic/probe.py
index 2a3d9b01e..1d8ecca6e 100644
--- a/positronic/probe.py
+++ b/positronic/probe.py
@@ -29,6 +29,7 @@
 
 import positronic.cfg.ds
 import positronic.cfg.policy as policy_cfg
+from positronic import keys
 from positronic.dataset.dataset import Dataset
 from positronic.drivers.roboarm.command import CartesianPosition, JointDelta
 from positronic.policy import Policy, Recorder, is_action
@@ -37,14 +38,14 @@
 # Tap name; the recorder logs each obs/action entity under ``{_TAP}/{key}`` (see recording.py).
 _TAP = 'raw'
 # Observation keys the endpoint expects, mirroring the inference harness, plus every image.*.
-_STATE_KEYS = ('robot_state.q', 'robot_state.dq', 'robot_state.ee_pose', 'grip')
+_STATE_KEYS = (keys.JOINTS, keys.JOINT_VEL, keys.EE_POSE, keys.GRIP)
 
 
 def _build_wire_obs(sample: dict, task: str | None, now_ns: int, recorded_ts: int) -> dict:
     obs = {k: sample[k] for k in _STATE_KEYS if k in sample}
     obs.update({k: v for k, v in sample.items() if k.startswith('image.')})
     if task:
-        obs['task'] = task
+        obs[keys.TASK] = task
     obs['wall_time_ns'] = now_ns  # rerun wall_time timeline
     obs['obs_time_ns'] = recorded_ts  # rerun obs_time + action_time anchor
     return obs
@@ -124,16 +125,16 @@ def main(
     ep = dataset[episode]
     ts = int(np.clip(ep.start_ts + int(at * 1e9), ep.start_ts, ep.last_ts))
     sample = ep.time[ts]
-    if 'robot_state.ee_pose' not in sample:
+    if keys.EE_POSE not in sample:
         raise ValueError('episode has no robot_state.ee_pose; cannot overlay actual pose')
-    task = task or ep.static.get('task')
+    task = task or ep.static.get(keys.TASK)
 
     now_ns = time.time_ns()
     obs = _build_wire_obs(sample, task, now_ns, ts)
     image_keys = [k for k in obs if k.startswith('image.')]
 
     rec = Recorder(pos3.sync(output_dir))
-    session = rec.tap(_TAP).wrap(policy).new_session({'task': task} if task else None, time.time)
+    session = rec.tap(_TAP).wrap(policy).new_session({keys.TASK: task} if task else None, time.time)
     meta = dict(session.meta)
     name = label or _recording_name(meta)
     try:
diff --git a/positronic/replay_record.py b/positronic/replay_record.py
index 3ea802768..c8056f5c8 100644
--- a/positronic/replay_record.py
+++ b/positronic/replay_record.py
@@ -11,7 +11,7 @@
 import pimm
 import positronic.cfg.ds
 import positronic.cfg.simulator
-from positronic import geom, wire
+from positronic import geom, keys, wire
 from positronic.dataset import Dataset, Episode, transforms
 from positronic.dataset.ds_player_agent import DsPlayerAgent, DsPlayerStartCommand
 from positronic.dataset.ds_writer_agent import DsWriterCommand, TimeMode
@@ -91,7 +91,7 @@ def parse_episodes(episodes: int | list[int] | str, dataset: Dataset) -> list[in
 
 @cfn.config(
     dataset=positronic.cfg.ds.local_all,
-    cameras={'image.wrist': 'handcam_left_ph', 'image.wrist_2': 'wrist_cam_ph', 'image.exterior': 'back_view_ph'},
+    cameras={keys.WRIST_IMAGE: 'handcam_left_ph', 'image.wrist_2': 'wrist_cam_ph', keys.EXTERIOR_IMAGE: 'back_view_ph'},
     mujoco_model_path=package_assets_path('assets/mujoco/franka_table.xml'),
     loaders=positronic.cfg.simulator.stack_cubes_loaders,
 )
diff --git a/positronic/server/positronic_server.py b/positronic/server/positronic_server.py
index c803bee0f..62b12ba61 100644
--- a/positronic/server/positronic_server.py
+++ b/positronic/server/positronic_server.py
@@ -26,7 +26,7 @@
 from starlette.requests import Request
 
 import positronic.cfg.ds
-from positronic import utils
+from positronic import keys, utils
 from positronic.dataset import CachedDataset, Dataset, Episode
 from positronic.dataset.local_dataset import LocalDataset
 from positronic.server.dataset_utils import get_dataset_root, get_episodes_list, stream_episode_rrd
@@ -208,7 +208,7 @@ def _make_serializable(obj, path=''):
             'episode_id': episode_id,
             'num_episodes': len(ds),
             'rerun_version': rr.__version__,
-            'task': episode.static.get('task', None),
+            'task': episode.static.get(keys.TASK, None),
             'repo_id': app_state['root'],
             'episode_path': meta.get('path'),
             'episode_size_mb': size_mb_display,
diff --git a/positronic/simulator/env_server/proxy.py b/positronic/simulator/env_server/proxy.py
index e8a9e9a84..49eb07014 100644
--- a/positronic/simulator/env_server/proxy.py
+++ b/positronic/simulator/env_server/proxy.py
@@ -14,6 +14,7 @@
 from typing import Any
 
 import pimm
+from positronic import keys
 from positronic.dataset.serializers import Serializers
 from positronic.drivers.roboarm import command as roboarm_command
 from positronic.eval import ROBOT_STATIC_META, Command, Embodiment, Observation
@@ -142,7 +143,7 @@ def remote_franka_embodiment(
     """
     observations = {
         'robot_state': Observation(proxy.observations['robot_state'], Serializers.robot_state),
-        'grip': Observation(proxy.observations['grip'], None),
+        keys.GRIP: Observation(proxy.observations[keys.GRIP], None),
         **{logical: Observation(proxy.observations[logical], Serializers.camera_images) for logical in camera_dict},
     }
     commands = {
diff --git a/positronic/simulator/env_server/tests/mujoco_env.py b/positronic/simulator/env_server/tests/mujoco_env.py
index 5a8248c14..59e89a327 100644
--- a/positronic/simulator/env_server/tests/mujoco_env.py
+++ b/positronic/simulator/env_server/tests/mujoco_env.py
@@ -17,7 +17,7 @@
 import pimm
 import positronic.cfg.simulator
 from pimm.world import LocalQueueEmitter, LocalQueueReceiver, VirtualClock
-from positronic import geom
+from positronic import geom, keys
 from positronic.drivers.roboarm import command as roboarm_command
 from positronic.eval import Eval, Observation, Task
 from positronic.simulator.env_server.adapter import WireCommandAdapter
@@ -172,7 +172,7 @@ def observations(self, raw_obs: dict[str, Any]) -> dict[str, Any]:
         ee_pose = geom.Transform3D(raw_obs['ee_pos'], geom.Rotation.from_quat(raw_obs['ee_quat']))
         state.encode(raw_obs['q'], raw_obs['dq'], ee_pose)
         state.array[14 + 7] = float(raw_obs['status'])
-        obs: dict[str, Any] = {'robot_state': state, 'grip': float(raw_obs['grip'])}
+        obs: dict[str, Any] = {'robot_state': state, keys.GRIP: float(raw_obs['grip'])}
         for logical, model_name in self._camera_dict.items():
             frame = raw_obs['cameras'][model_name]
             adapter = pimm.shared_memory.NumpySMAdapter(shape=frame.shape, dtype=frame.dtype)
diff --git a/positronic/simulator/libero/adapter.py b/positronic/simulator/libero/adapter.py
index 6e070a9a0..05711092b 100644
--- a/positronic/simulator/libero/adapter.py
+++ b/positronic/simulator/libero/adapter.py
@@ -11,7 +11,7 @@
 import numpy as np
 
 import pimm
-from positronic import geom
+from positronic import geom, keys
 from positronic.simulator.env_server.adapter import WireCommandAdapter
 from positronic.simulator.mujoco.sim import MujocoFrankaState
 
@@ -42,7 +42,7 @@ def observations(self, raw_obs: dict[str, Any]) -> dict[str, Any]:
         ee_pose = geom.Transform3D(raw_obs['eef_pos'], geom.Rotation.from_quat_xyzw(raw_obs['eef_quat']))
         state = MujocoFrankaState()
         state.encode(raw_obs['joint_pos'], raw_obs['joint_vel'], ee_pose)
-        obs: dict[str, Any] = {'robot_state': state, 'grip': float(raw_obs['grip'])}
+        obs: dict[str, Any] = {'robot_state': state, keys.GRIP: float(raw_obs['grip'])}
         for logical, env_key in self._camera_dict.items():
             # robosuite renders bottom-up; flip to standard top-down orientation (LIBERO's own video path
             # flips the same way).
diff --git a/positronic/simulator/robolab/adapter.py b/positronic/simulator/robolab/adapter.py
index 62eb71a1c..26d2643bf 100644
--- a/positronic/simulator/robolab/adapter.py
+++ b/positronic/simulator/robolab/adapter.py
@@ -11,7 +11,7 @@
 import numpy as np
 
 import pimm
-from positronic import geom
+from positronic import geom, keys
 from positronic.simulator.env_server.adapter import WireCommandAdapter
 from positronic.simulator.mujoco.sim import MujocoFrankaState
 
@@ -31,7 +31,7 @@ def observations(self, raw_obs: dict[str, Any]) -> dict[str, Any]:
         ee_pose = geom.Transform3D(raw_obs['eef_pos'], geom.Rotation.from_quat(raw_obs['eef_quat']))
         state = MujocoFrankaState()
         state.encode(raw_obs['joint_pos'], raw_obs['joint_vel'], ee_pose)
-        obs: dict[str, Any] = {'robot_state': state, 'grip': float(raw_obs['grip'])}
+        obs: dict[str, Any] = {'robot_state': state, keys.GRIP: float(raw_obs['grip'])}
         # TODO: honour a camera_dict naming any other RoboLab camera. env.py renders only the WRIST_LEFT
         # preset (over_shoulder_left + wrist) and hard-codes emitting those two, so a request for e.g.
         # over_shoulder_right_camera raises below. The full fix threads the requested set end-to-end: carry
diff --git a/positronic/tests/test_data_collection.py b/positronic/tests/test_data_collection.py
index 0828fcc45..71b6bf22b 100644
--- a/positronic/tests/test_data_collection.py
+++ b/positronic/tests/test_data_collection.py
@@ -5,7 +5,7 @@
 import pytest
 
 import pimm
-from positronic import wire
+from positronic import keys, wire
 from positronic.data_collection import DataCollectionController, OperatorPosition, controller_positions_serializer
 from positronic.dataset.ds_writer_agent import (
     DsWriterAgent,
@@ -13,6 +13,7 @@
     DsWriterCommandType,
     TrajectoryOverrideSerializer,
 )
+from positronic.dataset.episode import Episode
 from positronic.dataset.local_dataset import LocalDataset, LocalDatasetWriter
 from positronic.dataset.serializers import Serializers
 from positronic.geom import Rotation, Transform3D
@@ -75,7 +76,7 @@ def test_data_collection_records_task_metadata(tmp_path, world):
     def metadata_getter():
         nonlocal call_count
         call_count += 1
-        return {'task': 'stack-blocks'}
+        return {keys.TASK: 'stack-blocks'}
 
     (dc, agent, ctrl_em_dc, ctrl_em_agent, buttons_em, writer_cm, robot) = build_collection(
         world, tmp_path, metadata_getter=metadata_getter
@@ -109,7 +110,8 @@ def send_buttons(**kwargs):
     dataset = LocalDataset(tmp_path)
     assert len(dataset) == 1
     episode = dataset[0]
-    assert episode['task'] == 'stack-blocks'
+    assert isinstance(episode, Episode)
+    assert episode[keys.TASK] == 'stack-blocks'
 
 
 def test_data_collection_basic_recording(tmp_path, world):
@@ -164,7 +166,7 @@ def test_data_collection_with_mujoco_robot_gripper(tmp_path):
         agent.add_signal('robot_command', TrajectoryOverrideSerializer(Serializers.robot_command))
         agent.add_signal('controller_positions', controller_positions_serializer)
         agent.add_signal('robot_state', Serializers.robot_state)
-        agent.add_signal('grip')
+        agent.add_signal(keys.GRIP)
 
         world.connect(sim.state, dc.robot_state)
         world.connect(sim.state, agent.inputs['robot_state'])
@@ -172,7 +174,7 @@ def test_data_collection_with_mujoco_robot_gripper(tmp_path):
         world.connect(dc.robot_commands, agent.inputs['robot_command'])
         world.connect(dc.target_grip, sim.target_grip)
         world.connect(dc.target_grip, agent.inputs['target_grip'])
-        world.connect(sim.grip, agent.inputs['grip'])
+        world.connect(sim.grip, agent.inputs[keys.GRIP])
         world.connect(dc.ds_agent_commands, agent.command)
 
         ctrl_em_dc = world.pair(dc.controller_positions)
@@ -209,20 +211,14 @@ def stop_episode():
     ds = LocalDataset(tmp_path)
     assert len(ds) == 1
     ep = ds[0]
+    assert isinstance(ep, Episode)
 
-    expected = {
-        'target_grip',
-        'controller_positions.right',
-        'robot_state.q',
-        'robot_state.dq',
-        'robot_state.ee_pose',
-        'grip',
-    }
+    expected = {'target_grip', 'controller_positions.right', keys.JOINTS, keys.JOINT_VEL, keys.EE_POSE, keys.GRIP}
     assert expected.issubset(set(ep.keys()))
 
     # Robot/gripper signals should have at least one sample
-    robot_j = ep['robot_state.q']
-    grip_sig = ep['grip']
+    robot_j = ep[keys.JOINTS]
+    grip_sig = ep[keys.GRIP]
     assert len(robot_j) >= 1
     assert len(grip_sig) >= 1
 
@@ -242,7 +238,7 @@ def assert_strictly_increasing(sig):
         for i in range(1, len(sig)):
             assert sig[i][1] > sig[i - 1][1]
 
-    for name in ['robot_state.q', 'robot_state.dq', 'grip']:
+    for name in [keys.JOINTS, keys.JOINT_VEL, keys.GRIP]:
         assert_strictly_increasing(ep[name])
 
 
diff --git a/positronic/tests/test_inference_integration.py b/positronic/tests/test_inference_integration.py
index 517135157..837b2ce58 100644
--- a/positronic/tests/test_inference_integration.py
+++ b/positronic/tests/test_inference_integration.py
@@ -9,6 +9,7 @@
 
 import pimm
 import positronic.cfg.simulator
+from positronic import keys
 from positronic.cfg.eval.sim.positronic import stack_cubes
 from positronic.dataset.local_dataset import LocalDataset
 from positronic.dataset.serializers import Serializers
@@ -63,7 +64,7 @@ def close(self):
 
     policy = StubPolicy()
 
-    camera_dict = {'image.wrist': 'handcam_left_ph'}
+    camera_dict = {keys.WRIST_IMAGE: 'handcam_left_ph'}
 
     with pos3.mirror():
         ev = stack_cubes(
@@ -97,11 +98,11 @@ def close(self):
     signals = episode.signals
     assert 'robot_command.pose' in signals
     assert 'target_grip' in signals
-    assert 'image.wrist' in signals
+    assert keys.WRIST_IMAGE in signals
     # Privileged ground truth: the full sim state is recorded as a time-series signal.
     assert 'sim_state.mjSTATE_INTEGRATION' in signals
 
-    camera_samples = list(signals['image.wrist'])
+    camera_samples = list(signals[keys.WRIST_IMAGE])
     assert camera_samples, 'Camera signal for handcam_left is empty'
     first_image, _ = camera_samples[0]
     assert isinstance(first_image, np.ndarray)
@@ -137,10 +138,10 @@ def close(self):
 
     assert policy.observations, 'Policy did not receive any observations'
     last_obs = policy.observations[-1]
-    assert isinstance(last_obs['image.wrist'], np.ndarray)
-    assert 'robot_state.ee_pose' in last_obs
+    assert isinstance(last_obs[keys.WRIST_IMAGE], np.ndarray)
+    assert keys.EE_POSE in last_obs
     # The task's instruction is injected by the harness.
-    assert last_obs['task'] == 'integration-test'
+    assert last_obs[keys.TASK] == 'integration-test'
     assert last_obs['descriptor'] == 'mujoco.franka'
 
 
diff --git a/positronic/tests/test_keys.py b/positronic/tests/test_keys.py
new file mode 100644
index 000000000..2fed3df71
--- /dev/null
+++ b/positronic/tests/test_keys.py
@@ -0,0 +1,53 @@
+import ast
+from pathlib import Path
+
+from positronic import keys
+
+# Namespaced raw wire keys that denote an observation signal wherever they appear as a string literal.
+# Writing any of these as a bare literal instead of importing the constant is what this guard forbids —
+# the value must live once, in `positronic.keys`, so a rename stays a single-site change.
+_GUARDED = {keys.JOINTS, keys.JOINT_VEL, keys.EE_POSE, keys.WRIST_IMAGE, keys.EXTERIOR_IMAGE}
+# keys.GRIP and keys.TASK are deliberately not guarded: their values are bare tokens the wire reuses
+# across unrelated namespaces (action-command grip, vendor state-vectors, scene/reset tokens), so a
+# literal-value match cannot tell the observation key from those and would fire on legitimate code.
+
+_PACKAGE_ROOT = Path(__file__).resolve().parent.parent
+_REPO_ROOT = _PACKAGE_ROOT.parent
+_KEYS_MODULE = _PACKAGE_ROOT / 'keys.py'
+# First-party trees whose Python consumes the wire and must import the constants rather than
+# re-spell the literals — the package itself and the repo's `utilities/` scripts.
+_GUARDED_ROOTS = (_PACKAGE_ROOT, _REPO_ROOT / 'utilities')
+
+
+def _str_literals(tree: ast.AST):
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Constant) and isinstance(node.value, str):
+            yield node.value, node.lineno
+
+
+def test_no_raw_observation_key_literals():
+    offenders = []
+    for root in _GUARDED_ROOTS:
+        for path in sorted(root.rglob('*.py')):
+            if path == _KEYS_MODULE:
+                continue
+            tree = ast.parse(path.read_text(), filename=str(path))
+            for value, lineno in _str_literals(tree):
+                if value in _GUARDED:
+                    offenders.append(f'{path.relative_to(_REPO_ROOT)}:{lineno}: {value!r}')
+    assert not offenders, (
+        'Raw observation-key literals found — import the constant from `positronic.keys`:\n' + '\n'.join(offenders)
+    )
+
+
+def test_keys_module_imports_nothing():
+    # `positronic.keys` must stay a dependency-free leaf so an out-of-repo consumer can depend on it
+    # alone, without dragging in the rest of positronic (or its optional torch/lerobot deps). Any
+    # import statement appearing here breaks that contract.
+    tree = ast.parse(_KEYS_MODULE.read_text(), filename=str(_KEYS_MODULE))
+    imports = [
+        f'{_KEYS_MODULE.relative_to(_PACKAGE_ROOT.parent)}:{node.lineno}'
+        for node in ast.walk(tree)
+        if isinstance(node, (ast.Import, ast.ImportFrom))
+    ]
+    assert not imports, '`positronic.keys` must import nothing (dependency-free leaf module):\n' + '\n'.join(imports)
diff --git a/positronic/vendors/dreamzero/codecs.py b/positronic/vendors/dreamzero/codecs.py
index 9f7bf54e0..b3e281561 100644
--- a/positronic/vendors/dreamzero/codecs.py
+++ b/positronic/vendors/dreamzero/codecs.py
@@ -7,6 +7,7 @@
 import numpy as np
 from PIL import Image as PilImage
 
+from positronic import keys
 from positronic.cfg import codecs, wrappers
 from positronic.dataset import Signal, transforms
 from positronic.dataset.episode import Episode
@@ -36,8 +37,8 @@ class DreamZeroObservationCodec(Codec):
 
     def __init__(
         self,
-        wrist_camera: str = 'image.wrist',
-        exterior_camera_1: str = 'image.exterior',
+        wrist_camera: str = keys.WRIST_IMAGE,
+        exterior_camera_1: str = keys.EXTERIOR_IMAGE,
         exterior_camera_2: str | None = None,
         image_size: tuple[int, int] = (IMAGE_WIDTH, IMAGE_HEIGHT),
     ):
@@ -53,7 +54,7 @@ def __init__(
             'video.wrist_image_left': partial(self._derive_image, wrist_camera),
             'video.exterior_image_1_left': partial(self._derive_image, exterior_camera_1),
             'video.exterior_image_2_left': partial(self._derive_image, self._exterior_camera_2),
-            'task': Get('task', ''),
+            'task': Get(keys.TASK, ''),
         }
 
         self._training_meta = {
@@ -83,10 +84,10 @@ def __init__(
         }
 
     def _derive_joint_position(self, episode: Episode) -> Signal[Any]:
-        return transforms.astype(episode['robot_state.q'], np.float32)
+        return transforms.astype(episode[keys.JOINTS], np.float32)
 
     def _derive_gripper_position(self, episode: Episode) -> Signal[Any]:
-        return transforms.Elementwise(episode['grip'], _reshape_grip)
+        return transforms.Elementwise(episode[keys.GRIP], _reshape_grip)
 
     def _derive_image(self, input_key: str, episode: Episode) -> Signal[Any]:
         w, h = self._image_size
@@ -103,8 +104,8 @@ def _encode_image(self, input_key: str, inputs: dict[str, Any]) -> np.ndarray:
         return image.resize_with_pad_per_frame(w, h, PilImage.Resampling.BILINEAR, frame)
 
     def encode(self, inputs: dict[str, Any]) -> dict[str, Any]:
-        joint_pos = np.asarray(inputs['robot_state.q'], dtype=np.float32).reshape(-1)
-        grip = np.asarray(inputs['grip'], dtype=np.float32).reshape(-1)
+        joint_pos = np.asarray(inputs[keys.JOINTS], dtype=np.float32).reshape(-1)
+        grip = np.asarray(inputs[keys.GRIP], dtype=np.float32).reshape(-1)
 
         obs = {
             'observation/joint_position': joint_pos,
@@ -114,8 +115,8 @@ def encode(self, inputs: dict[str, Any]) -> dict[str, Any]:
             'observation/exterior_image_1_left': self._encode_image(self._exterior_camera_2, inputs),
         }
 
-        if 'task' in inputs:
-            obs['prompt'] = inputs['task']
+        if keys.TASK in inputs:
+            obs['prompt'] = inputs[keys.TASK]
 
         return obs
 
@@ -196,8 +197,8 @@ def training_encoder(self):
 
 
 @cfn.config(
-    wrist_camera='image.wrist',
-    exterior_camera_1='image.exterior',
+    wrist_camera=keys.WRIST_IMAGE,
+    exterior_camera_1=keys.EXTERIOR_IMAGE,
     exterior_camera_2=None,
     image_size=(IMAGE_WIDTH, IMAGE_HEIGHT),
 )
@@ -229,7 +230,7 @@ def dreamzero_action(tgt_joints_key: str, tgt_grip_key: str, num_joints: int):
 # 176-tall codec output at load, so only the DROID-serving codec needs the taller image.
 droid = codecs.compose.override(obs=dreamzero_obs.override(image_size=(IMAGE_WIDTH, 180)), action=_action, fps=15.0)
 
-_traj_action = dreamzero_action.override(tgt_joints_key='robot_state.q', tgt_grip_key='grip')
+_traj_action = dreamzero_action.override(tgt_joints_key=keys.JOINTS, tgt_grip_key=keys.GRIP)
 joints_traj = codecs.compose.override(obs=dreamzero_obs, action=_traj_action, fps=15.0)
 
 # IK variants: reconstruct joint targets from recorded EE targets via IK
@@ -252,5 +253,5 @@ def _ik_dreamzero_action(solver: str):
 # at stride 8 from the current frame with the oldest pinned to the window start → trained offsets
 # -23, -16, -8, 0 (test_client_AR.py RELATIVE_OFFSETS; -23 not -24 keeps the oldest inside the window).
 dreamzero_wrappers = wrappers.video_context_wrappers.override(
-    history_frames=23, stride=8, keys=('image.wrist', 'image.exterior')
+    history_frames=23, stride=8, keys=(keys.WRIST_IMAGE, keys.EXTERIOR_IMAGE)
 )
diff --git a/positronic/vendors/dreamzero/tests/test_codecs.py b/positronic/vendors/dreamzero/tests/test_codecs.py
index 4315eb639..e07665b99 100644
--- a/positronic/vendors/dreamzero/tests/test_codecs.py
+++ b/positronic/vendors/dreamzero/tests/test_codecs.py
@@ -3,6 +3,7 @@
 import numpy as np
 import pytest
 
+from positronic import keys
 from positronic.vendors.dreamzero.codecs import DreamZeroObservationCodec
 
 
@@ -10,11 +11,11 @@ class TestDreamZeroObservationCodec:
     @pytest.fixture
     def sample_inputs(self):
         return {
-            'robot_state.q': np.array([0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7]),
-            'grip': np.array([0.5]),
-            'image.wrist': np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8),
-            'image.exterior': np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8),
-            'task': 'pick up the cube',
+            keys.JOINTS: np.array([0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7]),
+            keys.GRIP: np.array([0.5]),
+            keys.WRIST_IMAGE: np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8),
+            keys.EXTERIOR_IMAGE: np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8),
+            keys.TASK: 'pick up the cube',
         }
 
     def test_encode_basic(self, sample_inputs):
@@ -30,7 +31,7 @@ def test_encode_basic(self, sample_inputs):
 
         assert result['observation/joint_position'].shape == (7,)
         assert result['observation/gripper_position'].shape == (1,)
-        assert np.allclose(result['observation/joint_position'], sample_inputs['robot_state.q'])
+        assert np.allclose(result['observation/joint_position'], sample_inputs[keys.JOINTS])
         assert result['prompt'] == 'pick up the cube'
 
     def test_encode_image_resize(self, sample_inputs):
@@ -43,7 +44,7 @@ def test_encode_image_resize(self, sample_inputs):
         assert result['observation/exterior_image_1_left'].shape == (176, 320, 3)
 
     def test_encode_missing_task(self, sample_inputs):
-        del sample_inputs['task']
+        del sample_inputs[keys.TASK]
         codec = DreamZeroObservationCodec()
         result = codec.encode(sample_inputs)
 
@@ -73,8 +74,8 @@ def test_meta(self):
         assert DreamZeroObservationCodec().meta == {'image_sizes': (320, 176)}
 
     def test_custom_camera_keys(self, sample_inputs):
-        sample_inputs['cam1'] = sample_inputs.pop('image.wrist')
-        sample_inputs['cam2'] = sample_inputs.pop('image.exterior')
+        sample_inputs['cam1'] = sample_inputs.pop(keys.WRIST_IMAGE)
+        sample_inputs['cam2'] = sample_inputs.pop(keys.EXTERIOR_IMAGE)
         sample_inputs['cam3'] = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
 
         codec = DreamZeroObservationCodec(wrist_camera='cam1', exterior_camera_1='cam2', exterior_camera_2='cam3')
@@ -85,7 +86,7 @@ def test_custom_camera_keys(self, sample_inputs):
         assert result['observation/exterior_image_1_left'].shape == (176, 320, 3)
 
     def test_non_square_input_image(self, sample_inputs):
-        sample_inputs['image.wrist'] = np.random.randint(0, 255, (100, 200, 3), dtype=np.uint8)
+        sample_inputs[keys.WRIST_IMAGE] = np.random.randint(0, 255, (100, 200, 3), dtype=np.uint8)
 
         codec = DreamZeroObservationCodec()
         result = codec.encode(sample_inputs)
diff --git a/positronic/vendors/gr00t/codecs.py b/positronic/vendors/gr00t/codecs.py
index d2e5a5aa4..1453878a9 100644
--- a/positronic/vendors/gr00t/codecs.py
+++ b/positronic/vendors/gr00t/codecs.py
@@ -7,7 +7,7 @@
 import numpy as np
 from PIL import Image as PilImage
 
-from positronic import geom
+from positronic import geom, keys
 from positronic.cfg import codecs
 from positronic.dataset import transforms
 from positronic.dataset import transforms as tf
@@ -33,8 +33,8 @@ def __init__(
         include_joints: bool = False,
         include_ee_pose: bool = True,
         image_size: tuple[int, int] = (224, 224),
-        exterior_camera: str = 'image.exterior',
-        wrist_camera: str = 'image.wrist',
+        exterior_camera: str = keys.EXTERIOR_IMAGE,
+        wrist_camera: str = keys.WRIST_IMAGE,
         num_joints: int = 7,
     ):
         self._rotation_rep = rotation_rep
@@ -49,7 +49,7 @@ def __init__(
             'grip': self._derive_grip,
             'wrist_image': partial(self._derive_image, wrist_camera),
             'exterior_image_1': partial(self._derive_image, exterior_camera),
-            'task': Get('task', ''),
+            'task': Get(keys.TASK, ''),
         }
 
         state_meta: dict[str, Any] = {'grip': {'start': 0, 'end': 1, 'original_key': 'grip'}}
@@ -82,7 +82,7 @@ def __init__(
         }
 
     def _derive_ee_pose(self, episode: Episode) -> Signal[Any]:
-        pose = episode['robot_state.ee_pose']
+        pose = episode[keys.EE_POSE]
         if self._rotation_rep is not None:
             pose = tf.recode_transform(RotRep.QUAT, self._rotation_rep, pose)
         return tf.astype(pose, np.float32)
@@ -92,17 +92,17 @@ def _reshape_to_1d(values):
             arr = np.asarray(values, dtype=np.float32)
             return arr.reshape(-1, 1)
 
-        return transforms.Elementwise(episode['grip'], _reshape_to_1d)
+        return transforms.Elementwise(episode[keys.GRIP], _reshape_to_1d)
 
     def _derive_joints(self, episode: Episode) -> Signal[Any]:
-        return tf.astype(episode['robot_state.q'], np.float32)
+        return tf.astype(episode[keys.JOINTS], np.float32)
 
     def _derive_image(self, input_key: str, episode: Episode) -> Signal[Any]:
         w, h = self._image_size
         return image.resize_with_pad(w, h, signal=episode[input_key])
 
     def _encode_ee_pose(self, inputs: dict[str, Any]) -> np.ndarray:
-        pose = np.asarray(inputs['robot_state.ee_pose'], dtype=np.float32).reshape(-1)
+        pose = np.asarray(inputs[keys.EE_POSE], dtype=np.float32).reshape(-1)
         if self._rotation_rep is not None:
             pose = geom.Transform3D.from_vector(pose, RotRep.QUAT).as_vector(self._rotation_rep).astype(np.float32)
         return pose
@@ -136,14 +136,14 @@ def _decode_single(self, data: dict, context: dict | None) -> dict:
         return {}
 
     def encode(self, inputs: dict[str, Any]) -> dict[str, Any]:
-        grip = np.asarray(inputs['grip'], dtype=np.float32).reshape(-1)
+        grip = np.asarray(inputs[keys.GRIP], dtype=np.float32).reshape(-1)
         state_dict: dict[str, Any] = {'grip': grip[np.newaxis, np.newaxis, ...]}
 
         if self._include_ee_pose:
             ee_pose = self._encode_ee_pose(inputs)
             state_dict['ee_pose'] = ee_pose[np.newaxis, np.newaxis, ...]
         if self._include_joints:
-            joints = np.asarray(inputs['robot_state.q'], dtype=np.float32).reshape(-1)
+            joints = np.asarray(inputs[keys.JOINTS], dtype=np.float32).reshape(-1)
             state_dict['joint_position'] = joints[np.newaxis, np.newaxis, ...]
 
         return {
@@ -152,7 +152,7 @@ def encode(self, inputs: dict[str, Any]) -> dict[str, Any]:
                 'exterior_image_1': self._encode_image(self._exterior_camera, inputs)[np.newaxis, np.newaxis, ...],
             },
             'state': state_dict,
-            'language': {'annotation.language.language_instruction': [[inputs.get('task', '')]]},
+            'language': {'annotation.language.language_instruction': [[inputs.get(keys.TASK, '')]]},
         }
 
     @property
@@ -224,18 +224,18 @@ def groot_action(base, action_key: str, action_dim: int):
 _traj_action = _ee_action.override(base=codecs.traj_ee_action)
 _rot6d_traj_action = _rot6d_action.override(base=codecs.traj_ee_action.override(rotation_rep='rot6d'))
 
-ee_quat_traj = codecs.compose.override(obs=groot_obs, action=_traj_action, binarize_grip=('grip',))
-ee_rot6d_traj = codecs.compose.override(obs=_rot6d_obs, action=_rot6d_traj_action, binarize_grip=('grip',))
+ee_quat_traj = codecs.compose.override(obs=groot_obs, action=_traj_action, binarize_grip=(keys.GRIP,))
+ee_rot6d_traj = codecs.compose.override(obs=_rot6d_obs, action=_rot6d_traj_action, binarize_grip=(keys.GRIP,))
 ee_quat_joints_traj = ee_quat_traj.override(**{'obs.include_joints': True})
 ee_rot6d_joints_traj = ee_rot6d_traj.override(**{'obs.include_joints': True})
 
 joints_traj = codecs.compose.override(
     obs=groot_obs.override(include_joints=True, include_ee_pose=False),
     action=groot_action.override(
-        base=codecs.absolute_joints_action.override(tgt_joints_key='robot_state.q', tgt_grip_key='grip'),
+        base=codecs.absolute_joints_action.override(tgt_joints_key=keys.JOINTS, tgt_grip_key=keys.GRIP),
         action_key='joint_position',
     ),
-    binarize_grip=('grip',),
+    binarize_grip=(keys.GRIP,),
 )
 
 # IK variants: GR00T obs (with joints) + IK joint-space action via groot_action wrapper
diff --git a/positronic/vendors/gr00t/tests/test_observation.py b/positronic/vendors/gr00t/tests/test_observation.py
index 6699b8f62..72a32e2c8 100644
--- a/positronic/vendors/gr00t/tests/test_observation.py
+++ b/positronic/vendors/gr00t/tests/test_observation.py
@@ -3,7 +3,7 @@
 import numpy as np
 import pytest
 
-from positronic import geom
+from positronic import geom, keys
 from positronic.vendors.gr00t.codecs import GrootObservationCodec
 
 RotRep = geom.Rotation.Representation
@@ -16,12 +16,12 @@ class TestGrootObservationCodec:
     def sample_inputs(self):
         """Sample raw inputs for inference encoding."""
         return {
-            'robot_state.ee_pose': np.array([0.1, 0.2, 0.3, 0.0, 0.0, 0.0, 1.0]),  # xyz + quat (w,x,y,z)
-            'grip': np.array([0.5]),
-            'robot_state.q': np.array([0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7]),
-            'image.wrist': np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8),
-            'image.exterior': np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8),
-            'task': 'pick up the cube',
+            keys.EE_POSE: np.array([0.1, 0.2, 0.3, 0.0, 0.0, 0.0, 1.0]),  # xyz + quat (w,x,y,z)
+            keys.GRIP: np.array([0.5]),
+            keys.JOINTS: np.array([0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7]),
+            keys.WRIST_IMAGE: np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8),
+            keys.EXTERIOR_IMAGE: np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8),
+            keys.TASK: 'pick up the cube',
         }
 
     # --- Inference encoding tests ---
@@ -56,9 +56,9 @@ def test_encode_with_rot6d(self, sample_inputs):
         assert result['state']['ee_pose'].shape == (1, 1, 9)
 
         ee_pose = result['state']['ee_pose'][0, 0]
-        assert np.allclose(ee_pose[:3], sample_inputs['robot_state.ee_pose'][:3])
+        assert np.allclose(ee_pose[:3], sample_inputs[keys.EE_POSE][:3])
 
-        expected_rot6d = geom.Rotation.from_quat(sample_inputs['robot_state.ee_pose'][3:7]).as_rot6d
+        expected_rot6d = geom.Rotation.from_quat(sample_inputs[keys.EE_POSE][3:7]).as_rot6d
         assert np.allclose(ee_pose[3:], expected_rot6d, atol=1e-6)
 
     def test_encode_with_joints(self, sample_inputs):
@@ -68,7 +68,7 @@ def test_encode_with_joints(self, sample_inputs):
 
         assert 'joint_position' in result['state']
         assert result['state']['joint_position'].shape == (1, 1, 7)
-        assert np.allclose(result['state']['joint_position'][0, 0], sample_inputs['robot_state.q'])
+        assert np.allclose(result['state']['joint_position'][0, 0], sample_inputs[keys.JOINTS])
 
     def test_encode_with_rot6d_and_joints(self, sample_inputs):
         """Test inference encoding with both rot6d and joints."""
@@ -81,7 +81,7 @@ def test_encode_with_rot6d_and_joints(self, sample_inputs):
 
     def test_encode_missing_task(self, sample_inputs):
         """Test inference encoding handles missing task gracefully."""
-        del sample_inputs['task']
+        del sample_inputs[keys.TASK]
         codec = GrootObservationCodec()
         result = codec.encode(sample_inputs)
 
@@ -91,7 +91,7 @@ def test_encode_missing_task(self, sample_inputs):
 
     def test_rot6d_identity_quaternion(self, sample_inputs):
         """Test rot6d conversion with identity quaternion."""
-        sample_inputs['robot_state.ee_pose'] = np.array([1.0, 2.0, 3.0, 1.0, 0.0, 0.0, 0.0])
+        sample_inputs[keys.EE_POSE] = np.array([1.0, 2.0, 3.0, 1.0, 0.0, 0.0, 0.0])
 
         codec = GrootObservationCodec(rotation_rep=RotRep.ROT6D)
         result = codec.encode(sample_inputs)
@@ -103,7 +103,7 @@ def test_rot6d_identity_quaternion(self, sample_inputs):
     def test_rot6d_90deg_rotation(self, sample_inputs):
         """Test rot6d conversion with 90 degree rotation around Z."""
         quat = np.array([0.0, 0.0, np.sin(np.pi / 4), np.cos(np.pi / 4)])
-        sample_inputs['robot_state.ee_pose'] = np.array([1.0, 2.0, 3.0, *quat])
+        sample_inputs[keys.EE_POSE] = np.array([1.0, 2.0, 3.0, *quat])
 
         codec = GrootObservationCodec(rotation_rep=RotRep.ROT6D)
         result = codec.encode(sample_inputs)
@@ -151,7 +151,7 @@ def test_training_meta_no_joints(self):
 
     def test_non_square_input_image(self, sample_inputs):
         """Test that non-square images are properly resized with padding."""
-        sample_inputs['image.wrist'] = np.random.randint(0, 255, (100, 200, 3), dtype=np.uint8)
+        sample_inputs[keys.WRIST_IMAGE] = np.random.randint(0, 255, (100, 200, 3), dtype=np.uint8)
 
         codec = GrootObservationCodec()
         result = codec.encode(sample_inputs)
@@ -179,8 +179,8 @@ def test_dummy_encoded_shape(self, rotation_rep, include_joints):
 
     def test_custom_camera_keys(self, sample_inputs):
         """Test custom camera key mapping."""
-        sample_inputs['cam1'] = sample_inputs.pop('image.wrist')
-        sample_inputs['cam2'] = sample_inputs.pop('image.exterior')
+        sample_inputs['cam1'] = sample_inputs.pop(keys.WRIST_IMAGE)
+        sample_inputs['cam2'] = sample_inputs.pop(keys.EXTERIOR_IMAGE)
 
         codec = GrootObservationCodec(wrist_camera='cam1', exterior_camera='cam2')
         result = codec.encode(sample_inputs)
diff --git a/positronic/vendors/lance/codecs.py b/positronic/vendors/lance/codecs.py
index d36d4a6dc..adcb51457 100644
--- a/positronic/vendors/lance/codecs.py
+++ b/positronic/vendors/lance/codecs.py
@@ -8,6 +8,7 @@
 
 import configuronic as cfn
 
+from positronic import keys
 from positronic.cfg import codecs as base
 from positronic.dataset.episode import Episode
 from positronic.dataset.transforms.episode import Derive, EpisodeTransform, Get
@@ -31,7 +32,7 @@ def training_encoder(self) -> EpisodeTransform:
 
 @cfn.config(fps=15.0, horizon=None, binarize_grip=None, uuid=False)
 def _compose(obs, action, fps: float, horizon: float | None, binarize_grip, uuid: bool):
-    derivations = {'current_task': Get('task', ''), 'language_instruction1': Get('task', '')}
+    derivations = {'current_task': Get(keys.TASK, ''), 'language_instruction1': Get(keys.TASK, '')}
     if uuid:
         derivations['uuid'] = _random_uuid
     inner = base.compose(obs=obs, action=action, fps=fps, horizon=horizon, binarize_grip=binarize_grip)
diff --git a/positronic/vendors/lerobot_0_3_3/codecs.py b/positronic/vendors/lerobot_0_3_3/codecs.py
index 7b05e31f2..ee243765d 100644
--- a/positronic/vendors/lerobot_0_3_3/codecs.py
+++ b/positronic/vendors/lerobot_0_3_3/codecs.py
@@ -1,18 +1,19 @@
 """LeRobot codecs (observation encoder | action decoder pairs)."""
 
+from positronic import keys
 from positronic.cfg import codecs
 
 ee = codecs.compose.override(obs=codecs.eepose_obs, action=codecs.absolute_pos_action, horizon=1.0)
 joints = ee.override(obs=codecs.joints_obs)
 
 # Trajectory variants: use actual robot trajectory as action target instead of commanded targets
-ee_traj = ee.override(action=codecs.traj_ee_action, binarize_grip=('grip',))
+ee_traj = ee.override(action=codecs.traj_ee_action, binarize_grip=(keys.GRIP,))
 
 # Pure joint-based trajectory variant (no commanded joint targets in recordings)
 joints_traj = codecs.compose.override(
     obs=codecs.joints_obs,
-    action=codecs.absolute_joints_action.override(tgt_joints_key='robot_state.q', tgt_grip_key='grip'),
-    binarize_grip=('grip',),
+    action=codecs.absolute_joints_action.override(tgt_joints_key=keys.JOINTS, tgt_grip_key=keys.GRIP),
+    binarize_grip=(keys.GRIP,),
     horizon=1.0,
 )
 
diff --git a/positronic/vendors/molmoact2/codecs.py b/positronic/vendors/molmoact2/codecs.py
index 21ab914ef..1bad775de 100644
--- a/positronic/vendors/molmoact2/codecs.py
+++ b/positronic/vendors/molmoact2/codecs.py
@@ -5,6 +5,7 @@
 import configuronic as cfn
 import numpy as np
 
+from positronic import keys
 from positronic.cfg import codecs
 from positronic.policy.codec import Codec
 
@@ -23,11 +24,11 @@ class MolmoAct2ObservationCodec(Codec):
 
     def __init__(
         self,
-        wrist_camera: str = 'image.wrist',
-        exterior_camera_1: str = 'image.exterior',
+        wrist_camera: str = keys.WRIST_IMAGE,
+        exterior_camera_1: str = keys.EXTERIOR_IMAGE,
         exterior_camera_2: str | None = None,
-        joint_key: str = 'robot_state.q',
-        grip_key: str = 'grip',
+        joint_key: str = keys.JOINTS,
+        grip_key: str = keys.GRIP,
     ):
         self._cameras = (exterior_camera_1, exterior_camera_2 or exterior_camera_1, wrist_camera)
         self._joint_key = joint_key
@@ -45,7 +46,7 @@ def encode(self, inputs: dict[str, Any]) -> dict[str, Any]:
         return {
             'images': [self._image(k, inputs) for k in self._cameras],
             'state': np.concatenate([joints, grip]).astype(np.float32),
-            'task': inputs.get('task', ''),
+            'task': inputs.get(keys.TASK, ''),
         }
 
     def dummy_encoded(self, data=None) -> dict[str, Any]:
diff --git a/positronic/vendors/openpi/codecs.py b/positronic/vendors/openpi/codecs.py
index 9ebf7957a..bf279e7c1 100644
--- a/positronic/vendors/openpi/codecs.py
+++ b/positronic/vendors/openpi/codecs.py
@@ -22,7 +22,7 @@
 import numpy as np
 from PIL import Image as PilImage
 
-from positronic import geom
+from positronic import geom, keys
 from positronic.cfg import codecs
 from positronic.dataset import Signal, transforms
 from positronic.dataset.episode import Episode
@@ -39,8 +39,8 @@ class ObservationCodec(Codec):
     def __init__(
         self,
         state_features: dict[str, int],
-        exterior_camera: str = 'image.exterior',
-        wrist_camera: str = 'image.wrist',
+        exterior_camera: str = keys.EXTERIOR_IMAGE,
+        wrist_camera: str = keys.WRIST_IMAGE,
         image_size: tuple[int, int] = (224, 224),
     ):
         self._state_features = state_features
@@ -52,7 +52,7 @@ def __init__(
             'observation.state': self._derive_state,
             'observation.images.left': partial(self._derive_image, wrist_camera),
             'observation.images.side': partial(self._derive_image, exterior_camera),
-            'task': Get('task', ''),
+            'task': Get(keys.TASK, ''),
         }
 
         state_dim = sum(state_features.values())
@@ -87,8 +87,8 @@ def encode(self, inputs: dict[str, Any]) -> dict[str, Any]:
             'observation/wrist_image': self._encode_image(self._wrist_camera, inputs),
             'observation/image': self._encode_image(self._exterior_camera, inputs),
         }
-        if 'task' in inputs:
-            obs['prompt'] = inputs['task']
+        if keys.TASK in inputs:
+            obs['prompt'] = inputs[keys.TASK]
         return obs
 
     def _encode_image(self, input_key: str, inputs: dict[str, Any]) -> np.ndarray:
@@ -121,9 +121,9 @@ def training_encoder(self):
 
 
 @cfn.config(
-    state_features={'robot_state.ee_pose': 7, 'grip': 1},
-    exterior_camera='image.exterior',
-    wrist_camera='image.wrist',
+    state_features={keys.EE_POSE: 7, keys.GRIP: 1},
+    exterior_camera=keys.EXTERIOR_IMAGE,
+    wrist_camera=keys.WRIST_IMAGE,
     image_size=(224, 224),
 )
 def observation(state_features: dict[str, int], exterior_camera: str, wrist_camera: str, image_size: tuple[int, int]):
@@ -134,17 +134,17 @@ def observation(state_features: dict[str, int], exterior_camera: str, wrist_came
 
 
 ee_obs = observation
-ee_joints_obs = observation.override(state_features={'robot_state.ee_pose': 7, 'grip': 1, 'robot_state.q': 7})
+ee_joints_obs = observation.override(state_features={keys.EE_POSE: 7, keys.GRIP: 1, keys.JOINTS: 7})
 
 
 # Pretrained DROID models read joints and gripper as separate observation keys and the language
 # prompt under `prompt` (see openpi `droid_policy.DroidInputs`).
 droid_obs = cfn.Config(
     GenericObservationCodec,
-    state={'observation/joint_position': {'robot_state.q': 7}, 'observation/gripper_position': {'grip': 1}},
+    state={'observation/joint_position': {keys.JOINTS: 7}, 'observation/gripper_position': {keys.GRIP: 1}},
     images={
-        'observation/wrist_image_left': ('image.wrist', (224, 224)),
-        'observation/exterior_image_1_left': ('image.exterior', (224, 224)),
+        'observation/wrist_image_left': (keys.WRIST_IMAGE, (224, 224)),
+        'observation/exterior_image_1_left': (keys.EXTERIOR_IMAGE, (224, 224)),
     },
     task_field='prompt',
 )
@@ -152,15 +152,15 @@ def observation(state_features: dict[str, int], exterior_camera: str, wrist_came
 ee = codecs.compose.override(obs=ee_obs, action=codecs.absolute_pos_action)
 ee_joints = ee.override(obs=ee_joints_obs)
 
-ee_traj = ee.override(action=codecs.traj_ee_action, binarize_grip=('grip',))
-ee_joints_traj = ee_joints.override(action=codecs.traj_ee_action, binarize_grip=('grip',))
+ee_traj = ee.override(action=codecs.traj_ee_action, binarize_grip=(keys.GRIP,))
+ee_joints_traj = ee_joints.override(action=codecs.traj_ee_action, binarize_grip=(keys.GRIP,))
 
 # Pure joint-based trajectory variant (no commanded joint targets in recordings)
-joints_obs = observation.override(state_features={'robot_state.q': 7, 'grip': 1})
+joints_obs = observation.override(state_features={keys.JOINTS: 7, keys.GRIP: 1})
 joints_traj = codecs.compose.override(
     obs=joints_obs,
-    action=codecs.absolute_joints_action.override(tgt_joints_key='robot_state.q', tgt_grip_key='grip'),
-    binarize_grip=('grip',),
+    action=codecs.absolute_joints_action.override(tgt_joints_key=keys.JOINTS, tgt_grip_key=keys.GRIP),
+    binarize_grip=(keys.GRIP,),
 )
 
 # IK variants: reconstruct joint targets from recorded EE targets via IK
@@ -179,8 +179,8 @@ def observation(state_features: dict[str, int], exterior_camera: str, wrist_came
 # after the full chunk executes, whatever each variant's length.
 droid_jointpos = codecs.compose.override(
     obs=droid_obs,
-    action=codecs.absolute_joints_action.override(tgt_joints_key='robot_state.q', tgt_grip_key='grip'),
-    binarize_grip=('grip',),
+    action=codecs.absolute_joints_action.override(tgt_joints_key=keys.JOINTS, tgt_grip_key=keys.GRIP),
+    binarize_grip=(keys.GRIP,),
 )
 
 
@@ -231,7 +231,7 @@ class LiberoObservationCodec(Codec):
     def __init__(
         self,
         exterior_camera: str = 'image.agentview',
-        wrist_camera: str = 'image.wrist',
+        wrist_camera: str = keys.WRIST_IMAGE,
         image_size: tuple[int, int] = (224, 224),
     ):
         self._exterior_camera = exterior_camera
@@ -244,12 +244,12 @@ def encode(self, inputs: dict[str, Any]) -> dict[str, Any]:
             'observation/wrist_image': self._encode_image(self._wrist_camera, inputs),
             'observation/image': self._encode_image(self._exterior_camera, inputs),
         }
-        if 'task' in inputs:
-            obs['prompt'] = inputs['task']
+        if keys.TASK in inputs:
+            obs['prompt'] = inputs[keys.TASK]
         return obs
 
     def _libero_state(self, inputs: dict[str, Any]) -> np.ndarray:
-        ee_pose = np.asarray(inputs['robot_state.ee_pose'], dtype=float)
+        ee_pose = np.asarray(inputs[keys.EE_POSE], dtype=float)
         hand_rot = geom.Rotation.from_quat(ee_pose[3:7]) * _GRIP_SITE_TO_HAND
         # Reproduce robosuite's axis-angle branch. Its `robot0_eef_quat` is MuJoCo's `body_xquat`, FK-continuous
         # from the tool-down home pose and thus consistently in the w<=0 hemisphere (angle >= pi) across the
@@ -258,7 +258,7 @@ def _libero_state(self, inputs: dict[str, Any]) -> np.ndarray:
         quat = np.asarray(hand_rot.to(geom.Rotation.Representation.QUAT))
         canonical = geom.Rotation.from_quat(quat if quat[0] <= 0 else -quat)
         axisangle = np.asarray(canonical.to(geom.Rotation.Representation.ROTVEC)).reshape(3)
-        closure = 1.0 - float(inputs['grip'])
+        closure = 1.0 - float(inputs[keys.GRIP])
         gripper_qpos = _GRIPPER_QPOS_CLOSED + closure * (_GRIPPER_QPOS_OPEN - _GRIPPER_QPOS_CLOSED)
         return np.concatenate([ee_pose[:3], axisangle, gripper_qpos]).astype(np.float32)
 
diff --git a/positronic/wire.py b/positronic/wire.py
index f2f32c2ba..026f94cfd 100644
--- a/positronic/wire.py
+++ b/positronic/wire.py
@@ -1,4 +1,5 @@
 import pimm
+from positronic import keys
 from positronic.dataset import DatasetWriter
 from positronic.dataset.ds_writer_agent import DsWriterAgent, TimeMode, TrajectoryOverrideSerializer
 from positronic.dataset.serializers import Serializers, StatefulSerializer
@@ -41,7 +42,7 @@ def wire(
             ds_agent.add_signal('robot_state', Serializers.robot_state)
         if gripper is not None:
             ds_agent.add_signal('target_grip', TrajectoryOverrideSerializer(None))
-            ds_agent.add_signal('grip')
+            ds_agent.add_signal(keys.GRIP)
 
         for signal_name, emitter in cameras.items():
             world.connect(emitter, ds_agent.inputs[signal_name])
@@ -50,7 +51,7 @@ def wire(
             world.connect(robot_arm.state, ds_agent.inputs['robot_state'])
         if gripper is not None:
             world.connect(harness.target_grip, ds_agent.inputs['target_grip'])
-            world.connect(gripper.grip, ds_agent.inputs['grip'])
+            world.connect(gripper.grip, ds_agent.inputs[keys.GRIP])
 
     if gui is not None:
         for signal_name, emitter in cameras.items():
diff --git a/utilities/convert_ds.py b/utilities/convert_ds.py
index 50cea57bc..c24910f12 100644
--- a/utilities/convert_ds.py
+++ b/utilities/convert_ds.py
@@ -24,6 +24,7 @@
 import configuronic as cfn
 import tqdm
 
+from positronic import keys
 from positronic.dataset import Dataset
 from positronic.dataset.local_dataset import LocalDataset, LocalDatasetWriter
 from positronic.dataset.signal import Kind
@@ -47,14 +48,14 @@ def update_v0_1_0(path: str):
             Derive(**{
                 'controller_positions.right': Concat('right_controller_translation', 'right_controller_quaternion'),
                 'robot_commands.pose': Concat('target_robot_position_translation', 'target_robot_position_quaternion'),
-                'robot_state.ee_pose': Concat('robot_position_translation', 'robot_position_quaternion'),
+                keys.EE_POSE: Concat('robot_position_translation', 'robot_position_quaternion'),
                 'task': FromValue('Pick up the green cube and place it on the red cube.'),
             }),
             Rename(**{
-                'robot_state.q': 'robot_state.joints',
-                'robot_state.dq': 'robot_state.joints_velocity',
-                'image.wrist': 'image.handcam_left',
-                'image.exterior': 'image.back_view',
+                keys.JOINTS: 'robot_state.joints',
+                keys.JOINT_VEL: 'robot_state.joints_velocity',
+                keys.WRIST_IMAGE: 'image.handcam_left',
+                keys.EXTERIOR_IMAGE: 'image.back_view',
             }),
             Identity(select=['grip', 'target_grip', 'mjSTATE_FULLPHYSICS', 'mjSTATE_INTEGRATION', 'mjSTATE_WARMSTART']),
         ),
diff --git a/utilities/fake_dataset_generator.py b/utilities/fake_dataset_generator.py
index f82dd6262..7da50f5af 100644
--- a/utilities/fake_dataset_generator.py
+++ b/utilities/fake_dataset_generator.py
@@ -5,7 +5,7 @@
 import pos3
 
 import pimm
-from positronic import geom
+from positronic import geom, keys
 from positronic.cfg.eval.real.tasks import SCISSORS_TASK, SPOONS_TASK, TOWELS_TASK
 from positronic.dataset.ds_writer_agent import DsWriterAgent, DsWriterCommand, TimeMode
 from positronic.dataset.local_dataset import LocalDatasetWriter
@@ -18,7 +18,7 @@
     'inference.observation.name': 'eepose',
     'inference.action.name': 'absolute_position',
     'inference.observation.lerobot_features': {
-        'observation.state': {'shape': (8,), 'names': ['robot_state.ee_pose', 'grip'], 'dtype': 'float32'},
+        'observation.state': {'shape': (8,), 'names': [keys.EE_POSE, 'grip'], 'dtype': 'float32'},
         'observation.images.left': {'shape': (240, 320, 3), 'names': ['height', 'width', 'channel'], 'dtype': 'video'},
         'observation.images.side': {'shape': (240, 320, 3), 'names': ['height', 'width', 'channel'], 'dtype': 'video'},
     },
@@ -220,14 +220,14 @@ def main(
         generator = FakeGenerator(num_episodes, fps, avg_run_per_item, meta, success_rate, min_items, max_items)
 
         # Wire generator to agent
-        agent.add_signal('image.wrist', Serializers.camera_images)
-        agent.add_signal('image.exterior', Serializers.camera_images)
+        agent.add_signal(keys.WRIST_IMAGE, Serializers.camera_images)
+        agent.add_signal(keys.EXTERIOR_IMAGE, Serializers.camera_images)
         agent.add_signal('robot_state')  # Already dict
         agent.add_signal('robot_command')  # Already dict
 
         world.connect(generator.command, agent.command)
-        world.connect(generator.image_wrist, agent.inputs['image.wrist'])
-        world.connect(generator.image_exterior, agent.inputs['image.exterior'])
+        world.connect(generator.image_wrist, agent.inputs[keys.WRIST_IMAGE])
+        world.connect(generator.image_exterior, agent.inputs[keys.EXTERIOR_IMAGE])
         world.connect(generator.robot_state, agent.inputs['robot_state'])
         world.connect(generator.robot_command, agent.inputs['robot_command'])