feat(adr-0243): Lane A — discovery wiring + live kappa-search caller

Wire the ADR-0242 §5 pinned primitives into the contemplation runner
(ADR-0243 §2.4): contemplate_surprise_history() gates the newest
caller-timed ADR-0239 dual-operator observation through
is_discovery_eligible, the measured history through
cross_band_discovery_gate (F5/F6/F7 persistence), and only then emits a
DiscoveryCandidate via the existing emit_surprise_discovery ->
DiscoveryCandidateSink. Proposal-plumbing only: the sole side effect is
sink.emit().

Surprise-signal sourcing (the Lane A open question): ContemplationFinding
carries no surprise measurement and the report-mining paths have no
geometry, so nothing is fabricated for them — the canonical carrier is
the ADR-0239 dual audit dict (SelfAuthorshipMiner / ADR-0240 harness),
consumed as explicit SurpriseObservations with caller-supplied logical
time (deterministic, no wall clock).

Give propose_kappa_line_search its live caller (ADR-0242 §5-P1):
evals/analogical_transfer/kappa_calibration.py calibrate_transfer_kappa,
confined to the ADR-0240 calibration pipeline per R-04; kappa_search_event
fires exactly once when a sink is passed. Proposal-only — no
GoldTetherMonitor mutation.

A-04 hardening found during review: chat/runtime.py idle_tick lazily
imports core.contemplation.runner inside the serve process, and the
package __init__ eagerly pulled wave_seam -> holographic_vault (a
serve-banned module) — a PRE-EXISTING process-level quarantine breach.
Closed by making the wave-seam re-exports lazy (PEP 562) and keeping the
multi_scale_energy gate import function-local; pinned by a new dynamic
idle-tick-edge test in test_serve_quarantine_transitive.py.

Pinned primitives untouched: is_discovery_eligible,
cross_band_discovery_gate, candidate_from_surprise_dual signatures
unchanged; no changes to cognitive_lifecycle.py or chat/runtime.py.

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,
+if TYPE_CHECKING:
-    WaveReconstructResult,
+    from .wave_seam import (
-    reconstruct_as_evidence,
+        WaveModeHypothesis,
-    reconstruct_as_hypothesis,
+        WaveReconstructResult,
-    speculative_seal_from_contemplation,
+        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",

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",
 ]

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"]

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 == []

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"

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;"