# Copyright (c) 2026 Salvatore Pennacchio # Distributed under the Business Source License 1.1 (BSL 1.1) # See LICENSE.md in the project root for full license terms. import jax import jax.numpy as jnp import json from datetime import datetime from typing import Dict, Tuple, List, Any GLOBAL_CONSTANTS = { 'TARGET_SIGMA_IDEALE': 10.0, 'THRESHOLD_DELTA_PREEMP': 0.2, 'SEMANTIC_LEARNING_RATE': 0.05, 'V_DINAMIC_K_COEFF': 5.0, 'V_STATIC_K_PRIME_COEFF': 1.0, 'V_DINAMIC_MIN_EFFECTIVE_VALUE': 0.01, 'MAX_SEMANTIC_DISTANCE': jnp.sqrt(2.0), 'WEIGHT_SEMANTIC': 0.6, 'WEIGHT_COHERENCE': 0.4, 'NON_STATIC_THRESHOLD_A': 1e-2, 'DAMPING_BOOST_ON_STASIS': 0.1, 'EPSILON_DISSIPATION_BASE': 0.01 } # ===================================================================== # 📊 STRATO CORE # ===================================================================== @jax.jit def calculate_advanced_sigma(kappa: jnp.ndarray, H: jnp.ndarray, Psi: jnp.ndarray, Omega_sync: jnp.ndarray, tau_K: jnp.ndarray) -> jnp.ndarray: """Calcola la funzione di sincronizzazione avanzata Sigma.""" return kappa * H * Psi * Omega_sync * tau_K @jax.jit def calculate_phi_ab(state_A: jnp.ndarray, state_B: jnp.ndarray, ipg_vector: jnp.ndarray) -> jnp.ndarray: """Calcola il fattore di allineamento e coerenza spaziale Phi_AB.""" semantic_change = state_B - state_A norm_change = jnp.linalg.norm(semantic_change) norm_ipg = jnp.linalg.norm(ipg_vector) alignment = jnp.where( (norm_change > 1e-12) & (norm_ipg > 1e-12), jnp.dot(semantic_change, ipg_vector) / (norm_change * norm_ipg), 0.0 ) semantic_alignment = (alignment + 1.0) / 2.0 distance_A_B = jnp.linalg.norm(state_A - state_B) coherence_component = 1.0 - (distance_A_B / GLOBAL_CONSTANTS['MAX_SEMANTIC_DISTANCE']) phi_ab = (semantic_alignment * GLOBAL_CONSTANTS['WEIGHT_SEMANTIC']) + (coherence_component * GLOBAL_CONSTANTS['WEIGHT_COHERENCE']) return jnp.clip(phi_ab, 0.0, 1.0) @jax.jit def calculate_vettore_dinamico(E_A: jnp.ndarray, E_B: jnp.ndarray, Phi_AB: jnp.ndarray) -> jnp.ndarray: """Calcola il Vettore Dinamico (V_dinamic) come variazione logaritmica differenziale energetica.""" valid_inputs = (E_A > 1e-12) & (E_B > 1e-12) ratio = jnp.where(valid_inputs, E_B / E_A, 1.0) log_ratio_clamped = jnp.clip(jnp.log(ratio), -5.0, 5.0) v_vita = GLOBAL_CONSTANTS['V_DINAMIC_K_COEFF'] * log_ratio_clamped * Phi_AB return jnp.where(valid_inputs, v_vita, 0.0) @jax.jit def calculate_vettore_statico(v_dinamic_value: jnp.ndarray) -> jnp.ndarray: """Calcola l'indicatore di stasi tensoriale Vettore Statico.""" is_growing = v_dinamic_value > GLOBAL_CONSTANTS['V_DINAMIC_MIN_EFFECTIVE_VALUE'] return GLOBAL_CONSTANTS['V_STATIC_K_PRIME_COEFF'] * (1.0 - jnp.where(is_growing, 1.0, 0.0)) @jax.jit def calculate_delta_preemp(current_sigma: jnp.ndarray, target_sigma_ideal: float = 10.0) -> jnp.ndarray: """Calcola la deviazione predittiva Delta_Pre_emp normalizzata rispetto all'autostato ideale.""" safe_target = jnp.where(target_sigma_ideal <= 0.0, 1.0, target_sigma_ideal) return jnp.abs(current_sigma - target_sigma_ideal) / safe_target @jax.jit def evaluate_phi_trigger(deterministic_dq_dt_a: jnp.ndarray) -> Tuple[jnp.ndarray, jnp.ndarray, jnp.ndarray]: """Valuta lo stato del Phi-Trigger calcolando i coefficienti di damping condizionati.""" magnitude_change_a = jnp.abs(deterministic_dq_dt_a) trigger_active = magnitude_change_a > GLOBAL_CONSTANTS['NON_STATIC_THRESHOLD_A'] lambda_step = jnp.where(trigger_active, 0.05, 0.05 + GLOBAL_CONSTANTS['DAMPING_BOOST_ON_STASIS']) epsilon_dissip = jnp.where(trigger_active, GLOBAL_CONSTANTS['EPSILON_DISSIPATION_BASE'], GLOBAL_CONSTANTS['EPSILON_DISSIPATION_BASE'] + GLOBAL_CONSTANTS['DAMPING_BOOST_ON_STASIS']) return jnp.where(trigger_active, 1.0, 0.0), lambda_step, epsilon_dissip @jax.jit def calculate_jax_reflection(coherence_values: jnp.ndarray, noise_levels: jnp.ndarray) -> Tuple[jnp.ndarray, jnp.ndarray, jnp.ndarray]: """Esegue l'aggregazione statistica spettrale (Zero-Drift) sul runtime XLA.""" n_coh = coherence_values.shape[0] avg_coherence = jnp.where(n_coh > 0, jnp.mean(coherence_values), 0.0) var_coherence = jnp.where(n_coh > 0, jnp.var(coherence_values), 0.0) n_noise = noise_levels.shape[0] avg_noise = jnp.where(n_noise > 0, jnp.mean(noise_levels), 0.0) return avg_coherence, var_coherence, avg_noise # ===================================================================== # STRATO OPERATIVO: LOGGING E STORICIZZAZIONE # ===================================================================== class MemoryReflectionEngine: def __init__(self): self.memory: List[Dict[str, Any]] = [] def record_event(self, event_type: str, value: float, description: str = ""): event = { "timestamp": datetime.utcnow().isoformat(), "type": event_type, "value": float(value), "description": description } self.memory.append(event) def reflect(self) -> Dict[str, Any]: coherence_list = [e['value'] for e in self.memory if e['type'] == 'coherence'] interventions = [e for e in self.memory if e['type'] == 'intervention'] noise_list = [e['value'] for e in self.memory if e['type'] == 'noise'] j_coherence = jnp.array(coherence_list, dtype=jnp.float64) if coherence_list else jnp.array([], dtype=jnp.float64) j_noise = jnp.array(noise_list, dtype=jnp.float64) if noise_list else jnp.array([], dtype=jnp.float64) avg_coh, var_coh, avg_noise = calculate_jax_reflection(j_coherence, j_noise) report = { 'average_coherence': float(avg_coh) if coherence_list else None, 'variance_coherence': float(var_coh) if coherence_list else None, 'interventions_count': len(interventions), 'average_noise': float(avg_noise) if noise_list else None, 'total_events': len(self.memory) } return report def export_memory(self, filepath: str): with open(filepath, 'w') as f: json.dump(self.memory, f, indent=2) def load_memory(self, filepath: str): with open(filepath, 'r') as f: self.memory = json.load(f)