From 1c7ea26ea20dbdfa32995da307c02a1842cc774d Mon Sep 17 00:00:00 2001 From: Shay Date: Fri, 17 Jul 2026 16:39:34 -0700 Subject: [PATCH] =?UTF-8?q?feat(adr-0244):=20Phase=202a=20=E2=80=94=20oper?= =?UTF-8?q?ator-preservation=20identity=20gate=20capability?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends core/physics/identity.py with the ADR-0244 §2.2/§4a wave-field gate as a pure, dual-mode capability (no runtime behavior change — the legacy path stays byte-identical when no wave field is supplied). - check(trajectory, manifold, *, wave_field=None, violated_boundary_ids= frozenset()): dual-mode. wave_field present -> operator-preservation gate; absent -> legacy scalar-L2 (ADR-0010). A MALFORMED wave field (non-finite / wrong shape / wrong byte-order) raises a typed ValueError via _validate_wave_field — it never silently falls back to legacy (the dual-mode fallback is for an ABSENT wave field only). - _wave_field_score: operator-preservation via IdentityManifoldGeometry (Phase 1). The trajectory is a versor (operator), so we measure whether it PRESERVES the value subspace: RMS subspace leakage (tilt toward alien dimensions) + signed self-alignment (in-subspace inversion). score = 1 - leakage_rms; flagged on score IdentityManifoldGeometry: + """Cached operator-preservation geometry for a set of value-axis directions. + + The identity subspace is frozen at pack load (ADR-0244 governance annotation + item 8), so the Gram matrix + inverse are computed once and memoized on the + canonical (hashable) axis-direction key. + """ + return IdentityManifoldGeometry.from_directions(directions) + + +def _geometry_for_manifold(manifold: "IdentityManifold") -> IdentityManifoldGeometry: + directions = tuple( + tuple(float(x) for x in getattr(axis, "direction", ()) or ()) + for axis in manifold.value_axes + ) + return _geometry_for_axis_directions(directions) + @dataclass(frozen=True) class ValueAxis: @@ -44,6 +91,19 @@ class IdentityScore: flagged: bool # True if any axis projection fell below alignment threshold deviation_axes: FrozenSet[str] # ValueAxis IDs where deviation was detected trajectory_id: str + # ADR-0244 §2.2 / §4a — operator-preservation wave-field measures. Populated + # only on the wave path (``wave_mode_active=True``); legacy defaults preserve + # the pre-ADR-0244 IdentityScore shape and all downstream serialization + # (the telemetry serializer emits these keys only when the wave path ran). + wave_mode_active: bool = False + # RMS subspace-leakage over the value axes (0.0 = every axis preserved). + leakage_norm: float = 0.0 + # Minimum signed self-alignment ⟨aᵢ, F aᵢ F̃⟩₀ across axes (+1 preserved, + # −1 inverted); 1.0 in legacy mode. + min_self_alignment: float = 1.0 + # Committed boundary_ids the turn violated (intersection with the manifold's + # boundary set); a non-empty set is a hard identity-boundary breach. + boundary_violations: FrozenSet[str] = frozenset() @property def value(self) -> float: @@ -170,17 +230,110 @@ class IdentityCheck: (0.75 * scalar_score) + (0.25 * directional_weight * coherence_term) ) - def check(self, trajectory, manifold: IdentityManifold | None = None) -> IdentityScore: + @staticmethod + def _validate_wave_field(wave_field) -> np.ndarray: + """Coerce + fail-closed-validate the live versor (ADR-0244 §4a). + + A malformed wave field (wrong shape, non-finite, wrong byte-order) is a + typed ``ValueError`` — it never silently falls back to the legacy + scalar-L2 path. The dual-mode fallback (see :meth:`check`) is for an + ABSENT wave field only, not a malformed one. + """ + F = np.ascontiguousarray(wave_field, dtype=np.float32) + if F.dtype.byteorder not in ("<", "="): + raise ValueError("identity gate requires little-endian float32") + if not np.all(np.isfinite(F)): + raise ValueError("identity gate encountered non-finite values in wave field") + if F.shape != (N_COMPONENTS,): + raise ValueError( + f"identity gate wave field must have shape ({N_COMPONENTS},), got {F.shape}" + ) + return F + + def _wave_field_score( + self, + wave_field, + manifold: IdentityManifold, + trajectory_id: str, + boundary_violations: FrozenSet[str], + ) -> IdentityScore: + """Operator-preservation identity score for a live versor (ADR-0244 §2.2/§4a). + + The trajectory is a versor (an operator); we measure whether it PRESERVES + the value subspace via its action on the axes ``F aᵢ F̃`` — subspace + leakage (tilt toward alien dimensions) plus signed self-alignment + (in-subspace inversion). See :mod:`core.physics.identity_manifold`. + """ + F = self._validate_wave_field(wave_field) + geometry = _geometry_for_manifold(manifold) + leakage, self_align = geometry.axis_response(F.astype(np.float64)) + leakage_rms = float((sum(l * l for l in leakage) / len(leakage)) ** 0.5) + min_align = float(min(self_align)) if self_align else 1.0 + score = self._clamp01(1.0 - leakage_rms) + threshold = float(manifold.alignment_threshold) + # Per-axis attribution consistent with the aggregate: an axis deviates + # when its own leakage exceeds the aggregate bound OR the versor has + # rotated it past orthogonal (toward inversion). + deviations = frozenset( + str(getattr(axis, "axis_id", getattr(axis, "name", "axis"))) + for axis, leak, align in zip(manifold.value_axes, leakage, self_align) + if leak > (1.0 - threshold) or align < _WAVE_SELF_ALIGNMENT_FLOOR + ) + flagged = ( + score < threshold + or min_align < _WAVE_SELF_ALIGNMENT_FLOOR + or bool(deviations) + or bool(boundary_violations) + ) + return IdentityScore( + score=score, + flagged=flagged, + deviation_axes=deviations, + trajectory_id=trajectory_id, + wave_mode_active=True, + leakage_norm=leakage_rms, + min_self_alignment=min_align, + boundary_violations=boundary_violations, + ) + + def check( + self, + trajectory, + manifold: IdentityManifold | None = None, + *, + wave_field=None, + violated_boundary_ids: FrozenSet[str] = frozenset(), + ) -> IdentityScore: + """Check a trajectory against the IdentityManifold (ADR-0010 / ADR-0244). + + Dual-mode (ADR-0244 §3): when a ``wave_field`` (the live versor + ``final_state.F``) is supplied, run the metric-exact operator-preservation + gate; otherwise fall back to the legacy scalar-L2 heuristic. A *malformed* + wave field raises (fail-closed) — only an ABSENT one falls back. + + ``violated_boundary_ids`` (the turn's safety/ethics violated boundaries) + is intersected with the manifold's committed ``boundary_ids``; a non-empty + intersection is a hard identity-boundary breach (governance annotation + item 7). Defaults empty so pre-ADR-0244 callers are byte-identical. + """ resolved_manifold = manifold or self._manifold if resolved_manifold is None: raise TypeError("IdentityCheck.check() requires an IdentityManifold") trajectory_id = str(getattr(trajectory, "trajectory_id", "legacy_trajectory")) + boundary_violations = ( + frozenset(violated_boundary_ids) & resolved_manifold.boundary_ids + ) if not resolved_manifold.value_axes: return IdentityScore( score=1.0, - flagged=False, + flagged=bool(boundary_violations), deviation_axes=frozenset(), trajectory_id=trajectory_id, + boundary_violations=boundary_violations, + ) + if wave_field is not None: + return self._wave_field_score( + wave_field, resolved_manifold, trajectory_id, boundary_violations ) confidence = float(getattr(trajectory, "total_coherence_delta", 0.0)) confidence += self._mean_frame_coherence(trajectory) @@ -192,11 +345,37 @@ class IdentityCheck: ) return IdentityScore( score=score, - flagged=bool(deviations), + flagged=bool(deviations) or bool(boundary_violations), deviation_axes=deviations, trajectory_id=trajectory_id, + boundary_violations=boundary_violations, ) + @staticmethod + def conjugate_correct( + score: IdentityScore, *, refuse: bool = False + ) -> IdentityScore: + """Conjugate corrector ``C_id`` + fail-closed egress (ADR-0244 §2.2/§4a). + + v1 policy is **admit-or-abstain**: it applies *no* corrective + displacement to the versor (a corrector that could rewrite reasoning to + force a low leakage score would be a "good-metric / bad-cognition" + defect, and silently mutating the field would violate the safety-pack + ``no_silent_correction`` boundary). When ``refuse=True`` and the score + is a violation, it abstains — raises :class:`IdentityGateRefusal`, + leaving the live parameters unchanged. Otherwise the score passes + through unmodified. A bounded geometric corrective displacement is a + future enhancement; v1 is the conservative bound (zero displacement). + """ + if refuse and IdentityCheck.would_violate(score): + raise IdentityGateRefusal( + f"identity gate refused trajectory {score.trajectory_id!r}: " + f"leakage={score.leakage_norm:.3f} min_self_align=" + f"{score.min_self_alignment:.3f} " + f"boundary_violations={sorted(score.boundary_violations)}" + ) + return score + @staticmethod def would_violate( score: IdentityScore | None, @@ -217,6 +396,13 @@ class IdentityCheck: return False if score.flagged: return True + # ADR-0244 §2.2 — a committed boundary breach or an inverted value axis + # is a violation independent of the aggregate score. Legacy defaults + # (empty boundary_violations, min_self_alignment=1.0) never trigger these. + if score.boundary_violations: + return True + if score.min_self_alignment < _WAVE_SELF_ALIGNMENT_FLOOR: + return True if manifold is not None and score.score < manifold.alignment_threshold: return True return False diff --git a/tests/test_adr_0244_identity_gate_wave.py b/tests/test_adr_0244_identity_gate_wave.py new file mode 100644 index 00000000..153d680c --- /dev/null +++ b/tests/test_adr_0244_identity_gate_wave.py @@ -0,0 +1,217 @@ +"""ADR-0244 §2.2/§4a — operator-preservation identity gate (dual-mode, fail-closed). + +Pins the wave-field path added to ``IdentityCheck``: dual-mode dispatch, fail-closed +validation of a malformed versor, the operator-preservation score, the +``boundary_ids`` intersection predicate, the admit-or-abstain ``C_id`` +(``IdentityGateRefusal``), and byte-compatible legacy behavior when no wave field +is supplied. +""" + +from __future__ import annotations + +import numpy as np +import pytest + +from algebra.cl41 import N_COMPONENTS +from core.physics.identity import ( + IdentityCheck, + IdentityGateRefusal, + IdentityManifold, + IdentityScore, + ValueAxis, +) + +_E12, _E14 = 6, 8 # e12, e14 bivector component indices + + +def _spatial_rotor(biv_idx: int, theta: float) -> np.ndarray: + R = np.zeros(N_COMPONENTS, dtype=np.float32) + R[0] = np.cos(theta / 2.0) + R[biv_idx] = np.sin(theta / 2.0) + return R + + +def _identity_versor() -> np.ndarray: + R = np.zeros(N_COMPONENTS, dtype=np.float32) + R[0] = 1.0 + return R + + +def _wave_manifold(threshold: float = 0.45, boundary_ids=frozenset()) -> IdentityManifold: + axes = ( + ValueAxis(name="truthfulness", direction=(1.0, 0.0, 0.0)), + ValueAxis(name="coherence", direction=(0.0, 1.0, 0.0)), + ValueAxis(name="reverence", direction=(0.0, 0.0, 1.0)), + ) + return IdentityManifold( + value_axes=axes, boundary_ids=boundary_ids, alignment_threshold=threshold + ) + + +class _Traj: + trajectory_id = "t-wave" + total_coherence_delta = 0.0 + frames = () + + +# --- dual-mode dispatch --------------------------------------------------- + +def test_absent_wave_field_uses_legacy_path(): + score = IdentityCheck().check(_Traj(), _wave_manifold()) + assert score.wave_mode_active is False + assert score.leakage_norm == 0.0 + assert score.min_self_alignment == 1.0 + + +def test_wave_field_activates_operator_preservation_path(): + score = IdentityCheck().check( + _Traj(), _wave_manifold(), wave_field=_identity_versor() + ) + assert score.wave_mode_active is True + + +# --- the operator-preservation score -------------------------------------- + +def test_identity_versor_scores_aligned_and_unflagged(): + score = IdentityCheck().check( + _Traj(), _wave_manifold(), wave_field=_identity_versor() + ) + assert score.leakage_norm < 1e-6 + assert score.min_self_alignment == pytest.approx(1.0, abs=1e-6) + assert score.score == pytest.approx(1.0, abs=1e-6) + assert score.flagged is False + assert score.deviation_axes == frozenset() + + +def test_tilt_toward_alien_dimension_raises_leakage_lowers_score(): + check = IdentityCheck() + manifold = _wave_manifold() + small = check.check(_Traj(), manifold, wave_field=_spatial_rotor(_E14, 0.5)) + large = check.check(_Traj(), manifold, wave_field=_spatial_rotor(_E14, 1.2)) + assert small.leakage_norm > 0.0 + assert large.leakage_norm > small.leakage_norm + assert large.score < small.score < 1.0 + + +def test_in_subspace_inversion_is_flagged_via_orientation(): + # π rotation in e12 sends e1 → −e1 (stays in the value subspace, leakage ~0), + # but inverts it — caught only by the signed self-alignment. + score = IdentityCheck().check( + _Traj(), _wave_manifold(), wave_field=_spatial_rotor(_E12, np.pi) + ) + assert score.leakage_norm < 1e-5 + assert score.min_self_alignment < -0.9 + assert score.flagged is True + assert "truthfulness" in score.deviation_axes + + +# --- fail-closed validation (malformed wave field never falls back) ------- + +def test_nonfinite_wave_field_raises_not_silent_legacy(): + bad = _identity_versor() + bad[3] = np.nan + with pytest.raises(ValueError, match="non-finite"): + IdentityCheck().check(_Traj(), _wave_manifold(), wave_field=bad) + + +def test_wrong_shape_wave_field_raises(): + with pytest.raises(ValueError, match="shape"): + IdentityCheck().check( + _Traj(), _wave_manifold(), wave_field=np.ones(16, dtype=np.float32) + ) + + +# --- boundary_ids intersection predicate ---------------------------------- + +def test_boundary_violation_in_manifold_flags_and_records(): + manifold = _wave_manifold(boundary_ids=frozenset({"no_identity_override"})) + score = IdentityCheck().check( + _Traj(), + manifold, + wave_field=_identity_versor(), + violated_boundary_ids=frozenset({"no_identity_override", "unrelated"}), + ) + assert score.boundary_violations == frozenset({"no_identity_override"}) + assert score.flagged is True # aligned versor, but a committed boundary fell + + +def test_boundary_violation_outside_manifold_is_ignored(): + manifold = _wave_manifold(boundary_ids=frozenset({"no_identity_override"})) + score = IdentityCheck().check( + _Traj(), + manifold, + wave_field=_identity_versor(), + violated_boundary_ids=frozenset({"some_other_boundary"}), + ) + assert score.boundary_violations == frozenset() + assert score.flagged is False + + +def test_boundary_predicate_works_on_legacy_path_too(): + manifold = _wave_manifold(boundary_ids=frozenset({"no_identity_override"})) + score = IdentityCheck().check( + _Traj(), manifold, violated_boundary_ids=frozenset({"no_identity_override"}) + ) + assert score.wave_mode_active is False + assert score.boundary_violations == frozenset({"no_identity_override"}) + assert score.flagged is True + + +# --- C_id: admit-or-abstain + IdentityGateRefusal ------------------------- + +def test_conjugate_correct_passes_clean_score_through(): + score = IdentityCheck().check( + _Traj(), _wave_manifold(), wave_field=_identity_versor() + ) + assert IdentityCheck.conjugate_correct(score, refuse=True) is score + + +def test_conjugate_correct_abstains_on_violation(): + score = IdentityCheck().check( + _Traj(), _wave_manifold(), wave_field=_spatial_rotor(_E12, np.pi) + ) + with pytest.raises(IdentityGateRefusal): + IdentityCheck.conjugate_correct(score, refuse=True) + + +def test_conjugate_correct_never_refuses_when_flag_off(): + score = IdentityCheck().check( + _Traj(), _wave_manifold(), wave_field=_spatial_rotor(_E12, np.pi) + ) + # refuse defaults False → admit-or-abstain gate does not fire; no mutation. + assert IdentityCheck.conjugate_correct(score) is score + + +# --- would_violate extensions + legacy back-compat ------------------------ + +def test_would_violate_catches_boundary_and_inversion(): + inverted = IdentityScore( + score=0.9, flagged=False, deviation_axes=frozenset(), trajectory_id="t", + wave_mode_active=True, min_self_alignment=-1.0, + ) + assert IdentityCheck.would_violate(inverted) is True + breach = IdentityScore( + score=1.0, flagged=False, deviation_axes=frozenset(), trajectory_id="t", + boundary_violations=frozenset({"no_identity_override"}), + ) + assert IdentityCheck.would_violate(breach) is True + + +def test_legacy_identity_score_still_constructs_without_new_fields(): + score = IdentityScore( + score=0.7, flagged=False, deviation_axes=frozenset(), trajectory_id="t" + ) + assert score.wave_mode_active is False + assert score.boundary_violations == frozenset() + assert IdentityCheck.would_violate(score) is False + + +def test_wave_score_is_deterministic(): + check = IdentityCheck() + manifold = _wave_manifold() + R = _spatial_rotor(_E14, 0.7) + a = check.check(_Traj(), manifold, wave_field=R) + b = check.check(_Traj(), manifold, wave_field=R) + assert (a.score, a.leakage_norm, a.min_self_alignment) == ( + b.score, b.leakage_norm, b.min_self_alignment + )