Add motor verdict enforcing gate

docs/decisions/ADR-0216-motor-verdict-lowering.md

@@ -0,0 +1,47 @@
+# ADR-0216: Motor Verdict Lowering Prerequisite
+
+**Status:** Proposed
+**Date:** 2026-06-06
+**Domains:** `sensorium/efferent.py`, `sensorium/registry.py`, future motor packs
+**Depends on:** ADR-0198, ADR-0211
+
+## Context
+
+ADR-0198 added the `ModalityRegistry.decode()` efferent path and intentionally
+left real motor emission fail-closed. The current `DefaultEfferentGate` checks
+capability and vector shape only; it does not lower a motor versor into the
+safety, ethics, and tool-scope verdicts required for real action.
+
+## Decision
+
+No physical motor decode is authorized until a dedicated implementation provides
+a verdict-enforcing gate:
+
+```text
+(32,) motor versor
+  -> MotorActionIntent
+  -> authority + safety + ethics + tool-scope verdicts
+  -> admit/refuse before SurfaceDecoder.decode()
+```
+
+`MotorActionIntent` is a semantic action predicate bundle, not an actuator
+command. A `VerdictEnforcingEfferentGate` must expose
+`enforces_action_verdicts=True` and fail closed on absent authority, absent
+verdict coverage, or any refusal.
+
+## Non-Goals
+
+- No actuator driver.
+- No robot interface.
+- No trajectory executor.
+- No sandbox opt-in on a physical path.
+- No ObservationFrame or CRDT write for emitted motor commands.
+
+## Required Proof
+
+- The gate refuses before decoder invocation on missing authority or failed
+  verdicts.
+- Traces include hashes, authority hash, policy hash, and verdict only.
+- No decoded command payload, trajectory, ndarray, or bytes object enters traces.
+- Physical action remains disabled until a later motor decoder ADR and lab
+  safety contract.

sensorium/efferent.py

@@ -6,7 +6,7 @@ from dataclasses import dataclass
 
 import numpy as np
 
-from sensorium.audio.checksum import sha256_json
+from sensorium.audio.checksum import sha256_array, sha256_json
 from sensorium.protocol import CL41_DIM, AuthorityToken, EfferentVerdict
 
 
@@ -34,6 +34,91 @@ class EfferentEmissionTrace:
         }
 
 
+@dataclass(frozen=True, slots=True)
+class MotorActionIntent:
+    """Hash-only semantic lowering of a motor versor into an action predicate."""
+
+    pack_id: str
+    predicate_id: str
+    vector_sha256: str
+    intent_sha256: str
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "pack_id": self.pack_id,
+            "predicate_id": self.predicate_id,
+            "vector_sha256": self.vector_sha256,
+            "intent_sha256": self.intent_sha256,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class ActionVerdictRecord:
+    """One pre-decode verdict over a lowered motor intent."""
+
+    intent_sha256: str
+    verdict_type: str
+    admitted: bool
+    reason: str
+    policy_sha256: str
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "intent_sha256": self.intent_sha256,
+            "verdict_type": self.verdict_type,
+            "admitted": self.admitted,
+            "reason": self.reason,
+            "policy_sha256": self.policy_sha256,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class MotorVerdictTrace:
+    """Trace-safe motor gate decision; no decoded payload or trajectory."""
+
+    pack_id: str
+    intent_sha256: str
+    admitted: bool
+    reason: str
+    authority_sha256: str
+    policy_sha256: str
+    required_verdicts: tuple[str, ...]
+    trace_sha256: str
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "pack_id": self.pack_id,
+            "intent_sha256": self.intent_sha256,
+            "admitted": self.admitted,
+            "reason": self.reason,
+            "authority_sha256": self.authority_sha256,
+            "policy_sha256": self.policy_sha256,
+            "required_verdicts": list(self.required_verdicts),
+            "trace_sha256": self.trace_sha256,
+        }
+
+
+def lower_motor_action_intent(pack_id: str, mv: np.ndarray) -> MotorActionIntent:
+    """Lower a motor versor to a semantic predicate, not an actuator command."""
+    vec = np.asarray(mv, dtype=np.float32)
+    if vec.shape != (CL41_DIM,):
+        raise ValueError(f"invalid motor vector shape: {vec.shape}")
+    vector_sha256 = sha256_array(vec)
+    predicate_id = f"motor.intent.{vector_sha256[:16]}"
+    payload = {
+        "kind": "MotorActionIntent",
+        "pack_id": pack_id,
+        "predicate_id": predicate_id,
+        "vector_sha256": vector_sha256,
+    }
+    return MotorActionIntent(
+        pack_id=pack_id,
+        predicate_id=predicate_id,
+        vector_sha256=vector_sha256,
+        intent_sha256=sha256_json(payload),
+    )
+
+
 @dataclass(frozen=True, slots=True)
 class DefaultEfferentGate:
     """Capability-scoped efferent gate.
@@ -111,3 +196,124 @@ class DefaultEfferentGate:
             capability=capability,
             trace_sha256=sha256_json(payload),
         )
+
+
+@dataclass(frozen=True, slots=True)
+class VerdictEnforcingEfferentGate:
+    """ADR-0198 §3 gate: authority plus explicit action verdict coverage."""
+
+    action_verdicts: tuple[ActionVerdictRecord, ...] = ()
+    policy_id: str = "verdict-enforcing-efferent-v1"
+    required_verdicts: tuple[str, ...] = ("safety", "ethics", "tool_scope")
+
+    @property
+    def enforces_action_verdicts(self) -> bool:
+        return True
+
+    @property
+    def policy_sha256(self) -> str:
+        return sha256_json({
+            "policy_id": self.policy_id,
+            "required_verdicts": list(self.required_verdicts),
+            "action_verdicts": [
+                verdict.as_dict()
+                for verdict in sorted(
+                    self.action_verdicts,
+                    key=lambda v: (v.intent_sha256, v.verdict_type, v.policy_sha256),
+                )
+            ],
+        })
+
+    def _verdict_index(self) -> dict[tuple[str, str], ActionVerdictRecord]:
+        return {
+            (verdict.intent_sha256, verdict.verdict_type): verdict
+            for verdict in self.action_verdicts
+        }
+
+    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)
+        if required not in caps and "decode:*" not in caps and "*" not in caps:
+            return EfferentVerdict(
+                admitted=False,
+                reason=f"missing capability: {required}",
+                authority_sha256=authority.authority_sha256,
+                policy_sha256=self.policy_sha256,
+            )
+        intent = lower_motor_action_intent(pack_id, vec)
+        verdicts = self._verdict_index()
+        for verdict_type in self.required_verdicts:
+            record = verdicts.get((intent.intent_sha256, verdict_type))
+            if record is None:
+                return EfferentVerdict(
+                    admitted=False,
+                    reason=f"missing action verdict coverage: {verdict_type}",
+                    authority_sha256=authority.authority_sha256,
+                    policy_sha256=self.policy_sha256,
+                )
+            if not record.admitted:
+                return EfferentVerdict(
+                    admitted=False,
+                    reason=f"action verdict refused: {verdict_type}: {record.reason}",
+                    authority_sha256=authority.authority_sha256,
+                    policy_sha256=self.policy_sha256,
+                )
+        return EfferentVerdict(
+            admitted=True,
+            reason="admitted",
+            authority_sha256=authority.authority_sha256,
+            policy_sha256=self.policy_sha256,
+        )
+
+    def trace(
+        self,
+        pack_id: str,
+        mv: np.ndarray,
+        authority: AuthorityToken,
+        verdict: EfferentVerdict,
+    ) -> MotorVerdictTrace:
+        intent = lower_motor_action_intent(pack_id, mv)
+        payload = {
+            "kind": "MotorVerdictTrace",
+            "pack_id": pack_id,
+            "intent_sha256": intent.intent_sha256,
+            "admitted": verdict.admitted,
+            "reason": verdict.reason,
+            "authority_sha256": authority.authority_sha256,
+            "policy_sha256": verdict.policy_sha256,
+            "required_verdicts": list(self.required_verdicts),
+        }
+        return MotorVerdictTrace(
+            pack_id=pack_id,
+            intent_sha256=intent.intent_sha256,
+            admitted=verdict.admitted,
+            reason=verdict.reason,
+            authority_sha256=authority.authority_sha256,
+            policy_sha256=verdict.policy_sha256,
+            required_verdicts=tuple(self.required_verdicts),
+            trace_sha256=sha256_json(payload),
+        )
+
+
+__all__ = [
+    "ActionVerdictRecord",
+    "DefaultEfferentGate",
+    "EfferentEmissionTrace",
+    "MotorActionIntent",
+    "MotorVerdictTrace",
+    "VerdictEnforcingEfferentGate",
+    "lower_motor_action_intent",
+]

tests/test_efferent_gate.py

@@ -3,7 +3,14 @@ from __future__ import annotations
 import numpy as np
 import pytest
 
-from sensorium.efferent import DefaultEfferentGate
+from pathlib import Path
+
+from sensorium.efferent import (
+    ActionVerdictRecord,
+    DefaultEfferentGate,
+    VerdictEnforcingEfferentGate,
+    lower_motor_action_intent,
+)
 from sensorium.protocol import (
     AuthorityToken,
     EfferentRefusal,
@@ -132,3 +139,88 @@ def test_registry_admits_decode_through_verdict_enforcing_gate():
     reg.mount(_pack(decoder))
     assert reg.decode("motor_test", _mv(), authority=_authority("decode:motor_test")) == "decoded"
     assert decoder.calls == 1
+
+
+def _admitted_records(intent_sha256: str) -> tuple[ActionVerdictRecord, ...]:
+    return (
+        ActionVerdictRecord(intent_sha256, "safety", True, "safe", "safety-policy"),
+        ActionVerdictRecord(intent_sha256, "ethics", True, "ethical", "ethics-policy"),
+        ActionVerdictRecord(intent_sha256, "tool_scope", True, "in scope", "tool-policy"),
+    )
+
+
+def test_adr_0216_motor_verdict_lowering_is_documented():
+    root = Path(__file__).resolve().parents[1]
+    adr = root / "docs" / "decisions" / "ADR-0216-motor-verdict-lowering.md"
+    assert adr.exists()
+    text = adr.read_text(encoding="utf-8")
+    assert "MotorActionIntent" in text
+    assert "VerdictEnforcingEfferentGate" in text
+    assert "No physical motor decode is authorized" in text
+
+
+def test_motor_action_intent_is_hash_only_and_stable():
+    intent = lower_motor_action_intent("motor_test", _mv())
+    same = lower_motor_action_intent("motor_test", _mv())
+    payload = intent.as_dict()
+    assert intent.intent_sha256 == same.intent_sha256
+    assert intent.predicate_id.startswith("motor.intent.")
+    assert "decoded" not in str(payload)
+    assert "trajectory" not in str(payload)
+    for value in payload.values():
+        assert not isinstance(value, (np.ndarray, bytes, bytearray))
+
+
+def test_verdict_enforcing_gate_refuses_missing_coverage_before_decoder():
+    decoder = _Decoder()
+    gate = VerdictEnforcingEfferentGate()
+    reg = ModalityRegistry(efferent_gate=gate)
+    reg.mount(_pack(decoder))
+    with pytest.raises(EfferentRefusal, match="missing action verdict coverage"):
+        reg.decode("motor_test", _mv(), authority=_authority("decode:motor_test"))
+    assert decoder.calls == 0
+
+
+def test_verdict_enforcing_gate_refuses_failed_verdict_before_decoder():
+    decoder = _Decoder()
+    intent = lower_motor_action_intent("motor_test", _mv())
+    records = (
+        ActionVerdictRecord(intent.intent_sha256, "safety", True, "safe", "safety-policy"),
+        ActionVerdictRecord(intent.intent_sha256, "ethics", False, "ethical refusal", "ethics-policy"),
+        ActionVerdictRecord(intent.intent_sha256, "tool_scope", True, "in scope", "tool-policy"),
+    )
+    reg = ModalityRegistry(efferent_gate=VerdictEnforcingEfferentGate(records))
+    reg.mount(_pack(decoder))
+    with pytest.raises(EfferentRefusal, match="action verdict refused: ethics"):
+        reg.decode("motor_test", _mv(), authority=_authority("decode:motor_test"))
+    assert decoder.calls == 0
+
+
+def test_verdict_enforcing_gate_admits_only_with_authority_and_verdicts():
+    decoder = _Decoder()
+    intent = lower_motor_action_intent("motor_test", _mv())
+    gate = VerdictEnforcingEfferentGate(_admitted_records(intent.intent_sha256))
+    reg = ModalityRegistry(efferent_gate=gate)
+    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
+
+
+def test_verdict_enforcing_trace_is_hash_only():
+    intent = lower_motor_action_intent("motor_test", _mv())
+    gate = VerdictEnforcingEfferentGate(_admitted_records(intent.intent_sha256))
+    authority = _authority("decode:motor_test")
+    verdict = gate.admit("motor_test", _mv(), authority)
+    trace = gate.trace("motor_test", _mv(), authority, verdict).as_dict()
+    assert trace["admitted"] is True
+    assert trace["intent_sha256"] == intent.intent_sha256
+    assert "decoded" not in str(trace)
+    assert "trajectory" not in str(trace)
+    assert "action_trace" not in str(trace)
+    for value in trace.values():
+        assert not isinstance(value, (np.ndarray, bytes, bytearray))