Implement Third-Door operators: - goldtether: coherence residual, dynamic floor, practice/serve autonomy - dynamic_manifold: signature-aware PCA, Procrustes, Cartan-Iwasawa - surprise: residual + dual with Procrustes - biography / temporal_gate / self_authorship: lifelong + proposal-only All versor outputs enforce versor_condition < 1e-6. Spin left-composition geodesic for supervised blend. Arena GoldTether (ADR-0199) untouched.
This commit is contained in:
Shay
parent
bad9b87bff
commit
a8e9977cbc
7 changed files with 1606 additions and 0 deletions
|
|
@ -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",
|
||||
]
|
||||
|
|
|
|||
107
core/physics/biography.py
Normal file
107
core/physics/biography.py
Normal file
|
|
@ -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)),
|
||||
}
|
||||
449
core/physics/dynamic_manifold.py
Normal file
449
core/physics/dynamic_manifold.py
Normal file
|
|
@ -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
|
||||
432
core/physics/goldtether.py
Normal file
432
core/physics/goldtether.py
Normal file
|
|
@ -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")
|
||||
190
core/physics/self_authorship.py
Normal file
190
core/physics/self_authorship.py
Normal file
|
|
@ -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)
|
||||
212
core/physics/surprise.py
Normal file
212
core/physics/surprise.py
Normal file
|
|
@ -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 <x, u_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,
|
||||
)
|
||||
130
core/physics/temporal_gate.py
Normal file
130
core/physics/temporal_gate.py
Normal file
|
|
@ -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),
|
||||
},
|
||||
)
|
||||
Loading…
Reference in a new issue