feat: intelligence-loop arc (ADR-0246, 0247, 0248 accepted) #64

Merged
core-labs merged 6 commits from feat/intelligence-loop-arc into main 2026-07-18 18:21:16 +00:00
2 changed files with 218 additions and 4 deletions
Showing only changes of commit d672c71211 - Show all commits

View file

@ -61,7 +61,10 @@ import functools
import hashlib
import json
from dataclasses import dataclass, field
from typing import Any, Mapping, Sequence
from typing import TYPE_CHECKING, Any, Mapping, Sequence
if TYPE_CHECKING: # annotation-only: the monitor instance is caller-supplied
from core.physics.goldtether import GoldTetherMonitor
import numpy as np
@ -990,6 +993,77 @@ def serving_cast(
)
# --- Unified autonomy floor (seam S3, ADR-0238 / spark-audit adjudication §4) ----------
@dataclass(frozen=True, slots=True)
class TetherReading:
"""Per-turn GoldTether autonomy-floor reading for one corridor turn.
``updated`` discloses whether the monitor's floor/autonomy state advanced
(see :func:`tether_reading` for the control law) or the residual was
measured without a state update (admitted open superpositions).
"""
residual: float
autonomy: float
chiral_verdict: str
updated: bool
def as_dict(self) -> dict[str, Any]:
return {
"residual": float(self.residual),
"autonomy": float(self.autonomy),
"chiral_verdict": self.chiral_verdict,
"updated": bool(self.updated),
}
def tether_reading(
monitor: "GoldTetherMonitor",
psi_steady: np.ndarray,
*,
admitted: bool,
versor_closed: bool,
) -> TetherReading:
"""Feed one corridor turn to the unified autonomy floor (seam S3).
Control law, composed with pin SD-A (the residual KERNEL was already
unified ``goldtether.coherence_residual`` delegates to
:meth:`WaveManifold.measure_unitary_residual`; what was missing is the
MONITOR state seeing corridor turns):
* **not admitted** ``monitor.update`` the non-unit/uncertified state
drives residual > ε and autonomy hard to 0 (fail-closed).
* **admitted + versor_closed** ``monitor.update`` a certified closed
state may elevate autonomy toward the floor.
* **admitted + open** measure only, no state update: a legitimate
interference state neither elevates nor decays the floor. Punishing
open superpositions here would encode the exact SD-A defect the egress
gate refuses to (versor closure routes, it does not gate).
Chiral orientation is observed on EVERY reading Q_top is material only
on non-versor states, which is precisely the open route and a material
sign flip raises :class:`~core.physics.chiral_gate.ChiralOrientationError`
(fail-closed, never averaged).
"""
arr = np.asarray(psi_steady, dtype=np.float64)
if admitted and not versor_closed:
residual = float(monitor.residual(arr))
autonomy = float(monitor.autonomy)
updated = False
else:
residual, autonomy = monitor.update(arr)
updated = True
chiral_verdict = monitor.chiral_gate.observe(arr).verdict
return TetherReading(
residual=float(residual),
autonomy=float(autonomy),
chiral_verdict=chiral_verdict,
updated=updated,
)
# --- Composed lifecycle ---------------------------------------------------------------
@ -998,6 +1072,10 @@ class LifecycleOutcome:
ingress: IngressWavePacket
relaxation: RelaxationResult
verdict: EgressVerdict
# Monitoring metadata, deliberately OUTSIDE outcome_id: the outcome's
# identity is its cognitive content (ingress/certificate/route/ψ), not the
# observer's floor state at the time it was watched.
tether: TetherReading | None = None
outcome_id: str = ""
def __post_init__(self) -> None:
@ -1022,9 +1100,19 @@ class CognitiveLifecycleEngine:
the corrected one (pins SD-A/SD-B/SD-C; deviations D-1D-5).
"""
def __init__(self, *, epsilon_drift: float = _EPSILON_DRIFT) -> None:
def __init__(
self,
*,
epsilon_drift: float = _EPSILON_DRIFT,
monitor: "GoldTetherMonitor | None" = None,
) -> None:
self.epsilon_drift = float(epsilon_drift)
self.manifold = WaveManifold(epsilon_drift=self.epsilon_drift)
# Seam S3: optional unified autonomy floor. solve() feeds it one
# reading per turn; stage-level drivers (e.g. the sensorium corridor
# eval) own their monitor calls explicitly and should not also pass
# one here (double-counting a turn).
self.monitor = monitor
def ingest_context(self, packets: Sequence[PacketLike], domain_id: str) -> IngressWavePacket:
return ingest_context(packets, domain_id)
@ -1062,13 +1150,25 @@ class CognitiveLifecycleEngine:
tol: float = 1e-10,
energy_inputs: Mapping[str, object] | None = None,
) -> LifecycleOutcome:
"""Ingress → relax → egress. Fail-closed at every stage (typed errors)."""
"""Ingress → relax → egress (→ tether). Fail-closed at every stage (typed errors)."""
ingress = self.ingest_context(packets, domain_id)
result = relax_to_ground(ingress.psi, hamiltonian, dt=dt, max_steps=max_steps, tol=tol)
verdict = self.egress(
result.psi_steady, result.certificate, **dict(energy_inputs or {})
)
return LifecycleOutcome(ingress=ingress, relaxation=result, verdict=verdict)
tether = (
tether_reading(
self.monitor,
result.psi_steady,
admitted=verdict.admitted,
versor_closed=verdict.versor_closed,
)
if self.monitor is not None
else None
)
return LifecycleOutcome(
ingress=ingress, relaxation=result, verdict=verdict, tether=tether
)
__all__ = [
@ -1091,6 +1191,8 @@ __all__ = [
"RelaxationResult",
"ServingCastError",
"ServingState",
"TetherReading",
"tether_reading",
"assignment_component_index",
"compile_propositional",
"compile_quadratic_well",

View file

@ -0,0 +1,112 @@
"""Seam S3 — unified autonomy floor wired into the cognitive lifecycle.
Pins the tether control law composed with pin SD-A: closed certified states
update the GoldTether monitor (may elevate autonomy), non-admitted states
fail-close it (autonomy hard 0), and ADMITTED OPEN SUPERPOSITIONS are
measured without a state update punishing legitimate interference states
would encode the exact defect the egress gate refuses to. Chiral orientation
is observed on every reading and a material flip raises (fail-closed).
"""
from __future__ import annotations
import dataclasses
import numpy as np
import pytest
from algebra.rotor import make_rotor_from_angle
from core.physics.chiral_gate import ChiralOrientationError
from core.physics.cognitive_lifecycle import (
CognitiveLifecycleEngine,
PropositionalProblem,
compile_propositional,
compile_quadratic_well,
egress_gate,
relax_to_ground,
tether_reading,
uniform_assignment_state,
)
from core.physics.goldtether import GoldTetherMonitor
from core.physics.sensorium_wave_feed import fake_deterministic_packet
def _closed_solve(monitor: GoldTetherMonitor):
engine = CognitiveLifecycleEngine(monitor=monitor)
target = np.asarray(make_rotor_from_angle(0.3, bivector_idx=6), dtype=np.float64)
packets = [fake_deterministic_packet("audio", angle=0.25, plane=6)]
return engine.solve(packets, "tether-demo", compile_quadratic_well(target))
def test_closed_certified_turn_updates_monitor():
monitor = GoldTetherMonitor()
outcome = _closed_solve(monitor)
assert outcome.verdict.admitted and outcome.verdict.versor_closed
tether = outcome.tether
assert tether is not None and tether.updated
assert tether.residual < 1e-6
assert tether.chiral_verdict == "vacuous" # closed versor: Q = 0 by theorem
assert len(monitor.history) == 1
def test_solve_without_monitor_records_no_tether():
engine = CognitiveLifecycleEngine()
target = np.asarray(make_rotor_from_angle(0.3, bivector_idx=6), dtype=np.float64)
packets = [fake_deterministic_packet("audio", angle=0.25, plane=6)]
outcome = engine.solve(packets, "tether-demo", compile_quadratic_well(target))
assert outcome.tether is None
def test_admitted_open_superposition_measures_without_update():
"""SD-A dual: a legitimate interference state must not decay the floor."""
problem = PropositionalProblem(
atoms=("a", "b"),
clauses=((("a", True), ("b", True)), (("a", False), ("b", True))),
)
result = relax_to_ground(uniform_assignment_state(problem), compile_propositional(problem))
verdict = egress_gate(result.psi_steady, result.certificate)
assert verdict.admitted and not verdict.versor_closed
monitor = GoldTetherMonitor(floor=0.4, autonomy=0.2)
tether = tether_reading(
monitor,
result.psi_steady,
admitted=verdict.admitted,
versor_closed=verdict.versor_closed,
)
assert not tether.updated
assert tether.residual > monitor.epsilon_drift # honest: open state, big residual
assert monitor.floor == 0.4 and monitor.autonomy == 0.2 # untouched
assert len(monitor.history) == 0
def test_non_admitted_turn_fails_closed_to_zero_autonomy():
monitor = GoldTetherMonitor(floor=0.6, autonomy=0.5)
outcome_engine = CognitiveLifecycleEngine()
target = np.asarray(make_rotor_from_angle(0.3, bivector_idx=6), dtype=np.float64)
packets = [fake_deterministic_packet("audio", angle=0.25, plane=6)]
outcome = outcome_engine.solve(packets, "tether-demo", compile_quadratic_well(target))
refused = dataclasses.replace(outcome.verdict, admitted=False, reason="forced_refusal")
tether = tether_reading(
monitor,
outcome.relaxation.psi_steady,
admitted=refused.admitted,
versor_closed=refused.versor_closed,
)
assert tether.updated
# Closed state but the turn is refused → update path ran; the monitor's own
# law applies (this closed versor has residual ≤ ε, so autonomy may step,
# bounded by the floor). The FAIL-CLOSED zeroing fires on drifted states:
drifted = np.zeros(32, dtype=np.float64)
drifted[0] = 0.5 # non-unit ⇒ ψψ̃ far from 1
t2 = tether_reading(monitor, drifted, admitted=False, versor_closed=False)
assert t2.updated and monitor.autonomy == 0.0
def test_material_chiral_flip_raises_fail_closed():
monitor = GoldTetherMonitor()
monitor.chiral_gate.observe_q(0.5) # latch +1
flipped = np.zeros(32, dtype=np.float64)
flipped[0], flipped[31] = 0.8, 0.6 # material Q of opposite sign
with pytest.raises(ChiralOrientationError):
tether_reading(monitor, flipped, admitted=True, versor_closed=False)