feat(adr-0244): Phase 2b — wire the operator-preservation gate into serve (flag-gated)
Wires the ADR-0244 §2.2 wave gate into the live turn loop behind a new
opt-in flag, byte-identical when off, plus a normalization fix found in
integration.
core/config.py: RuntimeConfig.identity_wave_gate (bool = False). OFF by
default — the leakage threshold is provisional (reuses alignment_threshold)
until calibrated to gamma_id in Phase 3, and flag-off is byte-identical to
the pre-ADR-0244 advisory path.
chat/runtime.py (main chat() path, two flag-gated touches):
- at the identity check, pass wave_field=result.final_state.F when the
flag is on (so hedging + telemetry both use the wave score); None when
off -> legacy scalar-L2 path.
- after the safety/ethics verdicts exist, intersect their violated
boundaries with the manifold's committed boundary_ids (supplement via
dataclasses.replace) and fold a fail-closed geometric IdentityGateRefusal
into the typed refusal surface (TYPED_REFUSAL_PREFIX). refusal_emitted is
derived once, after the block, preserving the (surface is not None) <=>
emitted invariant.
chat/telemetry.py: serialize_turn_event emits the wave-gate keys
(identity_wave_mode / _leakage_norm / _min_self_alignment /
_boundary_violations) only when identity_score.wave_mode_active — so the
wire format is byte-identical when the gate is off.
core/physics/identity_manifold.py — NORMALIZATION FIX (found in
integration): the live versor carries boost (e5) components, and a boost is
a unit versor (R R~ = 1) that does NOT preserve the Euclidean coefficient
norm, so ‖R a_i R~‖ > 1 and the un-normalized leakage/self-alignment ran
past their ranges (measured leakage 5.16, self_align -4.75 on a real turn).
axis_response now normalizes each measure by the rotated-axis magnitude:
leakage is a fraction in [0,1], self_align a signed cosine in [-1,1]. For a
norm-preserving spatial rotor the rotated axis is unit and normalization is
a no-op (Phase 1 results unchanged). §4a updated to match; new boost test
pins bounded measures.
tests/test_adr_0244_identity_gate_runtime.py (new): flag OFF -> legacy
score + no wave telemetry (byte-identical wire); flag ON -> wave gate active
on the live versor + wave telemetry present + bounded measures; flag-off
determinism. Seeds a short sequence to reach main-path (identity-checked)
turns (a fresh empty-vault runtime routes ungrounded inputs to disclosure).
[Verification]: 74 targeted (manifold+gate+runtime+legacy identity+
telemetry) passed; in-worktree smoke 176 passed; fast lane
(-m 'not quarantine and not slow' -n auto) 11883 passed, 109 skipped —
flag-off byte-identity confirmed across the whole suite.
This commit is contained in:
parent
3ea748b343
commit
c7e2b3b68f
7 changed files with 192 additions and 17 deletions
|
|
@ -28,6 +28,7 @@ from chat.teaching_grounding import (
|
|||
TEACHING_CORPUS_ID as _TEACHING_CORPUS_ID,
|
||||
)
|
||||
from chat.refusal import (
|
||||
TYPED_REFUSAL_PREFIX,
|
||||
build_hedge_prefix,
|
||||
build_refusal_surface,
|
||||
inject_hedge,
|
||||
|
|
@ -2676,7 +2677,19 @@ class ChatRuntime:
|
|||
# --- end articulation fidelity ---
|
||||
|
||||
reasoning_trajectory = _make_trajectory_from_result(result, self._context.turn)
|
||||
identity_score = self._identity_check.check(reasoning_trajectory, self.identity_manifold)
|
||||
# ADR-0244 §2.2 — operator-preservation identity gate (flag-gated). When
|
||||
# on, the check runs the metric-exact wave-field gate on the live versor
|
||||
# final_state.F; when off, wave_field=None selects the legacy scalar-L2
|
||||
# path (byte-identical). The boundary_ids intersection needs the
|
||||
# safety/ethics verdicts, which are computed below — it is supplemented
|
||||
# after those run.
|
||||
identity_score = self._identity_check.check(
|
||||
reasoning_trajectory,
|
||||
self.identity_manifold,
|
||||
wave_field=(
|
||||
result.final_state.F if self.config.identity_wave_gate else None
|
||||
),
|
||||
)
|
||||
flagged = identity_score.flagged
|
||||
cycle_cost = CycleCost(
|
||||
cycle_index=self._context.turn,
|
||||
|
|
@ -2738,6 +2751,30 @@ class ChatRuntime:
|
|||
refusal_surface = build_refusal_surface(
|
||||
safety_verdict, ethics_verdict, self.ethics_pack,
|
||||
)
|
||||
# ADR-0244 §2.2 — supplement the identity gate now that the safety/ethics
|
||||
# verdicts exist: intersect the turn's violated boundaries with the
|
||||
# manifold's committed boundary_ids, and fold a fail-closed geometric
|
||||
# IdentityGateRefusal into the typed refusal surface. Flag-gated OFF by
|
||||
# default → skipped → byte-identical to the pre-ADR-0244 path.
|
||||
# ``refusal_emitted`` is derived from ``refusal_surface`` after this
|
||||
# block so the (surface is not None) ⇔ emitted invariant is preserved.
|
||||
if self.config.identity_wave_gate:
|
||||
_violated_boundaries: frozenset[str] = (
|
||||
frozenset(getattr(safety_verdict, "violated_boundaries", ()) or ())
|
||||
| frozenset(getattr(ethics_verdict, "violated_commitments", ()) or ())
|
||||
)
|
||||
_boundary_breach = (
|
||||
_violated_boundaries & self.identity_manifold.boundary_ids
|
||||
)
|
||||
if _boundary_breach:
|
||||
identity_score = replace(
|
||||
identity_score,
|
||||
boundary_violations=_boundary_breach,
|
||||
flagged=True,
|
||||
)
|
||||
flagged = True
|
||||
if refusal_surface is None and IdentityCheck.would_violate(identity_score):
|
||||
refusal_surface = TYPED_REFUSAL_PREFIX + "identity:wave_gate"
|
||||
refusal_emitted = refusal_surface is not None
|
||||
hedge_injected = False
|
||||
warm_grounding_source: str | None = None
|
||||
|
|
|
|||
|
|
@ -137,6 +137,20 @@ def serialize_turn_event(
|
|||
out["identity_deviation_axes"] = sorted(
|
||||
getattr(identity_score, "deviation_axes", ()) or ()
|
||||
)
|
||||
# ADR-0244 §2.2 — operator-preservation wave-gate telemetry. Emitted only
|
||||
# when the wave path ran (config.identity_wave_gate on); absent otherwise,
|
||||
# so the pre-ADR-0244 wire format stays byte-identical when the gate is off.
|
||||
if getattr(identity_score, "wave_mode_active", False):
|
||||
out["identity_wave_mode"] = True
|
||||
out["identity_leakage_norm"] = float(
|
||||
getattr(identity_score, "leakage_norm", 0.0)
|
||||
)
|
||||
out["identity_min_self_alignment"] = float(
|
||||
getattr(identity_score, "min_self_alignment", 1.0)
|
||||
)
|
||||
out["identity_boundary_violations"] = sorted(
|
||||
getattr(identity_score, "boundary_violations", ()) or ()
|
||||
)
|
||||
if include_content:
|
||||
out["input_tokens"] = list(getattr(event, "input_tokens", ()))
|
||||
out["surface"] = str(getattr(event, "surface", ""))
|
||||
|
|
|
|||
|
|
@ -292,6 +292,17 @@ class RuntimeConfig:
|
|||
# wanting a hard identity-continuity guarantee opt in.
|
||||
strict_identity_continuity: bool = False
|
||||
|
||||
# ADR-0244 §2.2 / §4a — operator-preservation identity gate. When on, the
|
||||
# per-turn identity check runs the metric-exact wave-field gate on the live
|
||||
# versor (final_state.F): subspace-leakage + signed self-alignment via
|
||||
# F aᵢ F̃, plus the boundary_ids intersection with the turn's safety/ethics
|
||||
# violations, and a fail-closed IdentityGateRefusal folded into the typed
|
||||
# refusal surface. OFF by default: the leakage threshold is provisional
|
||||
# (reuses alignment_threshold) until calibrated to γ_id in D4 Phase 3, and
|
||||
# the flag-off path is byte-identical to the pre-ADR-0244 advisory behavior
|
||||
# (legacy scalar-L2 identity score, no geometric refusal).
|
||||
identity_wave_gate: bool = False
|
||||
|
||||
# Step B (inline realization) — when on, each turn ACCRUES knowledge into the
|
||||
# held self: a comprehensible declarative turn is realized into the session vault
|
||||
# (SPECULATIVE, as-told), and a comprehensible question turn is determined over
|
||||
|
|
|
|||
|
|
@ -187,31 +187,48 @@ class IdentityManifoldGeometry:
|
|||
) -> tuple[list[float], list[float]]:
|
||||
"""Per-axis operator-preservation measures for ``versor``.
|
||||
|
||||
Returns ``(leakage, self_align)`` — parallel lists over the value axes:
|
||||
Returns ``(leakage, self_align)`` — parallel lists over the value axes,
|
||||
both **scale-invariant** (normalized by the rotated axis magnitude):
|
||||
|
||||
* ``leakage[i]`` = ``‖R aᵢ R̃ − P_I(R aᵢ R̃)‖₂`` (subspace departure;
|
||||
catches tilt toward alien dimensions e4/e5).
|
||||
* ``self_align[i]`` = ``⟨aᵢ, R aᵢ R̃⟩₀`` (signed orientation; catches
|
||||
in-subspace inversion — ``e1 → −e1`` gives leakage 0 but −1 here).
|
||||
* ``leakage[i]`` = ``‖R aᵢ R̃ − P_I(R aᵢ R̃)‖₂ / ‖R aᵢ R̃‖₂`` — the
|
||||
*fraction* of the rotated axis outside the value subspace, in
|
||||
``[0, 1]``; catches tilt toward alien dimensions e4/e5.
|
||||
* ``self_align[i]`` = ``⟨aᵢ, R aᵢ R̃⟩₀ / (‖aᵢ‖₂ ‖R aᵢ R̃‖₂)`` — a signed
|
||||
cosine in ``[−1, 1]``; catches in-subspace inversion (``e1 → −e1``
|
||||
gives leakage 0 but −1 here).
|
||||
|
||||
Normalization is load-bearing: a versor with a boost (e5) component does
|
||||
NOT preserve the Euclidean coefficient norm (``‖R aᵢ R̃‖₂`` can exceed 1
|
||||
even though ``R R̃ = 1`` in the Minkowski sense), so an un-normalized
|
||||
magnitude would be unbounded. For a norm-preserving spatial rotor the
|
||||
rotated axis is unit and normalization is a no-op.
|
||||
"""
|
||||
versor = np.asarray(versor, dtype=np.float64)
|
||||
leakage: list[float] = []
|
||||
self_align: list[float] = []
|
||||
for axis in self.axes_psi:
|
||||
axis_norm = euclidean_norm(axis)
|
||||
rotated = sandwich(versor, axis)
|
||||
rot_norm = euclidean_norm(rotated)
|
||||
if rot_norm <= 0.0 or axis_norm <= 0.0:
|
||||
# Degenerate: the versor annihilated the axis — treat as full
|
||||
# leakage / zero alignment (fail-closed).
|
||||
leakage.append(1.0)
|
||||
self_align.append(0.0)
|
||||
continue
|
||||
rejection = rotated - subspace_project(
|
||||
rotated, self.axes_psi, self.gram_inv
|
||||
)
|
||||
leakage.append(euclidean_norm(rejection))
|
||||
self_align.append(_inner0(axis, rotated))
|
||||
leakage.append(euclidean_norm(rejection) / rot_norm)
|
||||
self_align.append(_inner0(axis, rotated) / (axis_norm * rot_norm))
|
||||
return leakage, self_align
|
||||
|
||||
def leakage_rms(self, versor: np.ndarray) -> float:
|
||||
"""Root-mean-square subspace leakage over all axes.
|
||||
"""Root-mean-square per-axis subspace-leakage fraction over all axes.
|
||||
|
||||
Each rotated axis is unit-norm (a versor preserves norm), so this is the
|
||||
aggregate subspace-departure fraction in ``[0, 1]``; the Phase-2 gate's
|
||||
``score`` is ``1 − leakage_rms``.
|
||||
Each per-axis leakage is a fraction in ``[0, 1]`` (see
|
||||
:meth:`axis_response`), so this aggregate is also in ``[0, 1]``; the
|
||||
Phase-2 gate's ``score`` is ``1 − leakage_rms``.
|
||||
"""
|
||||
leakage, _ = self.axis_response(versor)
|
||||
return float((sum(value * value for value in leakage) / len(leakage)) ** 0.5)
|
||||
|
|
|
|||
|
|
@ -475,14 +475,21 @@ def euclidean_norm(s: np.ndarray) -> float:
|
|||
|
||||
def axis_response(R, axes_psi, g_inv):
|
||||
"""Per-axis operator-preservation measures for versor R. For each value
|
||||
axis a_i:
|
||||
rotated_i = sandwich(R, a_i) # grade-1 unit vector
|
||||
axis a_i (both measures NORMALIZED by the rotated-axis magnitude, so they
|
||||
are scale-invariant):
|
||||
rotated_i = sandwich(R, a_i) # grade-1 vector
|
||||
rot_norm_i = euclidean_norm(rotated_i)
|
||||
rejection_i = rotated_i - subspace_project(rotated_i) # out-of-I component
|
||||
leakage_i = euclidean_norm(rejection_i) # subspace departure (Sec 2.2)
|
||||
self_align_i = <a_i, rotated_i>_0 # SIGNED orientation (item 4)
|
||||
leakage_i = euclidean_norm(rejection_i) / rot_norm_i # fraction in [0,1]
|
||||
self_align_i = <a_i, rotated_i>_0 / (norm(a_i)*rot_norm_i) # signed cosine [-1,1]
|
||||
Returns (leakage[], self_align[]). Both are needed and non-redundant:
|
||||
rejection catches tilt toward alien dimensions (e4/e5); self_align catches
|
||||
in-subspace inversion (e1 -> -e1: leakage 0 but self_align -1)."""
|
||||
in-subspace inversion (e1 -> -e1: leakage 0 but self_align -1).
|
||||
NORMALIZATION IS LOAD-BEARING: the live versor carries boost (e5)
|
||||
components, and a boost is a unit versor (R R~ = 1) that does NOT preserve
|
||||
the Euclidean coefficient norm (‖R a_i R~‖ > 1), so an un-normalized
|
||||
magnitude would be unbounded. For a norm-preserving spatial rotor the
|
||||
rotated axis is unit and normalization is a no-op."""
|
||||
leak, align = [], []
|
||||
for a in axes_psi:
|
||||
rot = sandwich(R, a)
|
||||
|
|
|
|||
73
tests/test_adr_0244_identity_gate_runtime.py
Normal file
73
tests/test_adr_0244_identity_gate_runtime.py
Normal file
|
|
@ -0,0 +1,73 @@
|
|||
"""ADR-0244 §2.2 — runtime wiring of the operator-preservation identity gate.
|
||||
|
||||
Validates the flag-gated wiring in ``chat/runtime.py``:
|
||||
* flag OFF (default) → legacy identity score, no wave telemetry (byte-identical
|
||||
wire format);
|
||||
* flag ON → the wave gate runs on the live versor ``final_state.F``, the score
|
||||
is wave-mode with real leakage/orientation, and the telemetry serializer
|
||||
surfaces the wave keys.
|
||||
|
||||
The per-turn identity gate lives on the main generation path; a fresh empty-vault
|
||||
runtime routes ungrounded inputs to the disclosure path (``identity_score=None``),
|
||||
so the tests seed a short sequence to reach main-path turns.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from chat.runtime import ChatRuntime
|
||||
from chat.telemetry import serialize_turn_event
|
||||
from core.config import RuntimeConfig
|
||||
|
||||
# A seeded sequence that reliably produces main-path (identity-checked) turns.
|
||||
_SEQUENCE = ("water boils", "water boils", "birds fly", "birds fly")
|
||||
|
||||
|
||||
def _main_path_events(flag: bool):
|
||||
runtime = ChatRuntime(
|
||||
config=RuntimeConfig(identity_wave_gate=flag), no_load_state=True
|
||||
)
|
||||
events = []
|
||||
for text in _SEQUENCE:
|
||||
runtime.chat(text)
|
||||
event = runtime.turn_log[-1]
|
||||
if event.identity_score is not None:
|
||||
events.append(event)
|
||||
assert events, "expected at least one main-path (identity-checked) turn"
|
||||
return events
|
||||
|
||||
|
||||
def test_flag_off_scores_are_legacy_no_wave_telemetry():
|
||||
for event in _main_path_events(False):
|
||||
score = event.identity_score
|
||||
assert score.wave_mode_active is False
|
||||
assert score.leakage_norm == 0.0
|
||||
assert score.min_self_alignment == 1.0
|
||||
payload = serialize_turn_event(event)
|
||||
assert "identity_wave_mode" not in payload
|
||||
assert "identity_leakage_norm" not in payload
|
||||
assert "identity_min_self_alignment" not in payload
|
||||
assert "identity_boundary_violations" not in payload
|
||||
# legacy identity telemetry unchanged
|
||||
assert "identity_alignment" in payload
|
||||
assert "identity_flagged" in payload
|
||||
|
||||
|
||||
def test_flag_on_activates_wave_gate_with_telemetry():
|
||||
for event in _main_path_events(True):
|
||||
score = event.identity_score
|
||||
assert score.wave_mode_active is True
|
||||
assert 0.0 <= score.leakage_norm <= 1.0
|
||||
assert -1.0 <= score.min_self_alignment <= 1.0
|
||||
payload = serialize_turn_event(event)
|
||||
assert payload.get("identity_wave_mode") is True
|
||||
assert "identity_leakage_norm" in payload
|
||||
assert "identity_min_self_alignment" in payload
|
||||
assert isinstance(payload["identity_boundary_violations"], list)
|
||||
|
||||
|
||||
def test_flag_off_is_deterministic_across_runs():
|
||||
# The flag-off path is byte-identical run to run (the fast lane pins that it
|
||||
# is also byte-identical to the pre-ADR-0244 baseline).
|
||||
first = [e.surface for e in _main_path_events(False)]
|
||||
second = [e.surface for e in _main_path_events(False)]
|
||||
assert first == second
|
||||
|
|
@ -202,6 +202,22 @@ def test_boost_toward_e5_leaks():
|
|||
assert leakage[0] > 0.05
|
||||
|
||||
|
||||
def test_boost_measures_stay_bounded_despite_non_unit_rotation():
|
||||
# A boost is a unit versor (b·b̃ = 1) but does NOT preserve the Euclidean
|
||||
# coefficient norm — ‖b aᵢ b̃‖₂ > 1. The per-axis normalization keeps the
|
||||
# leakage fraction in [0,1] and the signed alignment in [−1,1]; without it
|
||||
# (the earlier bug) leakage and |self_align| ran well past 1.
|
||||
geom = _geom()
|
||||
for theta in (0.5, 1.5, 2.5):
|
||||
leakage, self_align = geom.axis_response(_boost(_E15, theta))
|
||||
assert all(0.0 <= v <= 1.0 for v in leakage), (theta, leakage)
|
||||
assert all(-1.0 <= v <= 1.0 for v in self_align), (theta, self_align)
|
||||
# larger boost tilts the e1 axis further out of the value subspace
|
||||
small, _ = geom.axis_response(_boost(_E15, 0.5))
|
||||
large, _ = geom.axis_response(_boost(_E15, 1.5))
|
||||
assert large[0] > small[0]
|
||||
|
||||
|
||||
def test_larger_tilt_leaks_more():
|
||||
geom = _geom()
|
||||
small = geom.leakage_rms(_spatial_rotor(_E14, 0.5))
|
||||
|
|
|
|||
Loading…
Reference in a new issue