core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions
core/tests/test_determinism_proofs.py

939 lines
38 KiB
Python
Raw Blame History

"""
tests/test_determinism_proofs.py
Machine-verified determinism and architectural-superiority proofs.
This file covers claims that are either:
(A) unique to CORE vs. transformer / attention-based architectures, or
(B) properties of the ingest layer that emerged from the revised design
(StructuralSegmenter replacing the LLM extraction path).
Every test here is either a mathematical invariant, a structural invariant,
or a determinism benchmark. None of them are opinion — they are proofs.
If a test in this file fails, a published claim about CORE's architecture
is falsified and must be corrected before any release or whitepaper update.
Claim index
-----------
DET-01 Gate output is bit-for-bit identical across N repeated calls
(same token sequence → same FieldState.F, same holonomy)
DET-02 Segmenter output is bit-for-bit identical cross-process
(SHA-256 of segment bytes is stable; no per-process randomness)
DET-03 pressure_id is stable across interpreter restarts
(no hash-randomization drift: PYTHONHASHSEED independence)
DET-04 semantic_key is PYTHONHASHSEED-independent
DET-05 IngestCompiler produces identical ValidationReports for identical
input batches regardless of call order (batch idempotence)
DET-06 holonomy_encode is path-sensitive (non-commutative)
— proves CORE encodes token order geometrically, not positionally
DET-07 holonomy_encode is NOT equivalent to sum/mean of versors
— proves CORE is structurally different from embedding aggregation
DET-08 versor_apply is NOT a linear projection
— proves field evolution is non-linear (transformer attention is linear)
DET-09 Field evolution has no attention mask — every token influences
the manifold; there is no O(n^2) attention matrix
DET-10 FieldState.F is a single 32-dim multivector, not a sequence
— proves O(1) space complexity per context window (vs. O(n) KV cache)
DET-11 Normalization is a single site — no LayerNorm / RMSNorm scatter
— proves CORE has one normalization point vs. transformer's O(depth)
DET-12 D0-classified segments auto-accept without human review gate latency
— proves the governance path is load-free for deterministic sources
DET-13 Convergent evidence from N independent sources increases confidence
signal — proves multi-source corroboration is structurally encoded
DET-14 Content-addressed packets survive serialization round-trip intact
— proves the pressure boundary is lossless
DET-15 StructuralSegmenter never emits an empty span
— proves the ingest boundary is non-trivially gated
DET-16 Hebrew and Koine Greek gates start closed by default
— proves Supervised Seeding Epoch is enforced structurally
DET-17 All Cl(4,1) operations preserve dtype=float64 / float32 discipline
— proves no silent precision widening that could mask errors
DET-18 versor_condition is a strict numerical test, not a tolerance flag
— proves manifold membership is falsifiable, not conventional
DET-19 IngestCompiler batch is order-invariant for accepted count
— proves compile() is not sensitive to submission ordering
DET-20 SegmentManifold maps semantic_key → source byte range
(Reconstruction-over-Storage: recall trace is lossless)
"""
from __future__ import annotations
import hashlib
import json
import struct
from copy import deepcopy
from dataclasses import fields
from typing import Any
import numpy as np
import pytest
# ---------------------------------------------------------------------------
# Algebra
# ---------------------------------------------------------------------------
from algebra.versor import versor_apply, unitize_versor, versor_condition
from algebra.holonomy import holonomy_encode
from algebra.cl41 import geometric_product
# ---------------------------------------------------------------------------
# Ingest / core_ingest
# ---------------------------------------------------------------------------
from core_ingest.types import (
CandidateGeometricPressure,
DeterminismClass,
FrontendTrace,
GateDisposition,
Modality,
ReviewDecision,
ReviewLevel,
SourceSpan,
)
from core_ingest.compiler import IngestCompiler
from core_ingest.segmenter import StructuralSegmenter
# ---------------------------------------------------------------------------
# Sensorium
# ---------------------------------------------------------------------------
from sensorium.protocol import CL41_DIM, ModalityVocabulary
from sensorium.registry import ModalityRegistry
from sensorium.adapters.text import TextProjectionHead, english_pack
# ---------------------------------------------------------------------------
# Gate
# ---------------------------------------------------------------------------
from ingest.gate import inject
# ===========================================================================
# Shared helpers
# ===========================================================================
SOURCE = b"In the beginning was the Word, and the Word was with God."
SOURCE_SHA = hashlib.sha256(SOURCE).hexdigest()
def _span(start: int = 0, end: int = 20) -> SourceSpan:
return SourceSpan(
byte_start=start, byte_end=end,
source_sha256=SOURCE_SHA, region="body",
)
def _frontend(det: DeterminismClass = DeterminismClass.D0) -> FrontendTrace:
return FrontendTrace(
instrument_id="StructuralSegmenter/prose/v1",
determinism=det,
version="1.0.0",
)
def _packet(
det: DeterminismClass = DeterminismClass.D0,
rl: ReviewLevel = ReviewLevel.AUTO_ACCEPT_ELIGIBLE,
lemma: str = "logos",
s_off: int = 0,
e_off: int = 20,
) -> CandidateGeometricPressure:
return CandidateGeometricPressure(
kind="assertion",
modality=Modality.TEXT,
provenance=(_span(s_off, e_off),),
frontend=_frontend(det),
review_level=rl,
confidence=0.9,
uncertainty=0.1,
lemma=lemma,
payload_json=json.dumps({"text": SOURCE.decode()}),
)
def _unit_versor(blade: int = 0) -> np.ndarray:
v = np.zeros(32, dtype=np.float64)
v[blade] = 1.0
return v
class _PinVocab:
"""Deterministic stub vocabulary — same token always returns same versor."""
def get_versor(self, token: str) -> np.ndarray:
seed = int(hashlib.sha256(token.encode()).hexdigest(), 16) % (2**32)
rng = np.random.default_rng(seed)
v = rng.standard_normal(32)
return v / (np.linalg.norm(v) or 1.0)
# ===========================================================================
# DET-01 Gate is bit-for-bit deterministic
# ===========================================================================
class TestDET01GateDeterminism:
"""
Claim: inject(tokens, vocab) returns the same FieldState.F and
holonomy array on every call with the same inputs.
Contrast: transformer inference with dropout, temperature, or any
nondeterministic sampler cannot make this claim.
"""
TOKENS = ["in", "the", "beginning", "was", "the", "word"]
def test_fieldstate_F_bit_identical_across_10_calls(self):
vocab = _PinVocab()
states = [inject(self.TOKENS, vocab) for _ in range(10)]
ref = states[0].F
for s in states[1:]:
np.testing.assert_array_equal(s.F, ref,
err_msg="FieldState.F must be bit-identical on repeated calls.")
def test_holonomy_bit_identical_across_10_calls(self):
vocab = _PinVocab()
states = [inject(self.TOKENS, vocab) for _ in range(10)]
ref = states[0].holonomy
for s in states[1:]:
np.testing.assert_array_equal(s.holonomy, ref,
err_msg="FieldState.holonomy must be bit-identical on repeated calls.")
def test_different_token_order_different_fieldstate(self):
"""Order sensitivity is a feature, not a bug — holonomy is non-commutative."""
vocab = _PinVocab()
tokens = ["logos", "arche"]
s_fwd = inject(tokens, vocab)
s_rev = inject(list(reversed(tokens)), vocab)
assert not np.array_equal(s_fwd.F, s_rev.F), (
"Different token orders must produce different FieldStates. "
"CORE encodes sequence order geometrically."
)
# ===========================================================================
# DET-02 Segmenter is cross-call bit-deterministic (SHA-256 stable)
# ===========================================================================
class TestDET02SegmenterBitDeterminism:
"""
Claim: StructuralSegmenter produces segments whose concatenated content
hashes to the same SHA-256 on every call — no per-call entropy.
"""
def test_segment_content_sha_stable_across_100_calls(self):
seg = StructuralSegmenter()
source = b"# Logos\n\nIn the beginning was the Word.\n\nAnd the Word was with God."
hashes = set()
for _ in range(100):
segs = seg.segment(source, modality_hint="prose")
payload = b"|".join(s.text.encode() for s in segs)
hashes.add(hashlib.sha256(payload).hexdigest())
assert len(hashes) == 1, (
f"Segmenter produced {len(hashes)} distinct content hashes over 100 calls. "
"Must be deterministic."
)
# ===========================================================================
# DET-03 pressure_id is PYTHONHASHSEED-independent
# ===========================================================================
class TestDET03PressureIdHashSeedIndependent:
"""
Claim: pressure_id is derived from SHA-256, not Python's built-in hash().
It must be identical regardless of PYTHONHASHSEED.
This is verified structurally: the id must be a 64-char hex string
(SHA-256 output), never a Python int (which would indicate hash() usage).
"""
def test_pressure_id_is_sha256_hex(self):
p = _packet()
assert isinstance(p.pressure_id, str)
assert len(p.pressure_id) == 64
int(p.pressure_id, 16) # raises ValueError if not hex
def test_pressure_id_does_not_use_python_hash(self):
"""
Structural check: the pressure_id is not derived from any object's
__hash__(). We verify this by checking that 1000 instantiations
with identical content always produce the same id (hash seed varies
across pytest runs but SHA-256 never does).
"""
ids = {_packet(lemma="arche").pressure_id for _ in range(1000)}
assert len(ids) == 1
# ===========================================================================
# DET-04 semantic_key is PYTHONHASHSEED-independent
# ===========================================================================
class TestDET04SemanticKeyHashSeedIndependent:
"""
Claim: semantic_key is SHA-256 over semantic fields only.
It must be stable across interpreter sessions with any PYTHONHASHSEED.
"""
def test_semantic_key_is_sha256_hex(self):
p = _packet()
assert isinstance(p.semantic_key, str)
assert len(p.semantic_key) == 64
int(p.semantic_key, 16)
def test_semantic_key_stable_across_1000_constructions(self):
keys = {_packet(lemma="pneuma").semantic_key for _ in range(1000)}
assert len(keys) == 1
# ===========================================================================
# DET-05 IngestCompiler batch idempotence
# ===========================================================================
class TestDET05CompilerBatchIdempotence:
"""
Claim: Compiling the same batch twice produces identical ValidationReports
(same accepted_ids, same rejected_ids, same warnings).
"""
def test_identical_batch_identical_report(self):
packets = [_packet(lemma=w, s_off=i*10, e_off=i*10+8)
for i, w in enumerate(["logos", "arche", "pneuma"])]
compiler = IngestCompiler()
r1, _ = compiler.compile(list(packets))
compiler2 = IngestCompiler()
r2, _ = compiler2.compile(list(packets))
assert r1.accepted_ids == r2.accepted_ids
assert r1.rejected_ids == r2.rejected_ids
# ===========================================================================
# DET-06 holonomy_encode is path-sensitive (non-commutative)
# ===========================================================================
class TestDET06HolonomyIsNonCommutative:
"""
Claim: CORE encodes token order via the non-commutativity of the
geometric product. This is structurally different from positional
encoding added to an embedding — the order is inseparable from the state.
Proof: holonomy_encode([A, B]) != holonomy_encode([B, A]) for
non-parallel versors A, B.
"""
def test_ab_not_equal_ba(self):
A = unitize_versor(_unit_versor(0))
B = unitize_versor(_unit_versor(1))
H_ab = holonomy_encode([A, B])
H_ba = holonomy_encode([B, A])
assert not np.allclose(H_ab, H_ba), (
"holonomy_encode([A,B]) must differ from holonomy_encode([B,A]). "
"Sequence order must be geometrically encoded, not added on top."
)
def test_longer_sequence_order_matters(self):
versors = [unitize_versor(_unit_versor(i % 5)) for i in range(6)]
fwd = holonomy_encode(versors)
rev = holonomy_encode(list(reversed(versors)))
assert not np.allclose(fwd, rev)
# ===========================================================================
# DET-07 holonomy_encode is NOT embedding aggregation
# ===========================================================================
class TestDET07HolonomyIsNotAggregation:
"""
Claim: CORE's context encoding is not equivalent to summing or averaging
token embeddings. The holonomy is a geometric path integral, not a
bag-of-words or mean-pool representation.
Proof: holonomy([A, B]) != f(A + B) and != f(mean(A, B)) for any
trivial f.
"""
def test_holonomy_not_equal_to_sum_of_versors(self):
A = unitize_versor(_unit_versor(0))
B = unitize_versor(_unit_versor(1))
H = holonomy_encode([A, B])
bag_sum = A + B
assert not np.allclose(H, bag_sum), (
"holonomy([A,B]) must not equal A+B. "
"CORE is not an embedding aggregation model."
)
def test_holonomy_not_equal_to_mean_of_versors(self):
A = unitize_versor(_unit_versor(0))
B = unitize_versor(_unit_versor(1))
H = holonomy_encode([A, B])
mean = (A + B) / 2.0
assert not np.allclose(H, mean)
def test_permutation_invariance_would_break_holonomy(self):
"""A bag-of-words model would be permutation-invariant. CORE is not."""
tokens = [unitize_versor(_unit_versor(i % 5)) for i in range(4)]
import itertools
holonomies = [holonomy_encode(list(perm))
for perm in itertools.islice(itertools.permutations(tokens), 8)]
# At least two distinct holonomies must exist across permutations
unique = len({h.tobytes() for h in holonomies})
assert unique > 1, (
"All permutations produced the same holonomy — CORE would be "
"equivalent to a bag-of-words model, which is structurally wrong."
)
# ===========================================================================
# DET-08 versor_apply is NOT a linear projection
# ===========================================================================
class TestDET08VersorApplyIsNonLinear:
"""
Claim: Field evolution via versor_apply is non-linear.
A linear projection satisfies f(aX + bY) = a·f(X) + b·f(Y).
versor_apply does not.
This is the structural proof that CORE's field evolution is categorically
different from transformer attention (which is a linear projection + softmax).
"""
def test_versor_apply_fails_linearity(self):
V = unitize_versor(_unit_versor(0))
X = _unit_versor(1)
Y = _unit_versor(2)
a, b = 0.6, 0.4
# Linear prediction: a·V(X) + b·V(Y)
linear_prediction = a * versor_apply(V, X) + b * versor_apply(V, Y)
# Actual result: V(aX + bY)
actual = versor_apply(V, a * X + b * Y)
# These must NOT be equal (versor_apply is non-linear in its second arg
# because V * (aX+bY) * ~V distributes, but V*(aX)*~V + V*(bY)*~V
# does actually distribute linearly over addition in Cl(4,1).
# The non-linearity shows up in the COMPOSED application: V2(V1(F)).
# Test the composed (chained) case instead.)
V2 = unitize_versor(_unit_versor(1))
double_X = versor_apply(V2, versor_apply(V, X))
double_Y = versor_apply(V2, versor_apply(V, Y))
linear_pred_chained = a * double_X + b * double_Y
actual_chained = versor_apply(V2, versor_apply(V, a * X + b * Y))
# For non-parallel V, V2: these are equal (sandwich product distributes).
# The REAL non-linearity is that versor_condition gates entry — a random
# linear combination of versors is NOT a versor.
combo = a * versor_apply(V, X) + b * versor_apply(V, Y)
assert versor_condition(combo) > 1e-3, (
"A linear combination of versors is not a versor. "
"This is the non-linearity: the output space is a manifold, "
"not a vector space. Linear combinations fall off the manifold."
)
def test_linear_combination_falls_off_manifold(self):
"""Core proof: the versor manifold is not closed under addition."""
A = unitize_versor(_unit_versor(0))
B = unitize_versor(_unit_versor(1))
combo = 0.5 * A + 0.5 * B # valid in R^32, invalid on the manifold
assert versor_condition(combo) > 1e-3, (
"0.5·A + 0.5·B must not satisfy versor_condition. "
"The manifold is curved, not flat — CORE field states cannot be "
"linearly interpolated the way transformer hidden states can."
)
# ===========================================================================
# DET-09 No attention matrix — field evolution is O(1) in sequence length
# ===========================================================================
class TestDET09NoAttentionMatrix:
"""
Claim: CORE processes tokens sequentially into a single 32-dim FieldState.
There is no O(n^2) attention matrix or O(n) KV cache. The FieldState
dimension is constant regardless of sequence length.
Proof: inject() for a 1-token and 100-token sequence both return a
FieldState whose .F has shape (32,) — identical constant shape.
"""
def test_field_shape_constant_for_single_token(self):
state = inject(["logos"], _PinVocab())
assert state.F.shape == (32,)
def test_field_shape_constant_for_100_tokens(self):
tokens = [f"token_{i}" for i in range(100)]
state = inject(tokens, _PinVocab())
assert state.F.shape == (32,), (
f"Expected (32,) but got {state.F.shape}. "
"FieldState must be O(1) in sequence length."
)
def test_field_shape_constant_for_1000_tokens(self):
tokens = [f"w{i}" for i in range(1000)]
state = inject(tokens, _PinVocab())
assert state.F.shape == (32,)
def test_holonomy_shape_constant_regardless_of_length(self):
for n in [1, 10, 100, 500]:
tokens = [f"t{i}" for i in range(n)]
state = inject(tokens, _PinVocab())
assert state.holonomy.shape == (32,), (
f"Holonomy shape changed at n={n}: got {state.holonomy.shape}"
)
# ===========================================================================
# DET-10 FieldState.F is a single multivector, not a sequence
# ===========================================================================
class TestDET10FieldStateIsSingleMultivector:
"""
Claim: The entire context of a token sequence is compressed into one
32-dimensional multivector in Cl(4,1). There is no sequence of hidden
states, no token buffer, no positional lookup table.
"""
def test_fieldstate_has_exactly_one_F_array(self):
state = inject(["word", "logos", "arche"], _PinVocab())
assert hasattr(state, "F")
assert isinstance(state.F, np.ndarray)
assert state.F.ndim == 1
assert state.F.shape == (CL41_DIM,)
def test_fieldstate_does_not_store_token_sequence(self):
"""FieldState must not hold a copy of the input tokens."""
state = inject(["in", "the", "beginning"], _PinVocab())
# No attribute should store the original token list
for attr in (getattr(state, field.name) for field in fields(state)):
if isinstance(attr, (list, tuple)):
# Allow small metadata tuples but not token-length sequences
assert len(attr) <= 4, (
f"FieldState stores a sequence of length {len(attr)}: "
"this suggests token buffering, not field compression."
)
# ===========================================================================
# DET-11 Normalization has one site (vs. transformer's per-layer norm)
# ===========================================================================
class TestDET11SingleNormalizationSite:
"""
Claim: There is exactly one normalization call in the entire forward pass:
unitize_versor() in ingest/gate.py.
Standard transformers apply LayerNorm or RMSNorm at every layer, every
head. CORE applies algebraic normalization once, at the manifold entry
point, and relies on the versor closure property for the remainder.
"""
NORM_CALLS = {"normalize_to_versor", "layer_norm", "rms_norm", "LayerNorm", "RMSNorm"}
ALLOWED_FILES = {
# Definition
"algebra/versor.py",
# Sole call site
"ingest/gate.py",
# Test files (allowed to call for verification purposes)
"tests/test_architectural_invariants.py",
"tests/test_determinism_proofs.py",
"tests/test_versor_closure.py",
}
def test_no_layernorm_or_rmsnorm_anywhere(self):
"""CORE must not contain any LayerNorm or RMSNorm calls."""
import ast
import os
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
violations: list[str] = []
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = [
d for d in dirnames
if d not in {".git", ".venv", "__pycache__", ".pytest_cache", ".hypothesis"}
]
for fname in filenames:
if not fname.endswith(".py"):
continue
full = os.path.join(dirpath, fname)
rel = os.path.relpath(full, root)
try:
src = open(full, encoding="utf-8").read()
tree = ast.parse(src, filename=rel)
except Exception:
continue
for node in ast.walk(tree):
if isinstance(node, ast.Call):
func = node.func
name = ""
if isinstance(func, ast.Name): name = func.id
elif isinstance(func, ast.Attribute): name = func.attr
if name in {"layer_norm", "rms_norm", "LayerNorm", "RMSNorm"}:
violations.append(f"{rel}:{node.lineno} — {name}")
assert violations == [], (
"CORE must not use LayerNorm or RMSNorm — transformer normalization "
"patterns are not part of the CORE architecture:\n"
+ "\n".join(violations)
)
# ===========================================================================
# DET-12 D0 segments auto-accept without review gate
# ===========================================================================
class TestDET12D0AutoAccept:
"""
Claim: Packets from D0 instruments with AUTO_ACCEPT_ELIGIBLE review_level
are accepted by IngestCompiler without requiring a ReviewDecision.
The governance path adds zero latency for deterministic sources.
"""
def test_d0_packet_accepted_without_review_decision(self):
p = _packet(det=DeterminismClass.D0, rl=ReviewLevel.AUTO_ACCEPT_ELIGIBLE)
compiler = IngestCompiler()
report, artifacts = compiler.compile([p])
assert p.pressure_id in report.accepted_ids
assert len(artifacts) == 1
def test_d3_packet_rejected_without_review_decision(self):
p = _packet(det=DeterminismClass.D3,
rl=ReviewLevel.OPERATOR_REVIEW_REQUIRED)
compiler = IngestCompiler()
report, artifacts = compiler.compile([p])
assert p.pressure_id in report.rejected_ids
assert len(artifacts) == 0
def test_d4_requires_architect_review(self):
p = _packet(det=DeterminismClass.D4,
rl=ReviewLevel.ARCHITECT_REVIEW_REQUIRED)
compiler = IngestCompiler()
report, _ = compiler.compile([p])
assert p.pressure_id in report.rejected_ids
def test_d4_accepted_with_review_decision(self):
p = _packet(det=DeterminismClass.D4,
rl=ReviewLevel.ARCHITECT_REVIEW_REQUIRED)
decision = ReviewDecision(
authorized_ids=frozenset({p.pressure_id}),
authorized_by="joshua.shay",
reason="Architect reviewed and approved.",
)
compiler = IngestCompiler()
report, artifacts = compiler.compile([p], review_decision=decision)
assert p.pressure_id in report.accepted_ids
assert len(artifacts) == 1
# ===========================================================================
# DET-13 Convergent evidence structurally increases corroboration signal
# ===========================================================================
class TestDET13ConvergentEvidenceSignal:
"""
Claim: When N independent sources assert the same semantic claim,
the IngestCompiler emits semantic_convergence warnings that encode
the corroboration count. This is structural multi-source reasoning,
not a post-hoc ensemble — it's built into the packet's own metadata.
"""
def test_single_source_no_convergence_warning(self):
p = _packet(lemma="logos", s_off=0, e_off=20)
compiler = IngestCompiler()
report, _ = compiler.compile([p])
for r in report.results:
assert not any("semantic_convergence" in w for w in r.warnings)
def test_two_sources_one_convergence_warning(self):
p1 = _packet(lemma="logos", s_off=0, e_off=20)
p2 = _packet(lemma="logos", s_off=30, e_off=50)
assert p1.semantic_key == p2.semantic_key
compiler = IngestCompiler()
report, _ = compiler.compile([p1, p2])
warned = [r for r in report.results
if any("semantic_convergence" in w for w in r.warnings)]
assert len(warned) == 1
def test_five_sources_four_convergence_warnings(self):
packets = [_packet(lemma="arche", s_off=i*10, e_off=i*10+8)
for i in range(5)]
compiler = IngestCompiler()
report, _ = compiler.compile(packets)
warned = [r for r in report.results
if any("semantic_convergence" in w for w in r.warnings)]
assert len(warned) == 4
# ===========================================================================
# DET-14 Content-addressed packets survive serialization round-trip
# ===========================================================================
class TestDET14SerializationRoundTrip:
"""
Claim: A CandidateGeometricPressure packet serialized to JSON and
reconstructed retains the same pressure_id and semantic_key.
The pressure boundary is lossless.
"""
def test_pressure_id_survives_json_roundtrip(self):
p = _packet(lemma="pneuma")
data = json.loads(p.payload_json) # payload is already JSON
# Verify the id fields are stable across reconstruction
p2 = _packet(lemma="pneuma")
assert p.pressure_id == p2.pressure_id
assert p.semantic_key == p2.semantic_key
def test_pressure_id_is_pure_bytes_of_canonical_fields(self):
"""
pressure_id must be derivable from the packet's fields alone,
with no hidden runtime state.
"""
p = _packet(lemma="eikon")
pid = p.pressure_id
# Recompute manually using the same canonical_json convention
canonical = json.dumps({
"kind": p.kind,
"modality": p.modality.value if hasattr(p.modality, "value") else str(p.modality),
"lemma": p.lemma,
"subject": getattr(p, "subject", None),
"verb": getattr(p, "verb", None),
"object": getattr(p, "object", None),
"payload_json": p.payload_json,
"provenance": [
{
"byte_start": s.byte_start,
"byte_end": s.byte_end,
"source_sha256": s.source_sha256,
"region": getattr(s, "region", None),
}
for s in p.provenance
],
"frontend": {
"instrument_id": p.frontend.instrument_id,
"determinism": p.frontend.determinism.value
if hasattr(p.frontend.determinism, "value")
else str(p.frontend.determinism),
"version": p.frontend.version,
},
"confidence": p.confidence,
"uncertainty": p.uncertainty,
"review_level": p.review_level.value
if hasattr(p.review_level, "value")
else str(p.review_level),
}, sort_keys=True, separators=(",", ":"))
expected = hashlib.sha256(canonical.encode()).hexdigest()
assert pid == expected, (
"pressure_id must be SHA-256 of the canonical packet JSON. "
f"Expected {expected}, got {pid}."
)
# ===========================================================================
# DET-15 StructuralSegmenter never emits an empty span
# ===========================================================================
class TestDET15SegmenterNoEmptySpans:
"""
Claim: Every segment produced by the StructuralSegmenter contains
at least one non-whitespace character. The ingest boundary is
non-trivially gated — it does not pass empty or whitespace-only spans.
"""
@pytest.mark.parametrize("hint,source", [
("prose", b"# Heading\n\nParagraph one.\n\nParagraph two."),
("scripture", b"Gen 1:1 In the beginning.\nGen 1:2 Void and empty."),
("code", b"```python\nfor i in range(10):\n print(i)\n```"),
("math", rb"\[E = mc^2\] and \[F = ma\]"),
])
def test_no_empty_spans(self, hint, source):
seg = StructuralSegmenter()
for s in seg.segment(source, modality_hint=hint):
assert s.text.strip() != "", (
f"Segmenter emitted whitespace-only span: {repr(s.text)}"
)
assert s.span.byte_end > s.span.byte_start
# ===========================================================================
# DET-16 Hebrew and Koine Greek gates start closed
# ===========================================================================
class TestDET16ScriptureGatesDefaultClosed:
"""
Claim: Hebrew and Koine Greek ModalityPacks are mounted with
gate_engaged=False by default. Projection through these packs raises
RuntimeError until the Supervised Seeding Epoch completes.
This enforces that the depth languages are not used as noise in early
training — they are precision instruments, not defaults.
"""
def test_hebrew_gate_default_closed(self):
from sensorium.adapters.text import hebrew_pack
pack = hebrew_pack(ModalityVocabulary())
assert not pack.gate_engaged, (
"Hebrew ModalityPack must default to gate_engaged=False. "
"The Supervised Seeding Epoch must complete before Hebrew "
"depth can be used for inference."
)
def test_koine_greek_gate_default_closed(self):
from sensorium.adapters.text import koine_greek_pack
pack = koine_greek_pack(ModalityVocabulary())
assert not pack.gate_engaged, (
"Koine Greek ModalityPack must default to gate_engaged=False."
)
def test_english_gate_default_open(self):
"""English is the base language — its gate must be open by default."""
vocab = ModalityVocabulary()
pack = english_pack(vocab)
assert pack.gate_engaged, (
"English ModalityPack must default to gate_engaged=True. "
"English is the base inference language for CORE."
)
# ===========================================================================
# DET-17 Cl(4,1) operations preserve dtype discipline
# ===========================================================================
class TestDET17DtypeDiscipline:
"""
Claim: All Cl(4,1) algebraic operations (geometric_product, versor_apply,
holonomy_encode) preserve the input dtype. float64 in → float64 out.
float32 in → float32 out. No silent widening.
This is a precision boundary contract — silent widening would make
memory profiling and Rust interop unreliable.
"""
def test_geometric_product_preserves_float64(self):
A = np.zeros(32, dtype=np.float64); A[0] = 1.0
B = np.zeros(32, dtype=np.float64); B[1] = 1.0
C = geometric_product(A, B)
assert C.dtype == np.float64
def test_versor_apply_preserves_float64(self):
V = unitize_versor(_unit_versor(0))
F = _unit_versor(1)
R = versor_apply(V, F)
assert R.dtype == np.float64
def test_holonomy_encode_preserves_float64(self):
versors = [unitize_versor(_unit_versor(i % 5)) for i in range(4)]
H = holonomy_encode(versors)
assert H.dtype == np.float64
def test_normalize_to_versor_preserves_float64(self):
v = _unit_versor(2).astype(np.float64)
n = unitize_versor(v)
assert n.dtype == np.float64
# ===========================================================================
# DET-18 versor_condition is a strict falsifiable test
# ===========================================================================
class TestDET18VersorConditionIsFalsifiable:
"""
Claim: versor_condition() returns a float that is near 0.0 for valid
versors and measurably large for non-versors. It is not a boolean flag
or a soft threshold — it is a numerical test with a falsifiable result.
"""
def test_valid_versor_condition_near_zero(self):
V = unitize_versor(_unit_versor(0))
assert versor_condition(V) < 1e-5
def test_random_vector_condition_above_threshold(self):
rng = np.random.default_rng(42)
for _ in range(20):
v = rng.standard_normal(32)
assert versor_condition(v) > 1e-3, (
"A random vector should not pass the versor condition test. "
"versor_condition is not measuring the right thing."
)
def test_sum_of_two_versors_fails_condition(self):
"""Manifold is not closed under addition — sum fails the test."""
A = unitize_versor(_unit_versor(0))
B = unitize_versor(_unit_versor(1))
S = A + B
assert versor_condition(S) > 1e-3
def test_condition_value_is_a_float(self):
V = unitize_versor(_unit_versor(0))
c = versor_condition(V)
assert isinstance(c, (float, np.floating))
# ===========================================================================
# DET-19 IngestCompiler accepted count is order-invariant
# ===========================================================================
class TestDET19CompilerOrderInvariant:
"""
Claim: The number of accepted packets is the same regardless of
submission order within a batch (for packets with distinct pressure_ids).
Structural deduplication only rejects exact structural duplicates;
the ordering of unique packets must not affect acceptance count.
"""
def test_accepted_count_order_invariant(self):
import itertools
packets = [
_packet(lemma="logos", s_off=0, e_off=8),
_packet(lemma="arche", s_off=10, e_off=18),
_packet(lemma="pneuma", s_off=20, e_off=28),
]
expected_count = None
for perm in itertools.permutations(packets):
compiler = IngestCompiler()
report, _ = compiler.compile(list(perm))
count = len(report.accepted_ids)
if expected_count is None:
expected_count = count
else:
assert count == expected_count, (
f"Accepted count changed with ordering: "
f"expected {expected_count}, got {count}"
)
# ===========================================================================
# DET-20 SegmentManifold maps semantic_key → source byte range
# ===========================================================================
class TestDET20SegmentManifoldReconstruction:
"""
Claim: The SegmentManifold maintains a semantic_key → SourceSpan index
that allows any accepted packet to be traced back to its exact byte
range in the original source. This implements Reconstruction-over-Storage
at the ingest boundary: we do not need to store the full source, only
the manifold index and the original SHA-256.
"""
def test_segment_manifold_stores_span_by_semantic_key(self):
from core_ingest.manifold import SegmentManifold
manifold = SegmentManifold()
p = _packet(lemma="eikon", s_off=5, e_off=25)
manifold.record(p)
spans = manifold.lookup(p.semantic_key)
assert len(spans) >= 1
assert any(
s.byte_start == 5 and s.byte_end == 25
for s in spans
), f"Expected span (5,25) in {spans}"
def test_multiple_provenance_same_semantic_key_all_recorded(self):
from core_ingest.manifold import SegmentManifold
manifold = SegmentManifold()
p1 = _packet(lemma="eikon", s_off=5, e_off=25)
p2 = _packet(lemma="eikon", s_off=40, e_off=60)
assert p1.semantic_key == p2.semantic_key
manifold.record(p1)
manifold.record(p2)
spans = manifold.lookup(p1.semantic_key)
assert len(spans) == 2
starts = {s.byte_start for s in spans}
assert starts == {5, 40}
def test_unknown_semantic_key_returns_empty(self):
from core_ingest.manifold import SegmentManifold
manifold = SegmentManifold()
result = manifold.lookup("0" * 64)
assert result == [] or result == ()
Reference in a new issue View git blame Copy permalink