diff --git a/docs/decisions/ADR-0216-motor-verdict-lowering.md b/docs/decisions/ADR-0216-motor-verdict-lowering.md new file mode 100644 index 00000000..a5fefd04 --- /dev/null +++ b/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. diff --git a/sensorium/efferent.py b/sensorium/efferent.py index 0ae49173..3b819524 100644 --- a/sensorium/efferent.py +++ b/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", +] diff --git a/tests/test_efferent_gate.py b/tests/test_efferent_gate.py index e9074ab3..95cce8c7 100644 --- a/tests/test_efferent_gate.py +++ b/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))