"""Modality-Specific Synaptic Pathway Architecture (MSBA). MSBA is an embodiment-only, fail-closed contract below Neural Symbiosis. It models modality-specific gateway geometry, candidate plasticity mathematics, resource/energy accounting and adaptive allocation without importing or mutating the Brain-5D neural core. All learning and structural growth flags default to disabled. The functions in this module calculate candidate values only; applying them to a live gateway requires an explicit preregistered experiment. """ from __future__ import annotations import hashlib import json import math from dataclasses import dataclass from enum import StrEnum from .models import JSONValue MIN_PROJECTION_DIMENSIONS = 1 MAX_PROJECTION_DIMENSIONS = 32 class Modality(StrEnum): AUDIO = "audio" VISION = "vision" DIGITAL = "digital" class EnergyState(StrEnum): NORMAL = "normal" CONSERVE = "conserve" CRITICAL = "critical" SURVIVAL = "survival" def validate_projection_dimensions(value: int) -> int: """Validate an MSBA projection-space dimensionality without touching core IDs.""" if isinstance(value, bool): raise ValueError("projection_dimensions must be an integer") if not MIN_PROJECTION_DIMENSIONS <= value <= MAX_PROJECTION_DIMENSIONS: raise ValueError( f"projection_dimensions must be between {MIN_PROJECTION_DIMENSIONS} " f"and {MAX_PROJECTION_DIMENSIONS}" ) return value @dataclass(frozen=True, slots=True) class MSBAGatewayConfig: """Fail-closed configuration for modality-specific gateway experiments. ``projection_dimensions`` describes the external/adaptor projection space. It may be increased above five without changing the persisted 5D neuron-ID format. A future versioned core-format migration is required before the productive SNN itself can store more than five spatial coordinates. """ audio_bands: int = 32 audio_max_delay_ms: int = 20 vision_width: int = 100 vision_height: int = 100 vision_targets_per_input: int = 8 digital_population_size: int = 8 projection_dimensions: int = 5 allocation_beta: float = 5.0 base_energy_price: float = 0.1 pressure_energy_weight: float = 0.4 pressure_thermal_weight: float = 0.25 pressure_compute_weight: float = 0.2 pressure_memory_weight: float = 0.15 synaptic_plasticity_enabled: bool = False structural_plasticity_enabled: bool = False meta_gating_enabled: bool = False adaptive_allocation_enabled: bool = False def __post_init__(self) -> None: validate_projection_dimensions(self.projection_dimensions) def to_json(self) -> dict[str, JSONValue]: return { "audio_bands": self.audio_bands, "audio_max_delay_ms": self.audio_max_delay_ms, "vision_width": self.vision_width, "vision_height": self.vision_height, "vision_targets_per_input": self.vision_targets_per_input, "digital_population_size": self.digital_population_size, "projection_dimensions": self.projection_dimensions, "projection_dimensions_min": MIN_PROJECTION_DIMENSIONS, "projection_dimensions_max": MAX_PROJECTION_DIMENSIONS, "allocation_beta": self.allocation_beta, "base_energy_price": self.base_energy_price, "synaptic_plasticity_enabled": self.synaptic_plasticity_enabled, "structural_plasticity_enabled": self.structural_plasticity_enabled, "meta_gating_enabled": self.meta_gating_enabled, "adaptive_allocation_enabled": self.adaptive_allocation_enabled, "requires_preregistration": True, "requires_experiment_mode": True, "core_learning_rules_unchanged": True, } @dataclass(frozen=True, slots=True) class EnergyCoefficients: """Normalized energy-unit coefficients; not physical joule calibration.""" spike: float = 1.0 synaptic_event: float = 1.0 plasticity_update: float = 3.0 structural_event: float = 20.0 memory_byte: float = 0.0001 io_byte: float = 0.0002 adapter_base: float = 1.0 def to_json(self) -> dict[str, JSONValue]: """Return the normalized accounting coefficients as provenance data.""" return { "spike": self.spike, "synaptic_event": self.synaptic_event, "plasticity_update": self.plasticity_update, "structural_event": self.structural_event, "memory_byte": self.memory_byte, "io_byte": self.io_byte, "adapter_base": self.adapter_base, "units_are_not_physical_joules": True, } @dataclass(frozen=True, slots=True) class EnergyObservation: sensor_units: float = 0.0 encoder_units: float = 0.0 spikes: int = 0 synaptic_events: int = 0 plasticity_updates: int = 0 structural_events: int = 0 memory_bytes: int = 0 io_bytes: int = 0 adapter_units: float = 0.0 measured_joules: float | None = None @dataclass(frozen=True, slots=True) class EnergyEstimate: normalized_energy_units: float estimated_joules: float | None measured_joules: float | None component_units: dict[str, float] | None = None def to_json(self) -> dict[str, JSONValue]: return { "normalized_energy_units": self.normalized_energy_units, "estimated_joules": self.estimated_joules, "measured_joules": self.measured_joules, "component_units": self.component_units or {}, "provenance": { "normalized_energy_units": "NORMALIZED_MODEL_ESTIMATE", "estimated_joules": ( "CALIBRATED_CONVERSION" if self.estimated_joules is not None else "NOT_AVAILABLE" ), "measured_joules": ( "DIRECT_TELEMETRY" if self.measured_joules is not None else "NOT_AVAILABLE" ), }, "estimated_is_not_measured": True, } @dataclass(frozen=True, slots=True) class ResourcePressure: energy: float = 0.0 thermal: float = 0.0 compute: float = 0.0 memory: float = 0.0 fan_failure: bool = False thermal_safety_trip: bool = False persistence_failure: bool = False @dataclass(frozen=True, slots=True) class SymbolFrame: """Exact digital payload kept outside the lossy SNN representation.""" payload: bytes codec: str = "raw" sequence: int = 0 provenance: str = "" @property def checksum(self) -> str: return hashlib.sha256(self.payload).hexdigest() def deterministic_population(self, population_size: int = 8) -> tuple[int, ...]: """Map payload bytes to a deterministic, fixed-width population code.""" if population_size <= 0: raise ValueError("population_size must be positive") digest = hashlib.sha256(self.payload).digest() return tuple(digest[index] % 2 for index in range(population_size)) def to_json(self) -> dict[str, JSONValue]: """Persist digital provenance without replacing the exact payload.""" return { "codec": self.codec, "sequence": self.sequence, "provenance": self.provenance, "payload_size_bytes": len(self.payload), "checksum_algorithm": "sha256", "checksum": self.checksum, } @dataclass(frozen=True, slots=True) class EmbodimentTreatmentProvenance: """Immutable DATA record for an explicit adapter/gateway treatment.""" adapter_id: str adapter_architecture: str projection_mode: str = "structured" projection_dimensions: int = 5 gateway: MSBAGatewayConfig = MSBAGatewayConfig() energy: EnergyCoefficients = EnergyCoefficients() def __post_init__(self) -> None: if not self.adapter_id.strip(): raise ValueError("adapter_id must not be empty") if not self.adapter_architecture.strip(): raise ValueError("adapter_architecture must not be empty") if not self.projection_mode.strip(): raise ValueError("projection_mode must not be empty") validate_projection_dimensions(self.projection_dimensions) def to_json(self) -> dict[str, JSONValue]: """Return a deterministic, non-EVID treatment provenance record.""" record: dict[str, JSONValue] = { "schema_version": 1, "record_type": "embodiment_treatment_provenance", "status": "DATA_ONLY", "adapter": { "id": self.adapter_id, "architecture": self.adapter_architecture, }, "projection": { "mode": self.projection_mode, "dimensions": self.projection_dimensions, "productive_core_dimensions": 5, }, "gateway": self.gateway.to_json(), "energy": self.energy.to_json(), "scientific_boundary": { "requires_preregistration": True, "requires_human_review": True, "does_not_promote_to_evid": True, }, } canonical = json.dumps(record, sort_keys=True, separators=(",", ":")) record["record_sha256"] = hashlib.sha256(canonical.encode("utf-8")).hexdigest() return record @dataclass(frozen=True, slots=True) class ModalityProfile: modality: Modality pathway: str coordinate_features: tuple[str, ...] plasticity: str throttles: tuple[str, ...] exact_payload_outside_snn: bool = False def to_json(self) -> dict[str, JSONValue]: return { "modality": self.modality.value, "pathway": self.pathway, "coordinate_features": list(self.coordinate_features), "plasticity": self.plasticity, "throttles": list(self.throttles), "exact_payload_outside_snn": self.exact_payload_outside_snn, } def default_modality_profiles() -> tuple[ModalityProfile, ...]: """Return MSBA profiles without assigning semantics to fixed axes.""" return ( ModalityProfile( Modality.AUDIO, "temporal_coherence", ("band", "phase_or_envelope", "channel", "lag", "feature"), "phase_weighted_t_stdp_candidate", ("band_count", "update_rate", "delay_taps", "encoder_complexity"), ), ModalityProfile( Modality.VISION, "spatial_multiplex", ("x", "y", "feature", "scale", "frame"), "s_stdp_plus_structural_growth_candidate", ("fps", "resolution", "roi", "feature_channels", "plasticity"), ), ModalityProfile( Modality.DIGITAL, "quantized_high_fidelity", ("symbol", "sequence", "source", "context", "route"), "population_meta_gating_candidate", ("symbol_rate", "batching", "admission", "gateway_duty_cycle"), exact_payload_outside_snn=True, ), ) def energy_units( observation: EnergyObservation, coefficients: EnergyCoefficients = EnergyCoefficients(), *, joules_per_unit: float | None = None, ) -> EnergyEstimate: """Estimate resource cost while keeping measured and estimated joules distinct.""" values = ( observation.sensor_units, observation.encoder_units, observation.spikes, observation.synaptic_events, observation.plasticity_updates, observation.structural_events, observation.memory_bytes, observation.io_bytes, ) if any(value < 0 for value in values): raise ValueError("energy counters must be non-negative") units = ( observation.sensor_units + observation.encoder_units + coefficients.adapter_base + observation.adapter_units + coefficients.spike * observation.spikes + coefficients.synaptic_event * observation.synaptic_events + coefficients.plasticity_update * observation.plasticity_updates + coefficients.structural_event * observation.structural_events + coefficients.memory_byte * observation.memory_bytes + coefficients.io_byte * observation.io_bytes ) components = { "sensor": float(observation.sensor_units), "encoder": float(observation.encoder_units), "spikes": float(coefficients.spike * observation.spikes), "synaptic_events": float( coefficients.synaptic_event * observation.synaptic_events ), "plasticity": float( coefficients.plasticity_update * observation.plasticity_updates ), "structural": float( coefficients.structural_event * observation.structural_events ), "memory": float(coefficients.memory_byte * observation.memory_bytes), "io": float(coefficients.io_byte * observation.io_bytes), "adapter": float(coefficients.adapter_base + observation.adapter_units), } estimated = None if joules_per_unit is None else units * joules_per_unit return EnergyEstimate( float(units), estimated, observation.measured_joules, components ) def resource_pressure( pressure: ResourcePressure, config: MSBAGatewayConfig = MSBAGatewayConfig() ) -> float: """Return clipped soft resource pressure; fan failure is a hard safety override.""" if ( pressure.fan_failure or pressure.thermal_safety_trip or pressure.persistence_failure ): return 1.0 terms = ( config.pressure_energy_weight * _unit(pressure.energy) + config.pressure_thermal_weight * _unit(pressure.thermal) + config.pressure_compute_weight * _unit(pressure.compute) + config.pressure_memory_weight * _unit(pressure.memory) ) return _unit(terms) def energy_state(pressure: ResourcePressure) -> EnergyState: """Map resource pressure to a deterministic protection state.""" if ( pressure.fan_failure or pressure.thermal_safety_trip or pressure.persistence_failure ): return EnergyState.SURVIVAL level = resource_pressure(pressure) if level >= 0.85: return EnergyState.SURVIVAL if level >= 0.65: return EnergyState.CRITICAL if level >= 0.35: return EnergyState.CONSERVE return EnergyState.NORMAL def energy_price( pressure: ResourcePressure, config: MSBAGatewayConfig = MSBAGatewayConfig() ) -> float: """Return the dynamic internal price used by soft allocation experiments.""" return config.base_energy_price + resource_pressure(pressure) def allocation_gate( utility: float, cost: float, pressure: ResourcePressure, threshold: float = 0.0, config: MSBAGatewayConfig = MSBAGatewayConfig(), ) -> float: """Return a candidate gateway quota in [0, 1] without activating a gateway.""" if cost < 0.0: raise ValueError("cost must be non-negative") if ( pressure.fan_failure or pressure.thermal_safety_trip or pressure.persistence_failure ): return 0.0 drive = utility - energy_price(pressure, config) * cost - threshold return _sigmoid(config.allocation_beta * drive) def phase_coherence(delta_phase_rad: float) -> float: """Non-negative phase term preserving the sign of the underlying STDP rule.""" return 0.5 * (1.0 + math.cos(delta_phase_rad)) def phase_weighted_stdp(stdp_delta: float, delta_phase_rad: float) -> float: """Scale candidate temporal STDP by phase coherence without sign inversion.""" return stdp_delta * phase_coherence(delta_phase_rad) def visual_growth_probability( information_value: float, distance_5d: float, sigma: float, pressure: ResourcePressure, eta_growth: float = 0.001, ) -> float: """Candidate visual sprouting probability using utility, locality and energy.""" if sigma <= 0.0: raise ValueError("sigma must be positive") if distance_5d < 0.0: raise ValueError("distance_5d must be non-negative") locality = math.exp(-(distance_5d**2) / (2.0 * sigma**2)) availability = 1.0 - resource_pressure(pressure) return _unit(eta_growth * max(0.0, information_value) * locality * availability) def recommended_protection_policy(state: EnergyState) -> dict[str, JSONValue]: """Return deterministic throttling guidance; it never mutates runtime state.""" policies: dict[EnergyState, dict[str, JSONValue]] = { EnergyState.NORMAL: { "plasticity": "experiment_policy", "vision": "full_budget", "audio": "full_budget", "digital": "full_budget", }, EnergyState.CONSERVE: { "plasticity": "reduce_first", "vision": "reduce_fps_then_resolution", "audio": "reduce_bands_or_update_rate", "digital": "reduce_admission_rate", }, EnergyState.CRITICAL: { "plasticity": "freeze", "vision": "roi_low_resolution", "audio": "reduced_bands", "digital": "priority_symbols_only", }, EnergyState.SURVIVAL: { "plasticity": "freeze", "vision": "off_unless_safety_critical", "audio": "minimal_unless_safety_critical", "digital": "safety_and_persistence_only", "preserve": [ "thermal_sensing", "fan_monitoring", "core_persistence", "storage_integrity", "emergency_actuator_path", ], }, } return policies[state] def msba_contract() -> dict[str, JSONValue]: """Publish the complete read-only MSBA architecture contract.""" config = MSBAGatewayConfig() return { "name": "Modalitaets-spezifische Synaptische Bahn-Architektur", "acronym": "MSBA", "scope": "neural_symbiosis.embodiment", "status": "experimental_contract", "profiles": [profile.to_json() for profile in default_modality_profiles()], "gateway": config.to_json(), "energy_accounting": { "components": [ "sensor", "encoder", "spikes", "synaptic_events", "plasticity", "structural_growth", "memory", "io", "adapter", ], "reported_fields": [ "normalized_energy_units", "estimated_joules", "measured_joules", ], "provenance_classes": { "normalized_energy_units": "NORMALIZED_MODEL_ESTIMATE", "estimated_joules": "CALIBRATED_CONVERSION", "measured_joules": "DIRECT_TELEMETRY", "missing": "NOT_AVAILABLE", }, "estimated_is_not_measured": True, }, "topology": { "fixed_axis_semantics": False, "projection_required": True, "projection_dimensions": config.projection_dimensions, "projection_dimensions_range": [ MIN_PROJECTION_DIMENSIONS, MAX_PROJECTION_DIMENSIONS, ], "productive_core_dimensions": 5, "core_dimension_migration_required_above_five": True, "controls": [ "structured_projection", "shuffled_projection", "random_projection", "reduced_dimension_projection", "increased_dimension_projection", ], }, "scientific_boundary": { "core_imports": False, "core_mutation": False, "learning_disabled_by_default": True, "structural_growth_disabled_by_default": True, "digital_payload_outside_snn": True, "digital_checksum_persisted": True, "reachability_is_not_learned_use": True, "historical_data_unchanged": True, "activation_requires_preregistered_experiment": True, }, } def _unit(value: float) -> float: return min(1.0, max(0.0, float(value))) def _sigmoid(value: float) -> float: if value >= 0.0: inverse = math.exp(-value) return 1.0 / (1.0 + inverse) exponent = math.exp(value) return exponent / (1.0 + exponent) __all__ = [ "EnergyCoefficients", "EnergyEstimate", "EnergyObservation", "EnergyState", "MAX_PROJECTION_DIMENSIONS", "MIN_PROJECTION_DIMENSIONS", "MSBAGatewayConfig", "Modality", "ModalityProfile", "ResourcePressure", "SymbolFrame", "allocation_gate", "default_modality_profiles", "energy_price", "energy_state", "energy_units", "msba_contract", "phase_coherence", "phase_weighted_stdp", "recommended_protection_policy", "resource_pressure", "validate_projection_dimensions", "visual_growth_probability", ]