diff --git a/docs/NEXT_STATUS.md b/docs/NEXT_STATUS.md index 8993e23..9f7015a 100644 --- a/docs/NEXT_STATUS.md +++ b/docs/NEXT_STATUS.md @@ -17,7 +17,7 @@ not report those workstreams as implemented. The historical N1–N11 ledger and audit findings are retained after the roadmap. An old `shipped` designation applies to its named mechanism, not to every new acceptance criterion here. -### Implementation status — 2026-09-09 +### Implementation status — 2026-09-11 Work is merged per phase as PRs; nothing below claims a workstream finished unless every acceptance criterion is met. Partial means shipped mechanism @@ -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), semantic domain profiles for StreamSpec and CapabilitySpec (#66) | E4 media alignment (no test media; mp4 banned from VC); real robotics policy adapter | +| 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), profile-binding soundness closure: channel/schema conflation, identity pinning, geometry and calibration validation (#67) | 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) | diff --git a/src/or_audit/eval/contracts.py b/src/or_audit/eval/contracts.py index e05b6d7..0eebfe6 100644 --- a/src/or_audit/eval/contracts.py +++ b/src/or_audit/eval/contracts.py @@ -4,10 +4,19 @@ import hashlib import json +import math +from collections.abc import Mapping from enum import StrEnum -from typing import Annotated, Any, Self +from typing import Annotated, Any, NoReturn, Self -from pydantic import BaseModel, ConfigDict, Field, StringConstraints, model_validator +from pydantic import ( + BaseModel, + ConfigDict, + Field, + StringConstraints, + field_validator, + model_validator, +) from or_audit.errors import TaskContractError @@ -73,6 +82,102 @@ class _Frozen(BaseModel): ] +class _FrozenDict(dict[str, Any]): + """Deeply immutable JSON object used for pinned calibration payloads. + + Built only through :func:`_freeze_json`, which recursively replaces nested + mappings with this class and nested sequences with tuples, so no mutation + path — shallow or deep — survives construction. Equality stays + content-based (``dict.__eq__``), so a frozen and a plain mapping with the + same content still compare equal. + """ + + def __setitem__(self, key: str, value: Any) -> NoReturn: + raise TypeError("camera_calibration is deeply immutable") + + def __delitem__(self, key: str) -> NoReturn: + raise TypeError("camera_calibration is deeply immutable") + + def update(self, *args: Any, **kwargs: Any) -> NoReturn: + raise TypeError("camera_calibration is deeply immutable") + + def setdefault(self, key: str, default: Any = None) -> NoReturn: + raise TypeError("camera_calibration is deeply immutable") + + def pop(self, *args: Any) -> NoReturn: + raise TypeError("camera_calibration is deeply immutable") + + def popitem(self) -> NoReturn: + raise TypeError("camera_calibration is deeply immutable") + + def clear(self) -> NoReturn: + raise TypeError("camera_calibration is deeply immutable") + + # CPython resolves ``x |= y`` through the C-level ``dict.__ior__``, which + # mutates in place and bypasses ``__setitem__`` — the override is required + # for soundness. typeshed models ``dict.__or__`` as overloads no raising + # override can match, so the misc override diagnostic is suppressed here + # (the sibling ``__or__`` is inherited and produces a fresh dict anyway). + def __ior__(self, value: Any) -> NoReturn: # type: ignore[misc] + raise TypeError("camera_calibration is deeply immutable") + + +def _calibration_is_declared(value: Mapping[str, Any]) -> bool: + """Whether a calibration carries any meaningful calibration data. + + An empty mapping is "not declared". A mapping with keys is declared even + if a value is ``0`` or ``False`` — those are real calibration values + (e.g. zero principal-point offset, disabled skew), so emptiness is tested + by key count rather than by truthiness of the payload. + """ + return len(value) > 0 + + +def _calibration_equal(left: Mapping[str, Any], right: Mapping[str, Any]) -> bool: + """Content equality between two calibration trees. + + Re-canonicalizes both sides first: a tree that arrived through + ``model_copy(update=...)`` skipped validation and still carries lists + where a validated one holds tuples, and raw ``==`` would call that a + mismatch. A side that cannot even be canonicalized is not a calibration + anyone should bind against, so it compares unequal rather than raising + inside a predicate that selection loops (``bind.py``) call casually. + """ + try: + left_frozen: Any = _freeze_json(left, where="camera_calibration") + right_frozen: Any = _freeze_json(right, where="camera_calibration") + return bool(left_frozen == right_frozen) + except TaskContractError: + return False + + +def _freeze_json(value: Any, *, where: str) -> Any: + """Canonicalize ``value`` into a deeply immutable JSON tree. + + Rejects anything a JSON document could not carry: non-string object keys, + non-finite floats, and unsupported leaf types. Keys are sorted so two + calibrations with the same content canonicalize to the same order. + """ + if value is None or isinstance(value, bool | int | str): + return value + if isinstance(value, float): + if not math.isfinite(value): + raise TaskContractError(f"{where}: non-finite number {value!r} is not valid JSON") + return value + if isinstance(value, Mapping): + items: dict[str, Any] = {} + for key, sub in value.items(): + if not isinstance(key, str): + raise TaskContractError(f"{where}: object keys must be strings, got {key!r}") + items[key] = _freeze_json(sub, where=f"{where}.{key}") + return _FrozenDict(sorted(items.items())) + if isinstance(value, list | tuple): + return tuple(_freeze_json(item, where=f"{where}[]") for item in value) + raise TaskContractError( + f"{where}: value of type {type(value).__name__} has no JSON representation" + ) + + class StreamSpec(_Frozen): """One sensor/data channel a task presents to an agent. @@ -99,11 +204,25 @@ class StreamSpec(_Frozen): coordinate_frame: str = "" valid_range: tuple[float, float] | None = None controller_id: str = "" - camera_calibration: dict[str, Any] = Field(default_factory=dict) + #: Deeply immutable pinned calibration; see :func:`_freeze_json`. Declared as + #: ``Mapping`` so pydantic never hands back a caller-retained mutable dict. + camera_calibration: Mapping[str, Any] = Field(default_factory=_FrozenDict) joint_order: tuple[str, ...] = () invalid_depth_encoding: str = "" privileged: bool = False + @field_validator("shape", mode="before") + def _reject_impossible_shape(cls, value: Any) -> Any: + # ``tuple[int, ...]`` is lax in pydantic: ``True`` launders into ``1`` + # and a zero/negative extent passes the type check, so the raw input + # must be screened before coercion. A bool is not a dimension, and a + # non-positive axis is a buffer no tensor can be allocated against. + if isinstance(value, list | tuple): + for dim in value: + if isinstance(dim, bool) or (isinstance(dim, int) and dim <= 0): + raise TaskContractError(f"stream declares non-positive shape dimension {dim!r}") + return value + @model_validator(mode="after") def _plugin_needs_schema(self) -> Self: if not self.adapter: @@ -112,6 +231,69 @@ def _plugin_needs_schema(self) -> Self: raise TaskContractError(f"stream {self.id} requires an observation schema_id") return self + @model_validator(mode="after") + def _geometry_is_sane(self) -> Self: + # Shape soundness (bool laundering, non-positive axes) is screened on + # the raw input by ``_reject_impossible_shape`` before pydantic can + # coerce; what is left here is range semantics. + if self.valid_range is not None: + low, high = self.valid_range + if math.isnan(low) or math.isnan(high): + raise TaskContractError( + f"stream {self.id} valid_range {self.valid_range!r} contains a NaN bound; " + "every comparison against it is False, so the range could never be enforced" + ) + if low > high: + raise TaskContractError( + f"stream {self.id} valid_range {self.valid_range!r} is reversed" + ) + # A range is a constraint only if it rules something out. Fully + # finite ranges are the ordinary case and always admissible. + # One-sided ranges are legitimate physics (depth has no ceiling, a + # joint angle may be open below), so an infinite endpoint is fine + # while the other side is finite. But when *both* endpoints are + # infinite the range admits every value: it asserts nothing, so it + # must be omitted rather than declared-vacuous. + if not math.isfinite(low) and not math.isfinite(high): + raise TaskContractError( + f"stream {self.id} valid_range {self.valid_range!r} bounds neither side; " + "omit valid_range when the value is unconstrained" + ) + return self + + @field_validator("camera_calibration", mode="before") + def _pin_calibration(cls, value: Any) -> Any: + # A pinned calibration a caller can still mutate is not a pin: digest + # the stream, ship it, then edit the dict, and a binding that was + # refused now holds. Freeze deep, and reject anything JSON could not + # round-trip so the pinned form survives serialization unchanged. + # Screening happens before pydantic's own ``Mapping[str, Any]`` check + # so every malformed calibration surfaces as one TaskContractError, + # not sometimes-a-ValidationError depending on which field is bad. + if not isinstance(value, Mapping): + raise TaskContractError( + f"camera_calibration must be a mapping, got {type(value).__name__}" + ) + if type(value) is _FrozenDict: + return value + return _freeze_json(value, where="camera_calibration") + + @model_validator(mode="after") + def _refreeze_after_field_validation(self) -> Self: + # Safety net under the before-validator: ``Mapping[str, Any]`` field + # validation rebuilds the top-level container, so pydantic can hand + # back a plain dict even when the input was already canonical (the + # default_factory path also never runs the before-validator). Any + # container that is not the frozen type is re-canonicalized here, so + # a stored stream never exposes a mutable calibration — which is the + # whole point of pinning it. + if type(self.camera_calibration) is not _FrozenDict: + frozen: Any = _freeze_json( + self.camera_calibration, where=f"stream {self.id}.camera_calibration" + ) + object.__setattr__(self, "camera_calibration", frozen) + return self + class InterfaceSpec(_Frozen): """Requirements a task exposes to compatible agents.""" @@ -181,6 +363,21 @@ def _validate_capability(self) -> Self: seen_ids.add(s.id) return self + def _profile_index(self) -> tuple[dict[Slug, StreamSpec], dict[Slug, list[StreamSpec]]]: + """Index profiles by exact stream id and by schema id. + + Two maps, never one merged: keying a single dict by both ``s.id`` and + ``s.schema_id`` lets one channel's profile satisfy another channel's + requirements whenever a stream id collides with some schema name, which + is legal because both namespaces are open slugs. + """ + by_id: dict[Slug, StreamSpec] = {} + by_schema: dict[Slug, list[StreamSpec]] = {} + for s in self.stream_profiles: + by_id[s.id] = s + by_schema.setdefault(s.schema_id, []).append(s) + return by_id, by_schema + def satisfies(self, interface: InterfaceSpec) -> bool: """Return whether this declaration satisfies every task requirement.""" own_schemas = set(self.observations) | set(self.features) @@ -201,14 +398,12 @@ def satisfies(self, interface: InterfaceSpec) -> bool: 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 + + by_id, by_schema = self._profile_index() + interface_stream_ids = {stream.id for stream in interface.streams} for intf_stream in interface.streams: + if intf_stream.privileged and not self.accepts_privileged: + return False has_semantics = bool( intf_stream.unit or intf_stream.coordinate_frame @@ -218,52 +413,77 @@ def satisfies(self, interface: InterfaceSpec) -> bool: or intf_stream.joint_order or intf_stream.invalid_depth_encoding or intf_stream.valid_range is not None - or intf_stream.camera_calibration + or _calibration_is_declared(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 - ): + # Resolve the declaration that speaks for this channel. Exact channel + # identity wins. A schema-level route is admissible only under + # declared semantics and only when it cannot be confused with + # another channel, because profiles are declared per data shape and + # two channels may legitimately share one shape — guessing which + # channel a shape referred to is how a probe's profile ends up + # vouching for a camera. + matching_profile = by_id.get(intf_stream.id) + if matching_profile is None: + if not has_semantics: + continue + candidates = by_schema.get(intf_stream.schema_id) + if not candidates or len(candidates) != 1: return False - if intf_stream.dtype and matching_profile.dtype != intf_stream.dtype: + candidate = candidates[0] + # A profile named for a *different* channel of this same + # interface is that channel's declaration, not this one's. + if candidate.id != intf_stream.id and candidate.id in interface_stream_ids: 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: + matching_profile = candidate + + # Channel identity is compared unconditionally: adapter + digest + # are required fields on both sides, so "I serve this channel" + # means "through this pinned plugin content". A differing digest + # is a different plugin build, i.e. a different contract. + if ( + matching_profile.schema_id != intf_stream.schema_id + or matching_profile.adapter != intf_stream.adapter + or matching_profile.adapter_digest != intf_stream.adapter_digest + ): + return False + if matching_profile.source != intf_stream.source: + # ``source`` locates the observation slice the channel carries; + # whole-observation ("$") vs a pointer are different channels + # even under one id, so compare it on both sides. + return False + if intf_stream.role is not None and matching_profile.role != intf_stream.role: + 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 _calibration_is_declared(intf_stream.camera_calibration) and not _calibration_equal( + matching_profile.camera_calibration, intf_stream.camera_calibration + ): return False return True diff --git a/tests/test_multi_modality_contracts.py b/tests/test_multi_modality_contracts.py index fc5f1ab..678381b 100644 --- a/tests/test_multi_modality_contracts.py +++ b/tests/test_multi_modality_contracts.py @@ -690,17 +690,75 @@ def test_semantic_stream_profile_mismatches_refuse_binding() -> None: ) 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( + # 11. Schema-keyed lookup must not let a *different channel's* profile + # vouch for this channel. Interface declares two channels whose stream ids + # and schema names cross-reference each other (legal: both namespaces are + # open slugs). Under a merged id/schema index, B's profile is reachable + # through A's schema name, so A would bind on B's semantics. + chan_a = StreamSpec( + id="cam", + schema_id="probe-schema", + adapter="robotic-kinematics", + adapter_digest="a" * 64, + unit="mm", + coordinate_frame="world", + ) + chan_b = StreamSpec( + id="probe", + schema_id="cam", + adapter="robotic-kinematics", + adapter_digest="a" * 64, + unit="mm", + coordinate_frame="world", + ) + crossed = InterfaceSpec( + id="kinematics-control", + interaction_mode=InteractionMode.CLOSED_LOOP, + observations=("probe-schema", "cam"), + actions=("joint-cmd",), + streams=(chan_a, chan_b), + ) + # Capability declares only B's profile (correct for B). It carries no + # declaration for A at all, so A must refuse to bind. + only_b = CapabilitySpec( + interface="kinematics-control", + interaction_modes=(InteractionMode.CLOSED_LOOP,), + observations=("probe-schema", "cam"), + actions=("joint-cmd",), + modalities=("robotic-kinematics",), + stream_profiles=(chan_b,), + ) + assert not only_b.satisfies(crossed) + + # 11b. Legitimate schema-keyed fallback: exactly one profile carries this + # channel's schema and none of the interface's other channel ids collides + # with it, so the profile is unambiguously about this channel. + solo = InterfaceSpec( + id="kinematics-control", + interaction_mode=InteractionMode.CLOSED_LOOP, + observations=("kinematic-telemetry",), + actions=("joint-cmd",), + streams=(base_stream.model_copy(update={"id": "an-alternate-channel-name"}),), + ) + schema_keyed_cap = CapabilitySpec( interface="kinematics-control", interaction_modes=(InteractionMode.CLOSED_LOOP,), observations=("kinematic-telemetry",), actions=("joint-cmd",), modalities=("robotic-kinematics",), - stream_profiles=(schema_keyed_profile,), + stream_profiles=(base_stream,), ) - assert schema_cap.satisfies(interface) + assert schema_keyed_cap.satisfies(solo) + + # 11c. Two profiles competing for one schema make the schema-keyed route + # ambiguous; guessing either would bind on an unverified declaration. + twin = base_stream.model_copy(update={"id": "other-channel-name"}) + assert not schema_keyed_cap.model_copy( + update={"stream_profiles": (base_stream, twin)}, + ).satisfies(solo) + assert schema_keyed_cap.model_copy( + update={"stream_profiles": (twin,)}, + ).satisfies(solo) # "other-channel-name" is not an interface channel; schema route ok # 12. camera_calibration mismatch refuses binding, matching satisfies calib_stream = base_stream.model_copy(update={"camera_calibration": {"focal_length": 50.0}}) @@ -724,3 +782,153 @@ def test_semantic_stream_profile_mismatches_refuse_binding() -> None: observations=("kinematic-telemetry",), stream_profiles=(base_stream, base_stream), ) + + +def _stream(**kw: Any) -> StreamSpec: + base: dict[str, Any] = { + "id": "cam", + "schema_id": "video-obs", + "adapter": "video-laparoscopic", + "adapter_digest": "a" * 64, + } + base.update(kw) + return StreamSpec(**base) + + +def _cap(profiles: tuple[StreamSpec, ...], **kw: Any) -> CapabilitySpec: + base: dict[str, Any] = { + "interface": "k", + "interaction_modes": (InteractionMode.SINGLE_TURN,), + "observations": ("video-obs",), + "outputs": ("pred",), + "modalities": ("video-laparoscopic",), + "stream_profiles": profiles, + } + base.update(kw) + return CapabilitySpec(**base) + + +def _iface(streams: tuple[StreamSpec, ...]) -> InterfaceSpec: + return InterfaceSpec( + id="k", + interaction_mode=InteractionMode.SINGLE_TURN, + observations=("video-obs",), + outputs=("pred",), + streams=streams, + ) + + +def test_stream_rejects_impossible_geometry() -> None: + # A zero/negative axis is not a buffer anyone can allocate; a bool is not + # a dimension. pydantic's tuple[int, ...] silently coerces True -> 1, so + # the contract layer must reject it, not launder it. + with pytest.raises(TaskContractError, match="non-positive shape dimension"): + _stream(shape=(0,)) + with pytest.raises(TaskContractError, match="non-positive shape dimension"): + _stream(shape=(-1,)) + with pytest.raises(TaskContractError, match="non-positive shape dimension"): + _stream(shape=(True,)) + # NaN bounds make every comparison False: the range could never be + # enforced, so "declared" would be a lie. + with pytest.raises(TaskContractError, match="NaN"): + _stream(valid_range=(float("nan"), 2.0)) + # Reversed and fully unbounded ranges assert nothing; they must be + # omitted, not declared-vacuous. + with pytest.raises(TaskContractError, match="reversed"): + _stream(valid_range=(5.0, 1.0)) + with pytest.raises(TaskContractError, match="bounds neither side"): + _stream(valid_range=(float("-inf"), float("inf"))) + # One-sided ranges are real physics (depth has no ceiling): allowed. + assert _stream(valid_range=(0.0, float("inf"))).valid_range == (0.0, float("inf")) + assert _stream(valid_range=(float("-inf"), 0.0)).valid_range == (float("-inf"), 0.0) + # Fully finite ranges are the ordinary case; a guard that rejected them + # would make valid_range unusable, so pin the happy path explicitly. + assert _stream(valid_range=(0.0, 1.0)).valid_range == (0.0, 1.0) + assert _stream(valid_range=(-5.0, 5.0)).valid_range == (-5.0, 5.0) + assert _stream(valid_range=(2.0, 2.0)).valid_range == (2.0, 2.0) # degenerate but sound + + +def test_camera_calibration_is_deeply_immutable() -> None: + # A pinned calibration a caller can still edit is not a pin: digest the + # stream, ship it, then mutate the dict and the refused binding holds. + stream = _stream(camera_calibration={"intrinsics": {"fx": 50.0, "dist": [0.1, 0.2]}}) + # ``Any`` alias: the declared type is Mapping (read-only), but the whole + # point of this test is that mutation attempts raise even when attempted + # through the object the caller still holds. + cal: Any = stream.camera_calibration + assert isinstance(cal, dict) # it really is a dict subclass, so these are live paths + with pytest.raises(TypeError, match="deeply immutable"): + cal["fx"] = 999.0 + with pytest.raises(TypeError, match="deeply immutable"): + cal["intrinsics"]["fx"] = 999.0 + with pytest.raises(TypeError, match="deeply immutable"): + cal.update({"fx": 1.0}) + with pytest.raises(TypeError, match="deeply immutable"): + cal |= {"fx": 1.0} + with pytest.raises(TypeError, match="deeply immutable"): + del cal["intrinsics"] + with pytest.raises(TypeError, match="deeply immutable"): + cal.setdefault("fx", 1.0) + with pytest.raises(TypeError, match="deeply immutable"): + cal.pop("intrinsics") + with pytest.raises(TypeError, match="deeply immutable"): + cal.popitem() + with pytest.raises(TypeError, match="deeply immutable"): + cal.clear() + # Nothing leaked through the failed mutations, and equality stays + # content-based. The canonical form holds tuples where JSON has arrays + # (that is what makes it deep-immutable), so compare against that. + assert cal == {"intrinsics": {"fx": 50.0, "dist": (0.1, 0.2)}} + assert isinstance(cal["intrinsics"]["dist"], tuple) + # ``|`` without assignment stays available from dict and yields a fresh, + # ordinary dict — the pin itself is untouched either way. + merged = cal | {"extra": 1.0} + assert merged["extra"] == 1.0 + assert "extra" not in cal + + +def test_camera_calibration_rejects_non_json_values() -> None: + bad_values: list[Any] = [ + {"fx": float("nan")}, + {"fx": float("inf")}, + {1: "x"}, + {"ok": {"nested": {2.5: 1}}}, + ] + for bad in bad_values: + with pytest.raises(TaskContractError): + _stream(camera_calibration=bad) + + +def test_zero_valued_calibration_counts_as_declared() -> None: + # fx=0 / skew=False are real calibration values. Truthiness of the payload + # would read them as "not declared" and let an undeclared profile bind. + zeroed = _stream(camera_calibration={"fx": 0.0}) + iface = _iface((zeroed,)) + assert not _cap((_stream(),)).satisfies(iface) # profile declares nothing + assert _cap((_stream(camera_calibration={"fx": 0.0}),)).satisfies(iface) + + +def test_calibration_equality_ignores_key_order_and_uses_content() -> None: + a = _stream(camera_calibration={"a": 1.0, "b": {"c": [1, 2]}}) + b = _stream(camera_calibration={"b": {"c": [1, 2]}, "a": 1.0}) + iface = _iface((a,)) + assert _cap((b,)).satisfies(iface) # same content, different insertion order + c = _stream(camera_calibration={"a": 1.0, "b": {"c": [1, 3]}}) + assert not _cap((c,)).satisfies(iface) # deep content differs + + +def test_stream_identity_adapter_digest_and_source_are_enforced() -> None: + plain = _stream() + iface = _iface((plain,)) + # Same channel, same pinned plugin: binds. + assert _cap((plain,)).satisfies(iface) + # Different plugin build under the same channel is a different contract. + assert not _cap((_stream(adapter_digest="b" * 64),)).satisfies(iface) + assert not _cap((_stream(adapter="other-adapter"),)).satisfies(iface) + # ``source`` selects the observation slice the channel carries: "$" vs a + # pointer are different channels even under one id, and it is compared on + # both sides so neither party can silently re-locate the data. + assert not _cap((_stream(source="depth"),)).satisfies(iface) + pointer_iface = _iface((_stream(source="depth"),)) + assert _cap((_stream(source="depth"),)).satisfies(pointer_iface) + assert not _cap((_stream(),)).satisfies(pointer_iface)