Spaces:
Running
Running
| """Bounded PAN hyperstate runtime for non-canonical Playground sessions.""" | |
| from __future__ import annotations | |
| import math | |
| import random | |
| from collections.abc import Mapping, MutableMapping, Sequence | |
| from typing import Any | |
| from src.homeostasis.pan_contract import ( | |
| PAN_CONTRACT_ID, | |
| PAN_CONTRACT_STATUS, | |
| PANFormulaParameters, | |
| initialize_pan_state_mapping, | |
| ) | |
| from .candidates import pan_research_candidates | |
| from .hypervector import axis_schema, bundle | |
| PANState = MutableMapping[str, Any] | |
| def _clamp(value: float, low: float = 0.0, high: float = 1.0) -> float: | |
| return max(low, min(high, value)) | |
| def _sigmoid(value: float) -> float: | |
| value = max(-40.0, min(40.0, value)) | |
| return 1.0 / (1.0 + math.exp(-value)) | |
| class PANRuntime: | |
| """Exploratory PAN controller layered on the generic Playground engine.""" | |
| def __init__( | |
| self, | |
| *, | |
| n_neurons: int, | |
| dimensions: int, | |
| seed: int, | |
| feedback_gain: float, | |
| health_decay: float, | |
| apoptosis_threshold: float, | |
| closed_loop: bool, | |
| coordinates: Sequence[Sequence[float]], | |
| degree: Sequence[int], | |
| feedback_delay: int = 0, | |
| feedback_source: str = "population", | |
| feedback_target: str = "all", | |
| feedback_nonlinearity: str = "linear", | |
| feedback_threshold: float = 0.0, | |
| feedback_saturation: float = 100.0, | |
| ) -> None: | |
| if not 5 <= dimensions <= 32: | |
| raise ValueError("PAN dimensions must be between 5 and 32") | |
| self.n_neurons = n_neurons | |
| self.dimensions = dimensions | |
| self.feedback_gain = feedback_gain | |
| self.health_decay = health_decay | |
| self.apoptosis_threshold = apoptosis_threshold | |
| self.closed_loop = closed_loop | |
| self.coordinates = coordinates | |
| self.degree = degree | |
| self.feedback_delay = feedback_delay | |
| self.feedback_source = feedback_source | |
| self.feedback_target = feedback_target | |
| self.feedback_nonlinearity = feedback_nonlinearity | |
| self.feedback_threshold = feedback_threshold | |
| self.feedback_saturation = feedback_saturation | |
| rng = random.Random(seed ^ 0x50414E) | |
| scale = 1.0 / math.sqrt(max(dimensions, 1)) | |
| self.feedback_weights: list[list[float]] = [ | |
| [rng.uniform(-scale, scale) for _ in range(dimensions)] | |
| for _ in range(n_neurons) | |
| ] | |
| self.population_vector = [0.0 for _ in range(dimensions)] | |
| self.feedback_history: list[list[float]] = [] | |
| target_rng = random.Random(seed ^ 0xFADE) | |
| self.random_target_mask = [target_rng.random() < 0.5 for _ in range(n_neurons)] | |
| self.feedback_abs_total = 0.0 | |
| self.feedback_samples = 0 | |
| self.apoptosis_events: list[dict[str, object]] = [] | |
| self.formula_parameters = PANFormulaParameters() | |
| def initialize(self, states: Sequence[PANState]) -> None: | |
| """Attach bounded PAN state variables to existing neuron states.""" | |
| if len(states) != self.n_neurons: | |
| raise ValueError("PAN state count does not match n_neurons") | |
| for neuron_id, state in enumerate(states): | |
| initialize_pan_state_mapping( | |
| state, | |
| neuron_id=neuron_id, | |
| dimensions=self.dimensions, | |
| ) | |
| def feedback_vector(self) -> list[float]: | |
| if self.feedback_delay <= 0: | |
| vector = list(self.population_vector) | |
| elif len(self.feedback_history) > self.feedback_delay: | |
| vector = list(self.feedback_history[-1 - self.feedback_delay]) | |
| else: | |
| vector = [0.0 for _ in range(self.dimensions)] | |
| if self.feedback_source == "layer": | |
| cutoff = max(1, self.dimensions // 2) | |
| vector = [ | |
| value if index < cutoff else 0.0 for index, value in enumerate(vector) | |
| ] | |
| elif self.feedback_source == "subset": | |
| vector = [ | |
| value if index % 2 == 0 else 0.0 for index, value in enumerate(vector) | |
| ] | |
| elif self.feedback_source == "hypervector": | |
| vector = [math.tanh(value) for value in vector] | |
| return vector | |
| def feedback_currents(self) -> list[float]: | |
| """Project a selectable delayed PAN source back to target neurons.""" | |
| if not self.closed_loop: | |
| return [0.0 for _ in range(self.n_neurons)] | |
| vector = self.feedback_vector() | |
| currents: list[float] = [] | |
| layer_limit = max(1, self.n_neurons // 4) | |
| for neuron_id, row in enumerate(self.feedback_weights): | |
| enabled = True | |
| if self.feedback_target == "layer": | |
| enabled = neuron_id < layer_limit | |
| elif self.feedback_target == "random_subset": | |
| enabled = self.random_target_mask[neuron_id] | |
| # Fixed accumulation order across Python versions and CUDA. | |
| raw = 0.0 | |
| for weight, value in zip(row, vector): | |
| raw += weight * value | |
| if self.feedback_nonlinearity == "tanh": | |
| shaped = math.tanh(raw) | |
| elif self.feedback_nonlinearity == "sign": | |
| shaped = 1.0 if raw > 0.0 else (-1.0 if raw < 0.0 else 0.0) | |
| elif self.feedback_nonlinearity == "clip": | |
| shaped = max(-1.0, min(1.0, raw)) | |
| else: | |
| shaped = raw | |
| if abs(shaped) < self.feedback_threshold: | |
| shaped = 0.0 | |
| current = self.feedback_gain * shaped if enabled else 0.0 | |
| current = max( | |
| -self.feedback_saturation, | |
| min(self.feedback_saturation, current), | |
| ) | |
| currents.append(current) | |
| self.feedback_abs_total += abs(current) | |
| self.feedback_samples += 1 | |
| return currents | |
| def position_projection(self, neuron_id: int) -> float: | |
| if neuron_id >= len(self.coordinates): | |
| return 0.0 | |
| point = self.coordinates[neuron_id] | |
| if not point: | |
| return 0.0 | |
| return _clamp(sum(float(value) for value in point) / len(point)) | |
| def update( | |
| self, | |
| *, | |
| tick: int, | |
| dt_ms: float, | |
| states: Sequence[PANState], | |
| spiked_neurons: Sequence[int], | |
| plasticity_active: bool, | |
| ) -> None: | |
| """Advance PAN health, energy and hyperstate after one network tick.""" | |
| spiked = set(spiked_neurons) | |
| vectors: list[list[float]] = [] | |
| params = self.formula_parameters | |
| target_activity = params.target_activity | |
| dt_s = dt_ms / 1000.0 | |
| for neuron_id, state in enumerate(states): | |
| alive = bool(state.get("pan_alive", True)) | |
| if not alive: | |
| vectors.append([float(value) for value in state.get("pan_x_hd", [])]) | |
| continue | |
| did_spike = neuron_id in spiked | |
| activity = float(state.get("pan_activity_ema", 0.01)) | |
| activity = params.activity_decay * activity + params.activity_spike_gain * ( | |
| 1.0 if did_spike else 0.0 | |
| ) | |
| energy = float(state.get("pan_energy", 1.0)) | |
| energy += params.energy_recovery * (1.0 - energy) | |
| if did_spike: | |
| energy -= params.energy_spike_cost | |
| energy = _clamp(energy) | |
| p = _clamp(activity, 1e-6, 1.0 - 1e-6) | |
| surprise_bits = -math.log2(p if did_spike else (1.0 - p)) | |
| information_proxy = _clamp(surprise_bits / params.information_bits_scale) | |
| stress = abs(activity - target_activity) + max( | |
| 0.0, params.energy_stress_floor - energy | |
| ) | |
| health = float(state.get("pan_health", 1.0)) | |
| health -= self.health_decay * dt_s | |
| health -= params.health_stress_loss * stress | |
| health += params.health_recovery_gain * max(0.0, 1.0 - stress) | |
| health = _clamp(health) | |
| consolidation = float(state.get("pan_consolidation", 0.0)) | |
| consolidation = _clamp( | |
| params.consolidation_decay * consolidation | |
| + (params.consolidation_spike_gain if did_spike else 0.0) | |
| ) | |
| amplitude = _clamp( | |
| params.amplitude_floor + params.amplitude_health_gain * health, | |
| params.amplitude_floor, | |
| 1.0, | |
| ) | |
| membrane = float(state.get("v", -65.0)) | |
| threshold = float(state.get("pan_threshold", -50.0)) | |
| excitability = _sigmoid( | |
| (membrane - threshold) / params.excitability_scale_mv | |
| ) | |
| plasticity = health if plasticity_active else 0.0 | |
| period = params.neuromodulation_period_ticks | |
| neuromodulation = 0.5 + 0.5 * math.sin( | |
| 2.0 * math.pi * (tick % period) / period | |
| ) | |
| coupling = _clamp(self.degree[neuron_id] / max(self.n_neurons - 1, 1)) | |
| vector = [0.0 for _ in range(self.dimensions)] | |
| base_values = [ | |
| (tick + 1) * dt_s, | |
| excitability, | |
| plasticity, | |
| information_proxy, | |
| health, | |
| neuromodulation, | |
| energy, | |
| self.position_projection(neuron_id), | |
| consolidation, | |
| coupling, | |
| ] | |
| for index, value in enumerate(base_values[: self.dimensions]): | |
| vector[index] = float(value) | |
| state["pan_health"] = health | |
| state["pan_amplitude"] = amplitude | |
| state["pan_energy"] = energy | |
| state["pan_activity_ema"] = activity | |
| state["pan_consolidation"] = consolidation | |
| state["pan_information_proxy"] = information_proxy | |
| state["pan_x_hd"] = vector | |
| if health <= self.apoptosis_threshold: | |
| state["pan_alive"] = False | |
| self.apoptosis_events.append( | |
| { | |
| "tick": tick, | |
| "neuron_id": neuron_id, | |
| "health": health, | |
| } | |
| ) | |
| vectors.append(vector) | |
| if vectors: | |
| # Stable neuron order across Python versions and the CUDA reduction. | |
| self.population_vector = [] | |
| for index in range(self.dimensions): | |
| total = 0.0 | |
| for vector in vectors: | |
| total += vector[index] | |
| if not math.isfinite(total): | |
| raise ValueError("PAN population must remain finite") | |
| self.population_vector.append(total / len(vectors)) | |
| self.feedback_history.append(list(self.population_vector)) | |
| keep = max(2, self.feedback_delay + 2) | |
| if len(self.feedback_history) > keep: | |
| self.feedback_history = self.feedback_history[-keep:] | |
| def summary(self, states: Sequence[Mapping[str, Any]]) -> dict[str, object]: | |
| """Return descriptive PAN diagnostics with explicit evidence boundaries.""" | |
| health = [float(state.get("pan_health", 1.0)) for state in states] | |
| energy = [float(state.get("pan_energy", 1.0)) for state in states] | |
| alive = sum(1 for state in states if bool(state.get("pan_alive", True))) | |
| vectors = [ | |
| [float(value) for value in state.get("pan_x_hd", [])] for state in states | |
| ] | |
| sample = [ | |
| { | |
| "neuron_id": index, | |
| "alive": bool(state.get("pan_alive", True)), | |
| "health": float(state.get("pan_health", 1.0)), | |
| "energy": float(state.get("pan_energy", 1.0)), | |
| "amplitude": float(state.get("pan_amplitude", 1.0)), | |
| "x_hd": [float(value) for value in state.get("pan_x_hd", [])], | |
| } | |
| for index, state in enumerate(states[:128]) | |
| ] | |
| feedback_mean = ( | |
| self.feedback_abs_total / self.feedback_samples | |
| if self.feedback_samples | |
| else 0.0 | |
| ) | |
| return { | |
| "classification": "PLAYGROUND_PAN", | |
| "scientific_evidence": False, | |
| "pan_contract_id": PAN_CONTRACT_ID, | |
| "pan_contract_status": PAN_CONTRACT_STATUS, | |
| "evidence_eligible": False, | |
| "dimensions": self.dimensions, | |
| "axes": axis_schema(self.dimensions), | |
| "population_vector": list(self.population_vector), | |
| "population_bundle": bundle(vectors), | |
| "closed_loop": self.closed_loop, | |
| "feedback_gain": self.feedback_gain, | |
| "feedback_delay": self.feedback_delay, | |
| "feedback_source": self.feedback_source, | |
| "feedback_target": self.feedback_target, | |
| "feedback_nonlinearity": self.feedback_nonlinearity, | |
| "feedback_threshold": self.feedback_threshold, | |
| "feedback_saturation": self.feedback_saturation, | |
| "mean_abs_feedback_current": feedback_mean, | |
| "alive_neurons": alive, | |
| "apoptotic_neurons": self.n_neurons - alive, | |
| "apoptosis_events": list(self.apoptosis_events), | |
| "mean_health": sum(health) / len(health) if health else 0.0, | |
| "min_health": min(health, default=0.0), | |
| "mean_energy": sum(energy) / len(energy) if energy else 0.0, | |
| "information_axis": "local_surprise_proxy_not_PID", | |
| "pid_status": "NOT_IMPLEMENTED", | |
| "world_model_status": ( | |
| "exploratory_state_space_only_not_validated_world_model" | |
| ), | |
| "homeostasis": { | |
| "ultrafast_target_ms": 5.0, | |
| "fast_target_ms": 2000.0, | |
| "medium_target_ms": 300000.0, | |
| "slow_target_ms": 3600000.0, | |
| "note": ( | |
| "Continuous bounded approximation; no claim that these " | |
| "timescales reproduce biological or canonical MHRN dynamics." | |
| ), | |
| }, | |
| "state_sample": sample, | |
| "research_candidates": pan_research_candidates(), | |
| "promotion_path": ( | |
| "new hypothesis -> preregistration -> freeze -> new canonical " | |
| "DATA run -> Human Review -> optional EVID" | |
| ), | |
| } | |