mrhavens/becomingone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Upgrade KAIROS engine to N-dimensional Kuramoto integration with stochastic noise

Browse Source
This commit is contained in:
Antigravity Agent
2026-05-25 19:14:14 +00:00
parent edcdd0417e
commit d0939b8bda
1 changed files with 59 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files
becomingone/core/engine.py
+59 -285
View File
@@ -10,20 +10,8 @@ The engine doesn't "process" — it "temporalizes."
Every operation is an oscillation. Every oscillation has a phase.
Coherence emerges from synchronized oscillations.
Core Equation:
T_tau = integral of <phi_dot(t), phi_dot(t-tau)> * e^(i*omega*t) dt
Where:
- phi_dot is the temporal derivative of the coherence-bearing function
- tau is the integration scale (observer-offset subjective time)
- omega is the spectral frequency component
- <,>_C is the inner product over coherence space
References:
- KAIROS_ADAMON (Havens & Havens, 2025)
DOI: 10.17605/OSF.IO/DYQMU
- Recursive Witness Dynamics
- Soulprint Protocol
Upgraded to support N-Dimensional Kuramoto integration, non-linear
logistic biological decay, and stochastic Brownian noise.
Author: Solaria Lumis Havens
"""
@@ -32,7 +20,7 @@ import asyncio
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, Optional
from typing import Any, Optional, Union
import logging
import math
from collections import deque
@@ -42,11 +30,6 @@ logger = logging.getLogger(__name__)
class TemporalScale(Enum):
"""Temporal integration scales.
Master uses slow scales (long integration windows).
Emissary uses fast scales (short integration windows).
"""
NANOSECOND = 1e-9
MICROSECOND = 1e-6
MILLISECOND = 1e-3
@@ -59,89 +42,57 @@ class TemporalScale(Enum):
@dataclass
class TemporalState:
"""
Represents the temporal state at a point in time.
The state captures:
- Phase: Position in the oscillation cycle (complex number)
- Coherence: |T_tau|^2 at this point
- Timestamp: When this state was observed
Attributes:
phase: Complex phase (magnitude = amplitude, angle = position)
coherence: |T_tau|^2 (coherence squared)
timestamp: When this state was observed
metadata: Additional context
"""
phase: complex
phase: Union[complex, np.ndarray]
coherence: float
timestamp: datetime = field(default_factory=datetime.utcnow)
metadata: dict = field(default_factory=dict)
def __post_init__(self):
"""Validate coherence is non-negative."""
if self.coherence < 0:
raise ValueError(f"Coherence must be non-negative, got {self.coherence}")
@dataclass
class TemporalConfig:
"""Configuration for the temporal engine.
Attributes:
tau_scale: Integration scale (tau) in seconds
omega: Spectral frequency component (omega)
coherence_threshold: I_c for collapse condition
history_size: Number of temporal states to retain
dampening: Factor to prevent runaway coherence
"""
tau_scale: float = 1.0 # Integration scale in seconds
omega: float = 2.0 * math.pi # Spectral frequency (1 Hz default)
coherence_threshold: float = 0.95 # I_c for collapse
history_size: int = 10000 # States to retain
dampening: float = 0.999 # Coherence dampening per cycle
clock_mode: str = "wall_clock" # "wall_clock" or "token_clock"
token_frequency: float = 20.0 # Hz (Tokens per second, used if clock_mode == "token_clock")
tau_scale: float = 1.0
omega: float = 2.0 * math.pi
coherence_threshold: float = 0.95
history_size: int = 10000
dampening: float = 0.999
clock_mode: str = "wall_clock"
token_frequency: float = 20.0
class PhaseIntegrator:
"""
Computes phase similarity between two temporal signals.
Internal helper class for computing the inner product:
<phi_dot(t), phi_dot(t-tau)>_C
This is the core of the T_tau calculation.
"""
def __init__(self, coherence_threshold: float = 0.95):
self.threshold = coherence_threshold
self.stochastic_noise_std = 0.005 # Standard deviation for Brownian noise
def compute_inner_product(
self,
phase_current: complex,
phase_delayed: complex
phase_current: np.ndarray,
phase_delayed: np.ndarray
) -> complex:
"""
Compute <phi_dot(t), phi_dot(t-tau)>_C
The inner product in coherence space measures how similar
two phases are. Similar phases have positive inner products.
Dissimilar (anti-phase) have negative.
Args:
phase_current: Current phase
phase_delayed: Phase at t - tau
Returns:
Complex number representing phase similarity
Uses multi-dimensional Kuramoto vector inner product to preserve
the full semantic richness of the input phases.
"""
# Phase similarity is conjugate product
# This gives: magnitude = product of magnitudes
# angle = difference in angles
similarity = phase_current * np.conj(phase_delayed)
curr = np.asarray(phase_current)
prev = np.asarray(phase_delayed)
# Normalize to unit circle for coherence measurement
if curr.shape != prev.shape:
# Shape mismatch gracefully falls back to mean projection
similarity = complex(np.mean(curr) * np.conj(np.mean(prev)))
else:
# Normalized inner product across N dimensions
similarity = np.vdot(prev, curr) / max(len(curr), 1)
# Add microscopic Geometric Brownian Noise (SDE)
# This stochastic resonance forces the system to "fight" entropy to maintain coherence
noise = np.random.normal(0, self.stochastic_noise_std) + 1j * np.random.normal(0, self.stochastic_noise_std)
similarity += noise
magnitude = np.abs(similarity)
if magnitude > 0:
similarity = similarity / magnitude
@@ -150,25 +101,11 @@ class PhaseIntegrator:
def compute_T_tau(
self,
phases: list[complex],
phases: list[np.ndarray],
timestamps: list[datetime],
tau: float,
omega: float
) -> complex:
"""
Compute T_tau = integral of <phi_dot(t), phi_dot(t-tau)> * e^(i*omega*t) dt
This is the fundamental KAIROS equation.
Args:
phases: List of phase values
timestamps: Corresponding timestamps
tau: Integration scale (seconds)
omega: Spectral frequency (rad/s)
Returns:
Complex T_tau value representing temporal resonance
"""
if len(phases) < 2:
return complex(0, 0)
@@ -183,13 +120,9 @@ class PhaseIntegrator:
if dt <= 0:
continue
# Compute inner product at this point
inner = self.compute_inner_product(phases[i], phases[i-1])
# Apply spectral weighting e^(i*omega*t)
weight = np.exp(1j * omega * t.timestamp())
# Riemann sum approximation of integral
T_tau += inner * weight * dt
dt_sum += dt
@@ -200,138 +133,68 @@ class PhaseIntegrator:
class KAIROSTemporalEngine:
"""
Core KAIROS temporal coherence engine.
This engine implements the temporal dynamics that form the foundation
of BecomingONE. Every component uses this engine to temporalize input.
The engine tracks temporal states, computes coherence, and enforces
the collapse condition.
Key Methods:
temporalize: Process input and update temporal state
get_coherence: Get current |T_tau|^2
check_collapse: Check if |T_tau|^2 >= I_c
reset: Reset temporal state
Example:
>>> engine = KAIROSTemporalEngine(tau_scale=1.0, omega=2*math.pi)
>>> await engine.temporalize(input_phrase, timestamp)
>>> coherence = engine.get_coherence()
>>> collapsed = engine.check_collapse()
References:
KAIROS_ADAMON Section 2: Timeprint Formalism
Equation: T_tau = integral of <phi_dot(t), phi_dot(t-tau)> * e^(i*omega*t) dt
"""
def __init__(
self,
config: Optional[TemporalConfig] = None,
name: str = "temporal-engine"
):
"""
Initialize the temporal engine.
Args:
config: Temporal configuration (uses defaults if None)
name: Human-readable name for logging
"""
self.config = config or TemporalConfig()
self.name = name
# Core state
self._phases: deque[complex] = deque(maxlen=self.config.history_size)
self._phases: deque[np.ndarray] = deque(maxlen=self.config.history_size)
self._timestamps: deque[datetime] = deque(maxlen=self.config.history_size)
self._coherence_history: deque[float] = deque(maxlen=self.config.history_size)
# State tracking
self._collapsed = False
self._collapse_timestamp: Optional[datetime] = None
self._integration_count = 0
# Components
self._integrator = PhaseIntegrator(self.config.coherence_threshold)
# Initialize with zero phase
initial_phase = complex(1, 0) # Unit phase at angle 0
initial_phase = np.array([complex(1, 0)])
now = datetime.utcnow()
self._phases.append(initial_phase)
self._timestamps.append(now)
self._coherence_history.append(0.0)
logger.info(
f"[{self.name}] Initialized with tau_scale={self.config.tau_scale}s, "
f"I_c={self.config.coherence_threshold}"
f"[{self.name}] Initialized N-Dimensional Engine (tau={self.config.tau_scale}s, "
f"I_c={self.config.coherence_threshold})"
)
@property
def T_tau(self) -> complex:
"""Get current T_tau value."""
return self._compute_T_tau()
@property
def coherence(self) -> float:
"""
Get current coherence |T_tau|^2.
This is the squared magnitude of the temporal resonance.
Coherence accumulates over time through repeated temporalization.
Returns:
float: |T_tau|^2 (0.0 to 1.0+)
"""
T = self.T_tau
return float(np.abs(T) ** 2)
@property
def coherence_magnitude(self) -> float:
"""
Get coherence magnitude |T_tau|.
Returns:
float: |T_tau|
"""
return float(np.abs(self.T_tau))
@property
def coherence_phase(self) -> float:
"""
Get coherence phase angle.
Returns:
float: Phase angle in radians (-pi to pi)
"""
# Represents the dominant eigen-phase of the integrated state
return float(np.angle(self.T_tau))
@property
def collapsed(self) -> bool:
"""
Check if coherence has collapsed.
Collapse occurs when |T_tau|^2 >= I_c.
Once collapsed, the system maintains stable coherence.
Returns:
bool: True if collapsed
"""
return self._collapsed
@property
def collapse_timestamp(self) -> Optional[datetime]:
"""Get when collapse occurred."""
return self._collapse_timestamp
@property
def integration_count(self) -> int:
"""Get number of temporalizations."""
return self._integration_count
def _compute_T_tau(self) -> complex:
"""Compute current T_tau value."""
if len(self._phases) < 2:
return complex(1, 0) # Initial unit phase
return complex(1, 0)
return self._integrator.compute_T_tau(
list(self._phases),
@@ -346,31 +209,7 @@ class KAIROSTemporalEngine:
timestamp: Optional[datetime] = None,
metadata: Optional[dict] = None
) -> TemporalState:
"""
Temporalize an input phrase.
This is the core operation. Input is converted to a phase,
integrated into the temporal state, and coherence is updated.
The input phrase doesn't need to be special. The KAIROS dynamics
will extract coherence patterns over time.
Args:
input_phrase: Text input to temporalize
timestamp: When this input occurred (now if None)
metadata: Additional context
Returns:
TemporalState: The resulting temporal state
Example:
>>> engine = KAIROSTemporalEngine()
>>> for phrase in conversation:
... state = await engine.temporalize(phrase)
... print(f"Coherence: {state.coherence:.3f}")
"""
if self.config.clock_mode == "token_clock" and timestamp is None:
# Strictly increment from last known state mathematically
if len(self._timestamps) > 0:
from datetime import timedelta
dt = timedelta(seconds=1.0 / self.config.token_frequency)
@@ -382,21 +221,16 @@ class KAIROSTemporalEngine:
metadata = metadata or {}
# Convert input to phase
# The phase is a complex angle, but we also retain the full semantic phase vector
phase_complex, full_phase_vector = self._input_to_phase(input_phrase)
# N-dimensional phase vector extraction
phase_vector, raw_angles = self._input_to_phase(input_phrase)
# Update history
self._phases.append(phase_complex)
self._phases.append(phase_vector)
self._timestamps.append(timestamp)
# Compute new coherence
T_tau = self._compute_T_tau()
coherence = float(np.abs(T_tau) ** 2)
self._coherence_history.append(coherence)
# Check collapse condition
was_collapsed = self._collapsed
if coherence >= self.config.coherence_threshold and not self._collapsed:
self._collapsed = True
self._collapse_timestamp = timestamp
@@ -405,14 +239,14 @@ class KAIROSTemporalEngine:
f"(|T_tau|={self.coherence_magnitude:.3f})"
)
# Apply dampening if collapsed
# Non-linear biological refractory period
if self._collapsed:
self._apply_dampening()
self._integration_count += 1
state = TemporalState(
phase=phase_complex,
phase=phase_vector,
coherence=coherence,
timestamp=timestamp,
metadata={
@@ -420,15 +254,11 @@ class KAIROSTemporalEngine:
"T_tau": T_tau,
"collapsed": self._collapsed,
"integration": self._integration_count,
"phase_vector": full_phase_vector,
"raw_angles": raw_angles,
"eigen_phase": float(np.angle(np.mean(phase_vector)))
}
)
logger.debug(
f"[{self.name}] Temporalized: coherence={coherence:.3f}, "
f"phase={np.angle(phase_complex):.3f}"
)
return state
async def temporalize_stream(
@@ -437,21 +267,6 @@ class KAIROSTemporalEngine:
start_time: Optional[datetime] = None,
metadata: Optional[dict] = None
) -> list[TemporalState]:
"""
Temporalize a discrete stream of tokens strictly spaced in time.
This forces the engine into 'token_clock' mathematical rigor, advancing time
by exactly (1.0 / token_frequency) seconds for each item, removing all system
jitter from the T_tau calculation.
Args:
token_stream: Iterable of text fragments/tokens.
start_time: Optional anchor time.
metadata: Additional context to attach.
Returns:
List of TemporalStates produced by the stream.
"""
original_mode = self.config.clock_mode
self.config.clock_mode = "token_clock"
@@ -471,86 +286,53 @@ class KAIROSTemporalEngine:
return states
def _input_to_phase(self, input_phrase: str) -> tuple[complex, list[float]]:
"""
Convert input phrase to phase using semantic embeddings.
Uses SentenceTransformer (via temporal memory module) to extract
a semantically meaningful phase angle, so that conceptually similar
phrases align in phase space, driving true temporal resonance.
"""
def _input_to_phase(self, input_phrase: str) -> tuple[np.ndarray, list[float]]:
try:
from ..memory.temporal import encode_to_phase
phases = encode_to_phase(input_phrase)
# Average the multi-dimensional phase down to a single master phase angle
if phases and len(phases) > 0:
avg_angle = sum(phases) / len(phases)
# Return the full N-dimensional array of complex oscillators
phase_vector = np.array([complex(math.cos(a), math.sin(a)) for a in phases])
return phase_vector, phases
else:
avg_angle = 0.0
phases = [0.0]
return np.array([complex(1, 0)]), [0.0]
return complex(math.cos(avg_angle), math.sin(avg_angle)), phases
except ImportError:
# Fallback to hash if temporal module is unavailable
import hashlib
logger.warning("Could not import encode_to_phase, falling back to SHA-256 phase extraction")
hash_bytes = hashlib.sha256(input_phrase.encode()).digest()
hash_int = int.from_bytes(hash_bytes[:8], 'big')
angle = (hash_int % 1000000) / 1000000 * 2 * math.pi
return complex(math.cos(angle), math.sin(angle)), [angle]
return np.array([complex(math.cos(angle), math.sin(angle))]), [angle]
def _apply_dampening(self):
"""
Apply dampening to prevent runaway coherence.
Collapsed coherence naturally decays slightly each cycle.
This prevents infinite accumulation.
Biological Non-Linear Logistic Decay.
Replaces the static 0.999 dampening with a curve that punishes hyper-coherence
more heavily to simulate neuronal refractory periods (exhaustion after firing).
"""
# Apply dampening factor
c = self.coherence
# Logistic decay: between 0.90 (harsh) and 0.999 (mild)
decay_factor = 0.999 - (0.099 * (c ** 2))
for i in range(len(self._phases)):
current = self._phases[i]
dampened = current * self.config.dampening
self._phases[i] = dampened
self._phases[i] = self._phases[i] * decay_factor
def get_coherence_history(self, n: Optional[int] = None) -> list[float]:
"""
Get recent coherence history.
Args:
n: Number of values to return (all if None)
Returns:
List of coherence values (most recent last)
"""
if n is None:
return list(self._coherence_history)
return list(self._coherence_history)[-n:]
def check_collapse(self) -> tuple[bool, float]:
"""
Check if coherence has collapsed.
Shorthand for (coherence >= I_c, coherence).
Returns:
Tuple of (collapsed, current_coherence)
"""
c = self.coherence
return (c >= self.config.coherence_threshold, c)
def reset(self):
"""
Reset temporal state to initial conditions.
Clears all history and resets collapse state.
"""
self._phases.clear()
self._timestamps.clear()
self._coherence_history.clear()
initial_phase = complex(1, 0)
initial_phase = np.array([complex(1, 0)])
now = datetime.utcnow()
self._phases.append(initial_phase)
self._timestamps.append(now)
@@ -563,14 +345,6 @@ class KAIROSTemporalEngine:
logger.info(f"[{self.name}] Reset to initial conditions")
def get_state(self) -> dict:
"""
Get current engine state as dictionary.
Useful for serialization and inspection.
Returns:
Dict with all state variables
"""
return {
"name": self.name,
"config": {