Merge pull request 'feat(physics): ADR-0242 §5 carry seams — κ-cert telemetry + F5–F7 cross-band gate' (#49) from feat/adr-0242-carry-items into main

Reviewed-on: #49

core/physics/goldtether.py

@@ -542,6 +542,28 @@ class GoldTetherMonitor:
 # ---------------------------------------------------------------------------
 
 
+def kappa_search_event(kappa: float, result: object) -> dict[str, Any]:
+    """ADR-0242 §5-P1: JSONL-ready execution-telemetry event for a κ search.
+
+    The memo requires the certificate be "written to the execution telemetry"
+    (audit trail) with failures falling back to baseline κ. Both typed results
+    carry ``as_dict()``; this wraps them in a stable event envelope. Pure
+    serialization — no state, no COHERENT standing, no truth status (§6
+    sovereignty invariant).
+    """
+    from core.physics.fibonacci_search import FibonacciSearchCertificate
+
+    outcome = (
+        "certificate" if isinstance(result, FibonacciSearchCertificate) else "failure"
+    )
+    return {
+        "kind": "fibonacci_kappa_search",
+        "outcome": outcome,
+        "kappa": float(kappa),
+        "result": result.as_dict(),  # type: ignore[attr-defined]
+    }
+
+
 def propose_kappa_line_search(
     residual_fn,
     *,
@@ -550,13 +572,18 @@ def propose_kappa_line_search(
     evaluation_budget: int = 16,
     objective_id: str = "goldtether_kappa",
     objective_version: str = "v1",
+    sink: Any = None,
 ) -> tuple[float, object]:
     """Optional κ search via Fibonacci section (ADR-0242 Phase 1 seam).
 
     Returns ``(kappa, cert_or_failure)``. On failure, kappa is baseline 1.0.
     Does **not** mutate GoldTetherMonitor state, COHERENT standing, or serve
-    autonomy — caller may record the result as telemetry only.
+    autonomy. When ``sink`` (any object with ``emit(line: str)``, e.g. a
+    :class:`chat.telemetry.TurnEventSink`) is provided, the §5-P1 execution-
+    telemetry event is emitted as one deterministic JSONL line.
     """
+    import json
+
     from core.physics.fibonacci_search import (
         BoundedUnimodalObjective,
         fibonacci_section_search,
@@ -571,4 +598,9 @@ def propose_kappa_line_search(
         objective_version=str(objective_version),
     )
     result = fibonacci_section_search(objective, residual_fn)
-    return propose_kappa_from_search(result)
+    kappa, outcome = propose_kappa_from_search(result)
+    if sink is not None:
+        sink.emit(
+            json.dumps(kappa_search_event(kappa, outcome), sort_keys=True)
+        )
+    return kappa, outcome

core/physics/multi_scale_energy.py

@@ -16,6 +16,7 @@ Reuses ``fibonacci_number`` / ``fibonacci_tau_schedule`` — no parallel Fibonac
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from math import exp, isfinite
 from typing import Sequence
 
@@ -169,8 +170,104 @@ def schedule_mid_span_fraction(taus: Sequence[float], *, index: int | None = Non
     return float(vals[i] / peak)
 
 
+# --- ADR-0242 §5-P2: F5–F7 cross-band surprise persistence gate ---------------
+#
+# "Emits a DiscoveryCandidate in the contemplation loop *only* when the
+# surprise signal persists across multiple Fibonacci-scaled temporal bands
+# (F5 to F7), preventing transient noise from triggering ungrounded updates."
+#
+# Pure verdict function for the contemplation loop — PROPOSAL-side only.
+# Lives here (Tier-2, serve-quarantined) so the gate can never touch serving.
+
+_DISCOVERY_BANDS: tuple[int, int, int] = (5, 8, 13)  # F_5, F_6, F_7
+
+
+@dataclass(frozen=True, slots=True)
+class CrossBandVerdict:
+    """Typed persistence verdict; never emits or promotes anything itself."""
+
+    eligible: bool
+    bands: tuple[int, int, int]
+    band_energies: tuple[float, float, float]
+    gamma: float
+    reason: str  # "eligible" | "insufficient_span" | "band_below_gamma"
+
+
+def cross_band_discovery_gate(
+    events: Sequence[tuple[float, float]],
+    *,
+    now: float,
+    tau0: float = _DEFAULT_TAU0,
+    gamma: float,
+) -> CrossBandVerdict:
+    """Persistence gate over a surprise-event history.
+
+    ``events`` are ``(t, energy)`` samples with ``t <= now`` and
+    ``energy >= 0``. Each band accumulates decay-weighted surprise
+
+        E_band(now) = Σ_i energy_i · exp(-(now - t_i) / (F_band · τ0))
+
+    Eligible ⇔ the history spans at least the shortest band (F_5·τ0 —
+    a single fresh spike carries full weight in every band but has zero
+    temporal persistence) AND every band's accumulation ≥ ``gamma``.
+    Deterministic and pure; the caller (contemplation loop) decides whether
+    to emit a DiscoveryCandidate.
+    """
+    if not events:
+        raise ValueError("cross_band_discovery_gate: empty event history")
+    tau0 = _validate_tau0(tau0)
+    if not isfinite(gamma) or gamma <= 0.0:
+        raise ValueError("gamma must be finite and > 0")
+    times = []
+    for t, e in events:
+        t = float(t)
+        e = float(e)
+        if not (isfinite(t) and isfinite(e)):
+            raise ValueError("event times/energies must be finite")
+        if e < 0.0:
+            raise ValueError(f"negative surprise energy {e} at t={t}")
+        if t > float(now):
+            raise ValueError(f"event at t={t} is after now={now}")
+        times.append(t)
+
+    band_energies = tuple(
+        sum(
+            float(e) * exp(-(float(now) - float(t)) / (band * tau0))
+            for t, e in events
+        )
+        for band in _DISCOVERY_BANDS
+    )
+
+    span = max(times) - min(times)
+    if span < _DISCOVERY_BANDS[0] * tau0:
+        return CrossBandVerdict(
+            eligible=False,
+            bands=_DISCOVERY_BANDS,
+            band_energies=band_energies,  # type: ignore[arg-type]
+            gamma=float(gamma),
+            reason="insufficient_span",
+        )
+    if any(e < gamma for e in band_energies):
+        return CrossBandVerdict(
+            eligible=False,
+            bands=_DISCOVERY_BANDS,
+            band_energies=band_energies,  # type: ignore[arg-type]
+            gamma=float(gamma),
+            reason="band_below_gamma",
+        )
+    return CrossBandVerdict(
+        eligible=True,
+        bands=_DISCOVERY_BANDS,
+        band_energies=band_energies,  # type: ignore[arg-type]
+        gamma=float(gamma),
+        reason="eligible",
+    )
+
+
 __all__ = [
+    "CrossBandVerdict",
     "comparative_residual_separation",
+    "cross_band_discovery_gate",
     "dyadic_tau_schedule",
     "multi_scale_energy_for_schedule",
     "multi_scale_energy_vector",

tests/test_adr_0242_carry_seams.py

@@ -0,0 +1,187 @@
+"""ADR-0242 §5 carry seams — cert telemetry (P1) + F5–F7 cross-band gate (P2).
+
+RED until both seams land. These are the two staged items recorded in the
+acceptance packet §7 (ruled non-blocking at ratification, tracked work):
+
+P1 (memo §5 Phase 1): "The returned FibonacciSearchCertificate is written to
+the execution telemetry. If the certificate fails … defaults to the safe
+baseline kappa = 1.0 and logs an OptimizationFailure warning." The search
+returns typed results; nothing produced a telemetry-ready event.
+
+P2 (memo §5 Phase 2): "Emits a DiscoveryCandidate in the contemplation loop
+*only* when the surprise signal persists across multiple Fibonacci-scaled
+temporal bands (F5 to F7), preventing transient noise from triggering
+ungrounded updates." Band helpers existed; the persistence gate did not.
+
+Both stay PROPOSAL/telemetry-side: no serve import, no COHERENT promotion,
+no truth-status effect (memo §6 sovereignty invariant).
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from core.physics.goldtether import kappa_search_event, propose_kappa_line_search
+from core.physics.multi_scale_energy import (
+    CrossBandVerdict,
+    cross_band_discovery_gate,
+)
+
+# --- P1: κ-search certificate → execution telemetry event ---------------------
+
+
+def _good_objective(x: float) -> float:
+    return (x - 0.789) ** 2
+
+
+def _multimodal(x: float) -> float:
+    return x**4 - x**2
+
+
+class _CaptureSink:
+    def __init__(self) -> None:
+        self.lines: list[str] = []
+
+    def emit(self, line: str) -> None:
+        self.lines.append(line)
+
+
+def test_kappa_search_event_certificate_payload():
+    kappa, result = propose_kappa_line_search(_good_objective, evaluation_budget=16)
+    event = kappa_search_event(kappa, result)
+
+    assert event["kind"] == "fibonacci_kappa_search"
+    assert event["outcome"] == "certificate"
+    assert event["kappa"] == pytest.approx(kappa)
+    cert = event["result"]
+    assert cert["kind"] == "FibonacciSearchCertificate"
+    assert len(cert["cert_id"]) == 64  # content-addressed audit trail
+    assert cert["evaluations"] == 16
+    assert cert["objective_id"] == "goldtether_kappa"
+    # JSONL-ready: round-trips through json without custom encoders.
+    assert json.loads(json.dumps(event)) == event
+
+
+def test_kappa_search_event_failure_payload_falls_back_to_baseline():
+    kappa, result = propose_kappa_line_search(
+        _multimodal, lower=-2.0, upper=2.0, evaluation_budget=10
+    )
+    event = kappa_search_event(kappa, result)
+
+    assert event["outcome"] == "failure"
+    assert event["kappa"] == 1.0  # BASELINE_KAPPA — never a silent minimizer
+    assert event["result"]["kind"] == "OptimizationFailure"
+    assert "unimodality" in event["result"]["reason"]
+    assert json.loads(json.dumps(event)) == event
+
+
+def test_propose_kappa_line_search_emits_to_sink_when_provided():
+    sink = _CaptureSink()
+    kappa, result = propose_kappa_line_search(
+        _good_objective, evaluation_budget=12, sink=sink
+    )
+    assert len(sink.lines) == 1
+    event = json.loads(sink.lines[0])
+    assert event["kind"] == "fibonacci_kappa_search"
+    assert event["kappa"] == pytest.approx(kappa)
+    assert event["result"]["cert_id"]
+
+
+def test_kappa_search_event_deterministic():
+    _, result_a = propose_kappa_line_search(_good_objective, evaluation_budget=12)
+    _, result_b = propose_kappa_line_search(_good_objective, evaluation_budget=12)
+    kappa_a, _ = propose_kappa_line_search(_good_objective, evaluation_budget=12)
+    assert kappa_search_event(kappa_a, result_a) == kappa_search_event(
+        kappa_a, result_b
+    )
+
+
+# --- P2: F5–F7 cross-band surprise persistence gate ----------------------------
+#
+# Bands: tau_n = F_n · tau0 with (F5, F6, F7) = (5, 8, 13).
+# eligible ⇔ every band's decay-weighted accumulated surprise ≥ gamma AND the
+# event history spans at least the shortest band (temporal extent — a single
+# fresh spike has full weight in every band but zero persistence).
+
+
+def _sustained(tau0: float, *, per_event: float = 0.5, n: int = 14) -> list[tuple[float, float]]:
+    return [(float(i) * tau0, per_event) for i in range(n)]
+
+
+def test_sustained_signal_is_eligible_across_all_bands():
+    tau0 = 1.0
+    events = _sustained(tau0)
+    verdict = cross_band_discovery_gate(
+        events, now=13.0 * tau0, tau0=tau0, gamma=0.35
+    )
+    assert isinstance(verdict, CrossBandVerdict)
+    assert verdict.eligible is True
+    assert verdict.bands == (5, 8, 13)
+    assert all(e >= verdict.gamma for e in verdict.band_energies)
+
+
+def test_single_fresh_spike_is_not_eligible_no_temporal_extent():
+    tau0 = 1.0
+    verdict = cross_band_discovery_gate(
+        [(10.0, 5.0)], now=10.0, tau0=tau0, gamma=0.35
+    )
+    assert verdict.eligible is False
+    assert verdict.reason == "insufficient_span"
+
+
+def test_old_transient_is_not_eligible_short_band_decayed():
+    tau0 = 1.0
+    # Burst long ago: spans the extent requirement, but by `now` the F5 band
+    # (tau = 5·tau0) has decayed it below gamma while F7 may still hold mass.
+    events = [(0.0, 1.0), (2.0, 1.0), (4.0, 1.0), (6.0, 1.0)]
+    verdict = cross_band_discovery_gate(events, now=40.0, tau0=tau0, gamma=0.35)
+    assert verdict.eligible is False
+    assert verdict.reason == "band_below_gamma"
+    # The shortest band is the one that fails first.
+    assert verdict.band_energies[0] < verdict.gamma
+
+
+def test_gate_is_deterministic_and_pure():
+    tau0 = 2.0
+    events = _sustained(tau0, per_event=0.4)
+    a = cross_band_discovery_gate(events, now=30.0, tau0=tau0, gamma=0.2)
+    b = cross_band_discovery_gate(events, now=30.0, tau0=tau0, gamma=0.2)
+    assert a == b
+
+
+def test_gate_refuses_invalid_inputs():
+    with pytest.raises(ValueError):
+        cross_band_discovery_gate([], now=1.0, tau0=1.0, gamma=0.35)
+    with pytest.raises(ValueError):
+        cross_band_discovery_gate([(0.0, 1.0)], now=1.0, tau0=-1.0, gamma=0.35)
+    with pytest.raises(ValueError):
+        cross_band_discovery_gate([(2.0, 1.0)], now=1.0, tau0=1.0, gamma=0.35)  # event after now
+    with pytest.raises(ValueError):
+        cross_band_discovery_gate([(0.0, -1.0)], now=1.0, tau0=1.0, gamma=0.35)  # negative energy
+
+
+def test_gate_module_stays_off_serve():
+    """multi_scale_energy remains Tier-2: never imported by chat.runtime."""
+    import subprocess
+    import sys
+    from pathlib import Path
+
+    root = Path(__file__).resolve().parents[1]
+    probe = (
+        "import importlib, sys;"
+        "importlib.import_module('chat.runtime');"
+        "print('LEAK' if any(m == 'core.physics.multi_scale_energy' or "
+        "m.startswith('core.physics.multi_scale_energy.') 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"