diff --git a/core/contemplation/__init__.py b/core/contemplation/__init__.py index 435516bb..49b88322 100644 --- a/core/contemplation/__init__.py +++ b/core/contemplation/__init__.py @@ -7,7 +7,15 @@ ADR-0241 P9: Trace A wave seam may SPECULATIVE-seal standing-wave modes and emit RESONANT_MODE_CANDIDATE findings — never COHERENT, never serve-wired. """ -from .runner import contemplate_frontier_reports, run_contemplation +from typing import TYPE_CHECKING, Any + +from .runner import ( + SurpriseDiscoveryOutcome, + SurpriseObservation, + contemplate_frontier_reports, + contemplate_surprise_history, + run_contemplation, +) from .schema import ( ContemplationEvidenceRef, ContemplationFinding, @@ -15,23 +23,53 @@ from .schema import ( FindingKind, ) from .snapshot import ContemplationSubstrate -from .wave_seam import ( - WaveModeHypothesis, - WaveReconstructResult, - reconstruct_as_evidence, - reconstruct_as_hypothesis, - speculative_seal_from_contemplation, + +if TYPE_CHECKING: + from .wave_seam import ( + WaveModeHypothesis, + WaveReconstructResult, + reconstruct_as_evidence, + reconstruct_as_hypothesis, + speculative_seal_from_contemplation, + ) + +# A-04 process quarantine: chat/runtime.py's idle_tick lazily imports +# core.contemplation.runner inside the serve process, which imports this +# package __init__ first. wave_seam eagerly pulls +# core.physics.holographic_vault — a serve-banned Tier-2 module +# (tests/test_serve_quarantine_transitive.py) — so its re-exports are lazy +# (PEP 562): they load only when actually referenced, which never happens on +# the serve path. +_WAVE_SEAM_EXPORTS = frozenset( + { + "WaveModeHypothesis", + "WaveReconstructResult", + "reconstruct_as_evidence", + "reconstruct_as_hypothesis", + "speculative_seal_from_contemplation", + } ) + +def __getattr__(name: str) -> Any: + if name in _WAVE_SEAM_EXPORTS: + from core.contemplation import wave_seam + + return getattr(wave_seam, name) + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + __all__ = [ "ContemplationEvidenceRef", "ContemplationFinding", "ContemplationRun", "ContemplationSubstrate", "FindingKind", + "SurpriseDiscoveryOutcome", + "SurpriseObservation", "WaveModeHypothesis", "WaveReconstructResult", "contemplate_frontier_reports", + "contemplate_surprise_history", "reconstruct_as_evidence", "reconstruct_as_hypothesis", "run_contemplation", diff --git a/core/contemplation/runner.py b/core/contemplation/runner.py index dec35f87..da855dc9 100644 --- a/core/contemplation/runner.py +++ b/core/contemplation/runner.py @@ -2,9 +2,12 @@ from __future__ import annotations import hashlib import json +import math import os +from dataclasses import dataclass from pathlib import Path -from typing import Iterable +from types import MappingProxyType +from typing import TYPE_CHECKING, Any, Iterable, Mapping, Sequence from core.contemplation.miners.contradiction_detection import ( mine_contradiction_detection_report, @@ -16,8 +19,19 @@ from core.contemplation.schema import ( format_contemplation_finding_jsonl, ) from core.contemplation.snapshot import ContemplationSubstrate +from core.physics.surprise import DEFAULT_DISCOVERY_GAMMA, is_discovery_eligible +from teaching.discovery import DiscoveryCandidate, emit_surprise_discovery from teaching.discovery_sink import DiscoveryCandidateSink +if TYPE_CHECKING: + # A-04 process quarantine: chat/runtime.py's idle_tick lazily imports THIS + # module inside the serve process (frontier contemplation pass), and + # core.physics.multi_scale_energy is on the serve-banned list + # (tests/test_serve_quarantine_transitive.py). The gate import is + # function-local in contemplate_surprise_history; only the type name is + # needed here. + from core.physics.multi_scale_energy import CrossBandVerdict + def _config_hash(payload: dict[str, object]) -> str: encoded = json.dumps( @@ -55,6 +69,195 @@ def _emit_findings( sink.emit(format_contemplation_finding_jsonl(finding)) +# --- ADR-0243 §2.4 / ADR-0242 §5-P2 — surprise → discovery wiring ------------ +# +# Surprise-signal sourcing (Lane A open question, resolved by reading the +# substrate): `ContemplationFinding` carries no surprise measurement — the +# report-mining paths above have no geometry, and fabricating a per-finding +# number would be exactly the failure mode the brief warns against. The +# canonical surprise carrier is the ADR-0239 dual-operator audit dict +# (`dual_procrustes_surprise` / `dual_operator` output), produced live by +# `core.physics.self_authorship.SelfAuthorshipMiner` and the ADR-0240 +# harness. The wiring below therefore consumes explicit caller-timed +# observations of that dict; the report-mining finding paths are untouched. + + +@dataclass(frozen=True, slots=True) +class SurpriseObservation: + """One caller-timed ADR-0239 dual-operator audit measurement. + + ``t`` is caller-supplied logical time (same clock as ``now`` in + :func:`contemplate_surprise_history`) — never a wall clock, so the + pipeline stays deterministic and replayable. + """ + + t: float + dual: Mapping[str, Any] + source_turn_trace: str = "" + + def __post_init__(self) -> None: + if not math.isfinite(float(self.t)): + raise ValueError("SurpriseObservation.t must be finite") + if "surprise_norm" not in self.dual: + raise ValueError( + "SurpriseObservation.dual must carry 'surprise_norm' " + "(ADR-0239 dual-operator audit dict)" + ) + try: + norm = float(self.dual["surprise_norm"]) + except (TypeError, ValueError) as exc: + raise ValueError( + "SurpriseObservation.dual['surprise_norm'] must be numeric" + ) from exc + if math.isfinite(norm) and norm < 0.0: + raise ValueError( + "SurpriseObservation.dual['surprise_norm'] must be >= 0 " + "(it is a residual norm)" + ) + refused = self.dual.get("surprise_refused") + if refused is not None and not isinstance(refused, str): + raise ValueError( + "SurpriseObservation.dual['surprise_refused'] must be a string " + "refusal reason or None — a non-string truthy marker would " + "silently read as 'not refused'" + ) + # Snapshot the mapping (shallow) so the observation cannot be mutated + # out from under the deterministic gate after construction. + object.__setattr__(self, "dual", MappingProxyType(dict(self.dual))) + + +@dataclass(frozen=True, slots=True) +class SurpriseDiscoveryOutcome: + """Typed audit record of one discovery-gate pass. + + ``reason`` is one of: ``emitted``, ``not_discovery_eligible``, + ``insufficient_span``, ``band_below_gamma``, + ``candidate_not_constructed``. + """ + + candidate: DiscoveryCandidate | None + verdict: CrossBandVerdict | None + reason: str + surprise_norm: float + discovery_gamma: float + + +def _measured_surprise(dual: Mapping[str, Any]) -> tuple[float, str | None, bool]: + """Extract (surprise_norm, refusal, productive) from a dual audit dict. + + Types are guaranteed by ``SurpriseObservation.__post_init__`` — the dual + always carries a numeric ``surprise_norm`` and a ``str | None`` refusal. + """ + surprise_norm = float(dual["surprise_norm"]) + refusal = dual.get("surprise_refused") + productive = bool(dual.get("productive", False) or dual.get("transfer_accepted", False)) + return surprise_norm, refusal, productive + + +def contemplate_surprise_history( + observations: Sequence[SurpriseObservation], + *, + now: float, + tau0: float = 1.0, + discovery_gamma: float | None = None, + sink: DiscoveryCandidateSink | None = None, +) -> SurpriseDiscoveryOutcome: + """Gate a timed surprise history into at most one ``DiscoveryCandidate``. + + ADR-0243 §2.4: high surprise on a speculative wave is a discovery + signal, held as a ``DiscoveryCandidate`` and routed to the offline + review corridor. Two gates run in order, both ADR-0242 §5 pinned + primitives with unchanged signatures: + + 1. :func:`core.physics.surprise.is_discovery_eligible` on the newest + observation — measured residual above γ, no metric refusal, not a + productive transfer. + 2. :func:`core.physics.multi_scale_energy.cross_band_discovery_gate` + on the measured history — the signal must persist across the + F5/F6/F7 Fibonacci bands, so transient noise never emits. + + Only when both pass is a candidate built and emitted through the + existing sink protocol via + :func:`teaching.discovery.emit_surprise_discovery` — the only side + effect is ``sink.emit(...)`` (proposal plumbing, never a vault + write). Refused observations carry no measured energy and are + excluded from the band accumulation (exclusion can only lower band + energy — fail-closed); a refused *newest* observation never emits. + """ + obs = tuple(observations) + if not obs: + raise ValueError("contemplate_surprise_history: empty observation history") + for earlier, later in zip(obs, obs[1:]): + if float(later.t) < float(earlier.t): + raise ValueError( + "contemplate_surprise_history: observations must be in " + "non-decreasing time order" + ) + + current = obs[-1] + surprise_norm, refusal, productive = _measured_surprise(current.dual) + gamma = float( + discovery_gamma + if discovery_gamma is not None + else current.dual.get("discovery_gamma", DEFAULT_DISCOVERY_GAMMA) + ) + + if not is_discovery_eligible( + surprise_norm=surprise_norm, + productive_or_transfer=productive, + surprise_refused=refusal, + discovery_gamma=gamma, + ): + return SurpriseDiscoveryOutcome( + candidate=None, + verdict=None, + reason="not_discovery_eligible", + surprise_norm=surprise_norm, + discovery_gamma=gamma, + ) + + # Function-local on purpose (A-04): chat/runtime.py's idle_tick imports + # this MODULE inside the serve process; the serve-banned Tier-2 gate may + # only load when the discovery path actually runs (off-serving contexts). + from core.physics.multi_scale_energy import cross_band_discovery_gate + + events: list[tuple[float, float]] = [] + for o in obs: + norm, o_refusal, _ = _measured_surprise(o.dual) + if o_refusal is None and math.isfinite(norm): + events.append((float(o.t), norm)) + verdict = cross_band_discovery_gate(events, now=float(now), tau0=tau0, gamma=gamma) + if not verdict.eligible: + return SurpriseDiscoveryOutcome( + candidate=None, + verdict=verdict, + reason=verdict.reason, + surprise_norm=surprise_norm, + discovery_gamma=gamma, + ) + + # The runner has already judged eligibility with the effective γ; a stale + # measurement-time ``discovery_eligible`` flag baked into the dual dict + # must not veto (or resurrect) that verdict — force downstream + # recomputation against the same γ both gates used. + dual_for_emit = dict(current.dual) + dual_for_emit["discovery_gamma"] = gamma + dual_for_emit.pop("discovery_eligible", None) + candidate = emit_surprise_discovery( + dual_for_emit, + sink, + source_turn_trace=current.source_turn_trace, + discovery_gamma=gamma, + ) + return SurpriseDiscoveryOutcome( + candidate=candidate, + verdict=verdict, + reason="emitted" if candidate is not None else "candidate_not_constructed", + surprise_norm=surprise_norm, + discovery_gamma=gamma, + ) + + def contemplate_frontier_reports( report_paths: Iterable[str | Path], *, @@ -191,8 +394,11 @@ def write_contemplation_run(run: ContemplationRun, path: str | Path) -> None: __all__ = [ + "SurpriseDiscoveryOutcome", + "SurpriseObservation", "contemplate_contradiction_reports", "contemplate_frontier_reports", + "contemplate_surprise_history", "run_contemplation", "write_contemplation_run", ] diff --git a/evals/analogical_transfer/kappa_calibration.py b/evals/analogical_transfer/kappa_calibration.py new file mode 100644 index 00000000..4ad33069 --- /dev/null +++ b/evals/analogical_transfer/kappa_calibration.py @@ -0,0 +1,114 @@ +"""ADR-0242 §5-P1 — live κ line-search caller (calibration pipeline only). + +`core.physics.goldtether.propose_kappa_line_search` and its internal +`kappa_search_event` telemetry emission were fully implemented and pinned +(`tests/test_adr_0242_carry_seams.py`) but had zero live call sites. This +module is the live caller, confined to the ADR-0240 analogical-transfer +calibration pipeline per the master plan's R-04 mitigation ("trace +generation ... limited strictly to the calibration and training-loop +pipelines" — never a hot/serve path; `chat/runtime.py` must never import +this package). + +What is calibrated: in `core.physics.surprise.dual_operator` the +productive-transfer threshold is κ-scaled (``thr = productive_threshold / +κ``). The objective below proposes the in-bounds κ whose scaled threshold +sits closest to the worst measured Procrustes residual of the calibration +corpus — i.e. κ is tuned so the acceptance bar tracks the corpus's actual +structural-residual scale. The result is PROPOSAL-ONLY: nothing here +mutates `GoldTetherMonitor` state, COHERENT standing, or serve autonomy; +failures fall back to baseline κ = 1.0 inside the search itself. +""" + +from __future__ import annotations + +import math +from typing import Any, Sequence, cast + +import numpy as np + +from core.physics.dynamic_manifold import conformal_procrustes +from core.physics.goldtether import kappa_search_event, propose_kappa_line_search +from evals.analogical_transfer.harness import TransferCase, make_fixture_pair + + +def calibrate_transfer_kappa( + cases: Sequence[TransferCase] | None = None, + *, + productive_threshold: float = 0.35, + lower: float = 0.1, + upper: float = 2.0, + evaluation_budget: int = 16, + sink: Any | None = None, +) -> dict[str, Any]: + """Propose a κ for the transfer corridor via Fibonacci section search. + + Per-case Procrustes residuals are measured ONCE, outside the search + loop, so the κ objective is cheap arithmetic (R-04: bounded + calibration cost). Cases whose residual is not measurable + (Procrustes refusal or non-finite residual) are excluded and counted; + if none remain the calibration refuses (typed ``ValueError``) rather + than proposing κ from nothing. + + When ``sink`` is provided (any ``emit(line: str)`` object), the + §5-P1 ``kappa_search_event`` fires exactly once from inside + :func:`core.physics.goldtether.propose_kappa_line_search`. + """ + corpus = tuple(cases) if cases is not None else (make_fixture_pair(),) + if not corpus: + raise ValueError("calibrate_transfer_kappa: cases must be non-empty") + threshold = float(productive_threshold) + if not math.isfinite(threshold) or threshold <= 0.0: + raise ValueError("productive_threshold must be finite and > 0") + + residuals: list[float] = [] + refused = 0 + for case in corpus: + try: + # conformal_procrustes accepts sequences of 32-vectors (same call + # shape as the harness); its annotation is narrower than runtime. + proc_residual = float( + conformal_procrustes( + cast(np.ndarray, case.source), + cast(np.ndarray, case.target), + )[1] + ) + except ValueError: + refused += 1 + continue + if not math.isfinite(proc_residual): + refused += 1 + continue + residuals.append(proc_residual) + if not residuals: + raise ValueError( + "calibrate_transfer_kappa refused: no admissible cases " + f"({refused} of {len(corpus)} procrustes-refused or non-finite)" + ) + + # Worst measured structural residual is the binding acceptance constraint. + r_star = max(residuals) + + def residual_fn(kappa: float) -> float: + return (threshold / float(kappa) - r_star) ** 2 + + proposed_kappa, outcome = propose_kappa_line_search( + residual_fn, + lower=float(lower), + upper=float(upper), + evaluation_budget=int(evaluation_budget), + objective_id="analogical_transfer_kappa", + objective_version="v1", + sink=sink, + ) + return { + "kind": "analogical_transfer_kappa_calibration", + "proposed_kappa": float(proposed_kappa), + "productive_threshold": threshold, + "max_procrustes_residual": r_star, + "case_count": len(residuals), + "refused_case_count": refused, + "event": kappa_search_event(proposed_kappa, outcome), + } + + +__all__ = ["calibrate_transfer_kappa"] diff --git a/tests/test_adr_0243_lane_a_discovery_wiring.py b/tests/test_adr_0243_lane_a_discovery_wiring.py new file mode 100644 index 00000000..03fa513b --- /dev/null +++ b/tests/test_adr_0243_lane_a_discovery_wiring.py @@ -0,0 +1,287 @@ +"""ADR-0243 Phase 3 Lane A — contemplation runner discovery wiring. + +Wires the ADR-0242 §5 pinned primitives (`is_discovery_eligible`, +`cross_band_discovery_gate`) into the contemplation runner so a +high-surprise, band-persistent observation history emits a +`DiscoveryCandidate` through the existing `DiscoveryCandidateSink` +(ADR-0243 §2.4 discovery signal). + +Surprise-signal sourcing (the Lane A open question, resolved by reading +the substrate): `ContemplationFinding` carries NO surprise measurement — +the report-mining paths (frontier compare / contradiction detection) have +no geometry and are deliberately untouched. The canonical surprise +carrier is the ADR-0239 dual-operator audit dict +(`core.physics.surprise.dual_procrustes_surprise` / `dual_operator` +output), produced live by `SelfAuthorshipMiner` and the ADR-0240 +harness. The runner therefore consumes explicit caller-timed +`SurpriseObservation`s — never a fabricated per-finding number, never a +wall clock. +""" + +from __future__ import annotations + +import json + +import pytest + +from core.contemplation.runner import ( + SurpriseDiscoveryOutcome, + SurpriseObservation, + contemplate_surprise_history, +) +from core.physics.multi_scale_energy import CrossBandVerdict +from teaching.discovery import DiscoveryCandidate +from teaching.discovery_sink import DiscoveryBufferSink + + +def _dual( + surprise_norm: float, + *, + refused: str | None = None, + productive: bool = False, + gamma: float = 0.35, +) -> dict: + """ADR-0239 dual-operator audit dict shape (see dual_procrustes_surprise).""" + return { + "surprise_norm": surprise_norm, + "surprise_refused": refused, + "transfer_accepted": productive, + "discovery_eligible": ( + refused is None and not productive and surprise_norm > gamma + ), + "discovery_gamma": gamma, + "procrustes_residual": 0.5, + } + + +def _sustained( + surprise_norm: float = 0.5, + *, + n: int = 14, + tau0: float = 1.0, + trace: str = "trace-sustained", +) -> list[SurpriseObservation]: + """Mirror of the carry-seams `_sustained` history: eligible across bands.""" + return [ + SurpriseObservation( + t=float(i) * tau0, + dual=_dual(surprise_norm), + source_turn_trace=trace, + ) + for i in range(n) + ] + + +# --- emission path ------------------------------------------------------------ + + +def test_sustained_high_surprise_emits_candidate_on_sink() -> None: + sink = DiscoveryBufferSink() + out = contemplate_surprise_history( + _sustained(0.5), + now=13.0, + tau0=1.0, + sink=sink, + ) + assert isinstance(out, SurpriseDiscoveryOutcome) + assert out.reason == "emitted" + assert isinstance(out.candidate, DiscoveryCandidate) + assert out.candidate.trigger == "high_surprise" + assert out.candidate.review_state == "unreviewed" + assert out.candidate.proposed_chain["surprise_norm"] == pytest.approx(0.5) + assert isinstance(out.verdict, CrossBandVerdict) + assert out.verdict.eligible is True + assert len(sink.lines) == 1 + line = json.loads(sink.lines[0]) + assert line["trigger"] == "high_surprise" + assert line["source_turn_trace"] == "trace-sustained" + + +def test_no_sink_is_a_pure_factory() -> None: + out = contemplate_surprise_history(_sustained(0.5), now=13.0, tau0=1.0) + assert out.reason == "emitted" + assert isinstance(out.candidate, DiscoveryCandidate) + + +# --- point-eligibility gate (is_discovery_eligible) --------------------------- + + +def test_low_surprise_is_not_discovery_eligible() -> None: + sink = DiscoveryBufferSink() + out = contemplate_surprise_history( + _sustained(0.1), now=13.0, tau0=1.0, sink=sink + ) + assert out.candidate is None + assert out.verdict is None + assert out.reason == "not_discovery_eligible" + assert sink.lines == [] + + +def test_refused_current_observation_never_emits() -> None: + sink = DiscoveryBufferSink() + history = _sustained(0.5)[:-1] + [ + SurpriseObservation( + t=13.0, + dual=_dual(float("inf"), refused="degenerate_metric_span"), + ) + ] + out = contemplate_surprise_history(history, now=13.0, tau0=1.0, sink=sink) + assert out.candidate is None + assert out.reason == "not_discovery_eligible" + assert sink.lines == [] + + +def test_productive_transfer_is_not_discovery() -> None: + sink = DiscoveryBufferSink() + history = _sustained(0.5)[:-1] + [ + SurpriseObservation(t=13.0, dual=_dual(0.9, productive=True)) + ] + out = contemplate_surprise_history(history, now=13.0, tau0=1.0, sink=sink) + assert out.candidate is None + assert out.reason == "not_discovery_eligible" + assert sink.lines == [] + + +# --- persistence gate (cross_band_discovery_gate) ----------------------------- + + +def test_single_fresh_spike_is_blocked_by_persistence_gate() -> None: + sink = DiscoveryBufferSink() + out = contemplate_surprise_history( + [SurpriseObservation(t=10.0, dual=_dual(5.0))], + now=10.0, + tau0=1.0, + sink=sink, + ) + assert out.candidate is None + assert isinstance(out.verdict, CrossBandVerdict) + assert out.verdict.eligible is False + assert out.reason == "insufficient_span" + assert sink.lines == [] + + +def test_decayed_burst_is_blocked_band_below_gamma() -> None: + sink = DiscoveryBufferSink() + history = [ + SurpriseObservation(t=t, dual=_dual(1.0)) for t in (0.0, 2.0, 4.0, 6.0) + ] + out = contemplate_surprise_history(history, now=40.0, tau0=1.0, sink=sink) + assert out.candidate is None + assert out.reason == "band_below_gamma" + assert sink.lines == [] + + +def test_refused_history_events_contribute_no_energy() -> None: + """A refusal has no measured energy: excluding it can only lower band + energy (fail-closed direction), never raise it.""" + measured = _sustained(0.5) + with_refusals: list[SurpriseObservation] = [] + for obs in measured: + with_refusals.append( + SurpriseObservation( + t=obs.t, + dual=_dual(float("inf"), refused="degenerate_metric_span"), + ) + ) + with_refusals.append(obs) + baseline = contemplate_surprise_history(measured, now=13.0, tau0=1.0) + interleaved = contemplate_surprise_history(with_refusals, now=13.0, tau0=1.0) + assert baseline.verdict is not None and interleaved.verdict is not None + assert interleaved.verdict.band_energies == baseline.verdict.band_energies + # The interleaved current observation is refused, so only the baseline emits. + assert baseline.reason == "emitted" + + +# --- determinism and validation ---------------------------------------------- + + +def test_deterministic_and_pure() -> None: + sink_a = DiscoveryBufferSink() + sink_b = DiscoveryBufferSink() + a = contemplate_surprise_history(_sustained(0.5), now=13.0, tau0=1.0, sink=sink_a) + b = contemplate_surprise_history(_sustained(0.5), now=13.0, tau0=1.0, sink=sink_b) + assert a == b + assert sink_a.lines == sink_b.lines + assert a.candidate is not None and b.candidate is not None + assert a.candidate.candidate_id == b.candidate.candidate_id + + +def test_empty_observations_refused() -> None: + with pytest.raises(ValueError): + contemplate_surprise_history([], now=1.0) + + +def test_missing_surprise_norm_refused() -> None: + with pytest.raises(ValueError): + contemplate_surprise_history( + [SurpriseObservation(t=0.0, dual={"discovery_gamma": 0.35})], + now=1.0, + ) + + +def test_decreasing_time_order_refused() -> None: + history = [ + SurpriseObservation(t=5.0, dual=_dual(0.5)), + SurpriseObservation(t=1.0, dual=_dual(0.5)), + ] + with pytest.raises(ValueError): + contemplate_surprise_history(history, now=6.0) + + +def test_observation_after_now_refused_by_gate() -> None: + with pytest.raises(ValueError): + contemplate_surprise_history(_sustained(0.5), now=1.0, tau0=1.0) + + +def test_looser_gamma_override_still_emits() -> None: + """A stale measurement-time ``discovery_eligible: False`` baked into the + dual dict must not veto a verdict both runner gates reached with the + effective (looser) γ.""" + sink = DiscoveryBufferSink() + history = _sustained(0.2) # dual bakes gamma=0.35 → discovery_eligible False + assert history[-1].dual["discovery_eligible"] is False + out = contemplate_surprise_history( + history, now=13.0, tau0=1.0, discovery_gamma=0.1, sink=sink + ) + assert out.reason == "emitted" + assert isinstance(out.candidate, DiscoveryCandidate) + assert out.candidate.proposed_chain["discovery_gamma"] == pytest.approx(0.1) + assert len(sink.lines) == 1 + + +def test_non_string_refusal_marker_refused_at_boundary() -> None: + dual = _dual(0.5) + dual["surprise_refused"] = True # truthy non-string would read as "not refused" + with pytest.raises(ValueError): + SurpriseObservation(t=0.0, dual=dual) + + +def test_negative_surprise_norm_refused_at_boundary() -> None: + with pytest.raises(ValueError): + SurpriseObservation(t=0.0, dual=_dual(-0.5)) + + +def test_observation_dual_is_a_frozen_snapshot() -> None: + raw = _dual(0.5) + obs = SurpriseObservation(t=0.0, dual=raw) + raw["surprise_norm"] = 99.0 + assert obs.dual["surprise_norm"] == pytest.approx(0.5) + with pytest.raises(TypeError): + obs.dual["surprise_norm"] = 99.0 # type: ignore[index] + + +def test_explicit_gamma_overrides_dual_dict_gamma() -> None: + """The runner's gate runs first: a stricter caller γ blocks emission even + when the dual dict was computed against a looser γ.""" + sink = DiscoveryBufferSink() + out = contemplate_surprise_history( + _sustained(0.5), + now=13.0, + tau0=1.0, + discovery_gamma=0.75, + sink=sink, + ) + assert out.candidate is None + assert out.reason == "not_discovery_eligible" + assert out.discovery_gamma == pytest.approx(0.75) + assert sink.lines == [] diff --git a/tests/test_adr_0243_lane_a_kappa_calibration.py b/tests/test_adr_0243_lane_a_kappa_calibration.py new file mode 100644 index 00000000..7e0762d4 --- /dev/null +++ b/tests/test_adr_0243_lane_a_kappa_calibration.py @@ -0,0 +1,106 @@ +"""ADR-0243 Phase 3 Lane A — live `propose_kappa_line_search` caller. + +`propose_kappa_line_search` (ADR-0242 §5-P1) and its internal +`kappa_search_event` emission had zero live call sites — the seam was +tested but nothing invoked it. `calibrate_transfer_kappa` gives it a +live caller confined to the ADR-0240 analogical-transfer calibration +pipeline (R-04: trace generation is limited strictly to calibration and +training-loop pipelines; never a hot/serve path). +""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +import numpy as np +import pytest + +from evals.analogical_transfer.harness import make_fixture_pair +from evals.analogical_transfer.kappa_calibration import calibrate_transfer_kappa + +_ROOT = Path(__file__).resolve().parents[1] + + +class _CaptureSink: + def __init__(self) -> None: + self.lines: list[str] = [] + + def emit(self, line: str) -> None: + self.lines.append(line) + + +def test_calibration_fires_kappa_search_event_on_sink() -> None: + sink = _CaptureSink() + report = calibrate_transfer_kappa(sink=sink) + assert len(sink.lines) == 1 + event = json.loads(sink.lines[0]) + assert event["kind"] == "fibonacci_kappa_search" + assert event["kappa"] == pytest.approx(report["proposed_kappa"]) + assert report["event"] == event + + +def test_calibration_report_shape_and_bounds() -> None: + report = calibrate_transfer_kappa() + assert report["kind"] == "analogical_transfer_kappa_calibration" + assert report["case_count"] == 1 + assert report["refused_case_count"] == 0 + assert report["event"]["result"]["objective_id"] == "analogical_transfer_kappa" + assert 0.1 <= report["proposed_kappa"] <= 2.0 + assert np.isfinite(report["max_procrustes_residual"]) + # JSONL-ready without custom encoders. + assert json.loads(json.dumps(report, sort_keys=True)) == report + + +def test_calibration_without_sink_is_pure() -> None: + sink = _CaptureSink() + with_sink = calibrate_transfer_kappa(sink=sink) + without_sink = calibrate_transfer_kappa() + assert with_sink == without_sink + assert len(sink.lines) == 1 + + +def test_calibration_deterministic() -> None: + assert calibrate_transfer_kappa() == calibrate_transfer_kappa() + + +def test_empty_cases_fail_closed() -> None: + with pytest.raises(ValueError): + calibrate_transfer_kappa(cases=()) + + +def test_all_inadmissible_cases_fail_closed() -> None: + """Cases whose Procrustes residual is not measurable cannot calibrate.""" + base = make_fixture_pair() + poisoned = base.__class__( + case_id="poisoned", + source_domain=base.source_domain, + target_domain=base.target_domain, + source=[np.full(32, np.nan)], + target=[np.full(32, np.nan)], + novel_query=base.novel_query, + expected_novel=base.expected_novel, + ) + with pytest.raises(ValueError): + calibrate_transfer_kappa(cases=(poisoned,)) + + +def test_calibration_module_stays_off_serve() -> None: + """R-04 confinement: chat.runtime never loads the calibration caller.""" + probe = ( + "import importlib, sys;" + "importlib.import_module('chat.runtime');" + "print('LEAK' if any(m.startswith('evals.analogical_transfer')" + " for m in sys.modules) else 'CLEAN')" + ) + out = subprocess.run( + [sys.executable, "-c", probe], + cwd=str(_ROOT), + capture_output=True, + text=True, + env={"PYTHONPATH": str(_ROOT), "PATH": ""}, + ) + assert out.returncode == 0, out.stderr[-500:] + assert out.stdout.strip().splitlines()[-1] == "CLEAN" diff --git a/tests/test_serve_quarantine_transitive.py b/tests/test_serve_quarantine_transitive.py index 57945496..a8d7ed3f 100644 --- a/tests/test_serve_quarantine_transitive.py +++ b/tests/test_serve_quarantine_transitive.py @@ -36,6 +36,52 @@ _BANNED = ( ) +def test_idle_tick_contemplation_import_does_not_load_offserving_substrate(): + """The stated invariant is PROCESS-level, and the serve process executes + more than ``import chat.runtime``: the optional idle-tick frontier pass + lazily runs ``from core.contemplation.runner import run_contemplation, + write_contemplation_run`` inside the same always-on process that serves + turns (chat/runtime.py idle_tick, ``contemplate_frontier_during_idle``; + enabled by ``core/cli.py`` for the daemon). That import edge must stay + banned-module-clean too. + + Pre-ADR-0243-Lane-A this leaked ``core.physics.holographic_vault`` via the + package ``__init__``'s eager ``wave_seam`` import; Lane A made the + wave-seam re-exports lazy (PEP 562) and keeps the discovery-gate import + (``core.physics.multi_scale_energy``) function-local. The lazy re-export + must still resolve when actually referenced (checked last, in the same + probe, AFTER the leak assertion's snapshot). + """ + probe = ( + "import importlib, sys, json;" + "importlib.import_module('chat.runtime');" + # Mirror of chat/runtime.py idle_tick's lazy import, verbatim edge. + "from core.contemplation.runner import run_contemplation, write_contemplation_run;" + f"banned={list(_BANNED)!r};" + "leaked=sorted(m for m in sys.modules for b in banned if m==b or m.startswith(b+'.'));" + # Lazy re-export still works off-serve (loads wave_seam on demand). + "import core.contemplation as c;" + "assert c.WaveModeHypothesis is not None;" + "print(json.dumps(leaked))" + ) + result = subprocess.run( + [sys.executable, "-c", probe], + cwd=str(_ROOT), + capture_output=True, + text=True, + env={"PYTHONPATH": str(_ROOT), "PATH": ""}, + ) + assert result.returncode == 0, f"probe failed: {result.stderr[-2000:]}" + leaked = result.stdout.strip().splitlines()[-1] + import json as _json + + leaked_list = _json.loads(leaked) + assert not leaked_list, ( + "serve-process idle_tick contemplation import loaded off-serving " + f"modules (A-04 breach): {leaked_list}" + ) + + def test_import_chat_runtime_does_not_load_offserving_substrate(): probe = ( "import importlib, sys, json;"