feat(persistence): Shape B+ Phase A — bit-exact array codec + FieldState (de)serialize

Foundation for L10 resume-as-same-life persistence. Adds:
- core/array_codec.py: a leaf (numpy+base64) codec encoding arrays as
  {dtype, shape, b64(raw bytes)} — BIT-EXACT, never decimal. Float round-trips
  lose zero precision, so a restored versor keeps versor_condition < 1e-6 and a
  replayed turn keeps its trace_hash. dtype carries byte order; float32 is never
  conflated with float64.
- field/state.py: FieldState.to_dict/from_dict. Multivector arrays (F, holonomy)
  go through the byte codec; energy/valence round-trip exactly via JSON-safe
  helpers (lazy physics imports keep field/ cycle-free).

Exit gate (the scope's #1 risk, de-risked first): bit-exact round-trip AND
closure preserved — versor_condition(restored.F) == versor_condition(fs.F)
exactly. 10 codec/FieldState tests + 55 architectural-invariant/runtime tests
pass. Purely additive; no existing behavior changed.

Part of docs/analysis/L10-shapeBplus-persistence-scope-2026-06-05.md (Phase A).
This commit is contained in:
Shay 2026-06-05 11:38:58 -07:00
parent 457d1ae94e
commit 21e2ccb508
4 changed files with 319 additions and 1 deletions

54
core/array_codec.py Normal file
View file

@ -0,0 +1,54 @@
"""Bit-exact (de)serialization of numpy arrays for deterministic persistence.
A numpy array encodes to ``{"dtype", "shape", "b64"}`` where ``b64`` is base64 of
the array's raw bytes. This is **bit-exact**: every float round-trips with zero
precision loss, so a restored versor keeps ``versor_condition < 1e-6`` and a
replayed turn keeps its ``trace_hash``.
NEVER serialize field arrays as decimal/JSON floats. Decimal truncates the
mantissa and silently breaks both closure and deterministic replay the Cl(4,1)
float-truncation pitfall. ``dtype`` carries byte order (``'<f4'``/``'<f8'``), so
the encoding is portable, and ``float32`` is never conflated with ``float64``.
This module is a leaf: it imports only numpy + base64, so every layer (field,
vault, session, engine_state) can use it without an import cycle.
"""
from __future__ import annotations
import base64
from typing import Any
import numpy as np
def encode_array(arr: np.ndarray) -> dict[str, Any]:
"""Encode a numpy array to a bit-exact, JSON-safe dict."""
contiguous = np.ascontiguousarray(arr)
return {
"dtype": contiguous.dtype.str, # byte-order-aware, e.g. '<f4', '<f8', '<i4'
"shape": list(contiguous.shape),
"b64": base64.b64encode(contiguous.tobytes()).decode("ascii"),
}
def decode_array(payload: dict[str, Any]) -> np.ndarray:
"""Decode a payload produced by ``encode_array`` back to an exact array.
Returns a writable copy (``np.frombuffer`` is read-only) so the restored
array can be composed and mutated like a freshly-built one.
"""
dtype = np.dtype(payload["dtype"])
raw = base64.b64decode(payload["b64"])
flat = np.frombuffer(raw, dtype=dtype)
return flat.reshape(payload["shape"]).copy()
def encode_optional_array(arr: np.ndarray | None) -> dict[str, Any] | None:
"""Encode an array, or return ``None`` for ``None`` (e.g. optional holonomy)."""
return None if arr is None else encode_array(arr)
def decode_optional_array(payload: dict[str, Any] | None) -> np.ndarray | None:
"""Decode an optional-array payload, or return ``None`` for ``None``."""
return None if payload is None else decode_array(payload)

View file

@ -12,9 +12,16 @@ reference to the array passed in and expect coherence.
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, Any
import numpy as np import numpy as np
from core.array_codec import (
decode_array,
decode_optional_array,
encode_array,
encode_optional_array,
)
if TYPE_CHECKING: if TYPE_CHECKING:
from core.physics.energy import EnergyProfile from core.physics.energy import EnergyProfile
from core.physics.valence import ValenceBundle from core.physics.valence import ValenceBundle
@ -22,6 +29,82 @@ if TYPE_CHECKING:
_EXPECTED_COMPONENTS = 32 _EXPECTED_COMPONENTS = 32
def _encode_energy(energy: "EnergyProfile | None") -> dict[str, Any] | None:
if energy is None:
return None
return {
"raw": float(energy.raw),
"energy_class": energy.energy_class.value,
"convergence_density": int(energy.convergence_density),
"activation_count": int(energy.activation_count),
"last_activation_cycle": int(energy.last_activation_cycle),
"coherence_residual": float(energy.coherence_residual),
"aspect_weight": float(energy.aspect_weight),
"anchor_adjacent": bool(energy.anchor_adjacent),
}
def _decode_energy(payload: dict[str, Any] | None) -> "EnergyProfile | None":
if payload is None:
return None
from core.physics.energy import EnergyClass, EnergyProfile
return EnergyProfile(
raw=payload["raw"],
energy_class=EnergyClass(payload["energy_class"]),
convergence_density=payload["convergence_density"],
activation_count=payload["activation_count"],
last_activation_cycle=payload["last_activation_cycle"],
coherence_residual=payload["coherence_residual"],
aspect_weight=payload["aspect_weight"],
anchor_adjacent=payload["anchor_adjacent"],
)
def _encode_valence(valence: "ValenceBundle | None") -> dict[str, Any] | None:
if valence is None:
return None
return {
# sorted for deterministic serialization of the unordered frozenset
"affective": sorted(valence.affective),
"force": valence.force.value,
"emphasis": {
"focus_element": valence.emphasis.focus_element,
"mechanism": valence.emphasis.mechanism,
"degree": valence.emphasis.degree,
},
"polarity": {
"value": valence.polarity.value,
"kind": valence.polarity.kind,
},
"orientation": {
"direction": valence.orientation.direction,
"target": valence.orientation.target,
"preposition_source": valence.orientation.preposition_source,
},
}
def _decode_valence(payload: dict[str, Any] | None) -> "ValenceBundle | None":
if payload is None:
return None
from core.physics.valence import (
EmphasisProfile,
ForceClass,
OrientationSpec,
PolaritySpec,
ValenceBundle,
)
return ValenceBundle(
affective=frozenset(payload["affective"]),
force=ForceClass(payload["force"]),
emphasis=EmphasisProfile(**payload["emphasis"]),
polarity=PolaritySpec(**payload["polarity"]),
orientation=OrientationSpec(**payload["orientation"]),
)
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class FieldState: class FieldState:
F: np.ndarray # shape (32,) float32/float64 — Cl(4,1) multivector on the versor manifold F: np.ndarray # shape (32,) float32/float64 — Cl(4,1) multivector on the versor manifold
@ -70,6 +153,35 @@ class FieldState:
valence=self.valence, valence=self.valence,
) )
def to_dict(self) -> dict[str, Any]:
"""Serialize to a bit-exact, JSON-safe dict (Shape B+ persistence).
The multivector arrays (``F``, ``holonomy``) go through the byte-exact
array codec so ``versor_condition`` and ``trace_hash`` survive a
save/load cycle unchanged; scalar floats/strings on the energy/valence
side round-trip exactly through JSON.
"""
return {
"F": encode_array(self.F),
"node": int(self.node),
"step": int(self.step),
"holonomy": encode_optional_array(self.holonomy),
"energy": _encode_energy(self.energy),
"valence": _encode_valence(self.valence),
}
@classmethod
def from_dict(cls, payload: dict[str, Any]) -> FieldState:
"""Reconstruct a FieldState from ``to_dict`` output (exact round-trip)."""
return cls(
F=decode_array(payload["F"]),
node=int(payload["node"]),
step=int(payload["step"]),
holonomy=decode_optional_array(payload.get("holonomy")),
energy=_decode_energy(payload.get("energy")),
valence=_decode_valence(payload.get("valence")),
)
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class ManifoldState: class ManifoldState:

65
tests/test_array_codec.py Normal file
View file

@ -0,0 +1,65 @@
"""Bit-exact array codec — the foundation of Shape B+ persistence.
The whole resume-as-same-life flip depends on float arrays round-tripping with
ZERO precision loss: a versor that is valid (versor_condition < 1e-6) and a
trace_hash that is deterministic must both survive save->load unchanged. Decimal
JSON would truncate and break both, so the codec uses base64 of the raw bytes.
"""
from __future__ import annotations
import numpy as np
from core.array_codec import (
decode_array,
decode_optional_array,
encode_array,
encode_optional_array,
)
def test_float32_round_trips_bit_exact() -> None:
a = np.array([1.0, -2.5, 3.1415927, 1e-7, 1e30, 0.0], dtype=np.float32)
decoded = decode_array(encode_array(a))
assert decoded.dtype == np.float32
assert decoded.shape == a.shape
# Bit-exact: the raw bytes are identical, not just "close".
assert decoded.tobytes() == a.tobytes()
assert np.array_equal(decoded, a)
def test_float64_round_trips_bit_exact_and_preserves_dtype() -> None:
a = np.array([np.pi, np.e, 1e-300, -1e300], dtype=np.float64)
decoded = decode_array(encode_array(a))
assert decoded.dtype == np.float64 # float32 vs float64 must NOT be conflated
assert decoded.tobytes() == a.tobytes()
def test_int32_2d_shape_round_trips() -> None:
edges = np.array([[0, 1], [1, 2], [2, 0]], dtype=np.int32)
decoded = decode_array(encode_array(edges))
assert decoded.dtype == np.int32
assert decoded.shape == (3, 2)
assert np.array_equal(decoded, edges)
def test_encoding_is_not_decimal() -> None:
# The payload must carry base64 bytes, never decimal floats (which truncate).
payload = encode_array(np.array([0.1], dtype=np.float32))
assert set(payload.keys()) == {"dtype", "shape", "b64"}
assert isinstance(payload["b64"], str)
assert "0.1" not in payload["b64"]
def test_decoded_array_is_writable_copy() -> None:
# frombuffer returns a read-only view; the codec must hand back a copy so
# the restored field can be composed/mutated like a fresh one.
decoded = decode_array(encode_array(np.zeros(4, dtype=np.float32)))
decoded[0] = 1.0 # must not raise
def test_optional_array_handles_none() -> None:
assert encode_optional_array(None) is None
assert decode_optional_array(None) is None
a = np.array([1.0, 2.0], dtype=np.float32)
assert np.array_equal(decode_optional_array(encode_optional_array(a)), a)

View file

@ -0,0 +1,87 @@
"""FieldState round-trip — Shape B+ Phase A.
The restored field must be BIT-EXACT (so versor_condition < 1e-6 survives and the
replayed turn keeps its trace_hash) and FAITHFUL (node/step/holonomy/energy/
valence all preserved). Scalar floats and strings round-trip exactly through
JSON; only the multivector arrays use the byte-exact codec.
"""
from __future__ import annotations
import json
import numpy as np
from algebra.versor import versor_condition
from core.physics.energy import EnergyClass, EnergyProfile
from core.physics.valence import EmphasisProfile, ForceClass, ValenceBundle
from field.state import FieldState
def _identity_versor() -> np.ndarray:
# Scalar 1, rest 0 — a valid unit versor (versor_condition == 0 exactly).
f = np.zeros(32, dtype=np.float32)
f[0] = 1.0
return f
def _populated_fieldstate(dtype=np.float32) -> FieldState:
return FieldState(
F=_identity_versor().astype(dtype),
node=5,
step=3,
holonomy=np.full(32, 0.0123456789, dtype=dtype),
energy=EnergyProfile(
raw=1.5,
energy_class=EnergyClass.E2,
activation_count=2,
coherence_residual=3.5e-7,
anchor_adjacent=True,
),
valence=ValenceBundle(
affective=frozenset({"joy", "calm"}),
force=ForceClass.INTERROGATIVE,
emphasis=EmphasisProfile(focus_element="x", mechanism="cleft", degree="high"),
),
)
def test_fieldstate_round_trips_bit_exact_and_preserves_closure() -> None:
fs = _populated_fieldstate()
restored = FieldState.from_dict(fs.to_dict())
# Bit-exact arrays.
assert restored.F.tobytes() == fs.F.tobytes()
assert restored.F.dtype == fs.F.dtype
assert restored.holonomy is not None
assert restored.holonomy.tobytes() == fs.holonomy.tobytes()
# Faithful scalars / nested objects (frozen dataclass equality).
assert restored.node == fs.node
assert restored.step == fs.step
assert restored.energy == fs.energy
assert restored.valence == fs.valence
# Closure preserved EXACTLY (the load-bearing property).
assert versor_condition(restored.F) == versor_condition(fs.F)
assert versor_condition(restored.F) < 1e-6
def test_fieldstate_round_trip_is_json_safe() -> None:
fs = _populated_fieldstate()
restored = FieldState.from_dict(json.loads(json.dumps(fs.to_dict())))
assert restored.F.tobytes() == fs.F.tobytes()
assert restored.valence == fs.valence
def test_fieldstate_preserves_float64_dtype() -> None:
fs = _populated_fieldstate(dtype=np.float64)
restored = FieldState.from_dict(fs.to_dict())
assert restored.F.dtype == np.float64 # float32 must never be conflated
assert restored.F.tobytes() == fs.F.tobytes()
def test_fieldstate_round_trips_with_none_optionals() -> None:
fs = FieldState(F=_identity_versor(), node=0, step=0)
restored = FieldState.from_dict(fs.to_dict())
assert restored.holonomy is None
assert restored.energy is None
assert restored.valence is None
assert restored.F.tobytes() == fs.F.tobytes()