301 lines
9.9 KiB
Python
301 lines
9.9 KiB
Python
from __future__ import annotations
|
|
|
|
import builtins
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from sensorium.environment import (
|
|
ObservationUnitRef,
|
|
build_experiment_plan,
|
|
build_expected_observation_frame,
|
|
build_hypothesis_claim,
|
|
build_observation_frame,
|
|
compare_expected_to_observation,
|
|
run_falsification_scenario,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _Unit:
|
|
canonical_sha256: str
|
|
ir_sha256: str
|
|
pack_id: str
|
|
pack_manifest_sha256: str
|
|
projection_sha256: str
|
|
versor: np.ndarray
|
|
versor_condition: float = 0.0
|
|
|
|
@property
|
|
def merge_key(self) -> tuple[str, str, str]:
|
|
return (self.canonical_sha256, self.ir_sha256, self.projection_sha256)
|
|
|
|
|
|
def _unit(name: str, pack_id: str = "vision_core_v1") -> _Unit:
|
|
v = np.zeros(32, dtype=np.float32)
|
|
v[0] = 1.0
|
|
return _Unit(name, f"ir-{name}", pack_id, "manifest", f"proj-{name}", v)
|
|
|
|
|
|
def _ref(slot: str, name: str, pack_id: str = "vision_core_v1") -> ObservationUnitRef:
|
|
return ObservationUnitRef(slot, _unit(name, pack_id))
|
|
|
|
|
|
def test_adr_0211_contract_is_documented():
|
|
root = Path(__file__).resolve().parents[1]
|
|
adr = root / "docs" / "decisions" / "ADR-0211-conformal-falsification-bench.md"
|
|
runtime = root / "docs" / "runtime_contracts.md"
|
|
assert adr.exists()
|
|
adr_text = adr.read_text(encoding="utf-8")
|
|
runtime_text = runtime.read_text(encoding="utf-8")
|
|
assert "`ObservationFrame` remains afferent-only evidence" in adr_text
|
|
assert "SUPPORTED" in adr_text
|
|
assert "FALSIFIED" in adr_text
|
|
assert "Environmental falsification contract (ADR-0211)" in runtime_text
|
|
|
|
|
|
def test_expected_frame_hash_is_order_invariant_and_duplicate_safe():
|
|
a = _ref("slot:a", "a")
|
|
b = _ref("slot:b", "b")
|
|
f1 = build_expected_observation_frame(
|
|
monotonic_tick=1,
|
|
source_clock="fixture",
|
|
unit_refs=(a, b, a),
|
|
)
|
|
f2 = build_expected_observation_frame(
|
|
monotonic_tick=1,
|
|
source_clock="fixture",
|
|
unit_refs=(b, a),
|
|
)
|
|
assert f1.expected_sha256 == f2.expected_sha256
|
|
assert f1.expected_id == f2.expected_id
|
|
assert len(f1.unit_refs) == 2
|
|
|
|
|
|
def test_hypothesis_claim_is_speculative_and_basis_order_stable():
|
|
h1 = build_hypothesis_claim(
|
|
claim_id="claim-a",
|
|
claim_text="The anchored object repeats its contact evidence.",
|
|
domain="passive_tabletop",
|
|
basis_trace_hashes=("b", "a", "b"),
|
|
)
|
|
h2 = build_hypothesis_claim(
|
|
claim_id="claim-a",
|
|
claim_text="The anchored object repeats its contact evidence.",
|
|
domain="passive_tabletop",
|
|
basis_trace_hashes=("a", "b"),
|
|
)
|
|
assert h1.epistemic_status == "SPECULATIVE"
|
|
assert h1.basis_trace_hashes == ("a", "b")
|
|
assert h1.hypothesis_sha256 == h2.hypothesis_sha256
|
|
|
|
|
|
def test_experiment_plan_hash_is_order_invariant_and_duplicate_safe():
|
|
hypothesis = build_hypothesis_claim(
|
|
claim_id="claim-plan",
|
|
claim_text="Expected frames define the evidence envelope.",
|
|
domain="fixture",
|
|
)
|
|
first = build_expected_observation_frame(
|
|
monotonic_tick=10,
|
|
source_clock="fixture",
|
|
unit_refs=(_ref("slot:a", "a"),),
|
|
)
|
|
second = build_expected_observation_frame(
|
|
monotonic_tick=11,
|
|
source_clock="fixture",
|
|
unit_refs=(_ref("slot:b", "b"),),
|
|
)
|
|
p1 = build_experiment_plan(hypothesis=hypothesis, expected_frames=(first, second, first))
|
|
p2 = build_experiment_plan(hypothesis=hypothesis, expected_frames=(second, first))
|
|
assert p1.plan_sha256 == p2.plan_sha256
|
|
assert len(p1.expected_frames) == 2
|
|
|
|
|
|
def test_raw_payloads_are_rejected():
|
|
@dataclass(frozen=True, slots=True)
|
|
class BadUnit(_Unit):
|
|
pixels: bytes = b"raw"
|
|
|
|
with pytest.raises(TypeError, match="unsafe environmental evidence payload"):
|
|
ObservationUnitRef(
|
|
"bad",
|
|
BadUnit("a", "ir-a", "vision_core_v1", "manifest", "proj-a", np.zeros(32, dtype=np.float32)),
|
|
)
|
|
|
|
|
|
def test_efferent_and_motor_units_are_rejected():
|
|
@dataclass(frozen=True, slots=True)
|
|
class ActionUnit(_Unit):
|
|
efferent: bool = True
|
|
|
|
with pytest.raises(ValueError, match="efferent"):
|
|
ObservationUnitRef(
|
|
"action",
|
|
ActionUnit("m", "ir-m", "motor_test", "manifest", "proj-m", np.zeros(32, dtype=np.float32)),
|
|
)
|
|
with pytest.raises(ValueError, match="motor"):
|
|
ObservationUnitRef("motor", _unit("m", "motor_test"))
|
|
|
|
|
|
def test_exact_expected_vs_actual_match_is_supported():
|
|
refs = (_ref("slot:a", "a"), _ref("slot:b", "b"))
|
|
expected = build_expected_observation_frame(
|
|
monotonic_tick=2,
|
|
source_clock="fixture",
|
|
unit_refs=refs,
|
|
)
|
|
actual = build_observation_frame(
|
|
monotonic_tick=2,
|
|
source_clock="fixture",
|
|
units=[ref.unit for ref in refs],
|
|
)
|
|
run = compare_expected_to_observation(expected, actual, actual_refs=refs)
|
|
assert run.verdict == "SUPPORTED"
|
|
assert run.residual.matched == ("slot:a", "slot:b")
|
|
assert run.residual.missing == ()
|
|
assert run.residual.unexpected == ()
|
|
assert run.residual.changed == ()
|
|
|
|
|
|
def test_missing_unexpected_and_changed_slots_are_falsified():
|
|
expected_refs = (
|
|
_ref("slot:a", "a"),
|
|
_ref("slot:b", "b"),
|
|
_ref("slot:c", "c"),
|
|
)
|
|
actual_refs = (
|
|
_ref("slot:a", "a"),
|
|
_ref("slot:b", "b2"),
|
|
_ref("slot:d", "d"),
|
|
)
|
|
expected = build_expected_observation_frame(
|
|
monotonic_tick=3,
|
|
source_clock="fixture",
|
|
unit_refs=expected_refs,
|
|
)
|
|
actual = build_observation_frame(
|
|
monotonic_tick=3,
|
|
source_clock="fixture",
|
|
units=[ref.unit for ref in actual_refs],
|
|
)
|
|
run = compare_expected_to_observation(expected, actual, actual_refs=actual_refs)
|
|
assert run.verdict == "FALSIFIED"
|
|
assert run.residual.matched == ("slot:a",)
|
|
assert run.residual.missing == ("slot:c",)
|
|
assert run.residual.unexpected == ("slot:d",)
|
|
assert len(run.residual.changed) == 1
|
|
assert run.residual.changed[0].slot_id == "slot:b"
|
|
|
|
|
|
def test_scenario_supported_and_falsified_verdicts():
|
|
hypothesis = build_hypothesis_claim(
|
|
claim_id="claim-scenario",
|
|
claim_text="A scenario binds expected evidence to actual frames.",
|
|
domain="fixture",
|
|
)
|
|
refs = (_ref("slot:a", "a"),)
|
|
expected = build_expected_observation_frame(
|
|
monotonic_tick=6,
|
|
source_clock="fixture",
|
|
unit_refs=refs,
|
|
)
|
|
plan = build_experiment_plan(hypothesis=hypothesis, expected_frames=(expected,))
|
|
supported_actual = build_observation_frame(
|
|
monotonic_tick=6,
|
|
source_clock="fixture",
|
|
units=[ref.unit for ref in refs],
|
|
)
|
|
supported = run_falsification_scenario(
|
|
plan,
|
|
actual_frames_by_expected_id={expected.expected_id: supported_actual},
|
|
actual_refs_by_expected_id={expected.expected_id: refs},
|
|
)
|
|
assert supported.verdict == "SUPPORTED"
|
|
assert supported.total_count == 1
|
|
assert supported.supported_count == 1
|
|
|
|
changed_refs = (_ref("slot:a", "changed"),)
|
|
changed_actual = build_observation_frame(
|
|
monotonic_tick=6,
|
|
source_clock="fixture",
|
|
units=[ref.unit for ref in changed_refs],
|
|
)
|
|
falsified = run_falsification_scenario(
|
|
plan,
|
|
actual_frames_by_expected_id={expected.expected_id: changed_actual},
|
|
actual_refs_by_expected_id={expected.expected_id: changed_refs},
|
|
)
|
|
assert falsified.verdict == "FALSIFIED"
|
|
assert falsified.falsified_count == 1
|
|
assert falsified.runs[0].residual.changed[0].slot_id == "slot:a"
|
|
|
|
|
|
def test_scenario_missing_actual_frame_is_falsified():
|
|
hypothesis = build_hypothesis_claim(
|
|
claim_id="claim-missing",
|
|
claim_text="Missing actual evidence falsifies the scenario.",
|
|
domain="fixture",
|
|
)
|
|
expected = build_expected_observation_frame(
|
|
monotonic_tick=7,
|
|
source_clock="fixture",
|
|
unit_refs=(_ref("slot:a", "a"),),
|
|
)
|
|
plan = build_experiment_plan(hypothesis=hypothesis, expected_frames=(expected,))
|
|
report = run_falsification_scenario(plan, actual_frames_by_expected_id={})
|
|
assert report.verdict == "FALSIFIED"
|
|
assert report.falsified_count == 1
|
|
assert report.runs[0].actual_frame_id == "__missing_observation_frame__"
|
|
assert report.runs[0].residual.missing == ("slot:a",)
|
|
|
|
|
|
def test_run_trace_is_hash_only():
|
|
refs = (_ref("slot:a", "a"),)
|
|
expected = build_expected_observation_frame(
|
|
monotonic_tick=4,
|
|
source_clock="fixture",
|
|
unit_refs=refs,
|
|
)
|
|
actual = build_observation_frame(
|
|
monotonic_tick=4,
|
|
source_clock="fixture",
|
|
units=[ref.unit for ref in refs],
|
|
)
|
|
payload = compare_expected_to_observation(expected, actual, actual_refs=refs).as_dict()
|
|
assert "versor" not in str(payload)
|
|
assert "pixels" not in str(payload)
|
|
assert "command" not in str(payload)
|
|
|
|
|
|
def test_comparator_does_not_import_generate_or_call_decode(monkeypatch):
|
|
refs = (_ref("slot:a", "a"),)
|
|
expected = build_expected_observation_frame(
|
|
monotonic_tick=5,
|
|
source_clock="fixture",
|
|
unit_refs=refs,
|
|
)
|
|
actual = build_observation_frame(
|
|
monotonic_tick=5,
|
|
source_clock="fixture",
|
|
units=[ref.unit for ref in refs],
|
|
)
|
|
original_import = builtins.__import__
|
|
|
|
def guarded_import(name, *args, **kwargs):
|
|
if name.startswith("generate"):
|
|
raise AssertionError("falsification comparator must not import generate")
|
|
return original_import(name, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(builtins, "__import__", guarded_import)
|
|
from sensorium.registry import ModalityRegistry
|
|
|
|
monkeypatch.setattr(
|
|
ModalityRegistry,
|
|
"decode",
|
|
lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("decode called")),
|
|
)
|
|
run = compare_expected_to_observation(expected, actual, actual_refs=refs)
|
|
assert run.verdict == "SUPPORTED"
|