feat(adr-0244): Phase 2a — operator-preservation identity gate capability
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:
parent
c8c147a4e9
commit
1c7ea26ea2
2 changed files with 406 additions and 3 deletions
|
|
@ -12,11 +12,58 @@ CORE's identity is not a description of CORE. It is CORE, expressed geometricall
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
import functools
|
||||||
import math
|
import math
|
||||||
import warnings
|
import warnings
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import Dict, FrozenSet, List, Optional, Tuple
|
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)
|
@dataclass(frozen=True)
|
||||||
class ValueAxis:
|
class ValueAxis:
|
||||||
|
|
@ -44,6 +91,19 @@ class IdentityScore:
|
||||||
flagged: bool # True if any axis projection fell below alignment threshold
|
flagged: bool # True if any axis projection fell below alignment threshold
|
||||||
deviation_axes: FrozenSet[str] # ValueAxis IDs where deviation was detected
|
deviation_axes: FrozenSet[str] # ValueAxis IDs where deviation was detected
|
||||||
trajectory_id: str
|
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
|
@property
|
||||||
def value(self) -> float:
|
def value(self) -> float:
|
||||||
|
|
@ -170,17 +230,110 @@ class IdentityCheck:
|
||||||
(0.75 * scalar_score) + (0.25 * directional_weight * coherence_term)
|
(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
|
resolved_manifold = manifold or self._manifold
|
||||||
if resolved_manifold is None:
|
if resolved_manifold is None:
|
||||||
raise TypeError("IdentityCheck.check() requires an IdentityManifold")
|
raise TypeError("IdentityCheck.check() requires an IdentityManifold")
|
||||||
trajectory_id = str(getattr(trajectory, "trajectory_id", "legacy_trajectory"))
|
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:
|
if not resolved_manifold.value_axes:
|
||||||
return IdentityScore(
|
return IdentityScore(
|
||||||
score=1.0,
|
score=1.0,
|
||||||
flagged=False,
|
flagged=bool(boundary_violations),
|
||||||
deviation_axes=frozenset(),
|
deviation_axes=frozenset(),
|
||||||
trajectory_id=trajectory_id,
|
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 = float(getattr(trajectory, "total_coherence_delta", 0.0))
|
||||||
confidence += self._mean_frame_coherence(trajectory)
|
confidence += self._mean_frame_coherence(trajectory)
|
||||||
|
|
@ -192,11 +345,37 @@ class IdentityCheck:
|
||||||
)
|
)
|
||||||
return IdentityScore(
|
return IdentityScore(
|
||||||
score=score,
|
score=score,
|
||||||
flagged=bool(deviations),
|
flagged=bool(deviations) or bool(boundary_violations),
|
||||||
deviation_axes=deviations,
|
deviation_axes=deviations,
|
||||||
trajectory_id=trajectory_id,
|
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
|
@staticmethod
|
||||||
def would_violate(
|
def would_violate(
|
||||||
score: IdentityScore | None,
|
score: IdentityScore | None,
|
||||||
|
|
@ -217,6 +396,13 @@ class IdentityCheck:
|
||||||
return False
|
return False
|
||||||
if score.flagged:
|
if score.flagged:
|
||||||
return True
|
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:
|
if manifold is not None and score.score < manifold.alignment_threshold:
|
||||||
return True
|
return True
|
||||||
return False
|
return False
|
||||||
|
|
|
||||||
217
tests/test_adr_0244_identity_gate_wave.py
Normal file
217
tests/test_adr_0244_identity_gate_wave.py
Normal 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
|
||||||
|
)
|
||||||
Loading…
Reference in a new issue