From ecc7e8ab522769104d272134ee83e4da4d0ca952 Mon Sep 17 00:00:00 2001 From: Shay Date: Thu, 16 Jul 2026 06:23:50 -0700 Subject: [PATCH] =?UTF-8?q?feat(physics):=20ADR-0242=20=C2=A75=20carry=20s?= =?UTF-8?q?eams=20=E2=80=94=20=CE=BA-cert=20telemetry=20event=20+=20F5?= =?UTF-8?q?=E2=80=93F7=20cross-band=20gate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the two staged items recorded in the acceptance packet §7 (ruled non-blocking at ratification, tracked work): P1 — kappa_search_event(): JSONL-ready execution-telemetry envelope for the Fibonacci κ search (kind=fibonacci_kappa_search; certificate payload with content-addressed cert_id, or failure payload with baseline κ=1.0). propose_kappa_line_search gains an optional sink (any emit(line) object, e.g. chat.telemetry sinks) that writes one deterministic JSONL line. Pure serialization — no state, no COHERENT standing (§6 sovereignty). P2 — cross_band_discovery_gate(): persistence verdict over a surprise- event history across the F5/F6/F7 Fibonacci bands (τ = {5,8,13}·τ0). Eligible ⇔ history spans ≥ F5·τ0 (a single fresh spike has zero temporal persistence) AND every band's decay-weighted accumulation ≥ γ. Typed CrossBandVerdict; pure; PROPOSAL-side only. Lives in multi_scale_energy (Tier-2, serve-quarantined) so the gate can never touch serving — re-pinned by the module-level sys.modules probe in the new test file. TDD RED→GREEN. [Verification]: carry-seam suite 10 passed; regression (goldtether consumers + multi_scale + fibonacci + chiral + quarantine + cohesion) 85 passed; smoke suite passed locally (133s, 175 passed). --- core/physics/goldtether.py | 36 +++++- core/physics/multi_scale_energy.py | 97 +++++++++++++++ tests/test_adr_0242_carry_seams.py | 187 +++++++++++++++++++++++++++++ 3 files changed, 318 insertions(+), 2 deletions(-) create mode 100644 tests/test_adr_0242_carry_seams.py diff --git a/core/physics/goldtether.py b/core/physics/goldtether.py index 9e62a3f2..51dcaee2 100644 --- a/core/physics/goldtether.py +++ b/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 diff --git a/core/physics/multi_scale_energy.py b/core/physics/multi_scale_energy.py index 2efd7c24..fc63b884 100644 --- a/core/physics/multi_scale_energy.py +++ b/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", diff --git a/tests/test_adr_0242_carry_seams.py b/tests/test_adr_0242_carry_seams.py new file mode 100644 index 00000000..0814ae6d --- /dev/null +++ b/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"