core-labs/core
core-labs/core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

feat(adr-0244): Phase 2a — operator-preservation identity gate capability

Browse source 
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<threshold OR inversion OR boundary.
  - IdentityScore extended (wave_mode_active, leakage_norm,
    min_self_alignment, boundary_violations) — all defaulted, legacy shape
    and construction preserved.
  - boundary_ids predicate: violated_boundary_ids & manifold.boundary_ids;
    a non-empty intersection is a hard identity-boundary breach (activates
    the previously-dormant boundary_ids; item 7).
  - IdentityGateRefusal + conjugate_correct (C_id): v1 admit-or-abstain, NO
    silent correction of the versor (honors the safety-pack
    no_silent_correction boundary); abstains (raises) on violation.
  - would_violate extended for boundary + inversion (legacy defaults never
    trigger the new branches).

identity_manifold is now imported by identity.py (serve-safe pure algebra);
serve-quarantine transitive test stays green.

tests/test_adr_0244_identity_gate_wave.py (16 tests): dual-mode dispatch,
identity-versor alignment, alien-dimension tilt leakage, in-subspace
inversion caught via orientation, fail-closed malformed validation, the
boundary intersection predicate (in/out of manifold, legacy path too),
C_id admit-or-abstain, would_violate extensions, legacy back-compat,
determinism.

[Verification]: 16 new tests passed; legacy identity + quarantine +
telemetry-fanout suite 54 passed (legacy behavior + serve-quarantine
preserved); in-worktree smoke 176 passed; fast lane (-m 'not quarantine
and not slow' -n auto) 11879 passed, 109 skipped.
This commit is contained in:
Shay 2026-07-17 16:39:34 -07:00
parent c8c147a4e9
commit 1c7ea26ea2
2 changed files with 406 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

192
core/physics/identity.py
View file

@ -12,11 +12,58 @@ CORE's identity is not a description of CORE. It is CORE, expressed geometricall
"""
from __future__ import annotations
import functools
import math
import warnings
from dataclasses import dataclass
from typing import Dict, FrozenSet, List, Optional, Tuple
import numpy as np
from algebra.cl41 import N_COMPONENTS
from core.physics.identity_manifold import IdentityManifoldGeometry
# ADR-0244 §2.2 / §4a — provisional wave-gate thresholds. The leakage bound
# reuses ``IdentityManifold.alignment_threshold``; the orientation floor flags a
# value axis the versor has rotated *past orthogonal* (toward inversion). Both
# are calibrated against reference traces in D4 Phase 3 (γ_id); until then the
# wave gate is flag-gated off in the runtime so these provisional values never
# change live behavior.
_WAVE_SELF_ALIGNMENT_FLOOR: float = 0.0
class IdentityGateRefusal(Exception):
"""Fail-closed identity-gate refusal (ADR-0244 §2.2 / §4a).
Raised when the operator-preservation gate cannot admit a trajectory —
subspace leakage over bound, a value axis inverted, or a committed
``boundary_id`` violated — and the conjugate corrector ``C_id`` cannot
recover alignment within its bound. The live parameters are kept unchanged
(no silent correction — this honors the safety-pack ``no_silent_correction``
boundary).
"""
@functools.lru_cache(maxsize=32)
def _geometry_for_axis_directions(
directions: Tuple[Tuple[float, ...], ...]
) -> 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

217
tests/test_adr_0244_identity_gate_wave.py Normal file
View file

@ -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
)