diff --git a/field/operators.py b/field/operators.py new file mode 100644 index 00000000..1eea74c3 --- /dev/null +++ b/field/operators.py @@ -0,0 +1,66 @@ +""" +Manifold-level field operators — graph diffusion and protocol. + +Operators transform ManifoldState through algebraic transitions. +construction_seed_versor is used here as a construction primitive (building new +versors from damped blends), not as propagation repair. +""" + +from __future__ import annotations + +from typing import Protocol + +import numpy as np + +from algebra.backend import versor_apply +from algebra.rotor import word_transition_rotor +from algebra.versor import construction_seed_versor +from field.state import ManifoldState + + +class Operator(Protocol): + """Protocol for manifold field operators.""" + + def forward(self, state: ManifoldState) -> tuple[ManifoldState, float]: + """Apply operator, return (new_state, delta_norm).""" + ... + + def adjoint(self) -> Operator: + """Return the adjoint operator.""" + ... + + +class GraphDiffusionOperator: + """Propagate geometric pressure across graph edges via damped versor transitions. + + Self-adjoint: adjoint() returns self (symmetric diffusion). + Uses construction-tier construction_seed_versor for post-damping closure. + """ + + def __init__(self, damping: float = 0.5) -> None: + if not 0.0 < damping <= 1.0: + raise ValueError(f"damping must be in (0, 1], got {damping}") + self._damping = damping + + def forward(self, state: ManifoldState) -> tuple[ManifoldState, float]: + old_fields = state.fields + new_fields = old_fields.copy() + + for edge_idx in range(state.edges.shape[0]): + src, dst = int(state.edges[edge_idx, 0]), int(state.edges[edge_idx, 1]) + try: + V = word_transition_rotor(old_fields[src], old_fields[dst]) + except ValueError: + continue + diffused = versor_apply(V, old_fields[dst]) + blended = self._damping * diffused + (1.0 - self._damping) * old_fields[dst] + try: + new_fields[dst] = construction_seed_versor(blended) + except ValueError: + new_fields[dst] = old_fields[dst] + + delta = float(np.linalg.norm(new_fields - old_fields)) + return ManifoldState(fields=new_fields, edges=state.edges, step=state.step + 1), delta + + def adjoint(self) -> GraphDiffusionOperator: + return self diff --git a/field/state.py b/field/state.py index 7120a91c..9323aa9b 100644 --- a/field/state.py +++ b/field/state.py @@ -69,3 +69,54 @@ class FieldState: energy=self.energy, valence=self.valence, ) + + +@dataclass(frozen=True, slots=True) +class ManifoldState: + """Field over a graph topology — one versor per node, with edge connectivity. + + Invariant: versor_condition(fields[i]) < 1e-6 for every node i. + """ + + fields: np.ndarray # (N, 32) float32 — one Cl(4,1) versor per node + edges: np.ndarray # (E, 2) int32 — directed edge list + step: int = 0 + + def __post_init__(self) -> None: + from algebra.backend import versor_condition + + F = np.array(self.fields, dtype=np.float32).copy() + if F.ndim != 2 or F.shape[1] != _EXPECTED_COMPONENTS: + raise ValueError( + f"ManifoldState.fields must have shape (N, {_EXPECTED_COMPONENTS}), " + f"got {F.shape}." + ) + object.__setattr__(self, "fields", F) + + E = np.array(self.edges, dtype=np.int32).copy() + if E.ndim != 2 or E.shape[1] != 2: + raise ValueError( + f"ManifoldState.edges must have shape (E, 2), got {E.shape}." + ) + n_nodes = F.shape[0] + if E.size > 0 and (E.min() < 0 or E.max() >= n_nodes): + raise ValueError( + f"Edge indices must be in [0, {n_nodes}), " + f"got range [{E.min()}, {E.max()}]." + ) + object.__setattr__(self, "edges", E) + + for i in range(n_nodes): + vc = versor_condition(F[i]) + if vc >= 1e-6: + raise ValueError( + f"ManifoldState.fields[{i}] violates versor_condition: {vc:.2e} >= 1e-6." + ) + + def with_fields(self, new_fields: np.ndarray) -> ManifoldState: + """Return a new ManifoldState with updated field values.""" + return ManifoldState(fields=new_fields, edges=self.edges, step=self.step) + + def advance(self) -> ManifoldState: + """Return a new ManifoldState one step forward.""" + return ManifoldState(fields=self.fields, edges=self.edges, step=self.step + 1) diff --git a/scripts/run_pulse.py b/scripts/run_pulse.py new file mode 100644 index 00000000..4226ad49 --- /dev/null +++ b/scripts/run_pulse.py @@ -0,0 +1,79 @@ +""" +Vertical slice: one cognitive pulse from injection to token recall. + +Usage: + python -m scripts.run_pulse + python -m scripts.run_pulse "your input text" +""" + +from __future__ import annotations + +import sys + +import numpy as np + +from algebra.backend import vault_recall +from field.operators import GraphDiffusionOperator +from field.state import ManifoldState +from sensorium.adapters.text import deterministic_hash_versor + +CONVERGENCE_THRESHOLD = 1e-6 +MAX_STEPS = 2000 + +VOCAB_WORDS = [ + "truth", "light", "wisdom", "peace", "knowledge", + "word", "path", "life", "grace", "hope", +] + + +def build_initial_manifold(prompt_versor: np.ndarray) -> ManifoldState: + context_versor = deterministic_hash_versor("__context__") + output_versor = deterministic_hash_versor("__output__") + fields = np.stack([prompt_versor, context_versor, output_versor], axis=0) + edges = np.array([[0, 1], [1, 2], [0, 2]], dtype=np.int32) + return ManifoldState(fields=fields, edges=edges) + + +def build_mock_vault() -> tuple[list[np.ndarray], list[str]]: + versors = [deterministic_hash_versor(w) for w in VOCAB_WORDS] + return versors, list(VOCAB_WORDS) + + +def run_pulse(text: str) -> str: + prompt_versor = deterministic_hash_versor(text) + state = build_initial_manifold(prompt_versor) + op = GraphDiffusionOperator(damping=0.5) + + print(f"[pulse] input: {text!r}") + print(f"[pulse] nodes: 3, edges: {state.edges.shape[0]}") + + step = 0 + delta = float("inf") + while step < MAX_STEPS: + state, delta = op.forward(state) + step = state.step + if step <= 5 or step % 50 == 0: + print(f"[pulse] step {step:4d} delta={delta:.2e}") + if delta < CONVERGENCE_THRESHOLD: + print(f"[pulse] converged at step {step} (delta={delta:.2e})") + break + else: + print(f"[pulse] WARNING: max_steps ({MAX_STEPS}) reached without convergence (delta={delta:.2e})") + + output_versor = state.fields[2] + vault_versors, vault_words = build_mock_vault() + results = vault_recall(vault_versors, output_versor, top_k=1) + + if results: + best_idx, best_score = results[0] + resolved_word = vault_words[best_idx] + print(f"[pulse] output node -> vault recall: {resolved_word!r} (score={best_score:.6f})") + return resolved_word + + print("[pulse] vault recall returned no results") + return "" + + +if __name__ == "__main__": + input_text = " ".join(sys.argv[1:]) if len(sys.argv) > 1 else "hello world" + run_pulse(input_text) diff --git a/sensorium/adapters/text.py b/sensorium/adapters/text.py index c067c8a2..fd9f67ea 100644 --- a/sensorium/adapters/text.py +++ b/sensorium/adapters/text.py @@ -22,6 +22,9 @@ import logging import numpy as np +import hashlib + +from algebra.versor import construction_seed_versor, versor_condition from language_packs.schema import LanguageRole, OOVPolicy from sensorium.protocol import ( CL41_DIM, @@ -208,3 +211,29 @@ def koine_greek_pack(vocabulary: ModalityVocabulary | None = None) -> ModalityPa language_role=LanguageRole.DEPTH_RELATION, oov_policy=OOVPolicy.FAIL_CLOSED, ) + + +# --------------------------------------------------------------------------- +# Deterministic hash-to-versor stub for testing without vocabulary +# --------------------------------------------------------------------------- + + +def deterministic_hash_versor(text: str) -> np.ndarray: + """Map an arbitrary string to a valid Cl(4,1) versor, deterministically. + + Uses SHA-256 bytes mapped to bounded [-1, 1] floats to fill a dense + 32-component seed, then constructs a closed versor via the seed-to-rotor + path (construction tier). + """ + digest = hashlib.sha256(text.encode("utf-8")).digest() + seed = np.array( + [(b / 127.5) - 1.0 for b in digest], + dtype=np.float64, + ) + full_seed = np.zeros(32, dtype=np.float64) + full_seed[:32] = seed[:32] + versor = construction_seed_versor(full_seed) + vc = versor_condition(versor) + if vc >= 1e-6: + raise ValueError(f"deterministic_hash_versor: versor_condition {vc:.2e} >= 1e-6") + return versor.astype(np.float32) diff --git a/tests/test_deterministic_hash.py b/tests/test_deterministic_hash.py new file mode 100644 index 00000000..e132e389 --- /dev/null +++ b/tests/test_deterministic_hash.py @@ -0,0 +1,31 @@ +"""deterministic_hash_versor tests — determinism, uniqueness, closure.""" + +import numpy as np + +from algebra.backend import versor_condition +from sensorium.adapters.text import deterministic_hash_versor + + +class TestDeterministicHash: + def test_deterministic(self) -> None: + a = deterministic_hash_versor("hello") + b = deterministic_hash_versor("hello") + np.testing.assert_array_equal(a, b) + + def test_different_inputs_differ(self) -> None: + a = deterministic_hash_versor("alpha") + b = deterministic_hash_versor("beta") + assert not np.array_equal(a, b) + + def test_versor_condition(self) -> None: + v = deterministic_hash_versor("test string") + assert versor_condition(v) < 1e-6 + + def test_output_shape_and_dtype(self) -> None: + v = deterministic_hash_versor("shape check") + assert v.shape == (32,) + assert v.dtype == np.float32 + + def test_empty_string(self) -> None: + v = deterministic_hash_versor("") + assert versor_condition(v) < 1e-6 diff --git a/tests/test_graph_diffusion.py b/tests/test_graph_diffusion.py new file mode 100644 index 00000000..4f94d100 --- /dev/null +++ b/tests/test_graph_diffusion.py @@ -0,0 +1,65 @@ +"""GraphDiffusionOperator tests — convergence, closure, self-adjointness.""" + +import numpy as np +import pytest + +from algebra.backend import versor_condition +from algebra.rotor import make_rotor_from_angle +from field.operators import GraphDiffusionOperator +from field.state import ManifoldState + + +def _make_versors(n: int) -> np.ndarray: + return np.stack( + [make_rotor_from_angle(0.1 * (i + 1)).astype(np.float32) for i in range(n)], + axis=0, + ) + + +class TestGraphDiffusion: + def test_self_adjoint(self) -> None: + op = GraphDiffusionOperator(damping=0.5) + assert op.adjoint() is op + + def test_invalid_damping(self) -> None: + with pytest.raises(ValueError): + GraphDiffusionOperator(damping=0.0) + with pytest.raises(ValueError): + GraphDiffusionOperator(damping=1.5) + + def test_forward_returns_manifold_and_delta(self) -> None: + fields = _make_versors(2) + edges = np.array([[0, 1]], dtype=np.int32) + state = ManifoldState(fields=fields, edges=edges) + op = GraphDiffusionOperator(damping=0.5) + new_state, delta = op.forward(state) + assert isinstance(new_state, ManifoldState) + assert isinstance(delta, float) + assert new_state.step == 1 + + def test_versor_condition_preserved(self) -> None: + fields = _make_versors(3) + edges = np.array([[0, 1], [1, 2], [0, 2]], dtype=np.int32) + state = ManifoldState(fields=fields, edges=edges) + op = GraphDiffusionOperator(damping=0.5) + new_state, _ = op.forward(state) + for i in range(new_state.fields.shape[0]): + assert versor_condition(new_state.fields[i]) < 1e-6 + + def test_convergence_delta_nonnegative(self) -> None: + fields = _make_versors(3) + edges = np.array([[0, 1], [1, 2], [0, 2]], dtype=np.int32) + state = ManifoldState(fields=fields, edges=edges) + op = GraphDiffusionOperator(damping=0.5) + for _ in range(10): + state, delta = op.forward(state) + assert delta >= 0.0 + + def test_identical_nodes_small_delta(self) -> None: + v = make_rotor_from_angle(0.3).astype(np.float32) + fields = np.stack([v, v], axis=0) + edges = np.array([[0, 1]], dtype=np.int32) + state = ManifoldState(fields=fields, edges=edges) + op = GraphDiffusionOperator(damping=0.5) + _, delta = op.forward(state) + assert delta < 0.5 diff --git a/tests/test_manifold_state.py b/tests/test_manifold_state.py new file mode 100644 index 00000000..d8f258a5 --- /dev/null +++ b/tests/test_manifold_state.py @@ -0,0 +1,83 @@ +"""ManifoldState construction, validation, and immutability tests.""" + +import numpy as np +import pytest + +from algebra.rotor import make_rotor_from_angle +from field.state import ManifoldState + + +def _make_versors(n: int) -> np.ndarray: + return np.stack( + [make_rotor_from_angle(0.1 * (i + 1)).astype(np.float32) for i in range(n)], + axis=0, + ) + + +def _triangle_edges() -> np.ndarray: + return np.array([[0, 1], [1, 2], [0, 2]], dtype=np.int32) + + +class TestConstruction: + def test_valid_construction(self) -> None: + fields = _make_versors(3) + edges = _triangle_edges() + ms = ManifoldState(fields=fields, edges=edges) + assert ms.fields.shape == (3, 32) + assert ms.edges.shape == (3, 2) + assert ms.step == 0 + + def test_fields_are_copied(self) -> None: + fields = _make_versors(2) + edges = np.array([[0, 1]], dtype=np.int32) + ms = ManifoldState(fields=fields, edges=edges) + assert ms.fields is not fields + + def test_bad_field_shape_raises(self) -> None: + with pytest.raises(ValueError, match="shape"): + ManifoldState( + fields=np.ones((3, 16), dtype=np.float32), + edges=_triangle_edges(), + ) + + def test_bad_edge_shape_raises(self) -> None: + with pytest.raises(ValueError, match="shape"): + ManifoldState( + fields=_make_versors(3), + edges=np.array([[0, 1, 2]], dtype=np.int32), + ) + + def test_edge_out_of_bounds_raises(self) -> None: + with pytest.raises(ValueError, match="indices"): + ManifoldState( + fields=_make_versors(2), + edges=np.array([[0, 5]], dtype=np.int32), + ) + + def test_versor_condition_enforced(self) -> None: + bad_fields = np.random.randn(2, 32).astype(np.float32) + with pytest.raises(ValueError, match="versor_condition"): + ManifoldState( + fields=bad_fields, + edges=np.array([[0, 1]], dtype=np.int32), + ) + + +class TestMutation: + def test_frozen(self) -> None: + ms = ManifoldState(fields=_make_versors(2), edges=np.array([[0, 1]], dtype=np.int32)) + with pytest.raises(AttributeError): + ms.step = 5 # type: ignore[misc] + + def test_with_fields_returns_new(self) -> None: + ms = ManifoldState(fields=_make_versors(2), edges=np.array([[0, 1]], dtype=np.int32)) + new_fields = _make_versors(2) + ms2 = ms.with_fields(new_fields) + assert ms2 is not ms + assert ms.step == ms2.step + + def test_advance_increments_step(self) -> None: + ms = ManifoldState(fields=_make_versors(2), edges=np.array([[0, 1]], dtype=np.int32)) + ms2 = ms.advance() + assert ms2.step == ms.step + 1 + assert np.array_equal(ms2.fields, ms.fields) diff --git a/tests/test_pulse_integration.py b/tests/test_pulse_integration.py new file mode 100644 index 00000000..0d55e69e --- /dev/null +++ b/tests/test_pulse_integration.py @@ -0,0 +1,34 @@ +"""Integration test — full pulse cycle from injection to vault recall.""" + +import numpy as np + +from scripts.run_pulse import build_initial_manifold, build_mock_vault, run_pulse +from sensorium.adapters.text import deterministic_hash_versor + + +class TestPulseIntegration: + def test_full_cycle_completes(self) -> None: + word = run_pulse("hello world") + assert isinstance(word, str) + assert len(word) > 0 + + def test_output_node_changes(self) -> None: + prompt = deterministic_hash_versor("test input") + state = build_initial_manifold(prompt) + initial_output = state.fields[2].copy() + + from field.operators import GraphDiffusionOperator + op = GraphDiffusionOperator(damping=0.5) + for _ in range(20): + state, _ = op.forward(state) + assert not np.allclose(state.fields[2], initial_output, atol=1e-7) + + def test_vault_recall_returns_known_word(self) -> None: + word = run_pulse("wisdom seeker") + vault_versors, vault_words = build_mock_vault() + assert word in vault_words + + def test_different_inputs_may_differ(self) -> None: + w1 = run_pulse("alpha") + w2 = run_pulse("omega") + assert isinstance(w1, str) and isinstance(w2, str)