Phase E1: semantic domain profiles for StreamSpec and CapabilitySpec

docs/NEXT_STATUS.md

@@ -32,7 +32,7 @@ language — this table is the current status source of truth.
 | B | partial | Threat model + env scrubbing (#36), container backend (#38), bounded transfer (#39), remainder + attestation + CI image (#40), probes (#41) | B5 cloud mint/storage (blocked: cloud owner); registry image publication (commercial decision) |
 | C | partial | Dossier + cited gate (#37), orphan-label refusal + provenance + fault robustness (#42) | C4 world-native distributions (need SOFA/GPU world revisions); C5 phantom (blocked: partner) |
 | D | partial | Paired comparison + CIs (#43), compare CLI (#45), split manifests with patient/site grouping (#60-62), D5 explicit cohort subgroup validation, head-covered trial bindings, clustered continuous bootstrap, and cloud parity (#64, cloud #5) | D5 benchmark-level reporting with external study data |
-| E | partial | Obs/action contract + A3 (#44), interactive streams (#46), LeRobot reader + MONAI (#51), honest video adapter (#54) | E4 media alignment (no test media; mp4 banned from VC); semantic output schemas (needs E contract design) |
+| E | partial | Obs/action contract + A3 (#44), interactive streams (#46), LeRobot reader + MONAI (#51), honest video adapter (#54), semantic domain profiles for StreamSpec and CapabilitySpec (#66) | E4 media alignment (no test media; mp4 banned from VC); real robotics policy adapter |
 | F | partial | Prefix-replay branching proof + seed caveat (#47), trajectory-backed forecast task + null baseline (#56), planning utility measurement and recipe (#58) | F1 trajectory benchmark imports; F4 closed-loop simulator branching |
 | G | partial | SB3 PPO recipe + measurement (#48), runnable fixes (#50, #52), kernel-observed episode divergence (#57, #59), verifier-derived training rewards with divergence penalty (#65) | Prime interop |
 | H | partial | Episode resume (#49), crash-safe writes + partial recovery (#53), strict trial provenance + pre-write drift rejection (#59) | Fleet queue/autoscaling (blocked: hosted capacity decision); vectorized stepping (not justified by measured need) |

src/or_audit/eval/contracts.py

@@ -93,6 +93,16 @@ class StreamSpec(_Frozen):
     adapter_digest: SHA256Hex
     source: SourceLocator = "$"
     role: Slug | None = None
+    dtype: str = ""
+    shape: tuple[int, ...] = ()
+    unit: str = ""
+    coordinate_frame: str = ""
+    valid_range: tuple[float, float] | None = None
+    controller_id: str = ""
+    camera_calibration: dict[str, Any] = Field(default_factory=dict)
+    joint_order: tuple[str, ...] = ()
+    invalid_depth_encoding: str = ""
+    privileged: bool = False
 
     @model_validator(mode="after")
     def _plugin_needs_schema(self) -> Self:
@@ -155,11 +165,20 @@ class CapabilitySpec(_Frozen):
     features: tuple[Slug, ...] = ()
     modalities: tuple[Slug, ...] = ()
     schema_wildcard: bool = False
+    stream_profiles: tuple[StreamSpec, ...] = ()
+    accepts_privileged: bool = False
 
     @model_validator(mode="after")
-    def _non_empty_modes(self) -> Self:
+    def _validate_capability(self) -> Self:
         if not self.interaction_modes:
             raise TaskContractError(f"capability {self.interface} declares no interaction mode")
+        seen_ids: set[str] = set()
+        for s in self.stream_profiles:
+            if s.id in seen_ids:
+                raise TaskContractError(
+                    f"capability {self.interface} declares duplicate stream_profile id {s.id!r}"
+                )
+            seen_ids.add(s.id)
         return self
 
     def satisfies(self, interface: InterfaceSpec) -> bool:
@@ -175,12 +194,78 @@ def satisfies(self, interface: InterfaceSpec) -> bool:
                 for stream in interface.streams
             )
         )
-        return (
+        if not (
             self.interface == interface.id
             and interface.interaction_mode in self.interaction_modes
             and interface.protocol_version in self.protocol_versions
             and schemas_match
-        )
+        ):
+            return False
+        cap_profiles: dict[Slug, StreamSpec] = {}
+        if self.stream_profiles:
+            for s in self.stream_profiles:
+                if s.schema_id:
+                    cap_profiles[s.schema_id] = s
+                if s.id:
+                    cap_profiles[s.id] = s
+        for intf_stream in interface.streams:
+            has_semantics = bool(
+                intf_stream.unit
+                or intf_stream.coordinate_frame
+                or intf_stream.controller_id
+                or intf_stream.dtype
+                or intf_stream.shape
+                or intf_stream.joint_order
+                or intf_stream.invalid_depth_encoding
+                or intf_stream.valid_range is not None
+                or intf_stream.camera_calibration
+            )
+            if has_semantics:
+                matching_profile = cap_profiles.get(intf_stream.id) or cap_profiles.get(
+                    intf_stream.schema_id
+                )
+                if matching_profile is None:
+                    return False
+                if intf_stream.unit and matching_profile.unit != intf_stream.unit:
+                    return False
+                if (
+                    intf_stream.coordinate_frame
+                    and matching_profile.coordinate_frame != intf_stream.coordinate_frame
+                ):
+                    return False
+                if (
+                    intf_stream.controller_id
+                    and matching_profile.controller_id != intf_stream.controller_id
+                ):
+                    return False
+                if intf_stream.dtype and matching_profile.dtype != intf_stream.dtype:
+                    return False
+                if intf_stream.shape and matching_profile.shape != intf_stream.shape:
+                    return False
+                if (
+                    intf_stream.joint_order
+                    and matching_profile.joint_order != intf_stream.joint_order
+                ):
+                    return False
+                if (
+                    intf_stream.invalid_depth_encoding
+                    and matching_profile.invalid_depth_encoding
+                    != intf_stream.invalid_depth_encoding
+                ):
+                    return False
+                if (
+                    intf_stream.valid_range is not None
+                    and matching_profile.valid_range != intf_stream.valid_range
+                ):
+                    return False
+                if (
+                    intf_stream.camera_calibration
+                    and matching_profile.camera_calibration != intf_stream.camera_calibration
+                ):
+                    return False
+            if intf_stream.privileged and not self.accepts_privileged:
+                return False
+        return True
 
 
 class HarnessSpec(_Frozen):

tests/test_multi_modality_contracts.py

@@ -589,3 +589,138 @@ def test_interactive_history_never_carries_raw_fields(tmp_path: Path) -> None:
     hist_seen = second.output["history"][0]["observation"]
     assert "secret" not in json.dumps(hist_seen)
     assert hist_seen["cam"]["frame_index"] == 7
+
+
+def test_semantic_stream_profile_mismatches_refuse_binding() -> None:
+    base_stream = StreamSpec(
+        id="kinematics-stream",
+        schema_id="kinematic-telemetry",
+        adapter="robotic-kinematics",
+        adapter_digest="a" * 64,
+        unit="mm",
+        coordinate_frame="world",
+        controller_id="delta_pos_v1",
+        joint_order=("j1", "j2", "j3"),
+        dtype="float32",
+        shape=(3,),
+    )
+    interface = InterfaceSpec(
+        id="kinematics-control",
+        interaction_mode=InteractionMode.CLOSED_LOOP,
+        observations=("kinematic-telemetry",),
+        actions=("joint-cmd",),
+        streams=(base_stream,),
+    )
+
+    # 1. Identical semantic profile -> binds
+    matching_cap = CapabilitySpec(
+        interface="kinematics-control",
+        interaction_modes=(InteractionMode.CLOSED_LOOP,),
+        observations=("kinematic-telemetry",),
+        actions=("joint-cmd",),
+        modalities=("robotic-kinematics",),
+        stream_profiles=(base_stream,),
+    )
+    assert matching_cap.satisfies(interface)
+
+    # 2. Unit mismatch ("mm" vs "m") -> does not bind
+    unit_mismatch = matching_cap.model_copy(
+        update={"stream_profiles": (base_stream.model_copy(update={"unit": "m"}),)}
+    )
+    assert not unit_mismatch.satisfies(interface)
+
+    # 3. Coordinate frame mismatch ("world" vs "tool_tip") -> does not bind
+    frame_mismatch = matching_cap.model_copy(
+        update={
+            "stream_profiles": (base_stream.model_copy(update={"coordinate_frame": "tool_tip"}),)
+        }
+    )
+    assert not frame_mismatch.satisfies(interface)
+
+    # 4. Controller mismatch -> does not bind
+    controller_mismatch = matching_cap.model_copy(
+        update={
+            "stream_profiles": (
+                base_stream.model_copy(update={"controller_id": "absolute_pos_v1"}),
+            )
+        }
+    )
+    assert not controller_mismatch.satisfies(interface)
+
+    # 5. Joint order mismatch -> does not bind
+    joint_mismatch = matching_cap.model_copy(
+        update={
+            "stream_profiles": (base_stream.model_copy(update={"joint_order": ("j3", "j2", "j1")}),)
+        }
+    )
+    assert not joint_mismatch.satisfies(interface)
+
+    # 6. Privileged stream rejected unless accepts_privileged=True
+    priv_stream = base_stream.model_copy(update={"privileged": True})
+    priv_interface = interface.model_copy(update={"streams": (priv_stream,)})
+    assert not matching_cap.satisfies(priv_interface)
+    priv_cap = matching_cap.model_copy(update={"accepts_privileged": True})
+    assert priv_cap.satisfies(priv_interface)
+
+    # 7. Missing stream profile entirely when interface declares semantics -> does not bind
+    no_profile_cap = matching_cap.model_copy(update={"stream_profiles": ()})
+    assert not no_profile_cap.satisfies(interface)
+
+    # 8. Empty string unit on capability profile when interface declares unit="mm" -> does not bind
+    empty_unit_cap = matching_cap.model_copy(
+        update={"stream_profiles": (base_stream.model_copy(update={"unit": ""}),)}
+    )
+    assert not empty_unit_cap.satisfies(interface)
+
+    # 9. schema_wildcard=True cannot bypass semantic profile requirements
+    wildcard_no_profile = CapabilitySpec(
+        interface="kinematics-control",
+        interaction_modes=(InteractionMode.CLOSED_LOOP,),
+        schema_wildcard=True,
+        stream_profiles=(),
+    )
+    assert not wildcard_no_profile.satisfies(interface)
+
+    # 10. schema_wildcard=True with matching semantic profile satisfies
+    wildcard_matching_profile = CapabilitySpec(
+        interface="kinematics-control",
+        interaction_modes=(InteractionMode.CLOSED_LOOP,),
+        schema_wildcard=True,
+        stream_profiles=(base_stream,),
+    )
+    assert wildcard_matching_profile.satisfies(interface)
+
+    # 11. Profile keyed by schema_id (not stream id) matches when stream id differs
+    schema_keyed_profile = base_stream.model_copy(update={"id": "different-stream-id"})
+    schema_cap = CapabilitySpec(
+        interface="kinematics-control",
+        interaction_modes=(InteractionMode.CLOSED_LOOP,),
+        observations=("kinematic-telemetry",),
+        actions=("joint-cmd",),
+        modalities=("robotic-kinematics",),
+        stream_profiles=(schema_keyed_profile,),
+    )
+    assert schema_cap.satisfies(interface)
+
+    # 12. camera_calibration mismatch refuses binding, matching satisfies
+    calib_stream = base_stream.model_copy(update={"camera_calibration": {"focal_length": 50.0}})
+    calib_interface = interface.model_copy(update={"streams": (calib_stream,)})
+    calib_mismatch = matching_cap.model_copy(
+        update={
+            "stream_profiles": (
+                base_stream.model_copy(update={"camera_calibration": {"focal_length": 35.0}}),
+            )
+        }
+    )
+    assert not calib_mismatch.satisfies(calib_interface)
+    calib_match = matching_cap.model_copy(update={"stream_profiles": (calib_stream,)})
+    assert calib_match.satisfies(calib_interface)
+
+    # 13. Duplicate stream_profiles IDs on CapabilitySpec rejected with TaskContractError
+    with pytest.raises(TaskContractError, match="duplicate stream_profile id"):
+        CapabilitySpec(
+            interface="kinematics-control",
+            interaction_modes=(InteractionMode.CLOSED_LOOP,),
+            observations=("kinematic-telemetry",),
+            stream_profiles=(base_stream, base_stream),
+        )