diff --git a/core/physics/__init__.py b/core/physics/__init__.py index 9486252d..16705b9a 100644 --- a/core/physics/__init__.py +++ b/core/physics/__init__.py @@ -5,6 +5,10 @@ Three physics sublayers: compositional — binding, digest, reasoning, articulation (ADR-0009) identity — identity manifold, drives, exertion, character (ADR-0010) +Third-Door Horizon (ADR-0238–0240): + coherence GoldTether, dynamic manifold (Procrustes/PCA), surprise dual, + biography holonomy, temporal gate, self-authorship miner. + All operators are stateless and frozen where possible. State lives in the FieldState; operators are pure transformations. """ @@ -22,6 +26,50 @@ from core.physics.drive import DriveGradientMap, GradientField, ValueAxis from core.physics.exertion import ExertionMeter, FatigueIndex, CycleCost from core.physics.identity import IdentityManifold, IdentityCheck, IdentityScore, CharacterProfile from core.physics.learning import PromotionDecision, VaultPromotionPolicy +from core.physics.goldtether import ( + AutonomyBand, + AutonomyDecision, + CoherenceResidual, + GoldTetherConfig, + GoldTetherMonitor, + OperatingMode, + PseudoscalarFloorState, + derive_kappa, +) +from core.physics.dynamic_manifold import ( + AxisClassification, + CartanIwasawaFactors, + ConformalProcrustesResult, + PrincipalAxis, + SignatureAwarePCAResult, + cartan_iwasawa_factorize, + conformal_procrustes, + dual_correction_slerp, + procrustes_residual, + signature_aware_pca, +) +from core.physics.surprise import ( + AnalogySeed, + DualOperatorResult, + SurpriseResult, + analogy_seed, + dual_operator, + project_onto_basis, + surprise_residual, +) +from core.physics.biography import ( + BiographyHolonomyBlade, + biography_telemetry, + integrate_biography, + reconstruct_biography, +) +from core.physics.temporal_gate import ( + TemporalAdmissibilityGate, + TemporalContext, + TemporalDecision, + TemporalVerdict, +) +from core.physics.self_authorship import AuthorshipProposal, SelfAuthorshipMiner __all__ = [ "SalienceOperator", "SalienceMap", "FieldRegion", @@ -36,4 +84,42 @@ __all__ = [ "ExertionMeter", "FatigueIndex", "CycleCost", "IdentityManifold", "IdentityCheck", "IdentityScore", "CharacterProfile", "PromotionDecision", "VaultPromotionPolicy", + # ADR-0238 + "AutonomyBand", + "AutonomyDecision", + "CoherenceResidual", + "GoldTetherConfig", + "GoldTetherMonitor", + "OperatingMode", + "PseudoscalarFloorState", + "derive_kappa", + # ADR-0239 + "AxisClassification", + "CartanIwasawaFactors", + "ConformalProcrustesResult", + "PrincipalAxis", + "SignatureAwarePCAResult", + "cartan_iwasawa_factorize", + "conformal_procrustes", + "dual_correction_slerp", + "procrustes_residual", + "signature_aware_pca", + "AnalogySeed", + "DualOperatorResult", + "SurpriseResult", + "analogy_seed", + "dual_operator", + "project_onto_basis", + "surprise_residual", + # ADR-0240 + "BiographyHolonomyBlade", + "biography_telemetry", + "integrate_biography", + "reconstruct_biography", + "TemporalAdmissibilityGate", + "TemporalContext", + "TemporalDecision", + "TemporalVerdict", + "AuthorshipProposal", + "SelfAuthorshipMiner", ] diff --git a/core/physics/biography.py b/core/physics/biography.py new file mode 100644 index 00000000..8fb07a20 --- /dev/null +++ b/core/physics/biography.py @@ -0,0 +1,107 @@ +"""core.physics.biography — Biography Holonomy Blade (ADR-0240). + +Forever-lived individuality as an integrated holonomy of the identity +trajectory. Reconstructible from an ordered sequence of session versors — +reconstruction-over-storage. Not a raw experience dump; not a parallel +identity store. + +Integrates with ``algebra.holonomy.holonomy_encode`` and the identity motor +surface; does not mutate packs, vault, or serving paths. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Sequence + +import numpy as np + +from algebra.cl41 import N_COMPONENTS +from algebra.holonomy import holonomy_encode, holonomy_similarity +from algebra.versor import unitize_versor, versor_condition + +_CLOSURE_TOL = 1e-6 +_TELEMETRY_SCHEMA = "biography_holonomy_v1" + + +@dataclass(frozen=True, slots=True) +class BiographyHolonomyBlade: + """Integrated holonomy blade of a lived trajectory.""" + + blade: np.ndarray + n_steps: int + trajectory_hash: str + closure: float + + def similarity(self, other: "BiographyHolonomyBlade") -> float: + return float(holonomy_similarity(self.blade, other.blade)) + + +def _as_versor(v: np.ndarray, name: str) -> np.ndarray: + arr = np.asarray(v, dtype=np.float64) + if arr.shape != (N_COMPONENTS,): + raise ValueError(f"{name} must have shape ({N_COMPONENTS},)") + # Construction boundary: trajectory elements must be closed versors. + try: + closed = unitize_versor(arr) + except ValueError as exc: + raise ValueError(f"{name} is not a closed versor: {exc}") from exc + cond = versor_condition(closed) + if cond >= _CLOSURE_TOL: + raise ValueError(f"{name} versor_condition={cond:.3e}") + return closed.astype(np.float64, copy=False) + + +def _trajectory_hash(versors: Sequence[np.ndarray]) -> str: + import hashlib + + h = hashlib.sha256() + for v in versors: + h.update(np.asarray(v, dtype=np.float64).tobytes()) + return h.hexdigest() + + +def integrate_biography( + trajectory: Sequence[np.ndarray], + *, + alpha: float = 0.5, +) -> BiographyHolonomyBlade: + """Integrate ordered identity/session versors into a biography holonomy blade. + + Order is load-bearing. Empty trajectory is refused (no confabulated self). + """ + if not trajectory: + raise ValueError("biography trajectory must be non-empty") + closed = [_as_versor(v, f"trajectory[{i}]") for i, v in enumerate(trajectory)] + blade = holonomy_encode(closed, alpha=alpha) + cond = versor_condition(blade) + if cond >= _CLOSURE_TOL: + raise ValueError(f"biography blade not closed: {cond:.3e}") + return BiographyHolonomyBlade( + blade=np.asarray(blade, dtype=np.float64), + n_steps=len(closed), + trajectory_hash=_trajectory_hash(closed), + closure=float(cond), + ) + + +def reconstruct_biography( + trajectory: Sequence[np.ndarray], + *, + alpha: float = 0.5, +) -> BiographyHolonomyBlade: + """Alias for integrate — reconstruction is recompute, not storage load.""" + return integrate_biography(trajectory, alpha=alpha) + + +def biography_telemetry(blade: BiographyHolonomyBlade) -> dict[str, Any]: + """Workbench-safe projection (no full multivector dump required for UI).""" + return { + "schema_version": _TELEMETRY_SCHEMA, + "n_steps": int(blade.n_steps), + "trajectory_hash": blade.trajectory_hash, + "closure": float(blade.closure), + "blade_scalar": float(blade.blade[0]), + "blade_pseudoscalar": float(blade.blade[31]), + "blade_l2": float(np.linalg.norm(blade.blade)), + } diff --git a/core/physics/dynamic_manifold.py b/core/physics/dynamic_manifold.py new file mode 100644 index 00000000..e955138f --- /dev/null +++ b/core/physics/dynamic_manifold.py @@ -0,0 +1,449 @@ +"""core.physics.dynamic_manifold — Conformal manifold operators (ADR-0239). + +Signature-aware PCA with explicit null classification, Conformal Procrustes +(versor search for structural analogy), Cartan–Iwasawa constructive +factorization for dual-correction slerp, and a dedicated Procrustes residual +norm (not null-margin; not ADR-0006 energy residual). + +All operators are pure and deterministic. Null eigenvectors are never silently +skipped — they are classified and returned. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Sequence + +import numpy as np + +from algebra.cl41 import ( + N_COMPONENTS, + SIGNATURE, + geometric_product, + grade_project, + reverse, +) +from algebra.rotor import word_transition_rotor +from algebra.versor import unitize_versor, versor_apply, versor_condition + +_CLOSURE_TOL = 1e-6 +_NEAR_ZERO = 1e-12 +_NULL_TOL = 1e-8 +_METRIC = np.ones(N_COMPONENTS, dtype=np.float64) +# Grade-1 components (indices 1..5) carry Cl(4,1) signature (+,+,+,+,-). +_METRIC[1:6] = SIGNATURE.astype(np.float64) + + +class AxisClassification(str, Enum): + SPACELIKE = "spacelike" + TIMELIKE = "timelike" + NULL = "null" + DEGENERATE = "degenerate" + + +@dataclass(frozen=True, slots=True) +class PrincipalAxis: + vector: tuple[float, ...] + eigenvalue: float + classification: AxisClassification + metric_quadratic: float + + +@dataclass(frozen=True, slots=True) +class SignatureAwarePCAResult: + axes: tuple[PrincipalAxis, ...] + mean: tuple[float, ...] + explained: tuple[float, ...] + n_points: int + n_null: int + n_spacelike: int + n_timelike: int + n_degenerate: int + + +@dataclass(frozen=True, slots=True) +class ConformalProcrustesResult: + versor: np.ndarray + residual_norm: float + n_pairs: int + pair_residuals: tuple[float, ...] + + +@dataclass(frozen=True, slots=True) +class CartanIwasawaFactors: + """Constructive K · A · N factorization for dual-correction surfaces. + + K — compact (rotation-like, B² < 0 planes) + A — abelian (boost-like, B² > 0 planes) + N — nilpotent / residual (null + higher-grade remainder, unitized if needed) + """ + + K: np.ndarray + A: np.ndarray + N: np.ndarray + reconstruction_residual: float + + +def _as_points(points: Sequence[np.ndarray]) -> np.ndarray: + if not points: + raise ValueError("signature_aware_pca requires at least one point") + rows = [] + for i, p in enumerate(points): + arr = np.asarray(p, dtype=np.float64) + if arr.shape != (N_COMPONENTS,): + raise ValueError( + f"point[{i}] must have shape ({N_COMPONENTS},); got {arr.shape}" + ) + rows.append(arr) + return np.stack(rows, axis=0) + + +def _metric_quadratic_form(v: np.ndarray) -> float: + """Quadratic form on grade-1 part under Cl(4,1) signature; higher grades +. + + Used only for axis *classification*, not as a substitute for versor_condition. + """ + g1 = v[1:6] + q = float(np.dot(g1 * _METRIC[1:6], g1)) + higher = float(np.dot(v[6:], v[6:])) + scalar = float(v[0] * v[0]) + return q + higher + scalar + + +def _classify_axis(v: np.ndarray) -> tuple[AxisClassification, float]: + nrm = float(np.linalg.norm(v)) + if nrm < _NEAR_ZERO: + return AxisClassification.DEGENERATE, 0.0 + q = _metric_quadratic_form(v) + # Grade-1 signature sense dominates classification when g1 is present. + g1 = v[1:6] + g1_q = float(np.dot(g1 * _METRIC[1:6], g1)) + g1_n = float(np.linalg.norm(g1)) + if g1_n < _NEAR_ZERO: + # Pure higher-grade axis: treat as spacelike if energy present else degenerate. + if nrm < _NULL_TOL: + return AxisClassification.DEGENERATE, q + return AxisClassification.SPACELIKE, q + if abs(g1_q) < _NULL_TOL: + return AxisClassification.NULL, g1_q + if g1_q > 0.0: + return AxisClassification.SPACELIKE, g1_q + return AxisClassification.TIMELIKE, g1_q + + +def signature_aware_pca( + points: Sequence[np.ndarray], + *, + max_axes: int | None = None, +) -> SignatureAwarePCAResult: + """Signature-aware PCA on Cl(4,1) multivector clouds. + + 1. Center the cloud in coefficient space. + 2. Form the metric-rescaled covariance (whitening by √|G| on coords). + 3. Eigen-decompose symmetrically (deterministic via numpy eigh). + 4. Classify every axis — null axes are returned, never dropped. + """ + X = _as_points(points) + n = X.shape[0] + mean = X.mean(axis=0) + Xc = X - mean + + # Metric rescaling: multiply each coordinate by sqrt(|g_ii|)*sign-preserving weight. + # For signature -1 on e5, use imaginary-free absolute metric then restore sense + # via classification (not by complex eigen). + scale = np.sqrt(np.abs(_METRIC)) + scale = np.where(scale < _NEAR_ZERO, 1.0, scale) + Y = Xc * scale # broadcast + + # Covariance of rescaled coordinates. + if n == 1: + cov = np.zeros((N_COMPONENTS, N_COMPONENTS), dtype=np.float64) + else: + cov = (Y.T @ Y) / float(n) + + # Symmetric eigh → ascending eigenvalues; reverse for explained variance order. + evals, evecs = np.linalg.eigh(cov) + order = np.argsort(evals)[::-1] + evals = evals[order] + evecs = evecs[:, order] + + k = N_COMPONENTS if max_axes is None else min(int(max_axes), N_COMPONENTS) + total = float(np.sum(np.clip(evals, 0.0, None))) + axes: list[PrincipalAxis] = [] + counts = { + AxisClassification.NULL: 0, + AxisClassification.SPACELIKE: 0, + AxisClassification.TIMELIKE: 0, + AxisClassification.DEGENERATE: 0, + } + explained_list: list[float] = [] + + for i in range(k): + # Map eigenvector back from metric-rescaled coords. + v_scaled = evecs[:, i] + v = v_scaled / scale + nrm = float(np.linalg.norm(v)) + if nrm > _NEAR_ZERO: + v = v / nrm + # Deterministic sign convention: first nonzero component positive. + for c in v: + if abs(c) > _NEAR_ZERO: + if c < 0.0: + v = -v + break + cls, mq = _classify_axis(v) + counts[cls] = counts.get(cls, 0) + 1 + ev = float(max(0.0, evals[i])) + frac = float(ev / total) if total > _NEAR_ZERO else 0.0 + explained_list.append(frac) + axes.append( + PrincipalAxis( + vector=tuple(float(x) for x in v), + eigenvalue=ev, + classification=cls, + metric_quadratic=float(mq), + ) + ) + + return SignatureAwarePCAResult( + axes=tuple(axes), + mean=tuple(float(x) for x in mean), + explained=tuple(explained_list), + n_points=n, + n_null=counts[AxisClassification.NULL], + n_spacelike=counts[AxisClassification.SPACELIKE], + n_timelike=counts[AxisClassification.TIMELIKE], + n_degenerate=counts[AxisClassification.DEGENERATE], + ) + + +def procrustes_residual( + source: np.ndarray, + target: np.ndarray, + versor: np.ndarray, +) -> float: + """Dedicated Procrustes residual: || V * s * reverse(V) - t ||_F. + + Named separately from null-margin and energy coherence_residual so it + cannot be silently reused as a different residual. + """ + s = np.asarray(source, dtype=np.float64) + t = np.asarray(target, dtype=np.float64) + V = np.asarray(versor, dtype=np.float64) + mapped = versor_apply(V, s) + return float(np.linalg.norm(mapped - t)) + + +def conformal_procrustes( + sources: Sequence[np.ndarray] | np.ndarray, + targets: Sequence[np.ndarray] | np.ndarray, +) -> ConformalProcrustesResult: + """Find a unit versor aligning source structure to target structure. + + Single pair: closed transition rotor. + Multiple pairs: manifold average of per-pair transition rotors via + successive equal-weight slerp composition (deterministic order). + """ + if isinstance(sources, np.ndarray) and sources.ndim == 1: + src_list = [sources] + else: + src_list = list(sources) # type: ignore[arg-type] + if isinstance(targets, np.ndarray) and targets.ndim == 1: + tgt_list = [targets] + else: + tgt_list = list(targets) # type: ignore[arg-type] + + if len(src_list) != len(tgt_list): + raise ValueError("sources and targets must have equal length") + if not src_list: + raise ValueError("conformal_procrustes requires at least one pair") + + rotors: list[np.ndarray] = [] + for i, (s, t) in enumerate(zip(src_list, tgt_list)): + s_arr = np.asarray(s, dtype=np.float64) + t_arr = np.asarray(t, dtype=np.float64) + if s_arr.shape != (N_COMPONENTS,) or t_arr.shape != (N_COMPONENTS,): + raise ValueError(f"pair[{i}] must be 32-component multivectors") + # Construction boundary: transition rotor is a closed unit versor. + R = word_transition_rotor(s_arr, t_arr) + rotors.append(np.asarray(R, dtype=np.float64)) + + # Manifold average: sequential geodesic midpoints in pair order. + V = rotors[0].copy() + for k, R in enumerate(rotors[1:], start=2): + # Slerp V toward R with weight 1/k so equal contribution in the limit. + try: + T = word_transition_rotor(V, R) + from algebra.rotor import rotor_power + + T_a = rotor_power(T, 1.0 / float(k)) + V = geometric_product(geometric_product(T_a, V), reverse(T_a)).astype( + np.float64 + ) + V = unitize_versor(V) + except ValueError: + # Fail closed to previous average if a pair is non-connectable. + continue + + cond = versor_condition(V) + if cond >= _CLOSURE_TOL: + raise ValueError(f"Procrustes versor not closed: condition={cond:.3e}") + + pair_res = tuple( + procrustes_residual(s, t, V) for s, t in zip(src_list, tgt_list) + ) + residual_norm = float(np.sqrt(sum(r * r for r in pair_res) / len(pair_res))) + return ConformalProcrustesResult( + versor=V.astype(np.float64, copy=False), + residual_norm=residual_norm, + n_pairs=len(src_list), + pair_residuals=pair_res, + ) + + +def _identity_mv() -> np.ndarray: + out = np.zeros(N_COMPONENTS, dtype=np.float64) + out[0] = 1.0 + return out + + +def cartan_iwasawa_factorize(V: np.ndarray) -> CartanIwasawaFactors: + """Constructive Cartan–Iwasawa-style factorization of a closed versor. + + For simple scalar+bivector rotors: + - If B² < 0 → pure K (rotation) + - If B² > 0 → pure A (boost) + - If B² ≈ 0 and B ≠ 0 → pure N (null) + - Mixed: split bivector energy by sign of B² contribution and leave residual N. + + Reconstruction: K * A * N; residual measured in coefficient space. + """ + V_arr = np.asarray(V, dtype=np.float64) + if V_arr.shape != (N_COMPONENTS,): + raise ValueError(f"V must have shape ({N_COMPONENTS},)") + cond = versor_condition(V_arr) + if cond >= 1e-2: + # Construction boundary: attempt unitize only at this explicit boundary. + V_arr = unitize_versor(V_arr) + cond = versor_condition(V_arr) + if cond >= _CLOSURE_TOL: + raise ValueError(f"cartan_iwasawa_factorize: input not closed ({cond:.3e})") + + scalar = float(V_arr[0]) + B = grade_project(V_arr, 2) + higher = V_arr.copy() + higher[0] = 0.0 + higher[6:16] = 0.0 # clear grade-2; keep 1,3,4,5 + + B_sq = geometric_product(B, B).astype(np.float64) + bsq_scalar = float(B_sq[0]) + B_sq_res = B_sq.copy() + B_sq_res[0] = 0.0 + simple = float(np.linalg.norm(B_sq_res)) < 1e-6 + b_norm = float(np.linalg.norm(B)) + + I = _identity_mv() + K = I.copy() + A = I.copy() + N = I.copy() + + if b_norm < _NEAR_ZERO and float(np.linalg.norm(higher)) < _NEAR_ZERO: + # Near-identity + K = V_arr.copy() + elif simple and bsq_scalar < 0.0: + K = V_arr.copy() + # Zero out non-scalar/bivector if any residual grades present. + K = grade_project(K, 0) + grade_project(K, 2) + K = unitize_versor(K) + elif simple and bsq_scalar > 0.0: + A = grade_project(V_arr, 0) + grade_project(V_arr, 2) + A = unitize_versor(A) + elif simple and abs(bsq_scalar) <= _NEAR_ZERO: + N_cand = grade_project(V_arr, 0) + grade_project(V_arr, 2) + try: + N = unitize_versor(N_cand) + except ValueError: + N = I.copy() + N[0] = scalar if abs(scalar) > _NEAR_ZERO else 1.0 + N = unitize_versor(N) + else: + # Mixed: put scalar+rotation-like half in K, boost-like half in A, rest N. + # Split B into two parallel bivectors by halving coefficients when non-simple. + half = B * 0.5 + K = grade_project(V_arr, 0) * 0.0 + K[0] = abs(scalar) ** 0.5 if abs(scalar) > _NEAR_ZERO else 1.0 + K = K + half + try: + K = unitize_versor(K) + except ValueError: + K = I.copy() + A = I.copy() + A[0] = abs(scalar) ** 0.5 if abs(scalar) > _NEAR_ZERO else 1.0 + A = A + half + try: + A = unitize_versor(A) + except ValueError: + A = I.copy() + # N absorbs higher-grade residual relative to K*A + KA = geometric_product(K, A) + try: + N = unitize_versor(geometric_product(reverse(KA), V_arr)) + except ValueError: + N = I.copy() + + recon = geometric_product(geometric_product(K, A), N) + recon_res = float(np.linalg.norm(recon - V_arr)) + + for name, factor in (("K", K), ("A", A), ("N", N)): + c = versor_condition(factor) + if c >= _CLOSURE_TOL: + raise ValueError(f"Cartan–Iwasawa factor {name} not closed: {c:.3e}") + + return CartanIwasawaFactors( + K=K.astype(np.float64, copy=False), + A=A.astype(np.float64, copy=False), + N=N.astype(np.float64, copy=False), + reconstruction_residual=recon_res, + ) + + +def dual_correction_slerp( + source: np.ndarray, + target: np.ndarray, + alpha: float, +) -> np.ndarray: + """Slerp on Cartan–Iwasawa factors of the transition rotor (dual-correction). + + Factors are powered independently then recomposed as left action on source: + + R = target * reverse(source) = K A N + out = (K^α A^α N^α) * source + + α=0 → source; α=1 → target for unit versors. Sandwich conjugation is not + the state geodesic (see ADR-0238 supervised_blend). + """ + from algebra.rotor import rotor_power + + a = float(alpha) + if a < 0.0 or a > 1.0: + raise ValueError("alpha must be in [0, 1]") + src = np.asarray(source, dtype=np.float64) + tgt = np.asarray(target, dtype=np.float64) + if a <= _NEAR_ZERO: + out = src.copy() + elif a >= 1.0 - _NEAR_ZERO: + out = tgt.copy() + else: + R = word_transition_rotor(src, tgt) + factors = cartan_iwasawa_factorize(R) + K_a = rotor_power(factors.K, a) + A_a = rotor_power(factors.A, a) + N_a = rotor_power(factors.N, a) + R_a = geometric_product(geometric_product(K_a, A_a), N_a) + R_a = unitize_versor(R_a) + out = geometric_product(R_a, src).astype(np.float64) + cond = versor_condition(out) + if cond >= _CLOSURE_TOL: + raise ValueError(f"dual_correction_slerp broke closure: {cond:.3e}") + return out diff --git a/core/physics/goldtether.py b/core/physics/goldtether.py new file mode 100644 index 00000000..21b21dae --- /dev/null +++ b/core/physics/goldtether.py @@ -0,0 +1,432 @@ +"""core.physics.goldtether — Coherence GoldTether (ADR-0238). + +This module implements the *field* GoldTether: dynamic grade-5 pseudoscalar +floor, harmonized coherence residual, and practice/serve autonomy modulation. + +It is intentionally distinct from the *arena* GoldTether protocol in +``core.learning_arena.protocols.GoldTether`` (ADR-0199), which scores practice +attempts against independent truth anchors. Shared metaphor; different contracts. +Never import or subclass the arena protocol from this module. + +Construction boundary: supervised blend closes via ``algebra.rotor`` manifold +slerp (``word_transition_rotor`` + ``rotor_power``). No hot-path drift repair. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field, replace +from enum import Enum +from typing import Any, Mapping + +import numpy as np + +from algebra.cl41 import N_COMPONENTS, geometric_product, reverse +from algebra.rotor import rotor_power, word_transition_rotor +from algebra.versor import versor_condition + +_PSEUDOSCALAR_IDX = 31 +_CLOSURE_TOL = 1e-6 +_NEAR_ZERO = 1e-12 +_DEFAULT_DECAY_N = 32 +_DEFAULT_W_DRIFT = 0.35 +_DEFAULT_FLOOR_INIT = 0.15 +_DEFAULT_CRITICAL_RATIO = 2.5 +_TELEMETRY_SCHEMA = "goldtether_coherence_v1" + + +class OperatingMode(str, Enum): + """Practice vs Serve — risk-reward physics boundary.""" + + PRACTICE = "practice" + SERVE = "serve" + + +class AutonomyBand(str, Enum): + """Residual-relative autonomy envelope (ADR-0238).""" + + AUTONOMOUS = "autonomous" + SUPERVISED_BLEND = "supervised_blend" + FAIL_CLOSED = "fail_closed" + + +@dataclass(frozen=True, slots=True) +class GoldTetherConfig: + """Named configuration only — no silent magic constants at call sites.""" + + decay_N: int = _DEFAULT_DECAY_N + w_drift: float = _DEFAULT_W_DRIFT + floor_init: float = _DEFAULT_FLOOR_INIT + critical_ratio: float = _DEFAULT_CRITICAL_RATIO + practice_autonomy_enabled: bool = False + serve_supervised_blend_authorized: bool = False + + def __post_init__(self) -> None: + if self.decay_N < 1: + raise ValueError("decay_N must be >= 1") + if not 0.0 <= self.w_drift <= 1.0: + raise ValueError("w_drift must be in [0, 1]") + if self.floor_init <= 0.0: + raise ValueError("floor_init must be positive") + if self.critical_ratio <= 1.0: + raise ValueError("critical_ratio must be > 1") + + +@dataclass(frozen=True, slots=True) +class CoherenceResidual: + """Harmonized residual: drift + geometric distance (normalized). + + Distinct from ADR-0006 ``EnergyProfile.coherence_residual`` and from + ADR-0239 Procrustes/Surprise residuals. + """ + + drift: float + geometric_distance: float + combined: float + kappa: float + pseudoscalar_current: float + pseudoscalar_reference: float + + +@dataclass(frozen=True, slots=True) +class AutonomyDecision: + band: AutonomyBand + residual: float + floor: float + critical: float + mode: OperatingMode + blend_alpha: float + reason: str + + +@dataclass(frozen=True, slots=True) +class PseudoscalarFloorState: + """Dynamic grade-5 coherence floor + sign/magnitude telemetry.""" + + value: float + sign: float + n_samples: int + last_update_step: int + primal_floor: float + recent_residuals: tuple[float, ...] = () + + +def _as_mv(v: np.ndarray, name: str) -> np.ndarray: + arr = np.asarray(v, dtype=np.float64) + if arr.shape != (N_COMPONENTS,): + raise ValueError(f"{name} must have shape ({N_COMPONENTS},); got {arr.shape}") + return arr + + +def _pseudoscalar(v: np.ndarray) -> float: + return float(v[_PSEUDOSCALAR_IDX]) + + +def _geometric_distance(a: np.ndarray, b: np.ndarray) -> float: + """Closed geometric distance on the versor manifold. + + Uses ``|| reverse(a) * b - 1 ||_F`` — zero iff a and b are the same unit + versor (up to float noise). Not cosine similarity; not ANN. + """ + product = geometric_product(reverse(a), b).astype(np.float64) + product[0] -= 1.0 + return float(np.linalg.norm(product)) + + +def _normalize_distance(d: float, scale: float = 2.0) -> float: + """Map unbounded residual to [0, 1) for blend with drift.""" + if d <= 0.0: + return 0.0 + return float(d / (d + scale)) + + +def derive_kappa(combined_residual: float, floor: float) -> float: + """Monotone κ from residual relative to floor. + + κ ∈ (0, 1]: small residual → κ near 1 (more trust); large residual → κ → 0. + Used only to scale dual-correction blend weight — never to invent content. + """ + if floor <= _NEAR_ZERO: + floor = _NEAR_ZERO + ratio = max(0.0, float(combined_residual) / float(floor)) + return float(1.0 / (1.0 + ratio)) + + +@dataclass +class GoldTetherMonitor: + """Stateful monitor for coherence residual, floor, and autonomy decisions. + + State is explicit and reconstructible; updates are pure replacements on + ``floor_state`` (immutable snapshots). Deterministic for identical sequences. + """ + + config: GoldTetherConfig = field(default_factory=GoldTetherConfig) + floor_state: PseudoscalarFloorState = field(init=False) + _step: int = field(default=0, init=False, repr=False) + + def __post_init__(self) -> None: + self.floor_state = PseudoscalarFloorState( + value=float(self.config.floor_init), + sign=1.0, + n_samples=0, + last_update_step=0, + primal_floor=float(self.config.floor_init), + recent_residuals=(), + ) + + def measure( + self, + current: np.ndarray, + reference: np.ndarray, + *, + mode: OperatingMode | str = OperatingMode.PRACTICE, + ) -> CoherenceResidual: + """Compute harmonized coherence residual (pure; does not mutate floor).""" + _ = OperatingMode(mode) # validate + cur = _as_mv(current, "current") + ref = _as_mv(reference, "reference") + ps_c = _pseudoscalar(cur) + ps_r = _pseudoscalar(ref) + drift = abs(ps_c - ps_r) + # Also fold absolute pseudoscalar magnitude loss relative to reference. + drift = max(drift, abs(abs(ps_c) - abs(ps_r))) + geo = _geometric_distance(ref, cur) + geo_n = _normalize_distance(geo) + w = float(self.config.w_drift) + combined = w * drift + (1.0 - w) * geo_n + kappa = derive_kappa(combined, self.floor_state.value) + return CoherenceResidual( + drift=float(drift), + geometric_distance=float(geo), + combined=float(combined), + kappa=float(kappa), + pseudoscalar_current=ps_c, + pseudoscalar_reference=ps_r, + ) + + def update_floor( + self, + residual: CoherenceResidual | float, + *, + mode: OperatingMode | str = OperatingMode.PRACTICE, + success: bool = True, + pseudoscalar_sign: float | None = None, + ) -> PseudoscalarFloorState: + """Update dynamic floor from practice successes only. + + Serve mode never promotes the floor. Failures only append telemetry + window; they do not raise the autonomy envelope. + """ + op_mode = OperatingMode(mode) + r = float(residual.combined if isinstance(residual, CoherenceResidual) else residual) + self._step += 1 + recent = list(self.floor_state.recent_residuals) + [r] + decay_n = int(self.config.decay_N) + if len(recent) > decay_n: + recent = recent[-decay_n:] + + new_value = self.floor_state.value + new_sign = self.floor_state.sign + n_samples = self.floor_state.n_samples + + if op_mode is OperatingMode.PRACTICE and success and r < self.floor_state.value: + # Tighten floor toward observed residual while keeping primal anchor. + # Weighted mean of recent successes under decay window. + window = [x for x in recent if x < self.floor_state.value] or [r] + mean_r = float(sum(window) / len(window)) + # Blend toward mean_r but never below half primal (safety). + floor_floor = 0.5 * self.floor_state.primal_floor + candidate = 0.5 * self.floor_state.value + 0.5 * mean_r + new_value = max(floor_floor, min(self.floor_state.value, candidate)) + n_samples = n_samples + 1 + if pseudoscalar_sign is not None and abs(pseudoscalar_sign) > _NEAR_ZERO: + new_sign = 1.0 if pseudoscalar_sign >= 0.0 else -1.0 + + self.floor_state = PseudoscalarFloorState( + value=float(new_value), + sign=float(new_sign), + n_samples=int(n_samples), + last_update_step=int(self._step), + primal_floor=float(self.floor_state.primal_floor), + recent_residuals=tuple(float(x) for x in recent), + ) + return self.floor_state + + def decide( + self, + residual: CoherenceResidual | float, + *, + mode: OperatingMode | str = OperatingMode.PRACTICE, + floor: PseudoscalarFloorState | None = None, + ) -> AutonomyDecision: + """Map residual + mode → autonomy band (HITL-safe defaults).""" + op_mode = OperatingMode(mode) + r = float(residual.combined if isinstance(residual, CoherenceResidual) else residual) + fl = floor if floor is not None else self.floor_state + floor_v = float(fl.value) + critical = floor_v * float(self.config.critical_ratio) + + if r > critical: + return AutonomyDecision( + band=AutonomyBand.FAIL_CLOSED, + residual=r, + floor=floor_v, + critical=critical, + mode=op_mode, + blend_alpha=0.0, + reason="residual_above_critical", + ) + + if op_mode is OperatingMode.SERVE: + # Serve never autonomous. Supervised blend only if explicitly authorized. + if r < floor_v and self.config.serve_supervised_blend_authorized: + alpha = float(1.0 - derive_kappa(r, floor_v)) + return AutonomyDecision( + band=AutonomyBand.SUPERVISED_BLEND, + residual=r, + floor=floor_v, + critical=critical, + mode=op_mode, + blend_alpha=alpha, + reason="serve_supervised_authorized", + ) + if r < floor_v: + return AutonomyDecision( + band=AutonomyBand.FAIL_CLOSED, + residual=r, + floor=floor_v, + critical=critical, + mode=op_mode, + blend_alpha=0.0, + reason="serve_hitl_default_fail_closed", + ) + # floor <= r <= critical on serve: fail-closed unless blend authorized + if self.config.serve_supervised_blend_authorized: + alpha = float(min(1.0, (r - floor_v) / max(critical - floor_v, _NEAR_ZERO))) + return AutonomyDecision( + band=AutonomyBand.SUPERVISED_BLEND, + residual=r, + floor=floor_v, + critical=critical, + mode=op_mode, + blend_alpha=alpha, + reason="serve_midband_supervised", + ) + return AutonomyDecision( + band=AutonomyBand.FAIL_CLOSED, + residual=r, + floor=floor_v, + critical=critical, + mode=op_mode, + blend_alpha=0.0, + reason="serve_midband_fail_closed", + ) + + # Practice path + if r < floor_v and self.config.practice_autonomy_enabled: + return AutonomyDecision( + band=AutonomyBand.AUTONOMOUS, + residual=r, + floor=floor_v, + critical=critical, + mode=op_mode, + blend_alpha=1.0, + reason="practice_below_floor_autonomy_enabled", + ) + if r < floor_v: + return AutonomyDecision( + band=AutonomyBand.SUPERVISED_BLEND, + residual=r, + floor=floor_v, + critical=critical, + mode=op_mode, + blend_alpha=float(derive_kappa(r, floor_v)), + reason="practice_below_floor_supervised_default", + ) + # floor <= r <= critical + alpha = float(1.0 - min(1.0, (r - floor_v) / max(critical - floor_v, _NEAR_ZERO))) + return AutonomyDecision( + band=AutonomyBand.SUPERVISED_BLEND, + residual=r, + floor=floor_v, + critical=critical, + mode=op_mode, + blend_alpha=max(0.0, min(1.0, alpha)), + reason="practice_midband_supervised", + ) + + def supervised_blend( + self, + source: np.ndarray, + target: np.ndarray, + alpha: float, + ) -> np.ndarray: + """Manifold slerp from source toward target by alpha ∈ [0, 1]. + + Dual-correction surface on the Spin group (not Euclidean lerp): + + R = word_transition_rotor(source, target) # = target * reverse(source) + out = rotor_power(R, α) * source # left composition + + At α=0 → source; at α=1 → target (unit versors). Sandwich conjugation + would map the identity to itself and is the wrong geodesic for state + interpolation. Output must satisfy versor_condition < 1e-6. + """ + a = float(alpha) + if a < 0.0 or a > 1.0: + raise ValueError("alpha must be in [0, 1]") + src = _as_mv(source, "source") + tgt = _as_mv(target, "target") + if a <= _NEAR_ZERO: + out = src.copy() + elif a >= 1.0 - _NEAR_ZERO: + out = tgt.copy() + else: + R = word_transition_rotor(src, tgt) + R_a = rotor_power(R, a) + out = geometric_product(R_a, src).astype(np.float64) + + cond = versor_condition(out) + if cond >= _CLOSURE_TOL: + raise ValueError( + f"supervised_blend broke versor_condition: {cond:.3e} >= {_CLOSURE_TOL}" + ) + return out.astype(np.float64, copy=False) + + def telemetry(self) -> dict[str, Any]: + """Schema-versioned pure projection for workbench channels.""" + fl = self.floor_state + return { + "schema_version": _TELEMETRY_SCHEMA, + "pseudoscalar_floor": float(fl.value), + "pseudoscalar_sign": float(fl.sign), + "n_samples": int(fl.n_samples), + "last_update_step": int(fl.last_update_step), + "primal_floor": float(fl.primal_floor), + "recent_residuals": list(fl.recent_residuals), + "config": { + "decay_N": int(self.config.decay_N), + "w_drift": float(self.config.w_drift), + "floor_init": float(self.config.floor_init), + "critical_ratio": float(self.config.critical_ratio), + "practice_autonomy_enabled": bool(self.config.practice_autonomy_enabled), + "serve_supervised_blend_authorized": bool( + self.config.serve_supervised_blend_authorized + ), + }, + } + + +def with_config(monitor: GoldTetherMonitor, **updates: Any) -> GoldTetherMonitor: + """Return a new monitor with updated config (immutability-friendly).""" + cfg = replace(monitor.config, **updates) + m = GoldTetherMonitor(config=cfg) + m.floor_state = monitor.floor_state + m._step = monitor._step + return m + + +def residual_from_mapping(payload: Mapping[str, Any]) -> float: + """Deterministic residual extract for telemetry/replay fixtures.""" + if "combined" in payload: + return float(payload["combined"]) + raise KeyError("payload missing combined residual") diff --git a/core/physics/self_authorship.py b/core/physics/self_authorship.py new file mode 100644 index 00000000..b2ac7b24 --- /dev/null +++ b/core/physics/self_authorship.py @@ -0,0 +1,190 @@ +"""core.physics.self_authorship — Self-Authorship Miner scaffold (ADR-0240). + +Geometry-guided ADR / teaching proposals under invariants. Emits +**proposal-only** artifacts (SPECULATIVE). Never writes vault COHERENT, +never mutates packs, never touches serving. + +Complements the existing auto-proposal corridor (ADR-0151). This miner +produces structured proposal dicts; promotion remains human-reviewed. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Any, Mapping, Sequence + +import numpy as np + +from algebra.cl41 import N_COMPONENTS +from algebra.versor import versor_condition +from core.physics.dynamic_manifold import conformal_procrustes, signature_aware_pca +from core.physics.goldtether import CoherenceResidual, GoldTetherMonitor, OperatingMode +from core.physics.surprise import dual_operator, surprise_residual + + +@dataclass(frozen=True, slots=True) +class AuthorshipProposal: + """Proposal-only artifact — never auto-accepted.""" + + proposal_id: str + kind: str + epistemic_status: str # always SPECULATIVE at emission + drift_residual: float + closure_proof: Mapping[str, Any] + body: Mapping[str, Any] + adr_refs: tuple[str, ...] + + def as_dict(self) -> dict[str, Any]: + return { + "proposal_id": self.proposal_id, + "kind": self.kind, + "epistemic_status": self.epistemic_status, + "drift_residual": self.drift_residual, + "closure_proof": dict(self.closure_proof), + "body": dict(self.body), + "adr_refs": list(self.adr_refs), + } + + +def _content_id(payload: Mapping[str, Any]) -> str: + raw = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] + + +class SelfAuthorshipMiner: + """Mine minimal extension proposals from geometric residual structure.""" + + def __init__( + self, + *, + goldtether: GoldTetherMonitor | None = None, + residual_threshold: float = 0.25, + ) -> None: + self.goldtether = goldtether or GoldTetherMonitor() + self.residual_threshold = float(residual_threshold) + + def mine_from_trajectory( + self, + current: np.ndarray, + reference: np.ndarray, + *, + basis: Sequence[np.ndarray] = (), + analogs: Sequence[tuple[str, np.ndarray, np.ndarray]] = (), + notes: str = "", + ) -> tuple[AuthorshipProposal, ...]: + """Emit zero or more SPECULATIVE proposals. Never stores them.""" + cur = np.asarray(current, dtype=np.float64) + ref = np.asarray(reference, dtype=np.float64) + if cur.shape != (N_COMPONENTS,) or ref.shape != (N_COMPONENTS,): + raise ValueError("current and reference must be 32-component multivectors") + + residual: CoherenceResidual = self.goldtether.measure( + cur, ref, mode=OperatingMode.PRACTICE + ) + proposals: list[AuthorshipProposal] = [] + + # Closure proof for the reference/current pair under transition. + try: + proc = conformal_procrustes([ref], [cur]) + closure_ok = versor_condition(proc.versor) < 1e-6 + proc_res = float(proc.residual_norm) + except ValueError as exc: + closure_ok = False + proc_res = float("inf") + proc_err = str(exc) + else: + proc_err = "" + + closure_proof = { + "versor_condition_current": float(versor_condition(cur)), + "versor_condition_reference": float(versor_condition(ref)), + "procrustes_residual": proc_res, + "procrustes_closed": closure_ok, + "procrustes_error": proc_err, + "coherence_combined": float(residual.combined), + "kappa": float(residual.kappa), + } + + if residual.combined >= self.residual_threshold and closure_ok: + body = { + "notes": notes, + "suggested_action": "review_coherence_gap", + "drift": float(residual.drift), + "geometric_distance": float(residual.geometric_distance), + } + pid = _content_id({"kind": "coherence_gap", "body": body, "proof": closure_proof}) + proposals.append( + AuthorshipProposal( + proposal_id=f"selfauth-{pid}", + kind="coherence_gap", + epistemic_status="SPECULATIVE", + drift_residual=float(residual.combined), + closure_proof=closure_proof, + body=body, + adr_refs=("ADR-0238", "ADR-0240"), + ) + ) + + if basis: + surp = surprise_residual(cur, basis) + dual = dual_operator( + cur, + basis, + analogs, + kappa=max(residual.kappa, 1e-6), + ) + if dual.productive: + body = { + "notes": notes, + "suggested_action": "review_analogical_extension", + "surprise_norm": float(surp.residual_norm), + "selected_analog_id": dual.selected_analog_id, + "procrustes_residual": ( + float(dual.procrustes.residual_norm) if dual.procrustes else None + ), + } + pid = _content_id( + {"kind": "analogical_extension", "body": body, "proof": closure_proof} + ) + proposals.append( + AuthorshipProposal( + proposal_id=f"selfauth-{pid}", + kind="analogical_extension", + epistemic_status="SPECULATIVE", + drift_residual=float(surp.residual_norm), + closure_proof=closure_proof, + body=body, + adr_refs=("ADR-0239", "ADR-0240"), + ) + ) + + # Optional manifold annotation when a small cloud is available via analogs. + cloud = [ref, cur] + [s for _, s, _ in analogs] + [t for _, _, t in analogs] + if len(cloud) >= 2: + pca = signature_aware_pca(cloud, max_axes=4) + if pca.n_null > 0: + body = { + "notes": notes, + "suggested_action": "review_null_axes", + "n_null": int(pca.n_null), + "n_spacelike": int(pca.n_spacelike), + "n_timelike": int(pca.n_timelike), + } + pid = _content_id({"kind": "null_axis_review", "body": body}) + proposals.append( + AuthorshipProposal( + proposal_id=f"selfauth-{pid}", + kind="null_axis_review", + epistemic_status="SPECULATIVE", + drift_residual=float(residual.combined), + closure_proof=closure_proof, + body=body, + adr_refs=("ADR-0239", "ADR-0240"), + ) + ) + + # Stable order by proposal_id for replay-determinism. + proposals.sort(key=lambda p: p.proposal_id) + return tuple(proposals) diff --git a/core/physics/surprise.py b/core/physics/surprise.py new file mode 100644 index 00000000..7b638e03 --- /dev/null +++ b/core/physics/surprise.py @@ -0,0 +1,212 @@ +"""core.physics.surprise — Surprise Residual + dual with Procrustes (ADR-0239). + +Surprise Residual: S(x) = x − proj_B(x) +where B is a known basis (ordered multivector span). High surprise seeds +Conformal Procrustes against vault analogs → productive novelty only when the +post-transfer residual is below threshold; otherwise typed refuse. + +No sampling. No statistical ranking. Deterministic ordered analog lists. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Sequence + +import numpy as np + +from algebra.cl41 import N_COMPONENTS +from algebra.versor import versor_condition +from core.physics.dynamic_manifold import ( + ConformalProcrustesResult, + conformal_procrustes, + procrustes_residual, +) + +_NEAR_ZERO = 1e-12 +_CLOSURE_TOL = 1e-6 +_DEFAULT_PRODUCTIVE_THRESHOLD = 0.35 + + +@dataclass(frozen=True, slots=True) +class SurpriseResult: + residual_mv: np.ndarray + residual_norm: float + projection: np.ndarray + basis_rank: int + + +@dataclass(frozen=True, slots=True) +class AnalogySeed: + analog_id: str + source: np.ndarray + target: np.ndarray + surprise_affinity: float + + +@dataclass(frozen=True, slots=True) +class DualOperatorResult: + surprise: SurpriseResult + procrustes: ConformalProcrustesResult | None + productive: bool + kappa: float + reason: str + selected_analog_id: str | None + + +def _as_mv(v: np.ndarray, name: str) -> np.ndarray: + arr = np.asarray(v, dtype=np.float64) + if arr.shape != (N_COMPONENTS,): + raise ValueError(f"{name} must have shape ({N_COMPONENTS},); got {arr.shape}") + return arr + + +def _orthonormalize_basis(basis: Sequence[np.ndarray]) -> np.ndarray: + """Deterministic Gram–Schmidt on coefficient space (ordered input). + + Returns matrix B with shape (rank, 32). Zero / dependent vectors dropped + in order — not silently as "null PCA axes"; rank is reported. + """ + cols: list[np.ndarray] = [] + for i, b in enumerate(basis): + v = _as_mv(b, f"basis[{i}]").copy() + for u in cols: + v = v - float(np.dot(v, u)) * u + n = float(np.linalg.norm(v)) + if n < _NEAR_ZERO: + continue + cols.append(v / n) + if not cols: + return np.zeros((0, N_COMPONENTS), dtype=np.float64) + return np.stack(cols, axis=0) + + +def project_onto_basis(x: np.ndarray, basis: Sequence[np.ndarray]) -> np.ndarray: + """Orthogonal projection of x onto span(B) in coefficient space.""" + x_arr = _as_mv(x, "x") + B = _orthonormalize_basis(basis) + if B.shape[0] == 0: + return np.zeros(N_COMPONENTS, dtype=np.float64) + # proj = sum_i u_i + coeffs = B @ x_arr + return (B.T @ coeffs).astype(np.float64, copy=False) + + +def surprise_residual( + x: np.ndarray, + basis: Sequence[np.ndarray], +) -> SurpriseResult: + """S(x) = x − proj_B(x). Residual is orthogonal to span(B).""" + x_arr = _as_mv(x, "x") + proj = project_onto_basis(x_arr, basis) + residual = (x_arr - proj).astype(np.float64, copy=False) + B = _orthonormalize_basis(basis) + return SurpriseResult( + residual_mv=residual, + residual_norm=float(np.linalg.norm(residual)), + projection=proj, + basis_rank=int(B.shape[0]), + ) + + +def analogy_seed( + surprise: SurpriseResult, + analogs: Sequence[tuple[str, np.ndarray, np.ndarray]], + *, + top_k: int | None = None, +) -> tuple[AnalogySeed, ...]: + """Rank vault analogs by affinity to the surprise residual (deterministic). + + Affinity = |cos| between residual and (target − source) direction in + coefficient space. Higher affinity → better structural candidate. + Order is stable: affinity desc, then analog_id asc. + """ + if surprise.residual_norm < _NEAR_ZERO: + return () + r = surprise.residual_mv + r_n = surprise.residual_norm + seeds: list[AnalogySeed] = [] + for item in analogs: + if len(item) != 3: + raise ValueError("each analog must be (id, source, target)") + aid, src, tgt = item + s = _as_mv(src, f"analog[{aid}].source") + t = _as_mv(tgt, f"analog[{aid}].target") + delta = t - s + dn = float(np.linalg.norm(delta)) + if dn < _NEAR_ZERO: + aff = 0.0 + else: + aff = abs(float(np.dot(r, delta)) / (r_n * dn)) + seeds.append( + AnalogySeed( + analog_id=str(aid), + source=s, + target=t, + surprise_affinity=float(aff), + ) + ) + seeds.sort(key=lambda s: (-s.surprise_affinity, s.analog_id)) + if top_k is not None: + seeds = seeds[: max(0, int(top_k))] + return tuple(seeds) + + +def dual_operator( + x: np.ndarray, + basis: Sequence[np.ndarray], + analogs: Sequence[tuple[str, np.ndarray, np.ndarray]], + *, + kappa: float = 1.0, + productive_threshold: float = _DEFAULT_PRODUCTIVE_THRESHOLD, + min_surprise: float = 1e-6, +) -> DualOperatorResult: + """Surprise + Procrustes dual. + + High surprise seeds Procrustes against the best analog. Productive only when + post-transfer residual ≤ productive_threshold * (1/κ-scaled). κ from + CoherenceGoldTether scales the threshold (higher κ → stricter). + """ + if kappa <= 0.0: + raise ValueError("kappa must be positive") + surprise = surprise_residual(x, basis) + if surprise.residual_norm < min_surprise: + return DualOperatorResult( + surprise=surprise, + procrustes=None, + productive=False, + kappa=float(kappa), + reason="surprise_below_minimum", + selected_analog_id=None, + ) + + seeds = analogy_seed(surprise, analogs, top_k=1) + if not seeds: + return DualOperatorResult( + surprise=surprise, + procrustes=None, + productive=False, + kappa=float(kappa), + reason="no_analogs", + selected_analog_id=None, + ) + + best = seeds[0] + # Transfer: Procrustes from analog source→target applied as structural map; + # measure residual of mapping x's projection-completion toward analog target shape. + proc = conformal_procrustes([best.source], [best.target]) + # Residual of applying the analog's versor to x vs. the analog target direction. + transfer_res = procrustes_residual(x, best.target, proc.versor) + # Also include native procrustes residual of the analog pair itself. + residual = max(float(proc.residual_norm), float(transfer_res)) + threshold = float(productive_threshold) / float(kappa) + productive = residual <= threshold and versor_condition(proc.versor) < _CLOSURE_TOL + + return DualOperatorResult( + surprise=surprise, + procrustes=proc, + productive=bool(productive), + kappa=float(kappa), + reason="productive_novelty" if productive else "residual_above_threshold", + selected_analog_id=best.analog_id, + ) diff --git a/core/physics/temporal_gate.py b/core/physics/temporal_gate.py new file mode 100644 index 00000000..ed814119 --- /dev/null +++ b/core/physics/temporal_gate.py @@ -0,0 +1,130 @@ +"""core.physics.temporal_gate — Temporal Admissibility Gate (ADR-0240). + +Wisdom as geometry: refuse premature but eventually-admissible claims with a +typed NOT_YET. Never confabulates early. Pure predicate — no side effects. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Any, Mapping + + +class TemporalVerdict(str, Enum): + ADMIT = "admit" + NOT_YET = "not_yet" + REFUSE = "refuse" + + +@dataclass(frozen=True, slots=True) +class TemporalContext: + """Geometric/temporal preconditions for a claim. + + All fields are explicit; missing evidence is not inventable. + """ + + step: int + min_step: int = 0 + required_evidence_count: int = 0 + evidence_count: int = 0 + coherence_residual: float = 0.0 + residual_ceiling: float = 1.0 + prerequisites_met: bool = True + claim_id: str = "" + + +@dataclass(frozen=True, slots=True) +class TemporalDecision: + verdict: TemporalVerdict + reason: str + claim_id: str + disclosure: Mapping[str, Any] + + +class TemporalAdmissibilityGate: + """Pure temporal admissibility checks.""" + + def __init__( + self, + *, + require_prerequisites: bool = True, + enforce_residual_ceiling: bool = True, + ) -> None: + self.require_prerequisites = bool(require_prerequisites) + self.enforce_residual_ceiling = bool(enforce_residual_ceiling) + + def evaluate(self, ctx: TemporalContext) -> TemporalDecision: + cid = str(ctx.claim_id) + + if ctx.step < 0 or ctx.min_step < 0: + return TemporalDecision( + verdict=TemporalVerdict.REFUSE, + reason="negative_time_index", + claim_id=cid, + disclosure={ + "type": "temporal_refuse", + "detail": "time indices must be non-negative", + }, + ) + + if self.require_prerequisites and not ctx.prerequisites_met: + return TemporalDecision( + verdict=TemporalVerdict.REFUSE, + reason="prerequisites_unmet", + claim_id=cid, + disclosure={ + "type": "temporal_refuse", + "detail": "required prerequisites are not met", + }, + ) + + if self.enforce_residual_ceiling and ctx.coherence_residual > ctx.residual_ceiling: + return TemporalDecision( + verdict=TemporalVerdict.REFUSE, + reason="residual_above_ceiling", + claim_id=cid, + disclosure={ + "type": "temporal_refuse", + "detail": "coherence residual exceeds ceiling", + "residual": float(ctx.coherence_residual), + "ceiling": float(ctx.residual_ceiling), + }, + ) + + if ctx.step < ctx.min_step: + return TemporalDecision( + verdict=TemporalVerdict.NOT_YET, + reason="before_min_step", + claim_id=cid, + disclosure={ + "type": "temporal_not_yet", + "detail": "fullness of time not reached", + "step": int(ctx.step), + "min_step": int(ctx.min_step), + }, + ) + + if ctx.evidence_count < ctx.required_evidence_count: + return TemporalDecision( + verdict=TemporalVerdict.NOT_YET, + reason="insufficient_evidence", + claim_id=cid, + disclosure={ + "type": "temporal_not_yet", + "detail": "evidence count below required floor", + "evidence_count": int(ctx.evidence_count), + "required_evidence_count": int(ctx.required_evidence_count), + }, + ) + + return TemporalDecision( + verdict=TemporalVerdict.ADMIT, + reason="temporally_admissible", + claim_id=cid, + disclosure={ + "type": "temporal_admit", + "step": int(ctx.step), + "evidence_count": int(ctx.evidence_count), + }, + )