diff --git a/core/cli.py b/core/cli.py index 92b48a4a..638acec2 100644 --- a/core/cli.py +++ b/core/cli.py @@ -114,6 +114,26 @@ _TEST_SUITES: dict[str, tuple[str, ...]] = { "tests/test_versor_condition_rust_parity.py", "tests/test_versor_apply_rust_parity.py", ), + "sensorium": ( + "tests/test_sensorium_compiler_delta.py", + "tests/test_audio_compiler.py", + "tests/test_audio_crdt_merge.py", + "tests/test_audio_eval_gates.py", + "tests/test_audio_pack_manifest.py", + "tests/test_audio_sensorium_mount.py", + "tests/test_vision_compiler.py", + "tests/test_vision_crdt_merge.py", + "tests/test_vision_eval_gates.py", + "tests/test_vision_sensorium_mount.py", + "tests/test_sensorimotor_contract.py", + "tests/test_sensorimotor_pack_manifest.py", + "tests/test_observation_frame_contract.py", + "tests/test_observation_frame_harness.py", + "tests/test_environment_falsification.py", + "tests/test_environment_falsification_eval_cli.py", + "tests/test_sensorium_eval_cli.py", + "tests/test_efferent_gate.py", + ), "pulse": ( "tests/test_pulse_integration.py", "tests/test_graph_diffusion.py", @@ -2354,6 +2374,8 @@ def cmd_eval(args: argparse.Namespace) -> int: """Run an eval lane by name, or list available lanes.""" if getattr(args, "lane", None) == "sensorium": return cmd_eval_sensorium(args) + if getattr(args, "lane", None) == "environment-falsification": + return cmd_eval_environment_falsification(args) if getattr(args, "lane", None) == "math-contemplation": return cmd_eval_math_contemplation(args) @@ -2494,6 +2516,33 @@ def cmd_eval_sensorium(args: argparse.Namespace) -> int: return 0 if report["failed"] == 0 and report["gate_closed"] else 1 +def cmd_eval_environment_falsification(args: argparse.Namespace) -> int: + """Run deterministic environmental falsification replay reports.""" + from evals.environment_falsification import build_environment_falsification_report + + report = build_environment_falsification_report() + + if getattr(args, "json", False): + print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True)) + else: + print(f"lane : {report['lane']}") + print(f"version : {report['version']}") + print(f"cases : {report['total']}") + print(f"passed : {report['passed']}") + print(f"failed : {report['failed']}") + print(f"report_sha256 : {report['report_sha256']}") + + if getattr(args, "report", None): + report_path = Path(args.report) + report_path.parent.mkdir(parents=True, exist_ok=True) + report_path.write_text( + json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + ) + print(f"\nreport written: {report_path}", file=sys.stderr) + + return 0 if report["failed"] == 0 and report["expected_report_hash_ok"] else 1 + + # --------------------------------------------------------------------------- # ADR-0172 W3 — math-contemplation CLI lane # --------------------------------------------------------------------------- diff --git a/docs/decisions/ADR-0211-conformal-falsification-bench.md b/docs/decisions/ADR-0211-conformal-falsification-bench.md new file mode 100644 index 00000000..9189c040 --- /dev/null +++ b/docs/decisions/ADR-0211-conformal-falsification-bench.md @@ -0,0 +1,71 @@ +# ADR-0211: Conformal Falsification Bench + +**Status:** Accepted +**Date:** 2026-06-06 +**Domains:** `sensorium/environment/`, `evals/environment_falsification/` +**Depends on:** ADR-0198, ADR-0208, ADR-0209 + +## Context + +ADR-0208 established `ObservationFrame` as a deterministic bundle of already +compiled afferent units. ADR-0209 established sensorimotor feedback as afferent +evidence. ADR-0198 established that real motor emission is fail-closed until a +verdict-enforcing efferent gate exists. + +The missing contract was falsification: a replayable way to say, "this expected +environmental evidence was or was not observed," without turning observation +frames into fusion, memory mutation, hardware control, or learned truth. + +## Decision + +CORE will use the Conformal Falsification Bench as a Python-reference replay +contract: + +```text +hypothesis / plan + -> ExpectedObservationFrame + -> actual ObservationFrame + -> FalsificationRun + -> replay-stable report +``` + +The v1 bench is exact and binary: + +- `SUPPORTED` when every expected slot is present with the expected merge key and + no unexpected slot appears. +- `FALSIFIED` when any expected slot is missing, changed, or accompanied by + unexpected evidence. + +## Contract + +- `ObservationFrame` remains afferent-only evidence, not fusion and not a + mutable world model. +- `ExpectedObservationFrame` and `FalsificationRun` are replay artifacts, not + learned truth, reviewed memory, or pack mutation proposals. +- v1 verdicts are only `SUPPORTED` and `FALSIFIED`. +- No probabilistic confidence, numeric tolerance, hardware-noise envelope, or + learned latent is part of the v1 verdict. +- No motor/efferent unit, actuator trace, raw pixel buffer, PCM buffer, event + payload, decoded action payload, Vault mutation, or `generate/*` dependency is + allowed in the bench. +- Public API names are stable: + `ObservationUnitRef`, `ExpectedObservationFrame`, `FalsificationResidual`, + `FalsificationRun`, `build_expected_observation_frame`, and + `compare_expected_to_observation`. + +## Consequences + +The bench gives CORE a falsifiable environmental replay surface before any +hardware, motor, native backend, or learned world model is admitted. Later event +vision, witness-log import, tabletop lab, and motor governance work must feed or +consume this contract rather than bypass it. + +## Proof Obligations + +- Expected frame hashes are order-invariant and duplicate-safe. +- Raw payloads and efferent units are rejected before traces are built. +- Missing, unexpected, or changed evidence yields `FALSIFIED`. +- Exact matched evidence yields `SUPPORTED`. +- The bench does not import `generate`, mutate Vault, or call + `ModalityRegistry.decode`. +- `core eval environment-falsification --json` is hash-pinned. diff --git a/docs/runtime_contracts.md b/docs/runtime_contracts.md index af5206d7..125f8a04 100644 --- a/docs/runtime_contracts.md +++ b/docs/runtime_contracts.md @@ -290,6 +290,34 @@ not directly rewrite language packs, frames, identity axes, or operator code. Identity manifold mutation by user prompt or correction is forbidden. +## Environmental falsification contract (ADR-0211) + +`sensorium.environment.falsification` compares expected afferent evidence with +actual `ObservationFrame` evidence. It is a deterministic replay surface, not a +fusion layer, not reviewed memory, and not a mutable world model. + +The v1 verdict set is closed: + +```text +SUPPORTED | FALSIFIED +``` + +`SUPPORTED` means every expected slot matched by merge key and no unexpected +slot appeared. `FALSIFIED` means at least one expected slot was missing, +changed, or accompanied by unexpected evidence. Neither verdict promotes a +claim to reviewed memory or mutates packs, Vault state, identity axes, operator +code, or runtime policy. + +Forbidden in the falsification bench: + +- raw pixels, PCM, event streams, byte payloads, actuator traces, or decoded + action payloads in traces; +- motor/efferent units in `ExpectedObservationFrame` or `FalsificationRun`; +- learned latents as substrate; +- probabilistic confidence, hardware-noise envelopes, or tolerance thresholds in + v1 verdicts; +- `generate/*` dependencies, Vault mutation, or `ModalityRegistry.decode`. + ## Testing policy Tests should protect load-bearing behavior: diff --git a/evals/environment_falsification/__init__.py b/evals/environment_falsification/__init__.py new file mode 100644 index 00000000..94743edd --- /dev/null +++ b/evals/environment_falsification/__init__.py @@ -0,0 +1,5 @@ +"""Environment falsification eval lane.""" + +from evals.environment_falsification.report import build_environment_falsification_report + +__all__ = ["build_environment_falsification_report"] diff --git a/evals/environment_falsification/expected_hashes.json b/evals/environment_falsification/expected_hashes.json new file mode 100644 index 00000000..15fe8bd2 --- /dev/null +++ b/evals/environment_falsification/expected_hashes.json @@ -0,0 +1,3 @@ +{ + "report_sha256": "c97b2dca7282d0231f1b448add87256cc2f59d39c5cec5ac1350231541e07d0b" +} diff --git a/evals/environment_falsification/fixtures.json b/evals/environment_falsification/fixtures.json new file mode 100644 index 00000000..2ea88366 --- /dev/null +++ b/evals/environment_falsification/fixtures.json @@ -0,0 +1,91 @@ +{ + "fixtures": [ + { + "id": "supported_exact_multimodal", + "expected_verdict": "SUPPORTED", + "expected": { + "audio:left_tone": { + "modality": "audio", + "signal": {"id": "left_tone", "kind": "tone", "ms": 240, "hz": 180, "sweep": 0, "amp": 0.4} + }, + "vision:corner": { + "modality": "vision", + "signal": {"id": "corner", "kind": "corner", "size": 32} + }, + "sensorimotor:contact": { + "modality": "sensorimotor", + "signal": { + "id": "contact", + "pose_q": [10, -4, 3], + "velocity_q": [1, 0, -1], + "force_torque_q": [2, 3, 5], + "contact_q": [1, 0, 1, 0], + "actuator_state_q": [7, 8] + } + } + }, + "actual": "same" + }, + { + "id": "falsified_changed_contact", + "expected_verdict": "FALSIFIED", + "expected": { + "sensorimotor:contact": { + "modality": "sensorimotor", + "signal": { + "id": "contact_expected", + "pose_q": [10, -4, 3], + "velocity_q": [1, 0, -1], + "force_torque_q": [2, 3, 5], + "contact_q": [1, 0, 1, 0], + "actuator_state_q": [7, 8] + } + } + }, + "actual": { + "sensorimotor:contact": { + "modality": "sensorimotor", + "signal": { + "id": "contact_actual", + "pose_q": [10, -4, 3], + "velocity_q": [1, 0, -1], + "force_torque_q": [2, 3, 5], + "contact_q": [0, 0, 0, 0], + "actuator_state_q": [7, 8] + } + } + } + }, + { + "id": "falsified_missing_and_unexpected", + "expected_verdict": "FALSIFIED", + "expected": { + "audio:left_tone": { + "modality": "audio", + "signal": {"id": "left_tone_expected", "kind": "tone", "ms": 240, "hz": 180, "sweep": 0, "amp": 0.4} + }, + "vision:box": { + "modality": "vision", + "signal": {"id": "box_expected", "kind": "contour_box", "size": 32} + } + }, + "actual": { + "audio:left_tone": { + "modality": "audio", + "signal": {"id": "left_tone_actual", "kind": "tone", "ms": 240, "hz": 180, "sweep": 0, "amp": 0.4} + }, + "sensorimotor:unexpected_contact": { + "modality": "sensorimotor", + "signal": { + "id": "unexpected_contact", + "pose_q": [0, 0, 1], + "velocity_q": [0, 0, 0], + "force_torque_q": [1, 1, 1], + "contact_q": [1], + "actuator_state_q": [2] + } + } + } + } + ] +} diff --git a/evals/environment_falsification/report.py b/evals/environment_falsification/report.py new file mode 100644 index 00000000..7814d4ac --- /dev/null +++ b/evals/environment_falsification/report.py @@ -0,0 +1,135 @@ +"""Deterministic environmental falsification replay report.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import numpy as np + +from evals.audio_sensorium.synth import synthesize as synthesize_audio +from evals.sensorimotor_sensorium.synth import synthesize as synthesize_sensorimotor +from evals.vision_sensorium.synth import synthesize as synthesize_vision +from sensorium.audio.canonical import canonicalize as canonicalize_audio +from sensorium.audio.compiler import AudioCompiler +from sensorium.audio.checksum import sha256_json +from sensorium.environment import ( + ObservationUnitRef, + build_expected_observation_frame, + build_observation_frame, + compare_expected_to_observation, +) +from sensorium.sensorimotor.compiler import SensorimotorCompiler +from sensorium.vision import VisionCompiler, canonicalize_image +from sensorium.vision.grid import iter_tile_signals + +_ROOT = Path(__file__).resolve().parent +_AUDIO_SR = 24_000 + + +def _load_json(name: str) -> dict[str, Any]: + return json.loads((_ROOT / name).read_text(encoding="utf-8")) + + +def _trace_safe(value: object) -> bool: + if isinstance(value, (np.ndarray, bytes, bytearray)): + return False + if isinstance(value, dict): + return all(_trace_safe(child) for child in value.values()) + if isinstance(value, (list, tuple)): + return all(_trace_safe(child) for child in value) + return True + + +def _compile_unit(spec: dict[str, Any]): + modality = spec["modality"] + signal = spec["signal"] + if modality == "audio": + return AudioCompiler().compile_signal( + canonicalize_audio(synthesize_audio(signal), _AUDIO_SR) + ) + if modality == "vision": + image = canonicalize_image(synthesize_vision(signal)) + tile = iter_tile_signals(image)[0] + return VisionCompiler().compile_tile(tile) + if modality == "sensorimotor": + return SensorimotorCompiler().compile_signal(synthesize_sensorimotor(signal)) + raise ValueError(f"unsupported falsification fixture modality: {modality!r}") + + +def _refs(spec: dict[str, Any]) -> tuple[ObservationUnitRef, ...]: + return tuple( + ObservationUnitRef(slot_id=slot_id, unit=_compile_unit(unit_spec)) + for slot_id, unit_spec in sorted(spec.items()) + ) + + +def _actual_spec(case: dict[str, Any]) -> dict[str, Any]: + actual = case["actual"] + if actual == "same": + return case["expected"] + return actual + + +def _case_report(index: int, case: dict[str, Any]) -> dict[str, object]: + expected_refs = _refs(case["expected"]) + actual_refs = _refs(_actual_spec(case)) + expected = build_expected_observation_frame( + monotonic_tick=index, + source_clock="environment-falsification-fixture", + unit_refs=expected_refs, + causal_parent_ids=(), + ) + actual = build_observation_frame( + monotonic_tick=index, + source_clock="environment-falsification-fixture", + units=tuple(ref.unit for ref in actual_refs), + causal_parent_ids=(expected.expected_id,), + ) + run = compare_expected_to_observation(expected, actual, actual_refs=actual_refs) + expected_verdict = str(case["expected_verdict"]) + row = { + "id": case["id"], + "expected_verdict": expected_verdict, + "actual_verdict": run.verdict, + "verdict_ok": run.verdict == expected_verdict, + "trace_hygiene_ok": _trace_safe(run.as_dict()), + "expected_sha256": expected.expected_sha256, + "actual_trace_hash": actual.trace_hash, + "run_trace_hash": run.trace_hash, + "residual": run.residual.as_dict(), + } + return row + + +def _report_hash(report_without_hash: dict[str, object]) -> str: + return sha256_json(report_without_hash) + + +def build_environment_falsification_report() -> dict[str, object]: + fixtures = _load_json("fixtures.json")["fixtures"] + cases = [_case_report(idx, case) for idx, case in enumerate(fixtures)] + passed = sum( + 1 + for case in cases + if case["verdict_ok"] is True and case["trace_hygiene_ok"] is True + ) + report = { + "lane": "environment-falsification", + "version": "v1", + "total": len(cases), + "passed": passed, + "failed": len(cases) - passed, + "cases": cases, + } + report["report_sha256"] = _report_hash(report) + expected_hashes = _load_json("expected_hashes.json") + report["expected_report_sha256"] = expected_hashes["report_sha256"] + report["expected_report_hash_ok"] = report["report_sha256"] == expected_hashes["report_sha256"] + if not report["expected_report_hash_ok"]: + report["failed"] = int(report["failed"]) + 1 + return report + + +__all__ = ["build_environment_falsification_report"] diff --git a/sensorium/environment/__init__.py b/sensorium/environment/__init__.py index 9c909434..5291209c 100644 --- a/sensorium/environment/__init__.py +++ b/sensorium/environment/__init__.py @@ -1,6 +1,26 @@ """Environmental observation contracts for sensorium units.""" +from sensorium.environment.falsification import ( + ChangedSlot, + ExpectedObservationFrame, + FalsificationResidual, + FalsificationRun, + ObservationUnitRef, + build_expected_observation_frame, + compare_expected_to_observation, +) from sensorium.environment.frame import ObservationFrame, build_observation_frame from sensorium.environment.harness import build_fixture_observation_frame -__all__ = ["ObservationFrame", "build_fixture_observation_frame", "build_observation_frame"] +__all__ = [ + "ChangedSlot", + "ExpectedObservationFrame", + "FalsificationResidual", + "FalsificationRun", + "ObservationFrame", + "ObservationUnitRef", + "build_expected_observation_frame", + "build_fixture_observation_frame", + "build_observation_frame", + "compare_expected_to_observation", +] diff --git a/sensorium/environment/falsification.py b/sensorium/environment/falsification.py new file mode 100644 index 00000000..e2b6be2a --- /dev/null +++ b/sensorium/environment/falsification.py @@ -0,0 +1,307 @@ +"""Deterministic expected-vs-actual environmental falsification. + +This module compares already-compiled afferent units. It does not compile raw +signals, decode motor commands, fuse modalities, mutate Vault state, or create a +world model. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Iterable + +import numpy as np + +from sensorium.audio.checksum import sha256_json +from sensorium.compiler.protocol import CompilationUnitLike, MergeKey +from sensorium.environment.frame import ObservationFrame + +_UNSAFE_ATTRS = ("pixels", "samples", "pcm", "waveform", "raw_bytes", "action_trace") + + +def _reject_unsafe_unit(unit: CompilationUnitLike) -> None: + if bool(getattr(unit, "efferent", False)): + raise ValueError("efferent action traces are not environmental evidence units") + if str(getattr(unit, "pack_id", "")).startswith("motor"): + raise ValueError("motor/efferent packs are not environmental evidence units") + for attr in _UNSAFE_ATTRS: + if hasattr(unit, attr): + value = getattr(unit, attr) + if isinstance(value, (np.ndarray, bytes, bytearray)): + raise TypeError(f"unsafe environmental evidence payload on unit: {attr}") + + +def _merge_key_record(key: MergeKey) -> list[str]: + return [str(key[0]), str(key[1]), str(key[2])] + + +def _unit_record(unit: CompilationUnitLike) -> dict[str, object]: + _reject_unsafe_unit(unit) + return { + "merge_key": _merge_key_record(unit.merge_key), + "canonical_sha256": unit.canonical_sha256, + "ir_sha256": unit.ir_sha256, + "pack_id": unit.pack_id, + "pack_manifest_sha256": unit.pack_manifest_sha256, + "projection_sha256": unit.projection_sha256, + "versor_condition": float(unit.versor_condition), + } + + +@dataclass(frozen=True, slots=True) +class ObservationUnitRef: + """Slot-labelled reference to one compiled afferent unit.""" + + slot_id: str + unit: CompilationUnitLike + + def __post_init__(self) -> None: + if not self.slot_id.strip(): + raise ValueError("ObservationUnitRef.slot_id is required") + _reject_unsafe_unit(self.unit) + + @property + def merge_key(self) -> MergeKey: + return self.unit.merge_key + + def as_dict(self) -> dict[str, object]: + return { + "slot_id": self.slot_id, + "unit": _unit_record(self.unit), + } + + +def _canonical_refs(refs: Iterable[ObservationUnitRef]) -> tuple[ObservationUnitRef, ...]: + ordered = sorted(tuple(refs), key=lambda r: (r.slot_id, r.merge_key)) + deduped: list[ObservationUnitRef] = [] + seen_pairs: set[tuple[str, MergeKey]] = set() + seen_slots: dict[str, MergeKey] = {} + for ref in ordered: + key = (ref.slot_id, ref.merge_key) + if key in seen_pairs: + continue + if ref.slot_id in seen_slots and seen_slots[ref.slot_id] != ref.merge_key: + raise ValueError(f"conflicting units for observation slot: {ref.slot_id}") + seen_pairs.add(key) + seen_slots[ref.slot_id] = ref.merge_key + deduped.append(ref) + return tuple(deduped) + + +@dataclass(frozen=True, slots=True) +class ExpectedObservationFrame: + """Replay-stable expectation over afferent observation slots.""" + + expected_id: str + monotonic_tick: int + source_clock: str + unit_refs: tuple[ObservationUnitRef, ...] + causal_parent_ids: tuple[str, ...] + expected_sha256: str + + def as_dict(self) -> dict[str, object]: + return { + "expected_id": self.expected_id, + "monotonic_tick": self.monotonic_tick, + "source_clock": self.source_clock, + "unit_refs": [ref.as_dict() for ref in self.unit_refs], + "causal_parent_ids": list(self.causal_parent_ids), + "expected_sha256": self.expected_sha256, + } + + +def build_expected_observation_frame( + *, + monotonic_tick: int, + source_clock: str, + unit_refs: Iterable[ObservationUnitRef], + causal_parent_ids: tuple[str, ...] = (), +) -> ExpectedObservationFrame: + if monotonic_tick < 0: + raise ValueError("monotonic_tick must be non-negative") + canonical_refs = _canonical_refs(unit_refs) + payload = { + "kind": "ExpectedObservationFrame", + "monotonic_tick": int(monotonic_tick), + "source_clock": str(source_clock), + "causal_parent_ids": list(causal_parent_ids), + "unit_refs": [ref.as_dict() for ref in canonical_refs], + } + expected_sha256 = sha256_json(payload) + expected_id = sha256_json({ + "kind": "ExpectedObservationFrame.id", + "monotonic_tick": int(monotonic_tick), + "source_clock": str(source_clock), + "expected_sha256": expected_sha256, + }) + return ExpectedObservationFrame( + expected_id=expected_id, + monotonic_tick=int(monotonic_tick), + source_clock=str(source_clock), + unit_refs=canonical_refs, + causal_parent_ids=tuple(causal_parent_ids), + expected_sha256=expected_sha256, + ) + + +@dataclass(frozen=True, slots=True) +class ChangedSlot: + """One slot whose compiled evidence changed.""" + + slot_id: str + expected_merge_key: MergeKey + actual_merge_key: MergeKey + + def as_dict(self) -> dict[str, object]: + return { + "slot_id": self.slot_id, + "expected_merge_key": _merge_key_record(self.expected_merge_key), + "actual_merge_key": _merge_key_record(self.actual_merge_key), + } + + +@dataclass(frozen=True, slots=True) +class FalsificationResidual: + """Exact slot/set delta between expectation and observation.""" + + matched: tuple[str, ...] + missing: tuple[str, ...] + unexpected: tuple[str, ...] + changed: tuple[ChangedSlot, ...] + residual_sha256: str + + @property + def is_supported(self) -> bool: + return not self.missing and not self.unexpected and not self.changed + + def as_dict(self) -> dict[str, object]: + return { + "matched": list(self.matched), + "missing": list(self.missing), + "unexpected": list(self.unexpected), + "changed": [change.as_dict() for change in self.changed], + "residual_sha256": self.residual_sha256, + } + + +@dataclass(frozen=True, slots=True) +class FalsificationRun: + """Trace-safe result of one expected-vs-actual comparison.""" + + expected_id: str + actual_frame_id: str + verdict: str + residual: FalsificationResidual + expected_sha256: str + actual_trace_hash: str + trace_hash: str + + def as_dict(self) -> dict[str, object]: + return { + "expected_id": self.expected_id, + "actual_frame_id": self.actual_frame_id, + "verdict": self.verdict, + "residual": self.residual.as_dict(), + "expected_sha256": self.expected_sha256, + "actual_trace_hash": self.actual_trace_hash, + "trace_hash": self.trace_hash, + } + + +def _refs_by_slot(refs: Iterable[ObservationUnitRef]) -> dict[str, ObservationUnitRef]: + return {ref.slot_id: ref for ref in _canonical_refs(refs)} + + +def _actual_refs_from_frame(actual: ObservationFrame) -> tuple[ObservationUnitRef, ...]: + return tuple( + ObservationUnitRef( + slot_id="unassigned:" + ":".join(unit.merge_key), + unit=unit, + ) + for unit in actual.units + ) + + +def compare_expected_to_observation( + expected: ExpectedObservationFrame, + actual: ObservationFrame, + *, + actual_refs: Iterable[ObservationUnitRef] | None = None, +) -> FalsificationRun: + """Compare expected slot evidence with an actual afferent observation frame.""" + actual_ref_tuple = ( + _actual_refs_from_frame(actual) + if actual_refs is None + else _canonical_refs(actual_refs) + ) + actual_frame_keys = {unit.merge_key for unit in actual.units} + for ref in actual_ref_tuple: + if ref.merge_key not in actual_frame_keys: + raise ValueError(f"actual ref is not present in ObservationFrame: {ref.slot_id}") + + expected_by_slot = _refs_by_slot(expected.unit_refs) + actual_by_slot = _refs_by_slot(actual_ref_tuple) + for unit in actual.units: + if unit.merge_key not in {ref.merge_key for ref in actual_ref_tuple}: + synthetic = ObservationUnitRef( + slot_id="unassigned:" + ":".join(unit.merge_key), + unit=unit, + ) + actual_by_slot[synthetic.slot_id] = synthetic + + matched: list[str] = [] + missing: list[str] = [] + changed: list[ChangedSlot] = [] + for slot_id, exp_ref in expected_by_slot.items(): + act_ref = actual_by_slot.get(slot_id) + if act_ref is None: + missing.append(slot_id) + elif act_ref.merge_key == exp_ref.merge_key: + matched.append(slot_id) + else: + changed.append(ChangedSlot(slot_id, exp_ref.merge_key, act_ref.merge_key)) + + unexpected = sorted(set(actual_by_slot) - set(expected_by_slot)) + residual_payload = { + "matched": sorted(matched), + "missing": sorted(missing), + "unexpected": unexpected, + "changed": [change.as_dict() for change in sorted(changed, key=lambda c: c.slot_id)], + } + residual = FalsificationResidual( + matched=tuple(residual_payload["matched"]), + missing=tuple(residual_payload["missing"]), + unexpected=tuple(unexpected), + changed=tuple(sorted(changed, key=lambda c: c.slot_id)), + residual_sha256=sha256_json(residual_payload), + ) + verdict = "SUPPORTED" if residual.is_supported else "FALSIFIED" + trace_payload = { + "kind": "FalsificationRun", + "expected_id": expected.expected_id, + "actual_frame_id": actual.frame_id, + "expected_sha256": expected.expected_sha256, + "actual_trace_hash": actual.trace_hash, + "verdict": verdict, + "residual_sha256": residual.residual_sha256, + } + return FalsificationRun( + expected_id=expected.expected_id, + actual_frame_id=actual.frame_id, + verdict=verdict, + residual=residual, + expected_sha256=expected.expected_sha256, + actual_trace_hash=actual.trace_hash, + trace_hash=sha256_json(trace_payload), + ) + + +__all__ = [ + "ChangedSlot", + "ExpectedObservationFrame", + "FalsificationResidual", + "FalsificationRun", + "ObservationUnitRef", + "build_expected_observation_frame", + "compare_expected_to_observation", +] diff --git a/tests/test_environment_falsification.py b/tests/test_environment_falsification.py new file mode 100644 index 00000000..7d3ab3ef --- /dev/null +++ b/tests/test_environment_falsification.py @@ -0,0 +1,196 @@ +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_expected_observation_frame, + build_observation_frame, + compare_expected_to_observation, +) + + +@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_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_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" diff --git a/tests/test_environment_falsification_eval_cli.py b/tests/test_environment_falsification_eval_cli.py new file mode 100644 index 00000000..b968ad17 --- /dev/null +++ b/tests/test_environment_falsification_eval_cli.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import json + +from core.cli import main +from evals.environment_falsification import build_environment_falsification_report + + +def test_environment_falsification_report_passes_expected_fixtures(): + report = build_environment_falsification_report() + assert report["lane"] == "environment-falsification" + assert report["failed"] == 0 + assert report["expected_report_hash_ok"] is True + assert {case["actual_verdict"] for case in report["cases"]} == {"SUPPORTED", "FALSIFIED"} + + +def test_core_eval_environment_falsification_json(capsys): + assert main(["eval", "environment-falsification", "--json"]) == 0 + out = capsys.readouterr().out + report = json.loads(out) + assert report["lane"] == "environment-falsification" + assert report["failed"] == 0 + + +def test_core_eval_environment_falsification_text_summary(capsys): + assert main(["eval", "environment-falsification"]) == 0 + out = capsys.readouterr().out + assert "lane : environment-falsification" in out + assert "failed : 0" in out