feat(adr-0243): §2.3 linguistic wave readback — seam S1 closed (off-serving)
Egress route readback_eligible now flows into geometric token selection (resonance = <psi psi~_T>_0 via the I-04 phase correlation; cga_inner is the un-reversed grade-0 product and is wrong for versor states — pinned in tests) plus the hearing-ourselves-think round-trip: articulated tokens are re-ingested through the same sensorium boundary and phase-locked agreement is measured with the same metric. Fail-closed typed ReadbackRefusal when no token resonates (decoding, not generating — no fallback strings). Vocab access is structural (VocabLike) to avoid a core.physics->vocab cycle; the real VocabManifold satisfies it (proven in tests). 10 new pins, lifecycle suite untouched (48 passed).
This commit is contained in:
Shay
parent
fcea2d3a4f
commit
19d5731a23
2 changed files with 564 additions and 0 deletions
328
core/physics/linguistic_readback.py
Normal file
328
core/physics/linguistic_readback.py
Normal file
|
|
@ -0,0 +1,328 @@
|
|||
"""ADR-0243 §2.3 — Linguistic wave readback (Tier-2, OFF-SERVING).
|
||||
|
||||
Closes seam S1 (docs/research/spark-audit-adjudication-2026-07-18.md §4): the
|
||||
egress route ``readback_eligible`` now flows into geometric token selection
|
||||
over a vocabulary of Cl(4,1) versors,
|
||||
|
||||
token_t = argmax_T ⟨ψ_steady ψ̃_T⟩_0 (descending resonance spectrum),
|
||||
|
||||
followed by the "hearing ourselves think" round-trip: the articulated tokens
|
||||
are re-ingested through the same sensorium construction boundary
|
||||
(:func:`~core.physics.cognitive_lifecycle.ingest_context`) and the
|
||||
phase-locked agreement with the pre-articulation steady state is measured via
|
||||
the I-04 sanctioned metric :meth:`WaveManifold.phase_correlation`
|
||||
(ρ = ⟨ψ_A ψ̃_B + ψ_B ψ̃_A⟩_0; agreement = ρ/2). No cosine, no ANN.
|
||||
|
||||
Design pins:
|
||||
|
||||
* **Decoding, not generating**: a state with no resonant token above the
|
||||
caller's ``min_resonance`` raises a typed :class:`ReadbackRefusal` — never a
|
||||
fallback string. Sparse vocabulary coverage surfaces as honest refusal.
|
||||
* **Layering**: vocabulary access is structural (:class:`VocabLike`), because
|
||||
the vocab package imports ``core.physics.energy`` — a reverse import here
|
||||
would cycle through the ``core.physics`` barrel. The real ``VocabManifold``
|
||||
satisfies the protocol as-is (proven in the test suite).
|
||||
* **Policy is caller-supplied**: ``min_resonance`` / ``max_tokens`` are
|
||||
explicit keyword-only inputs — never invented here (same doctrine as the
|
||||
egress energy axes).
|
||||
* **Certificate binding**: readback re-asserts ψ↔certificate digest identity
|
||||
(defense-in-depth mirroring :func:`~core.physics.cognitive_lifecycle.serving_cast`);
|
||||
a borrowed certificate refuses.
|
||||
|
||||
Serve quarantine (A-04): never imported by ``chat/runtime.py``; purity pinned
|
||||
by ``tests/test_adr_0243_linguistic_readback.py``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Protocol
|
||||
|
||||
import numpy as np
|
||||
|
||||
from core.physics.cognitive_lifecycle import (
|
||||
CognitiveLifecycleError,
|
||||
EgressVerdict,
|
||||
LifecycleOutcome,
|
||||
RelaxationCertificate,
|
||||
_as_psi,
|
||||
_content_id,
|
||||
_psi_digest,
|
||||
ingest_context,
|
||||
)
|
||||
from core.physics.sensorium_wave_feed import ModalityPacket
|
||||
from core.physics.wave_manifold import WaveManifold
|
||||
|
||||
|
||||
class ReadbackRefusal(CognitiveLifecycleError):
|
||||
"""Typed fail-closed readback refusal (§2.3) — never a fallback string."""
|
||||
|
||||
|
||||
class VocabLike(Protocol):
|
||||
"""Structural view of ``VocabManifold`` (indexed word/versor access only)."""
|
||||
|
||||
def __len__(self) -> int: ...
|
||||
|
||||
def get_versor_at(self, idx: int) -> np.ndarray: ...
|
||||
|
||||
def get_word_at(self, idx: int) -> str: ...
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ReadbackToken:
|
||||
"""One decoded token: word, vocabulary index, grade-0 resonance score."""
|
||||
|
||||
word: str
|
||||
index: int
|
||||
resonance: float
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {"word": self.word, "index": int(self.index), "resonance": float(self.resonance)}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class LinguisticReadback:
|
||||
"""Ordered resonance-spectrum readback of one certified ψ_steady."""
|
||||
|
||||
tokens: tuple[ReadbackToken, ...]
|
||||
psi_digest: str
|
||||
certificate_id: str
|
||||
min_resonance: float
|
||||
max_tokens: int
|
||||
vocab_size: int
|
||||
readback_id: str = ""
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
object.__setattr__(
|
||||
self,
|
||||
"readback_id",
|
||||
"readback-"
|
||||
+ _content_id(
|
||||
{
|
||||
"psi": self.psi_digest,
|
||||
"certificate": self.certificate_id,
|
||||
"tokens": [t.as_dict() for t in self.tokens],
|
||||
"min_resonance": float(self.min_resonance),
|
||||
"max_tokens": int(self.max_tokens),
|
||||
"vocab_size": int(self.vocab_size),
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
@property
|
||||
def words(self) -> tuple[str, ...]:
|
||||
return tuple(t.word for t in self.tokens)
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"readback_id": self.readback_id,
|
||||
"psi_digest": self.psi_digest,
|
||||
"certificate_id": self.certificate_id,
|
||||
"tokens": [t.as_dict() for t in self.tokens],
|
||||
"min_resonance": float(self.min_resonance),
|
||||
"max_tokens": int(self.max_tokens),
|
||||
"vocab_size": int(self.vocab_size),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RoundTripReport:
|
||||
""""Hearing ourselves think" — phase-locked agreement of the re-ingested
|
||||
articulation with the pre-articulation steady state (agreement = ρ/2)."""
|
||||
|
||||
agreement: float
|
||||
phase_correlation: float
|
||||
n_tokens: int
|
||||
domain_id: str
|
||||
psi_steady_digest: str
|
||||
psi_roundtrip_digest: str
|
||||
report_id: str = ""
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
object.__setattr__(
|
||||
self,
|
||||
"report_id",
|
||||
"roundtrip-"
|
||||
+ _content_id(
|
||||
{
|
||||
"steady": self.psi_steady_digest,
|
||||
"roundtrip": self.psi_roundtrip_digest,
|
||||
"phase_correlation": float(self.phase_correlation),
|
||||
"domain": self.domain_id,
|
||||
"n_tokens": int(self.n_tokens),
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"report_id": self.report_id,
|
||||
"agreement": float(self.agreement),
|
||||
"phase_correlation": float(self.phase_correlation),
|
||||
"n_tokens": int(self.n_tokens),
|
||||
"domain_id": self.domain_id,
|
||||
"psi_steady_digest": self.psi_steady_digest,
|
||||
"psi_roundtrip_digest": self.psi_roundtrip_digest,
|
||||
}
|
||||
|
||||
|
||||
def linguistic_readback(
|
||||
psi_steady: np.ndarray,
|
||||
certificate: RelaxationCertificate,
|
||||
verdict: EgressVerdict,
|
||||
vocab: VocabLike,
|
||||
*,
|
||||
min_resonance: float,
|
||||
max_tokens: int,
|
||||
) -> LinguisticReadback:
|
||||
"""Decode a certified hot state into its resonant token spectrum.
|
||||
|
||||
Admission chain (each failure is a typed refusal, in this order): policy
|
||||
validation → state validation → egress route must be ``readback_eligible``
|
||||
→ ψ↔certificate digest binding → non-empty vocabulary → at least one token
|
||||
with resonance ≥ ``min_resonance``.
|
||||
"""
|
||||
mr = float(min_resonance)
|
||||
if not np.isfinite(mr) or mr <= 0.0:
|
||||
raise ReadbackRefusal("min_resonance_not_positive", min_resonance=mr)
|
||||
mt = int(max_tokens)
|
||||
if mt < 1:
|
||||
raise ReadbackRefusal("max_tokens_not_positive", max_tokens=mt)
|
||||
|
||||
arr = _as_psi(psi_steady, "ψ_steady", error=ReadbackRefusal)
|
||||
if not verdict.admitted or verdict.route != "readback_eligible":
|
||||
raise ReadbackRefusal(
|
||||
"route_not_readback_eligible", route=verdict.route, verdict_reason=verdict.reason
|
||||
)
|
||||
if _psi_digest(arr) != certificate.psi_digest:
|
||||
raise ReadbackRefusal(
|
||||
"certificate_state_mismatch", certificate_id=certificate.certificate_id
|
||||
)
|
||||
n = len(vocab)
|
||||
if n == 0:
|
||||
raise ReadbackRefusal("empty_vocabulary")
|
||||
|
||||
# Resonance = ⟨ψ_steady ψ̃_T⟩_0 via the I-04 sanctioned symmetrized
|
||||
# correlation (ρ/2; reversion preserves grade-0, so the symmetrization is
|
||||
# exact — selection and the round-trip agreement share ONE metric). Note
|
||||
# ``cga_inner`` is the UN-reversed ⟨X Y⟩_0 — correct for null-vector word
|
||||
# points where reversion is identity, wrong for general versor states.
|
||||
m = WaveManifold()
|
||||
best_resonance = -np.inf
|
||||
scored: list[tuple[float, int]] = []
|
||||
for i in range(n):
|
||||
versor = np.asarray(vocab.get_versor_at(i), dtype=np.float64)
|
||||
score = float(m.phase_correlation(arr, versor)) / 2.0
|
||||
if not np.isfinite(score):
|
||||
raise ReadbackRefusal("nonfinite_resonance", index=i, word=vocab.get_word_at(i))
|
||||
if score > best_resonance:
|
||||
best_resonance = score
|
||||
if score >= mr:
|
||||
scored.append((score, i))
|
||||
if not scored:
|
||||
raise ReadbackRefusal(
|
||||
"no_resonant_token",
|
||||
best_resonance=float(best_resonance),
|
||||
min_resonance=mr,
|
||||
vocab_size=n,
|
||||
)
|
||||
# Descending resonance; ties break to the lower index (deterministic, the
|
||||
# same convention as VocabManifold.nearest's strict `>`).
|
||||
scored.sort(key=lambda pair: (-pair[0], pair[1]))
|
||||
tokens = tuple(
|
||||
ReadbackToken(word=vocab.get_word_at(i), index=i, resonance=s) for s, i in scored[:mt]
|
||||
)
|
||||
return LinguisticReadback(
|
||||
tokens=tokens,
|
||||
psi_digest=certificate.psi_digest,
|
||||
certificate_id=certificate.certificate_id,
|
||||
min_resonance=mr,
|
||||
max_tokens=mt,
|
||||
vocab_size=n,
|
||||
)
|
||||
|
||||
|
||||
def readback_packets(readback: LinguisticReadback, vocab: VocabLike) -> tuple[ModalityPacket, ...]:
|
||||
"""Lift the decoded tokens back into sensorium packets (one per token)."""
|
||||
return tuple(
|
||||
ModalityPacket(
|
||||
modality_id=f"linguistic:{t.word}",
|
||||
coefficients=np.asarray(vocab.get_versor_at(t.index), dtype=np.float64),
|
||||
)
|
||||
for t in readback.tokens
|
||||
)
|
||||
|
||||
|
||||
def hearing_ourselves_think(
|
||||
psi_steady: np.ndarray,
|
||||
readback: LinguisticReadback,
|
||||
vocab: VocabLike,
|
||||
*,
|
||||
domain_id: str,
|
||||
manifold: WaveManifold | None = None,
|
||||
) -> RoundTripReport:
|
||||
"""Re-ingest the articulated tokens and measure phase-locked agreement.
|
||||
|
||||
The articulated output is fed back through the SAME ingress construction
|
||||
boundary as external input (superpose → normalize; degenerate cancellation
|
||||
refuses there), then compared to the pre-articulation steady state with
|
||||
the metric-exact phase correlation. ``agreement = ρ/2`` so identical
|
||||
unit states score 1.0.
|
||||
"""
|
||||
arr = _as_psi(psi_steady, "ψ_steady", error=ReadbackRefusal)
|
||||
if _psi_digest(arr) != readback.psi_digest:
|
||||
raise ReadbackRefusal("readback_state_mismatch", readback_id=readback.readback_id)
|
||||
roundtrip = ingest_context(readback_packets(readback, vocab), domain_id)
|
||||
m = manifold if manifold is not None else WaveManifold()
|
||||
rho = float(m.phase_correlation(arr, roundtrip.psi))
|
||||
return RoundTripReport(
|
||||
agreement=rho / 2.0,
|
||||
phase_correlation=rho,
|
||||
n_tokens=len(readback.tokens),
|
||||
domain_id=roundtrip.domain_id,
|
||||
psi_steady_digest=readback.psi_digest,
|
||||
psi_roundtrip_digest=_psi_digest(roundtrip.psi),
|
||||
)
|
||||
|
||||
|
||||
def articulate_outcome(
|
||||
outcome: LifecycleOutcome,
|
||||
vocab: VocabLike,
|
||||
*,
|
||||
min_resonance: float,
|
||||
max_tokens: int,
|
||||
domain_id: str | None = None,
|
||||
) -> tuple[LinguisticReadback, RoundTripReport]:
|
||||
"""Composed seam-S1 stage: readback a lifecycle outcome, then round-trip it.
|
||||
|
||||
The round-trip re-ingests under the outcome's own domain unless the caller
|
||||
supplies one — the feedback loop hears itself in the same domain it spoke.
|
||||
"""
|
||||
rb = linguistic_readback(
|
||||
outcome.relaxation.psi_steady,
|
||||
outcome.relaxation.certificate,
|
||||
outcome.verdict,
|
||||
vocab,
|
||||
min_resonance=min_resonance,
|
||||
max_tokens=max_tokens,
|
||||
)
|
||||
report = hearing_ourselves_think(
|
||||
outcome.relaxation.psi_steady,
|
||||
rb,
|
||||
vocab,
|
||||
domain_id=outcome.ingress.domain_id if domain_id is None else str(domain_id),
|
||||
)
|
||||
return rb, report
|
||||
|
||||
|
||||
__all__ = [
|
||||
"LinguisticReadback",
|
||||
"ReadbackRefusal",
|
||||
"ReadbackToken",
|
||||
"RoundTripReport",
|
||||
"VocabLike",
|
||||
"articulate_outcome",
|
||||
"hearing_ourselves_think",
|
||||
"linguistic_readback",
|
||||
"readback_packets",
|
||||
]
|
||||
236
tests/test_adr_0243_linguistic_readback.py
Normal file
236
tests/test_adr_0243_linguistic_readback.py
Normal file
|
|
@ -0,0 +1,236 @@
|
|||
"""ADR-0243 §2.3 — linguistic wave readback pins (seam S1 closure).
|
||||
|
||||
Ground truth: docs/research/spark-audit-adjudication-2026-07-18.md §4 (S1) and
|
||||
ADR-0243 §2.3 — egress route ``readback_eligible`` must flow into geometric
|
||||
token selection over a versor vocabulary (token_t = argmax_T ⟨ψ_steady ψ̃_T⟩_0),
|
||||
then the "hearing ourselves think" round-trip: re-ingest the articulated tokens
|
||||
through the sensorium boundary and measure phase-locked agreement with the
|
||||
I-04 sanctioned metric (WaveManifold.phase_correlation — no cosine/ANN).
|
||||
|
||||
Fail-closed doctrine (decoding, not generating): no resonant token ⇒ typed
|
||||
ReadbackRefusal, never a fallback string. The real VocabManifold is used in
|
||||
these tests to prove the structural VocabLike protocol matches it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from algebra.rotor import make_rotor_from_angle
|
||||
from core.physics.cognitive_lifecycle import (
|
||||
CognitiveLifecycleEngine,
|
||||
compile_quadratic_well,
|
||||
)
|
||||
from core.physics.linguistic_readback import (
|
||||
LinguisticReadback,
|
||||
ReadbackRefusal,
|
||||
RoundTripReport,
|
||||
articulate_outcome,
|
||||
hearing_ourselves_think,
|
||||
linguistic_readback,
|
||||
readback_packets,
|
||||
)
|
||||
from core.physics.sensorium_wave_feed import fake_deterministic_packet
|
||||
from vocab.manifold import VocabManifold
|
||||
|
||||
# Hot-band energy inputs matching the E3/E4 precedent
|
||||
# (test_egress_routes_hot_state_to_readback_eligible).
|
||||
_HOT_ENERGY = {
|
||||
"convergence_density": 8,
|
||||
"activation_count": 8,
|
||||
"current_cycle": 1,
|
||||
"last_activation_cycle": 1,
|
||||
"morphology_features": {"mood": "imperative"},
|
||||
}
|
||||
|
||||
|
||||
def _target_rotor() -> np.ndarray:
|
||||
return np.asarray(make_rotor_from_angle(0.3, bivector_idx=6), dtype=np.float64)
|
||||
|
||||
|
||||
def _hot_outcome():
|
||||
"""Ingress → relax → egress with hot energy axes ⇒ route readback_eligible."""
|
||||
engine = CognitiveLifecycleEngine()
|
||||
target = _target_rotor()
|
||||
ham = compile_quadratic_well(target)
|
||||
packets = [fake_deterministic_packet("audio", angle=0.25, plane=6)]
|
||||
outcome = engine.solve(packets, "readback-demo", ham, energy_inputs=_HOT_ENERGY)
|
||||
assert outcome.verdict.route == "readback_eligible"
|
||||
return target, outcome
|
||||
|
||||
|
||||
def _vocab(target: np.ndarray) -> VocabManifold:
|
||||
"""Real VocabManifold: the target mode, a kindred same-plane rotor, and two
|
||||
far large-angle rotors in other planes (resonance ≈ cos(0.15)·cos(θ/2) ≪ 0.5)."""
|
||||
v = VocabManifold()
|
||||
v.add("resonant", target)
|
||||
v.add("kindred", np.asarray(make_rotor_from_angle(0.5, bivector_idx=6), dtype=np.float64))
|
||||
v.add("far-a", np.asarray(make_rotor_from_angle(2.8, bivector_idx=7), dtype=np.float64))
|
||||
v.add("far-b", np.asarray(make_rotor_from_angle(2.9, bivector_idx=8), dtype=np.float64))
|
||||
return v
|
||||
|
||||
|
||||
def test_readback_selects_most_resonant_token_first():
|
||||
target, outcome = _hot_outcome()
|
||||
vocab = _vocab(target)
|
||||
rb = linguistic_readback(
|
||||
outcome.relaxation.psi_steady,
|
||||
outcome.relaxation.certificate,
|
||||
outcome.verdict,
|
||||
vocab,
|
||||
min_resonance=0.5,
|
||||
max_tokens=4,
|
||||
)
|
||||
assert isinstance(rb, LinguisticReadback)
|
||||
assert tuple(t.word for t in rb.tokens) == ("resonant", "kindred")
|
||||
assert rb.tokens[0].resonance > 0.999 # ψ_steady locked onto the target mode
|
||||
resonances = [t.resonance for t in rb.tokens]
|
||||
assert resonances == sorted(resonances, reverse=True)
|
||||
assert all(r >= 0.5 for r in resonances)
|
||||
json.dumps(rb.as_dict()) # JSON-safe artifact
|
||||
|
||||
|
||||
def test_readback_respects_max_tokens_bound():
|
||||
target, outcome = _hot_outcome()
|
||||
rb = linguistic_readback(
|
||||
outcome.relaxation.psi_steady,
|
||||
outcome.relaxation.certificate,
|
||||
outcome.verdict,
|
||||
_vocab(target),
|
||||
min_resonance=0.5,
|
||||
max_tokens=1,
|
||||
)
|
||||
assert tuple(t.word for t in rb.tokens) == ("resonant",)
|
||||
|
||||
|
||||
def test_readback_refuses_when_route_not_eligible():
|
||||
engine = CognitiveLifecycleEngine()
|
||||
target = _target_rotor()
|
||||
ham = compile_quadratic_well(target)
|
||||
packets = [fake_deterministic_packet("audio", angle=0.25, plane=6)]
|
||||
cold = engine.solve(packets, "readback-demo", ham) # cold ⇒ crystallization route
|
||||
assert cold.verdict.route != "readback_eligible"
|
||||
with pytest.raises(ReadbackRefusal) as exc:
|
||||
linguistic_readback(
|
||||
cold.relaxation.psi_steady,
|
||||
cold.relaxation.certificate,
|
||||
cold.verdict,
|
||||
_vocab(target),
|
||||
min_resonance=0.5,
|
||||
max_tokens=4,
|
||||
)
|
||||
assert exc.value.reason == "route_not_readback_eligible"
|
||||
|
||||
|
||||
def test_readback_refuses_without_resonant_token_no_fallback():
|
||||
_target, outcome = _hot_outcome()
|
||||
sparse = VocabManifold()
|
||||
sparse.add("far-a", np.asarray(make_rotor_from_angle(2.8, bivector_idx=7), dtype=np.float64))
|
||||
sparse.add("far-b", np.asarray(make_rotor_from_angle(2.9, bivector_idx=8), dtype=np.float64))
|
||||
with pytest.raises(ReadbackRefusal) as exc:
|
||||
linguistic_readback(
|
||||
outcome.relaxation.psi_steady,
|
||||
outcome.relaxation.certificate,
|
||||
outcome.verdict,
|
||||
sparse,
|
||||
min_resonance=0.5,
|
||||
max_tokens=4,
|
||||
)
|
||||
assert exc.value.reason == "no_resonant_token"
|
||||
assert 0.0 < exc.value.disclosure["best_resonance"] < 0.5
|
||||
|
||||
|
||||
def test_readback_refuses_certificate_state_mismatch():
|
||||
target, outcome = _hot_outcome()
|
||||
foreign = np.asarray(make_rotor_from_angle(1.0, bivector_idx=7), dtype=np.float64)
|
||||
with pytest.raises(ReadbackRefusal) as exc:
|
||||
linguistic_readback(
|
||||
foreign,
|
||||
outcome.relaxation.certificate,
|
||||
outcome.verdict,
|
||||
_vocab(target),
|
||||
min_resonance=0.5,
|
||||
max_tokens=4,
|
||||
)
|
||||
assert exc.value.reason == "certificate_state_mismatch"
|
||||
|
||||
|
||||
def test_readback_refuses_empty_vocabulary_and_bad_policy():
|
||||
target, outcome = _hot_outcome()
|
||||
args = (
|
||||
outcome.relaxation.psi_steady,
|
||||
outcome.relaxation.certificate,
|
||||
outcome.verdict,
|
||||
)
|
||||
with pytest.raises(ReadbackRefusal) as exc:
|
||||
linguistic_readback(*args, VocabManifold(), min_resonance=0.5, max_tokens=4)
|
||||
assert exc.value.reason == "empty_vocabulary"
|
||||
with pytest.raises(ReadbackRefusal):
|
||||
linguistic_readback(*args, _vocab(target), min_resonance=0.0, max_tokens=4)
|
||||
with pytest.raises(ReadbackRefusal):
|
||||
linguistic_readback(*args, _vocab(target), min_resonance=0.5, max_tokens=0)
|
||||
|
||||
|
||||
def test_round_trip_agreement_above_099():
|
||||
"""Hearing ourselves think: re-ingested articulation stays phase-locked."""
|
||||
target, outcome = _hot_outcome()
|
||||
vocab = _vocab(target)
|
||||
rb = linguistic_readback(
|
||||
outcome.relaxation.psi_steady,
|
||||
outcome.relaxation.certificate,
|
||||
outcome.verdict,
|
||||
vocab,
|
||||
min_resonance=0.5,
|
||||
max_tokens=4,
|
||||
)
|
||||
report = hearing_ourselves_think(
|
||||
outcome.relaxation.psi_steady, rb, vocab, domain_id=outcome.ingress.domain_id
|
||||
)
|
||||
assert isinstance(report, RoundTripReport)
|
||||
assert report.n_tokens == 2
|
||||
assert report.agreement > 0.99
|
||||
assert report.agreement == pytest.approx(report.phase_correlation / 2.0)
|
||||
json.dumps(report.as_dict())
|
||||
|
||||
|
||||
def test_readback_packets_carry_token_versors():
|
||||
target, outcome = _hot_outcome()
|
||||
vocab = _vocab(target)
|
||||
rb = linguistic_readback(
|
||||
outcome.relaxation.psi_steady,
|
||||
outcome.relaxation.certificate,
|
||||
outcome.verdict,
|
||||
vocab,
|
||||
min_resonance=0.5,
|
||||
max_tokens=1,
|
||||
)
|
||||
(pkt,) = readback_packets(rb, vocab)
|
||||
assert pkt.modality_id == "linguistic:resonant"
|
||||
np.testing.assert_allclose(pkt.coefficients, target, atol=1e-6) # f32 store rounding
|
||||
|
||||
|
||||
def test_articulate_outcome_composes_and_is_deterministic():
|
||||
target, outcome = _hot_outcome()
|
||||
vocab = _vocab(target)
|
||||
rb1, rt1 = articulate_outcome(outcome, vocab, min_resonance=0.5, max_tokens=4)
|
||||
rb2, rt2 = articulate_outcome(outcome, vocab, min_resonance=0.5, max_tokens=4)
|
||||
assert rb1.readback_id == rb2.readback_id
|
||||
assert rt1.report_id == rt2.report_id
|
||||
assert rt1.agreement == rt2.agreement
|
||||
assert rb1.psi_digest == outcome.relaxation.certificate.psi_digest
|
||||
|
||||
|
||||
def test_module_is_pure_offserving_and_vocab_decoupled():
|
||||
import core.physics.linguistic_readback as m
|
||||
|
||||
with open(m.__file__, encoding="utf-8") as fh:
|
||||
src = fh.read()
|
||||
assert "chat.runtime" not in src
|
||||
assert "import chat" not in src
|
||||
# Layering pin: vocab access is structural (VocabLike protocol) — importing
|
||||
# the vocab package from core.physics would cycle through the barrel.
|
||||
assert "from vocab" not in src
|
||||
assert "import vocab" not in src
|
||||
Loading…
Reference in a new issue