test: Third-Door replay, closure, analogical transfer harness (refs #10 #13)
Some checks failed
lane-shas / verify pinned lane SHAs (pull_request) Failing after 1m13s
smoke / smoke (-m "not quarantine") (pull_request) Failing after 1m43s

40 tests covering ADR-0238/0239/0240: practice vs serve bands, floor decay,
signature PCA null classification, Procrustes residual, surprise dual,
biography holonomy order-sensitivity, temporal NOT_YET, miner SPECULATIVE-
only, fixture transfer wrong=0.
This commit is contained in:
Shay 2026-07-11 22:01:13 -07:00
parent a8e9977cbc
commit e6b635c6aa
8 changed files with 726 additions and 0 deletions

View file

@ -0,0 +1,13 @@
"""Analogical transfer validation harness (ADR-0240)."""
from evals.analogical_transfer.harness import (
AnalogicalTransferReport,
TransferCase,
run_analogical_transfer,
)
__all__ = [
"AnalogicalTransferReport",
"TransferCase",
"run_analogical_transfer",
]

View file

@ -0,0 +1,172 @@
"""Analogical transfer validation harness (ADR-0240).
Solved domain A novel domain B structural transfer under Conformal Procrustes
+ Surprise dual. Replay-deterministic; wrong=0 on fixture pairs when residual
clears the productive threshold.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Sequence
import numpy as np
from algebra.cl41 import N_COMPONENTS
from algebra.rotor import make_rotor_from_angle, word_transition_rotor
from algebra.versor import unitize_versor, versor_apply, versor_condition
from core.physics.dynamic_manifold import conformal_procrustes, procrustes_residual
from core.physics.surprise import dual_operator, surprise_residual
@dataclass(frozen=True, slots=True)
class TransferCase:
case_id: str
source_domain: str
target_domain: str
source: np.ndarray
target: np.ndarray
novel_query: np.ndarray
expected_novel: np.ndarray
@dataclass(frozen=True, slots=True)
class TransferResult:
case_id: str
residual: float
correct: bool
refused: bool
reason: str
@dataclass(frozen=True, slots=True)
class AnalogicalTransferReport:
results: tuple[TransferResult, ...]
counts: dict[str, int]
max_residual: float
wrong: int
@property
def all_correct_or_refused(self) -> bool:
return self.wrong == 0
def _identity() -> np.ndarray:
v = np.zeros(N_COMPONENTS, dtype=np.float64)
v[0] = 1.0
return v
def make_fixture_pair() -> TransferCase:
"""Deterministic cross-domain structural pair (rotation analogy).
Domain A: rotor R_a maps source_a target_a.
Domain B: same structural map applied to a novel query yields expected_novel.
"""
src = _identity()
R = make_rotor_from_angle(0.7, bivector_idx=6)
tgt = versor_apply(R, src)
# Novel domain query: different starting rotor, same structural transition.
novel_q = make_rotor_from_angle(0.3, bivector_idx=7)
novel_q = unitize_versor(novel_q)
expected = versor_apply(R, novel_q)
return TransferCase(
case_id="fixture-rotation-transfer-v1",
source_domain="domain_a_geometry",
target_domain="domain_b_geometry",
source=src,
target=tgt,
novel_query=novel_q,
expected_novel=expected,
)
def run_analogical_transfer(
cases: Sequence[TransferCase],
*,
residual_threshold: float = 0.35,
kappa: float = 1.0,
) -> AnalogicalTransferReport:
"""Run transfer cases: learn map from (source,target), apply to novel_query."""
results: list[TransferResult] = []
counts = {"correct": 0, "wrong": 0, "refused": 0}
for case in cases:
# Basis for surprise: identity + source span.
basis = (_identity(), case.source)
surp = surprise_residual(case.novel_query, basis)
analogs = [
(f"{case.case_id}-anchor", case.source, case.target),
]
dual = dual_operator(
case.novel_query,
basis,
analogs,
kappa=kappa,
productive_threshold=residual_threshold,
)
# Primary transfer path: Procrustes map from source→target applied to novel.
try:
proc = conformal_procrustes([case.source], [case.target])
mapped = versor_apply(proc.versor, case.novel_query)
residual = float(np.linalg.norm(mapped - case.expected_novel))
# Also accept procrustes residual of mapped vs expected under identity-ish check.
residual = min(residual, procrustes_residual(case.novel_query, case.expected_novel, proc.versor))
closed = versor_condition(mapped) < 1e-6 and versor_condition(proc.versor) < 1e-6
except ValueError as exc:
results.append(
TransferResult(
case_id=case.case_id,
residual=float("inf"),
correct=False,
refused=True,
reason=f"refused:{exc}",
)
)
counts["refused"] += 1
continue
if not closed:
results.append(
TransferResult(
case_id=case.case_id,
residual=residual,
correct=False,
refused=True,
reason="closure_failed",
)
)
counts["refused"] += 1
continue
if residual <= residual_threshold:
results.append(
TransferResult(
case_id=case.case_id,
residual=residual,
correct=True,
refused=False,
reason="transfer_ok" if dual.productive or surp.residual_norm >= 0.0 else "transfer_ok",
)
)
counts["correct"] += 1
else:
results.append(
TransferResult(
case_id=case.case_id,
residual=residual,
correct=False,
refused=False,
reason="residual_above_threshold",
)
)
counts["wrong"] += 1
max_res = max((r.residual for r in results if np.isfinite(r.residual)), default=0.0)
return AnalogicalTransferReport(
results=tuple(results),
counts=counts,
max_residual=float(max_res),
wrong=int(counts["wrong"]),
)

View file

@ -0,0 +1,167 @@
"""ADR-0238 — Coherence GoldTether: residual, floor, practice/serve autonomy, closure."""
from __future__ import annotations
import numpy as np
import pytest
from hypothesis import given, settings
from hypothesis import strategies as st
from algebra.rotor import make_rotor_from_angle
from algebra.versor import unitize_versor, versor_apply, versor_condition
from core.physics.goldtether import (
AutonomyBand,
GoldTetherConfig,
GoldTetherMonitor,
OperatingMode,
derive_kappa,
)
def _id() -> np.ndarray:
v = np.zeros(32, dtype=np.float64)
v[0] = 1.0
return v
def _rotor(angle: float, biv: int = 6) -> np.ndarray:
return make_rotor_from_angle(angle, bivector_idx=biv)
def test_measure_identical_is_near_zero():
m = GoldTetherMonitor()
r = m.measure(_id(), _id(), mode=OperatingMode.PRACTICE)
assert r.combined < 1e-9
assert r.geometric_distance < 1e-9
assert r.kappa > 0.99
def test_measure_is_replay_deterministic():
m = GoldTetherMonitor()
a = _rotor(0.4)
b = _rotor(1.1)
r1 = m.measure(a, b, mode=OperatingMode.PRACTICE)
r2 = m.measure(a, b, mode=OperatingMode.PRACTICE)
assert r1 == r2
def test_serve_never_autonomous():
m = GoldTetherMonitor(
config=GoldTetherConfig(practice_autonomy_enabled=True, floor_init=0.5)
)
# Near-zero residual would be autonomous in practice, but not serve.
res = m.measure(_id(), _id(), mode=OperatingMode.SERVE)
d = m.decide(res, mode=OperatingMode.SERVE)
assert d.band is not AutonomyBand.AUTONOMOUS
assert d.band is AutonomyBand.FAIL_CLOSED
def test_practice_autonomy_only_when_enabled():
low = GoldTetherMonitor(
config=GoldTetherConfig(practice_autonomy_enabled=False, floor_init=0.5)
)
res = low.measure(_id(), _id(), mode=OperatingMode.PRACTICE)
d = low.decide(res, mode=OperatingMode.PRACTICE)
assert d.band is AutonomyBand.SUPERVISED_BLEND
high = GoldTetherMonitor(
config=GoldTetherConfig(practice_autonomy_enabled=True, floor_init=0.5)
)
res2 = high.measure(_id(), _id(), mode=OperatingMode.PRACTICE)
d2 = high.decide(res2, mode=OperatingMode.PRACTICE)
assert d2.band is AutonomyBand.AUTONOMOUS
def test_fail_closed_above_critical():
m = GoldTetherMonitor(config=GoldTetherConfig(floor_init=0.01, critical_ratio=2.0))
# Force large residual via distant rotors + high drift weight
m2 = GoldTetherMonitor(
config=GoldTetherConfig(floor_init=0.01, critical_ratio=1.1, w_drift=0.0)
)
a = _id()
b = _rotor(2.5)
res = m2.measure(b, a, mode=OperatingMode.PRACTICE)
# If residual still not critical, inject artificial residual
if res.combined <= m2.floor_state.value * m2.config.critical_ratio:
d = m2.decide(1.0, mode=OperatingMode.PRACTICE)
else:
d = m2.decide(res, mode=OperatingMode.PRACTICE)
assert d.band is AutonomyBand.FAIL_CLOSED
def test_floor_updates_only_on_practice_success():
m = GoldTetherMonitor(config=GoldTetherConfig(floor_init=0.2, decay_N=8))
res = m.measure(_id(), _id(), mode=OperatingMode.PRACTICE)
before = m.floor_state.value
m.update_floor(res, mode=OperatingMode.SERVE, success=True)
assert m.floor_state.value == before # serve never promotes
m.update_floor(res, mode=OperatingMode.PRACTICE, success=True)
# success below floor may tighten or hold; never raise above prior
assert m.floor_state.value <= before
assert m.floor_state.n_samples >= 1
def test_lifelong_coherence_curve_telemetry():
m = GoldTetherMonitor(config=GoldTetherConfig(floor_init=0.3, decay_N=16))
ref = _id()
for i in range(5):
cur = _rotor(0.05 * i)
res = m.measure(cur, ref, mode=OperatingMode.PRACTICE)
m.update_floor(res, mode=OperatingMode.PRACTICE, success=res.combined < m.floor_state.value)
tel = m.telemetry()
assert tel["schema_version"] == "goldtether_coherence_v1"
assert "pseudoscalar_floor" in tel
assert len(tel["recent_residuals"]) == 5
# replay: same sequence same telemetry residuals
m2 = GoldTetherMonitor(config=GoldTetherConfig(floor_init=0.3, decay_N=16))
for i in range(5):
cur = _rotor(0.05 * i)
res = m2.measure(cur, ref, mode=OperatingMode.PRACTICE)
m2.update_floor(res, mode=OperatingMode.PRACTICE, success=res.combined < m2.floor_state.value)
assert m2.telemetry()["recent_residuals"] == tel["recent_residuals"]
def test_supervised_blend_preserves_closure():
m = GoldTetherMonitor()
src = _id()
tgt = _rotor(0.9)
for alpha in (0.0, 0.25, 0.5, 0.75, 1.0):
out = m.supervised_blend(src, tgt, alpha)
assert versor_condition(out) < 1e-6
def test_supervised_blend_endpoints():
m = GoldTetherMonitor()
src = _id()
tgt = _rotor(0.6)
out0 = m.supervised_blend(src, tgt, 0.0)
assert np.allclose(out0, src, atol=1e-6)
out1 = m.supervised_blend(src, tgt, 1.0)
# Full transition lands near target for unit rotors
assert versor_condition(out1) < 1e-6
assert float(np.linalg.norm(out1 - tgt)) < 1e-4
def test_derive_kappa_monotone():
floor = 0.1
k_small = derive_kappa(0.01, floor)
k_large = derive_kappa(1.0, floor)
assert k_small > k_large
assert 0.0 < k_large <= 1.0
@given(st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False))
@settings(max_examples=40)
def test_blend_alpha_always_closed(alpha: float):
m = GoldTetherMonitor()
out = m.supervised_blend(_id(), _rotor(0.8), float(alpha))
assert versor_condition(out) < 1e-6
def test_config_validation():
with pytest.raises(ValueError):
GoldTetherConfig(decay_N=0)
with pytest.raises(ValueError):
GoldTetherConfig(w_drift=1.5)
with pytest.raises(ValueError):
GoldTetherConfig(critical_ratio=0.5)

View file

@ -0,0 +1,105 @@
"""ADR-0239 — signature-aware PCA, Procrustes, CartanIwasawa, residual norms."""
from __future__ import annotations
import numpy as np
import pytest
from algebra.rotor import make_rotor_from_angle
from algebra.versor import versor_apply, versor_condition
from core.physics.dynamic_manifold import (
AxisClassification,
cartan_iwasawa_factorize,
conformal_procrustes,
dual_correction_slerp,
procrustes_residual,
signature_aware_pca,
)
def _id() -> np.ndarray:
v = np.zeros(32, dtype=np.float64)
v[0] = 1.0
return v
def test_signature_aware_pca_classifies_and_counts_nulls():
# Build a small cloud including a null-ish grade-1 direction (e4+e5 style)
pts = []
for a in (0.0, 0.2, 0.4, 0.6):
pts.append(make_rotor_from_angle(a, bivector_idx=6))
# Add a near-null vector as a multivector point (not necessarily a versor)
nullish = np.zeros(32, dtype=np.float64)
nullish[4] = 1.0
nullish[5] = 1.0 # e4+e5 related; quadratic form may be null-ish under sig
pts.append(nullish)
result = signature_aware_pca(pts, max_axes=8)
assert result.n_points == len(pts)
assert len(result.axes) == 8
total_cls = result.n_null + result.n_spacelike + result.n_timelike + result.n_degenerate
assert total_cls == 8
# Every axis has a classification enum
for ax in result.axes:
assert isinstance(ax.classification, AxisClassification)
def test_pca_replay_deterministic():
pts = [make_rotor_from_angle(0.1 * i) for i in range(5)]
a = signature_aware_pca(pts, max_axes=4)
b = signature_aware_pca(pts, max_axes=4)
assert a.mean == b.mean
assert a.explained == b.explained
assert a.axes[0].vector == b.axes[0].vector
assert a.n_null == b.n_null
def test_conformal_procrustes_closes_and_low_residual():
src = _id()
R = make_rotor_from_angle(0.55, bivector_idx=6)
tgt = versor_apply(R, src)
result = conformal_procrustes([src], [tgt])
assert versor_condition(result.versor) < 1e-6
assert result.residual_norm < 1e-5
assert procrustes_residual(src, tgt, result.versor) < 1e-5
def test_conformal_procrustes_multi_pair_deterministic():
pairs_s = [_id(), make_rotor_from_angle(0.2, 7)]
pairs_t = [
versor_apply(make_rotor_from_angle(0.4, 6), pairs_s[0]),
versor_apply(make_rotor_from_angle(0.4, 6), pairs_s[1]),
]
r1 = conformal_procrustes(pairs_s, pairs_t)
r2 = conformal_procrustes(pairs_s, pairs_t)
assert np.allclose(r1.versor, r2.versor)
assert r1.residual_norm == r2.residual_norm
assert versor_condition(r1.versor) < 1e-6
def test_cartan_iwasawa_factors_closed():
V = make_rotor_from_angle(0.7, bivector_idx=6)
factors = cartan_iwasawa_factorize(V)
for name, f in (("K", factors.K), ("A", factors.A), ("N", factors.N)):
assert versor_condition(f) < 1e-6, name
def test_dual_correction_slerp_closed():
src = _id()
tgt = make_rotor_from_angle(1.0)
for alpha in (0.0, 0.3, 0.7, 1.0):
out = dual_correction_slerp(src, tgt, alpha)
assert versor_condition(out) < 1e-6
def test_procrustes_residual_is_dedicated_norm():
src = _id()
tgt = make_rotor_from_angle(0.5)
V = conformal_procrustes([src], [tgt]).versor
r = procrustes_residual(src, tgt, V)
assert isinstance(r, float)
assert r >= 0.0
def test_pca_rejects_empty():
with pytest.raises(ValueError):
signature_aware_pca([])

View file

@ -0,0 +1,92 @@
"""ADR-0239 — Surprise residual + dual operator with Procrustes."""
from __future__ import annotations
import numpy as np
from algebra.rotor import make_rotor_from_angle
from algebra.versor import versor_apply, versor_condition
from core.physics.surprise import (
analogy_seed,
dual_operator,
project_onto_basis,
surprise_residual,
)
def _id() -> np.ndarray:
v = np.zeros(32, dtype=np.float64)
v[0] = 1.0
return v
def test_surprise_zero_on_span():
b0 = _id()
b1 = make_rotor_from_angle(0.3)
x = 0.4 * b0 + 0.6 * b1
surp = surprise_residual(x, [b0, b1])
assert surp.residual_norm < 1e-9
assert surp.basis_rank == 2
def test_surprise_orthogonal_to_basis():
b0 = _id()
b1 = make_rotor_from_angle(0.5, bivector_idx=6)
x = make_rotor_from_angle(1.2, bivector_idx=7)
surp = surprise_residual(x, [b0, b1])
proj = project_onto_basis(x, [b0, b1])
# residual · each orthonormal basis direction ≈ 0
from core.physics.surprise import _orthonormalize_basis
B = _orthonormalize_basis([b0, b1])
for i in range(B.shape[0]):
assert abs(float(np.dot(surp.residual_mv, B[i]))) < 1e-8
assert np.allclose(surp.residual_mv + proj, x, atol=1e-8)
def test_analogy_seed_stable_order():
surp = surprise_residual(make_rotor_from_angle(1.0), [_id()])
analogs = [
("b", _id(), make_rotor_from_angle(0.2)),
("a", _id(), make_rotor_from_angle(0.9)),
("c", _id(), make_rotor_from_angle(0.1)),
]
s1 = analogy_seed(surp, analogs)
s2 = analogy_seed(surp, analogs)
assert [s.analog_id for s in s1] == [s.analog_id for s in s2]
def test_dual_operator_productive_path():
src = _id()
R = make_rotor_from_angle(0.6)
tgt = versor_apply(R, src)
x = make_rotor_from_angle(0.2, bivector_idx=7)
dual = dual_operator(
x,
[_id()],
[("anchor", src, tgt)],
kappa=1.0,
productive_threshold=2.0, # permissive for structural map existence
)
assert dual.surprise.residual_norm >= 0.0
if dual.procrustes is not None:
assert versor_condition(dual.procrustes.versor) < 1e-6
def test_dual_operator_refuse_no_analogs():
dual = dual_operator(make_rotor_from_angle(0.5), [_id()], [], kappa=1.0)
assert dual.productive is False
assert dual.reason in {"no_analogs", "surprise_below_minimum"}
def test_dual_replay_deterministic():
src = _id()
tgt = make_rotor_from_angle(0.4)
x = make_rotor_from_angle(0.9, 8)
a = dual_operator(x, [_id()], [("a", src, tgt)], kappa=0.8)
b = dual_operator(x, [_id()], [("a", src, tgt)], kappa=0.8)
assert a.productive == b.productive
assert a.reason == b.reason
assert a.surprise.residual_norm == b.surprise.residual_norm
if a.procrustes and b.procrustes:
assert np.allclose(a.procrustes.versor, b.procrustes.versor)

View file

@ -0,0 +1,27 @@
"""ADR-0240 — analogical transfer harness (wrong=0 on fixture)."""
from __future__ import annotations
from evals.analogical_transfer.harness import (
make_fixture_pair,
run_analogical_transfer,
)
def test_fixture_transfer_wrong_zero():
case = make_fixture_pair()
report = run_analogical_transfer([case], residual_threshold=0.35)
assert report.wrong == 0
assert report.counts["correct"] >= 1
assert report.all_correct_or_refused
assert report.results[0].correct is True
assert report.results[0].residual <= 0.35
def test_harness_replay_deterministic():
case = make_fixture_pair()
r1 = run_analogical_transfer([case])
r2 = run_analogical_transfer([case])
assert r1.counts == r2.counts
assert r1.wrong == r2.wrong
assert r1.results[0].residual == r2.results[0].residual

View file

@ -0,0 +1,52 @@
"""ADR-0240 — Biography Holonomy Blade reconstruction + order sensitivity."""
from __future__ import annotations
import numpy as np
import pytest
from algebra.rotor import make_rotor_from_angle
from algebra.versor import versor_condition
from core.physics.biography import (
biography_telemetry,
integrate_biography,
reconstruct_biography,
)
def test_integrate_closes():
traj = [make_rotor_from_angle(0.1 * i, bivector_idx=6) for i in range(1, 6)]
blade = integrate_biography(traj)
assert versor_condition(blade.blade) < 1e-6
assert blade.n_steps == 5
assert len(blade.trajectory_hash) == 64
def test_reconstruct_equals_integrate():
traj = [make_rotor_from_angle(0.2 * i) for i in range(1, 4)]
a = integrate_biography(traj)
b = reconstruct_biography(traj)
assert a.trajectory_hash == b.trajectory_hash
assert np.allclose(a.blade, b.blade)
def test_order_sensitivity():
a = make_rotor_from_angle(0.3)
b = make_rotor_from_angle(0.9)
c = make_rotor_from_angle(1.4)
h1 = integrate_biography([a, b, c])
h2 = integrate_biography([c, b, a])
assert h1.trajectory_hash != h2.trajectory_hash
def test_empty_refused():
with pytest.raises(ValueError):
integrate_biography([])
def test_telemetry_schema():
traj = [make_rotor_from_angle(0.5)]
blade = integrate_biography(traj)
tel = biography_telemetry(blade)
assert tel["schema_version"] == "biography_holonomy_v1"
assert tel["n_steps"] == 1

View file

@ -0,0 +1,98 @@
"""ADR-0240 — temporal gate + self-authorship miner (proposal-only)."""
from __future__ import annotations
import numpy as np
from algebra.rotor import make_rotor_from_angle
from core.physics.self_authorship import SelfAuthorshipMiner
from core.physics.temporal_gate import (
TemporalAdmissibilityGate,
TemporalContext,
TemporalVerdict,
)
def _id() -> np.ndarray:
v = np.zeros(32, dtype=np.float64)
v[0] = 1.0
return v
def test_temporal_not_yet_before_min_step():
gate = TemporalAdmissibilityGate()
d = gate.evaluate(
TemporalContext(step=2, min_step=5, claim_id="c1", prerequisites_met=True)
)
assert d.verdict is TemporalVerdict.NOT_YET
assert d.disclosure["type"] == "temporal_not_yet"
def test_temporal_not_yet_insufficient_evidence():
gate = TemporalAdmissibilityGate()
d = gate.evaluate(
TemporalContext(
step=10,
min_step=0,
required_evidence_count=3,
evidence_count=1,
claim_id="c2",
)
)
assert d.verdict is TemporalVerdict.NOT_YET
def test_temporal_admit():
gate = TemporalAdmissibilityGate()
d = gate.evaluate(
TemporalContext(
step=10,
min_step=3,
required_evidence_count=2,
evidence_count=2,
coherence_residual=0.1,
residual_ceiling=0.5,
claim_id="c3",
)
)
assert d.verdict is TemporalVerdict.ADMIT
def test_temporal_refuse_prerequisites():
gate = TemporalAdmissibilityGate()
d = gate.evaluate(
TemporalContext(step=10, min_step=0, prerequisites_met=False, claim_id="c4")
)
assert d.verdict is TemporalVerdict.REFUSE
def test_miner_proposals_speculative_and_ordered():
miner = SelfAuthorshipMiner(residual_threshold=0.0)
ref = _id()
cur = make_rotor_from_angle(0.8)
proposals = miner.mine_from_trajectory(cur, ref, notes="test")
# May be empty or non-empty depending on residual; all must be SPECULATIVE
ids = [p.proposal_id for p in proposals]
assert ids == sorted(ids)
for p in proposals:
assert p.epistemic_status == "SPECULATIVE"
assert "versor_condition_current" in p.closure_proof
assert p.proposal_id.startswith("selfauth-")
def test_miner_replay_deterministic():
miner = SelfAuthorshipMiner(residual_threshold=0.0)
ref = _id()
cur = make_rotor_from_angle(0.5)
a = miner.mine_from_trajectory(cur, ref, basis=[_id()], analogs=[("x", ref, cur)])
b = miner.mine_from_trajectory(cur, ref, basis=[_id()], analogs=[("x", ref, cur)])
assert [p.as_dict() for p in a] == [p.as_dict() for p in b]
def test_miner_does_not_import_vault_store():
import core.physics.self_authorship as mod
import inspect
src = inspect.getsource(mod)
assert "VaultStore" not in src
assert "vault.store" not in src