Merge pull request #541 from AssetOverflow/codex/sensorium-runtime-eval-governance

Add sensorium eval and governance runway

core/cli.py

@@ -2349,6 +2349,8 @@ def cmd_doctor(args: argparse.Namespace) -> int:
 
 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) == "math-contemplation":
         return cmd_eval_math_contemplation(args)
 
@@ -2456,6 +2458,39 @@ def cmd_eval(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_eval_sensorium(args: argparse.Namespace) -> int:
+    """Run deterministic sensorium modality evidence reports."""
+    from evals.sensorium import build_sensorium_report
+
+    modality = getattr(args, "modality", "vision") or "vision"
+    try:
+        report = build_sensorium_report(modality)
+    except ValueError as exc:
+        _die(str(exc), code=2)
+
+    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"modality       : {report['modality']}")
+        print(f"pack_id        : {report['pack_id']}")
+        print(f"gate_engaged   : {report['gate_engaged']}")
+        print(f"gate_closed    : {report['gate_closed']}")
+        print(f"cases          : {report['total']}")
+        print(f"passed         : {report['passed']}")
+        print(f"failed         : {report['failed']}")
+
+    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["gate_closed"] else 1
+
+
 # ---------------------------------------------------------------------------
 # ADR-0172 W3 — math-contemplation CLI lane
 # ---------------------------------------------------------------------------
@@ -4889,6 +4924,12 @@ def build_parser() -> argparse.ArgumentParser:
     eval_cmd.add_argument("--json", action="store_true", help="emit machine-readable JSON")
     eval_cmd.add_argument("--save", action="store_true", help="write result to lane results/ directory")
     eval_cmd.add_argument("--report", metavar="PATH", help="write JSON report to file")
+    eval_cmd.add_argument(
+        "--modality",
+        choices=["audio", "vision", "sensorimotor"],
+        default="vision",
+        help="sensorium lane modality to evaluate (default: vision)",
+    )
     eval_cmd.add_argument(
         "--audit",
         metavar="PATH",

evals/sensorimotor_sensorium/__init__.py

@@ -0,0 +1 @@
+"""Deterministic sensorimotor compiler eval lane."""

evals/sensorimotor_sensorium/expected_ir.jsonl

@@ -0,0 +1,3 @@
+{"canonical_sha256": "b4f4b643018ef6dd7ffdaf02f83c25a7c58daa2b5b13c0be1406546d6e4a70a4", "event_count": 5, "event_types": ["proprio.pose", "proprio.velocity", "haptic.force_torque", "haptic.contact", "actuator.state"], "id": "resting_pose", "ir_sha256": "297bc879305553c1c1eb39153cf48e7222a10f94b279b3122cd2d6e64c92bb14"}
+{"canonical_sha256": "1edc1e022efdf1e1208f574329ae4a82d94ada16a225029dab1fd13c2d92f98c", "event_count": 5, "event_types": ["proprio.pose", "proprio.velocity", "haptic.force_torque", "haptic.contact", "actuator.state"], "id": "moving_contact", "ir_sha256": "1f24bcd14847aef44ed3c4baccfddbc5cc71eb6e1a0c741403c348e707225dbb"}
+{"canonical_sha256": "fade3485eb40896ace76afefc4397ba872ff9e2e17b2c9c818fe696ec5096b93", "event_count": 5, "event_types": ["proprio.pose", "proprio.velocity", "haptic.force_torque", "haptic.contact", "actuator.state"], "id": "force_spike", "ir_sha256": "d75fd4d67f7a7484db900b81d1fa2b2f8499458feaa8a9593feddd1d43881e63"}

evals/sensorimotor_sensorium/expected_projection.json

@@ -0,0 +1,113 @@
+{
+  "force_spike": {
+    "projection_sha256": "989f9381e35fef12f30f64b7153d647cd1159bb3b0029d8d535033aeb9d31d5c",
+    "reference_versor": [
+      0.8165946006774902,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      -0.053018417209386826,
+      0.23315252363681793,
+      0.33966025710105896,
+      0.0,
+      0.19017764925956726,
+      0.3096834123134613,
+      0.0,
+      -0.16899415850639343,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0016558070201426744,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0
+    ]
+  },
+  "moving_contact": {
+    "projection_sha256": "a35c264e112f57df6cda1d4f8056963cb3ea7d207981eb17d639222aa8ce2168",
+    "reference_versor": [
+      0.8357952833175659,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.00914173573255539,
+      0.24707065522670746,
+      0.2951143980026245,
+      0.0,
+      0.1899629831314087,
+      0.3036644458770752,
+      0.0,
+      -0.15599043667316437,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      -0.024398060515522957,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0
+    ]
+  },
+  "resting_pose": {
+    "projection_sha256": "da6a31469c1d6e4a7ba827b6728e7d2f4348b389cc81e4ffdb96cd85ba037f56",
+    "reference_versor": [
+      0.8604249954223633,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      -0.0035402218345552683,
+      0.22752714157104492,
+      0.24097634851932526,
+      0.0,
+      0.21282944083213806,
+      0.2927039563655853,
+      0.0,
+      -0.1361951380968094,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      -0.017234591767191887,
+      0.0,
+      0.0,
+      0.0,
+      0.0,
+      0.0
+    ]
+  }
+}

evals/sensorimotor_sensorium/fixtures.json

@@ -0,0 +1,32 @@
+{
+  "comment": "Deterministic quantized proprioceptive fixture specs. These are afferent feedback cases, never motor commands.",
+  "fixtures": [
+    {
+      "id": "resting_pose",
+      "pose_q": [0, 0, 0],
+      "velocity_q": [0, 0, 0],
+      "force_torque_q": [0, 0, 0],
+      "contact_q": [0, 0],
+      "actuator_state_q": [1, 1],
+      "expect": "stationary proprioceptive state"
+    },
+    {
+      "id": "moving_contact",
+      "pose_q": [10, -4, 3],
+      "velocity_q": [2, 0, -1],
+      "force_torque_q": [3, 5, 8],
+      "contact_q": [1, 0, 1],
+      "actuator_state_q": [7, 8],
+      "expect": "moving state with contact"
+    },
+    {
+      "id": "force_spike",
+      "pose_q": [1, 1, 1],
+      "velocity_q": [0, 0, 0],
+      "force_torque_q": [20, -15, 4],
+      "contact_q": [1, 1],
+      "actuator_state_q": [9, 9],
+      "expect": "force torque evidence"
+    }
+  ]
+}

evals/sensorimotor_sensorium/generate_expected.py

@@ -0,0 +1,41 @@
+"""Regenerate frozen expected artifacts for sensorimotor eval."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from evals.sensorimotor_sensorium.synth import synthesize
+from sensorium.sensorimotor import SensorimotorCompiler
+
+_HERE = Path(__file__).resolve().parent
+
+
+def main() -> None:
+    spec = json.loads((_HERE / "fixtures.json").read_text())
+    compiler = SensorimotorCompiler()
+    lines: list[str] = []
+    projection: dict[str, dict] = {}
+    for fx in spec["fixtures"]:
+        signal = synthesize(fx)
+        unit = compiler.compile_signal(signal)
+        lines.append(json.dumps({
+            "id": fx["id"],
+            "canonical_sha256": unit.canonical_sha256,
+            "ir_sha256": unit.ir_sha256,
+            "event_count": len(unit.sensorimotor_ir.events),
+            "event_types": [event.event_type for event in unit.sensorimotor_ir.events],
+        }, sort_keys=True))
+        projection[fx["id"]] = {
+            "projection_sha256": unit.projection_sha256,
+            "reference_versor": [float(x) for x in unit.versor.tolist()],
+        }
+    (_HERE / "expected_ir.jsonl").write_text("\n".join(lines) + "\n")
+    (_HERE / "expected_projection.json").write_text(
+        json.dumps(projection, indent=2, sort_keys=True) + "\n"
+    )
+    print(f"wrote expected artifacts for {len(spec['fixtures'])} fixtures")
+
+
+if __name__ == "__main__":
+    main()

evals/sensorimotor_sensorium/synth.py

@@ -0,0 +1,16 @@
+"""Fixture conversion for quantized sensorimotor eval specs."""
+
+from __future__ import annotations
+
+from sensorium.sensorimotor import ProprioceptiveSignal, canonicalize_proprioception
+
+
+def synthesize(spec: dict) -> ProprioceptiveSignal:
+    return canonicalize_proprioception(
+        pose_q=tuple(int(v) for v in spec.get("pose_q", ())),
+        velocity_q=tuple(int(v) for v in spec.get("velocity_q", ())),
+        force_torque_q=tuple(int(v) for v in spec.get("force_torque_q", ())),
+        contact_q=tuple(int(v) for v in spec.get("contact_q", ())),
+        actuator_state_q=tuple(int(v) for v in spec.get("actuator_state_q", ())),
+        source_id=str(spec["id"]),
+    )

evals/sensorium/__init__.py

@@ -0,0 +1,5 @@
+"""Unified sensorium eval report surface."""
+
+from evals.sensorium.report import build_sensorium_report
+
+__all__ = ["build_sensorium_report"]

evals/sensorium/report.py

@@ -0,0 +1,219 @@
+"""Deterministic sensorium eval reports for modality compiler lanes."""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+from pathlib import Path
+from typing import Literal
+
+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.adapters.audio import make_audio_pack
+from sensorium.adapters.sensorimotor import make_sensorimotor_pack
+from sensorium.adapters.vision import make_vision_pack
+from sensorium.audio.canonical import canonicalize as canonicalize_audio
+from sensorium.audio.compiler import AudioCompiler
+from sensorium.audio.trace import audio_evidence_trace
+from sensorium.audio.types import AudioIR
+from sensorium.registry import ModalityRegistry
+from sensorium.sensorimotor import SensorimotorCompiler, sensorimotor_evidence_trace
+from sensorium.vision import VisionCompiler, canonicalize_image, vision_evidence_trace
+from sensorium.vision.grid import iter_tile_signals
+from sensorium.vision.types import VisionIR
+
+ModalityName = Literal["audio", "vision", "sensorimotor"]
+_ROOT = Path(__file__).resolve().parents[2]
+_AUDIO_DIR = _ROOT / "evals" / "audio_sensorium"
+_VISION_DIR = _ROOT / "evals" / "vision_sensorium"
+_SENSORIMOTOR_DIR = _ROOT / "evals" / "sensorimotor_sensorium"
+_AUDIO_SR = 24_000
+_TOL = 1e-6
+
+
+def _json(path: Path):
+    return json.loads(path.read_text())
+
+
+def _jsonl_by_id(path: Path) -> dict[str, dict]:
+    out: dict[str, dict] = {}
+    for line in path.read_text().splitlines():
+        if line.strip():
+            row = json.loads(line)
+            out[row["id"]] = row
+    return out
+
+
+def _audio_counts(ir: AudioIR) -> dict[str, int]:
+    events = (
+        *ir.speech_spans,
+        *ir.pause_spans,
+        *ir.prosody_arcs,
+        *ir.turn_events,
+        *ir.non_speech_events,
+        *ir.content_anchors,
+    )
+    return dict(sorted(Counter(e.event_type for e in events).items()))
+
+
+def _vision_counts(ir: VisionIR) -> dict[str, int]:
+    events = (
+        *ir.regions,
+        *ir.contour_arcs,
+        *ir.orient_events,
+        *ir.texture_atoms,
+        *ir.salient_events,
+        *ir.content_anchors,
+    )
+    return dict(sorted(Counter(e.event_type for e in events).items()))
+
+
+def _trace_safe(trace: dict[str, object]) -> bool:
+    return all(not isinstance(value, (np.ndarray, bytes, bytearray)) for value in trace.values())
+
+
+def _audio_report() -> dict[str, object]:
+    fixtures = _json(_AUDIO_DIR / "fixtures.json")["fixtures"]
+    expected_ir = _jsonl_by_id(_AUDIO_DIR / "expected_ir.jsonl")
+    expected_proj = _json(_AUDIO_DIR / "expected_projection.json")
+    compiler = AudioCompiler()
+    cases: list[dict[str, object]] = []
+    for fx in fixtures:
+        fid = fx["id"]
+        unit = compiler.compile_signal(canonicalize_audio(synthesize_audio(fx), _AUDIO_SR))
+        replay = compiler.compile_ir(unit.audio_ir)
+        ref = np.asarray(expected_proj[fid]["reference_versor"], dtype=np.float32)
+        cases.append({
+            "id": fid,
+            "canonical_sha256": unit.canonical_sha256,
+            "ir_sha256": unit.ir_sha256,
+            "projection_sha256": unit.projection_sha256,
+            "shape_ok": unit.versor.shape == (32,),
+            "dtype_ok": unit.versor.dtype == np.float32,
+            "replay_ok": bool(np.array_equal(unit.versor, replay.versor)),
+            "expected_ir_ok": unit.ir_sha256 == expected_ir[fid]["ir_sha256"],
+            "expected_projection_ok": bool(np.allclose(unit.versor, ref, atol=_TOL)),
+            "event_counts_ok": _audio_counts(unit.audio_ir) == expected_ir[fid]["event_type_counts"],
+            "trace_hygiene_ok": _trace_safe(audio_evidence_trace(unit)),
+            "versor_condition": unit.versor_condition,
+        })
+    reg = ModalityRegistry()
+    sample = canonicalize_audio(synthesize_audio(fixtures[0]), _AUDIO_SR)
+    reg.mount(make_audio_pack("audio_core_v1"), sample=sample)
+    gate_closed = False
+    try:
+        reg.project("audio_core_v1", sample)
+    except RuntimeError:
+        gate_closed = True
+    return _report("audio", "audio_core_v1", cases, gate_closed)
+
+
+def _vision_report() -> dict[str, object]:
+    fixtures = _json(_VISION_DIR / "fixtures.json")["fixtures"]
+    expected_ir = _jsonl_by_id(_VISION_DIR / "expected_ir.jsonl")
+    expected_proj = _json(_VISION_DIR / "expected_projection.json")
+    compiler = VisionCompiler()
+    cases: list[dict[str, object]] = []
+    for fx in fixtures:
+        fid = fx["id"]
+        image = canonicalize_image(synthesize_vision(fx))
+        units = compiler.compile_image(image)
+        counts = Counter()
+        units_ok = True
+        projection_ok = True
+        trace_ok = True
+        for idx, unit in enumerate(units):
+            replay = compiler.compile_ir(unit.vision_ir)
+            units_ok = units_ok and unit.versor.shape == (32,) and unit.versor.dtype == np.float32
+            units_ok = units_ok and np.array_equal(unit.versor, replay.versor)
+            ref = np.asarray(expected_proj[fid][idx]["reference_versor"], dtype=np.float32)
+            projection_ok = projection_ok and unit.projection_sha256 == expected_proj[fid][idx]["projection_sha256"]
+            projection_ok = projection_ok and np.allclose(unit.versor, ref, atol=_TOL)
+            trace_ok = trace_ok and _trace_safe(vision_evidence_trace(unit))
+            counts.update(_vision_counts(unit.vision_ir))
+        cases.append({
+            "id": fid,
+            "canonical_sha256": image.canonical_sha256,
+            "unit_count": len(units),
+            "unit_count_ok": len(units) == expected_ir[fid]["unit_count"],
+            "units_ok": bool(units_ok),
+            "expected_projection_ok": bool(projection_ok),
+            "event_counts_ok": dict(sorted(counts.items())) == expected_ir[fid]["event_type_counts"],
+            "trace_hygiene_ok": bool(trace_ok),
+        })
+    reg = ModalityRegistry()
+    sample = iter_tile_signals(canonicalize_image(synthesize_vision(fixtures[0])))[0]
+    reg.mount(make_vision_pack("vision_core_v1"), sample=sample)
+    gate_closed = False
+    try:
+        reg.project("vision_core_v1", sample)
+    except RuntimeError:
+        gate_closed = True
+    return _report("vision", "vision_core_v1", cases, gate_closed)
+
+
+def _sensorimotor_report() -> dict[str, object]:
+    fixtures = _json(_SENSORIMOTOR_DIR / "fixtures.json")["fixtures"]
+    expected_ir = _jsonl_by_id(_SENSORIMOTOR_DIR / "expected_ir.jsonl")
+    expected_proj = _json(_SENSORIMOTOR_DIR / "expected_projection.json")
+    compiler = SensorimotorCompiler()
+    cases: list[dict[str, object]] = []
+    for fx in fixtures:
+        fid = fx["id"]
+        unit = compiler.compile_signal(synthesize_sensorimotor(fx))
+        replay = compiler.compile_ir(unit.sensorimotor_ir)
+        ref = np.asarray(expected_proj[fid]["reference_versor"], dtype=np.float32)
+        cases.append({
+            "id": fid,
+            "canonical_sha256": unit.canonical_sha256,
+            "ir_sha256": unit.ir_sha256,
+            "projection_sha256": unit.projection_sha256,
+            "shape_ok": unit.versor.shape == (32,),
+            "dtype_ok": unit.versor.dtype == np.float32,
+            "replay_ok": bool(np.array_equal(unit.versor, replay.versor)),
+            "expected_ir_ok": unit.ir_sha256 == expected_ir[fid]["ir_sha256"],
+            "expected_projection_ok": bool(np.allclose(unit.versor, ref, atol=_TOL)),
+            "event_types_ok": [e.event_type for e in unit.sensorimotor_ir.events] == expected_ir[fid]["event_types"],
+            "trace_hygiene_ok": _trace_safe(sensorimotor_evidence_trace(unit)),
+            "versor_condition": unit.versor_condition,
+        })
+    reg = ModalityRegistry()
+    sample = synthesize_sensorimotor(fixtures[0])
+    reg.mount(make_sensorimotor_pack("sensorimotor_core_v1"), sample=sample)
+    gate_closed = False
+    try:
+        reg.project("sensorimotor_core_v1", sample)
+    except RuntimeError:
+        gate_closed = True
+    return _report("sensorimotor", "sensorimotor_core_v1", cases, gate_closed)
+
+
+def _report(modality: str, pack_id: str, cases: list[dict[str, object]], gate_closed: bool) -> dict[str, object]:
+    pass_count = sum(
+        1 for case in cases
+        if all(value is True for key, value in case.items() if key.endswith("_ok"))
+    )
+    return {
+        "lane": "sensorium",
+        "modality": modality,
+        "pack_id": pack_id,
+        "gate_engaged": False,
+        "gate_closed": gate_closed,
+        "total": len(cases),
+        "passed": pass_count,
+        "failed": len(cases) - pass_count,
+        "cases": cases,
+    }
+
+
+def build_sensorium_report(modality: ModalityName) -> dict[str, object]:
+    if modality == "audio":
+        return _audio_report()
+    if modality == "vision":
+        return _vision_report()
+    if modality == "sensorimotor":
+        return _sensorimotor_report()
+    raise ValueError(f"unsupported sensorium modality: {modality!r}")

packs/sensorimotor/__init__.py

@@ -0,0 +1 @@
+"""Sensorimotor pack artifacts."""

packs/sensorimotor/loader.py

@@ -0,0 +1,85 @@
+"""Sensorimotor pack loader with fail-closed checksum verification."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from pathlib import Path
+
+from sensorium.audio.checksum import sha256_json
+
+_PACKS_ROOT = Path(__file__).resolve().parent
+
+
+class SensorimotorPackError(ValueError):
+    """Raised when a sensorimotor pack is missing, malformed, or tampered."""
+
+
+def _validate_pack_id(pack_id: object) -> str:
+    from core._safe_display import safe_pack_id as _disp
+
+    if not isinstance(pack_id, str):
+        raise SensorimotorPackError(f"pack_id must be a string, got {_disp(pack_id)!r}")
+    if pack_id == "":
+        raise SensorimotorPackError("pack_id must not be empty")
+    if ".." in pack_id:
+        raise SensorimotorPackError(f"pack_id must not contain '..': {_disp(pack_id)!r}")
+    if "/" in pack_id or "\\" in pack_id:
+        raise SensorimotorPackError(f"pack_id must be a simple pack id, not a path: {_disp(pack_id)!r}")
+    if pack_id.startswith("."):
+        raise SensorimotorPackError(f"pack_id must not start with '.': {_disp(pack_id)!r}")
+    for ch in pack_id:
+        if not (ch.isascii() and (ch.isalnum() or ch in {"_", "-"})):
+            raise SensorimotorPackError(f"pack_id must be alphanumeric/_/-, got {_disp(pack_id)!r}")
+    return pack_id
+
+
+@dataclass(frozen=True, slots=True)
+class LoadedSensorimotorPack:
+    pack_id: str
+    manifest: dict
+    manifest_sha256: str
+    basis_map: dict
+
+
+def _verify_checksums(pack_dir: Path) -> None:
+    checks_path = pack_dir / "checksums.json"
+    if not checks_path.is_file():
+        raise SensorimotorPackError(f"checksums.json missing for pack at {pack_dir.name}")
+    checks = json.loads(checks_path.read_text())
+    for fname, expected in checks.get("files", {}).items():
+        fpath = pack_dir / fname
+        if not fpath.is_file():
+            raise SensorimotorPackError(f"pack file '{fname}' named in checksums.json is missing")
+        actual = "sha256:" + hashlib.sha256(fpath.read_bytes()).hexdigest()
+        if actual != expected:
+            raise SensorimotorPackError(
+                f"checksum mismatch for '{fname}': expected {expected}, got {actual}"
+            )
+
+
+def load_sensorimotor_pack(
+    pack_id: str = "sensorimotor_core_v1",
+    *,
+    packs_root: Path | None = None,
+    verify: bool = True,
+) -> LoadedSensorimotorPack:
+    safe_id = _validate_pack_id(pack_id)
+    root = packs_root if packs_root is not None else _PACKS_ROOT
+    pack_dir = (root / safe_id).resolve()
+    if not str(pack_dir).startswith(str(root.resolve())):
+        raise SensorimotorPackError(f"resolved pack path escapes packs root: {safe_id!r}")
+    if not pack_dir.is_dir():
+        raise SensorimotorPackError(f"no sensorimotor pack mounted at {safe_id!r}")
+    if verify:
+        _verify_checksums(pack_dir)
+
+    manifest = json.loads((pack_dir / "manifest.json").read_text())
+    basis_map = json.loads((pack_dir / "basis_map.json").read_text())
+    manifest_sha256 = sha256_json({
+        "pack_id": safe_id,
+        "basis_version": manifest.get("basis_version", "sensorimotor-basis-v1"),
+        "events": list(manifest.get("event_order", ())),
+    })
+    return LoadedSensorimotorPack(safe_id, manifest, manifest_sha256, basis_map)

packs/sensorimotor/sensorimotor_core_v1/basis_map.json

@@ -0,0 +1,26 @@
+{
+  "basis_version": "sensorimotor-basis-v1",
+  "comment": "Afferent proprioceptive facts lower to elliptic bivector rotors only; this pack contains no efferent decoder or command surface.",
+  "events": {
+    "proprio.pose": {
+      "blade_index": 6,
+      "base_theta_q": 48
+    },
+    "proprio.velocity": {
+      "blade_index": 7,
+      "base_theta_q": 64
+    },
+    "haptic.force_torque": {
+      "blade_index": 8,
+      "base_theta_q": 80
+    },
+    "haptic.contact": {
+      "blade_index": 10,
+      "base_theta_q": 96
+    },
+    "actuator.state": {
+      "blade_index": 11,
+      "base_theta_q": 112
+    }
+  }
+}

packs/sensorimotor/sensorimotor_core_v1/checksums.json

@@ -0,0 +1,6 @@
+{
+  "files": {
+    "basis_map.json": "sha256:e0c174988bae3f3480258b474fd4ec79bfb70fca46210faa29f8db10a1e3cc9e",
+    "manifest.json": "sha256:073768936418c3a8ee3e49a65de3d0351c387b37f95a4499c7cc1d1ee93ee822"
+  }
+}

packs/sensorimotor/sensorimotor_core_v1/manifest.json

@@ -0,0 +1,22 @@
+{
+  "pack_id": "sensorimotor_core_v1",
+  "modality": "sensorimotor",
+  "compiler_version": "sensorimotor-compiler-v1",
+  "basis_version": "sensorimotor-basis-v1",
+  "cl41_dim": 32,
+  "gate_engaged": false,
+  "projection_unit": "quantized_proprioceptive_signal",
+  "efferent": false,
+  "event_order": [
+    "proprio.pose",
+    "proprio.velocity",
+    "haptic.force_torque",
+    "haptic.contact",
+    "actuator.state"
+  ],
+  "rules": [
+    "sensorimotor/proprioception is afferent feedback",
+    "motor commands remain efferent and governed by EfferentGate",
+    "no robot, actuator, trajectory executor, tool invocation, or skill invocation is present"
+  ]
+}

sensorium/__init__.py

@@ -27,6 +27,7 @@ from sensorium.protocol import (
     ProjectionHead,
     SurfaceDecoder,
 )
+from sensorium.efferent import DefaultEfferentGate, EfferentEmissionTrace
 from sensorium.registry import ModalityRegistry
 
 __all__ = [
@@ -35,6 +36,8 @@ __all__ = [
     "EfferentGate",
     "EfferentRefusal",
     "EfferentVerdict",
+    "DefaultEfferentGate",
+    "EfferentEmissionTrace",
     "ProjectionHead",
     "SurfaceDecoder",
     "ModalityVocabulary",

sensorium/adapters/sensorimotor.py

@@ -0,0 +1,71 @@
+"""Sensorimotor modality adapter.
+
+Sensorimotor v1 is afferent proprioceptive feedback only. It provides a
+ProjectionHead so compiled feedback can enter the shared manifold, but it does
+not provide a SurfaceDecoder or any motor command path.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import numpy as np
+
+from sensorium.protocol import CL41_DIM, Modality, ModalityPack, ModalityVocabulary
+from sensorium.sensorimotor.compiler import SensorimotorCompiler
+from sensorium.sensorimotor.types import ProprioceptiveSignal
+
+
+@dataclass(frozen=True, slots=True)
+class SensorimotorProjectionHead:
+    """ProjectionHead for quantized afferent ProprioceptiveSignal values."""
+
+    compiler: SensorimotorCompiler
+    modality: Modality = Modality.SENSORIMOTOR
+
+    @property
+    def embedding_dim(self) -> int:
+        return CL41_DIM
+
+    def project(self, signal: ProprioceptiveSignal) -> np.ndarray:
+        out = self.compiler.compile_signal(signal).versor
+        if out.shape != (CL41_DIM,):
+            raise ValueError(f"expected ({CL41_DIM},), got {out.shape}")
+        if out.dtype != np.float32:
+            raise TypeError(f"expected float32, got {out.dtype}")
+        return out
+
+    def project_batch(self, signals: list[ProprioceptiveSignal]) -> np.ndarray:
+        return np.stack([self.project(signal) for signal in signals], axis=0)
+
+    def verify_unitarity(self, sample: ProprioceptiveSignal) -> bool:
+        try:
+            return self.compiler.compile_signal(sample).versor_condition < 1e-6
+        except Exception:
+            return False
+
+
+def make_sensorimotor_pack(
+    pack_id: str = "sensorimotor_core_v1",
+    *,
+    gate_engaged: bool = False,
+    checksum_verified: bool = False,
+    packs_root=None,
+) -> ModalityPack:
+    from packs.sensorimotor.loader import load_sensorimotor_pack
+
+    loaded = load_sensorimotor_pack(pack_id, packs_root=packs_root)
+    compiler = SensorimotorCompiler(
+        loaded.pack_id,
+        pack_manifest_sha256=loaded.manifest_sha256,
+    )
+    return ModalityPack(
+        pack_id=loaded.pack_id,
+        modality_type=Modality.SENSORIMOTOR,
+        projection=SensorimotorProjectionHead(compiler),
+        decoder=None,
+        vocabulary=ModalityVocabulary(),
+        grammar_scaffold=None,
+        checksum_verified=checksum_verified,
+        gate_engaged=gate_engaged,
+    )

sensorium/efferent.py

@@ -0,0 +1,104 @@
+"""Concrete efferent gate policy and trace-safe decision records."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import numpy as np
+
+from sensorium.audio.checksum import sha256_json
+from sensorium.protocol import CL41_DIM, AuthorityToken, EfferentVerdict
+
+
+@dataclass(frozen=True, slots=True)
+class EfferentEmissionTrace:
+    """Trace-safe record of an efferent admission or refusal."""
+
+    pack_id: str
+    admitted: bool
+    reason: str
+    authority_sha256: str
+    policy_sha256: str
+    capability: str
+    trace_sha256: str
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "pack_id": self.pack_id,
+            "admitted": self.admitted,
+            "reason": self.reason,
+            "authority_sha256": self.authority_sha256,
+            "policy_sha256": self.policy_sha256,
+            "capability": self.capability,
+            "trace_sha256": self.trace_sha256,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class DefaultEfferentGate:
+    """Capability-scoped efferent gate.
+
+    Admission requires a valid ``(32,)`` vector and one of:
+    ``decode:<pack_id>``, ``decode:*``, or ``*`` in the authority token.
+    """
+
+    policy_id: str = "default-efferent-v1"
+
+    @property
+    def policy_sha256(self) -> str:
+        return sha256_json({
+            "policy_id": self.policy_id,
+            "required_capability": "decode:<pack_id>",
+            "wildcards": ["decode:*", "*"],
+            "shape": [CL41_DIM],
+        })
+
+    def admit(
+        self,
+        pack_id: str,
+        mv: np.ndarray,
+        authority: AuthorityToken,
+    ) -> EfferentVerdict:
+        vec = np.asarray(mv, dtype=np.float32)
+        if vec.shape != (CL41_DIM,):
+            return EfferentVerdict(
+                admitted=False,
+                reason=f"invalid efferent vector shape: {vec.shape}",
+                authority_sha256=authority.authority_sha256,
+                policy_sha256=self.policy_sha256,
+            )
+        required = f"decode:{pack_id}"
+        caps = set(authority.capabilities)
+        admitted = required in caps or "decode:*" in caps or "*" in caps
+        return EfferentVerdict(
+            admitted=admitted,
+            reason="admitted" if admitted else f"missing capability: {required}",
+            authority_sha256=authority.authority_sha256,
+            policy_sha256=self.policy_sha256,
+        )
+
+    def trace(
+        self,
+        pack_id: str,
+        authority: AuthorityToken,
+        verdict: EfferentVerdict,
+    ) -> EfferentEmissionTrace:
+        capability = f"decode:{pack_id}"
+        payload = {
+            "kind": "EfferentEmissionTrace",
+            "pack_id": pack_id,
+            "admitted": verdict.admitted,
+            "reason": verdict.reason,
+            "authority_sha256": authority.authority_sha256,
+            "policy_sha256": verdict.policy_sha256,
+            "capability": capability,
+        }
+        return EfferentEmissionTrace(
+            pack_id=pack_id,
+            admitted=verdict.admitted,
+            reason=verdict.reason,
+            authority_sha256=authority.authority_sha256,
+            policy_sha256=verdict.policy_sha256,
+            capability=capability,
+            trace_sha256=sha256_json(payload),
+        )

sensorium/environment/__init__.py

@@ -1,5 +1,6 @@
 """Environmental observation contracts for sensorium units."""
 
 from sensorium.environment.frame import ObservationFrame, build_observation_frame
+from sensorium.environment.harness import build_fixture_observation_frame
 
-__all__ = ["ObservationFrame", "build_observation_frame"]
+__all__ = ["ObservationFrame", "build_fixture_observation_frame", "build_observation_frame"]

sensorium/environment/harness.py

@@ -0,0 +1,54 @@
+"""Deterministic fixture harness for environmental observation frames."""
+
+from __future__ import annotations
+
+from sensorium.environment.frame import ObservationFrame, build_observation_frame
+
+_AUDIO_FIXTURE = {"id": "env_tone", "kind": "tone", "ms": 300, "hz": 150, "sweep": 90, "amp": 0.5}
+_VISION_FIXTURE = {"id": "env_corner", "kind": "corner", "size": 32}
+_SENSORIMOTOR_FIXTURE = {
+    "id": "env_contact",
+    "pose_q": [10, -4, 3],
+    "velocity_q": [2, 0, -1],
+    "force_torque_q": [3, 5, 8],
+    "contact_q": [1, 0, 1],
+    "actuator_state_q": [7, 8],
+}
+
+
+def build_fixture_observation_frame(
+    *,
+    monotonic_tick: int = 0,
+    source_clock: str = "fixture-clock",
+    causal_parent_ids: tuple[str, ...] = (),
+) -> ObservationFrame:
+    """Build a deterministic mixed-modality ObservationFrame.
+
+    The frame contains already-compiled afferent units only: one audio unit,
+    one vision tile unit, and one sensorimotor feedback unit.
+    """
+
+    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.sensorimotor.compiler import SensorimotorCompiler
+    from sensorium.vision import VisionCompiler, canonicalize_image
+    from sensorium.vision.grid import iter_tile_signals
+
+    audio_unit = AudioCompiler().compile_signal(
+        canonicalize_audio(synthesize_audio(_AUDIO_FIXTURE), 24_000)
+    )
+    vision_image = canonicalize_image(synthesize_vision(_VISION_FIXTURE))
+    vision_tile = iter_tile_signals(vision_image)[0]
+    vision_unit = VisionCompiler().compile_tile(vision_tile)
+    sensorimotor_unit = SensorimotorCompiler().compile_signal(
+        synthesize_sensorimotor(_SENSORIMOTOR_FIXTURE)
+    )
+    return build_observation_frame(
+        monotonic_tick=monotonic_tick,
+        source_clock=source_clock,
+        units=(audio_unit, vision_unit, sensorimotor_unit),
+        causal_parent_ids=causal_parent_ids,
+    )

sensorium/sensorimotor/compiler.py

@@ -130,9 +130,14 @@ class SensorimotorCompiler:
 
     modality = "sensorimotor"
 
-    def __init__(self, pack_id: str = "sensorimotor_core_v1") -> None:
+    def __init__(
+        self,
+        pack_id: str = "sensorimotor_core_v1",
+        *,
+        pack_manifest_sha256: str | None = None,
+    ) -> None:
         self._pack_id = pack_id
-        self._manifest_sha256 = sha256_json({
+        self._manifest_sha256 = pack_manifest_sha256 or sha256_json({
             "pack_id": pack_id,
             "basis_version": "sensorimotor-basis-v1",
             "events": list(_EVENT_ORDER),

tests/test_efferent_gate.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+from sensorium.efferent import DefaultEfferentGate
+from sensorium.protocol import AuthorityToken, EfferentRefusal, Modality, ModalityPack, ModalityVocabulary
+from sensorium.registry import ModalityRegistry
+
+
+class _Decoder:
+    modality = Modality.MOTOR
+
+    def __init__(self) -> None:
+        self.calls = 0
+
+    def decode(self, mv: np.ndarray) -> str:
+        self.calls += 1
+        return "decoded"
+
+    def decode_batch(self, mvs: np.ndarray) -> list[str]:
+        self.calls += len(mvs)
+        return ["decoded" for _ in range(len(mvs))]
+
+
+def _mv() -> np.ndarray:
+    out = np.zeros(32, dtype=np.float32)
+    out[0] = 1.0
+    return out
+
+
+def _authority(*capabilities: str) -> AuthorityToken:
+    return AuthorityToken(
+        principal_id="test-principal",
+        capabilities=tuple(capabilities),
+        issued_at_revision="test-revision",
+    )
+
+
+def _pack(decoder: _Decoder) -> ModalityPack:
+    return ModalityPack(
+        pack_id="motor_test",
+        modality_type=Modality.MOTOR,
+        vocabulary=ModalityVocabulary(),
+        grammar_scaffold=None,
+        checksum_verified=True,
+        decoder=decoder,
+        gate_engaged=True,
+    )
+
+
+def test_default_efferent_gate_admits_exact_and_wildcard_capabilities():
+    gate = DefaultEfferentGate()
+    assert gate.admit("motor_test", _mv(), _authority("decode:motor_test")).admitted
+    assert gate.admit("motor_test", _mv(), _authority("decode:*")).admitted
+    assert gate.admit("motor_test", _mv(), _authority("*")).admitted
+
+
+def test_default_efferent_gate_denies_missing_capability_and_bad_shape():
+    gate = DefaultEfferentGate()
+    denied = gate.admit("motor_test", _mv(), _authority("decode:other"))
+    assert denied.admitted is False
+    assert "missing capability" in denied.reason
+
+    malformed = gate.admit("motor_test", np.zeros(31, dtype=np.float32), _authority("decode:motor_test"))
+    assert malformed.admitted is False
+    assert "invalid efferent vector shape" in malformed.reason
+
+
+def test_default_efferent_trace_is_hash_only():
+    gate = DefaultEfferentGate()
+    authority = _authority("decode:motor_test")
+    verdict = gate.admit("motor_test", _mv(), authority)
+    trace = gate.trace("motor_test", authority, verdict).as_dict()
+    assert trace["admitted"] is True
+    assert trace["capability"] == "decode:motor_test"
+    assert "mv" not in trace
+    for value in trace.values():
+        assert not isinstance(value, (np.ndarray, bytes, bytearray))
+
+
+def test_registry_uses_default_efferent_gate_before_decoder():
+    decoder = _Decoder()
+    reg = ModalityRegistry(efferent_gate=DefaultEfferentGate())
+    reg.mount(_pack(decoder))
+
+    with pytest.raises(EfferentRefusal, match="missing capability"):
+        reg.decode("motor_test", _mv(), authority=_authority("decode:other"))
+    assert decoder.calls == 0
+
+    assert reg.decode("motor_test", _mv(), authority=_authority("decode:motor_test")) == "decoded"
+    assert decoder.calls == 1

tests/test_observation_frame_harness.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import numpy as np
+
+from sensorium.environment import build_fixture_observation_frame
+
+
+def test_fixture_observation_frame_is_mixed_modality_and_deterministic():
+    f1 = build_fixture_observation_frame(monotonic_tick=7, source_clock="test-clock")
+    f2 = build_fixture_observation_frame(monotonic_tick=7, source_clock="test-clock")
+    assert f1.frame_id == f2.frame_id
+    assert f1.environment_sha256 == f2.environment_sha256
+    assert f1.trace_hash == f2.trace_hash
+    assert {unit.pack_id for unit in f1.units} == {
+        "audio_core_v1",
+        "sensorimotor_core_v1",
+        "vision_core_v1",
+    }
+    for unit in f1.units:
+        assert unit.versor.shape == (32,)
+        assert unit.versor.dtype == np.float32
+        assert unit.versor_condition < 1e-6
+
+
+def test_fixture_observation_frame_tick_changes_identity_not_unit_set():
+    f1 = build_fixture_observation_frame(monotonic_tick=7)
+    f2 = build_fixture_observation_frame(monotonic_tick=8)
+    assert f1.trace_hash != f2.trace_hash
+    assert [unit.merge_key for unit in f1.units] == [unit.merge_key for unit in f2.units]

tests/test_sensorimotor_pack_manifest.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from evals.sensorimotor_sensorium.synth import synthesize
+from packs.sensorimotor.loader import SensorimotorPackError, load_sensorimotor_pack
+from sensorium.adapters.sensorimotor import SensorimotorProjectionHead, make_sensorimotor_pack
+from sensorium.protocol import Modality
+from sensorium.registry import ModalityRegistry
+from sensorium.sensorimotor import SensorimotorCompiler
+
+
+def _fixture_signal():
+    return synthesize({
+        "id": "pack_probe",
+        "pose_q": [1, 2, 3],
+        "velocity_q": [0, 0, 1],
+        "force_torque_q": [5, 8, 13],
+        "contact_q": [1, 0],
+        "actuator_state_q": [3, 5],
+    })
+
+
+def test_sensorimotor_pack_loads_and_mounts_closed_by_default():
+    loaded = load_sensorimotor_pack("sensorimotor_core_v1")
+    assert loaded.pack_id == "sensorimotor_core_v1"
+    assert loaded.manifest["modality"] == "sensorimotor"
+    assert loaded.manifest["gate_engaged"] is False
+
+    pack = make_sensorimotor_pack("sensorimotor_core_v1")
+    assert pack.modality_type is Modality.SENSORIMOTOR
+    assert pack.decoder is None
+    assert pack.gate_engaged is False
+
+    reg = ModalityRegistry()
+    reg.mount(pack, sample=_fixture_signal())
+    with pytest.raises(RuntimeError, match="gate is not engaged"):
+        reg.project("sensorimotor_core_v1", _fixture_signal())
+
+
+def test_sensorimotor_projection_head_is_deterministic_when_engaged():
+    sample = _fixture_signal()
+    head = SensorimotorProjectionHead(SensorimotorCompiler())
+    assert head.verify_unitarity(sample)
+    mv = head.project(sample)
+    assert mv.shape == (32,)
+    assert mv.dtype == np.float32
+    assert np.array_equal(mv, head.project(sample))
+
+    reg = ModalityRegistry()
+    reg.mount(
+        make_sensorimotor_pack(
+            "sensorimotor_core_v1",
+            gate_engaged=True,
+            checksum_verified=True,
+        ),
+        sample=sample,
+    )
+    assert np.array_equal(reg.project("sensorimotor_core_v1", sample), mv)
+
+
+def test_sensorimotor_pack_rejects_path_traversal_and_checksum_mismatch(tmp_path: Path):
+    with pytest.raises(SensorimotorPackError):
+        load_sensorimotor_pack("../sensorimotor_core_v1")
+
+    src = Path("packs/sensorimotor/sensorimotor_core_v1")
+    root = tmp_path / "packs"
+    shutil.copytree(src, root / "sensorimotor_core_v1")
+    manifest = root / "sensorimotor_core_v1" / "manifest.json"
+    manifest.write_text(manifest.read_text().replace('"cl41_dim": 32', '"cl41_dim": 31'))
+    with pytest.raises(SensorimotorPackError, match="checksum mismatch"):
+        load_sensorimotor_pack("sensorimotor_core_v1", packs_root=root)

tests/test_sensorium_eval_cli.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+import json
+
+from core.cli import main
+
+
+def test_core_eval_sensorium_json_reports_selected_modality(capsys):
+    assert main(["eval", "sensorium", "--modality", "sensorimotor", "--json"]) == 0
+    out = capsys.readouterr().out
+    report = json.loads(out)
+    assert report["lane"] == "sensorium"
+    assert report["modality"] == "sensorimotor"
+    assert report["pack_id"] == "sensorimotor_core_v1"
+    assert report["gate_closed"] is True
+    assert report["failed"] == 0
+
+
+def test_core_eval_sensorium_text_summary(capsys):
+    assert main(["eval", "sensorium", "--modality", "vision"]) == 0
+    out = capsys.readouterr().out
+    assert "lane           : sensorium" in out
+    assert "modality       : vision" in out
+    assert "gate_closed    : True" in out
+    assert "failed         : 0" in out

tests/test_sensorium_eval_report.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+import json
+
+from evals.sensorium import build_sensorium_report
+
+
+def test_sensorium_reports_are_deterministic_and_gate_closed():
+    for modality in ("audio", "vision", "sensorimotor"):
+        first = build_sensorium_report(modality)
+        second = build_sensorium_report(modality)
+        assert json.dumps(first, sort_keys=True) == json.dumps(second, sort_keys=True)
+        assert first["lane"] == "sensorium"
+        assert first["modality"] == modality
+        assert first["gate_engaged"] is False
+        assert first["gate_closed"] is True
+        assert first["total"] > 0
+        assert first["failed"] == 0
+        assert first["passed"] == first["total"]