Spaces:
Paused
Paused
| """Runtime adapter that connects HomeostasisSignal to the structural coordinator. | |
| This adapter is the bridge between live runtime measurements and the | |
| canonical structural mutation path. It runs as a post-tick hook and: | |
| 1. Captures the real HomeostasisSignal from HomeostasisEngine | |
| 2. Feeds it through SelfOrganizationPolicy.analyze() | |
| 3. Publishes the resulting PolicyReport to the SelfOrganizationCoordinator | |
| It does NOT mutate the network. Mutation remains exclusively in: | |
| Proposal -> Approval -> StructuralPlasticityEngine -> Manipulator | |
| This is the missing link that closes Gate A for the production path. | |
| Config-authoritative: | |
| The adapter reads its interval from config and passes the policy config | |
| from the YAML ``self_organization`` section. Hardcoded defaults are | |
| never used when production config exists. | |
| Error visibility: | |
| The adapter captures structured RuntimeErrorEvent on failure. The error | |
| is observable through the error buffer and dashboard — never silently | |
| swallowed. | |
| """ | |
| from __future__ import annotations | |
| import traceback | |
| from collections.abc import Sequence | |
| from dataclasses import dataclass | |
| from typing import Any | |
| from src.homeostasis.engine import HomeostasisEngine | |
| from src.self_organization.coordinator import SelfOrganizationCoordinator | |
| from src.self_organization.policy import ( | |
| SelfOrganizationPolicy, | |
| SelfOrganizationPolicyConfig, | |
| ) | |
| # ============================================================================ | |
| # Structured error event for scientific integrity | |
| # ============================================================================ | |
| class RuntimeErrorEvent: | |
| """Canonical structured error event for the runtime adapter. | |
| Attributes: | |
| timestamp: Monotonic time (time.monotonic_ns()) when the error occurred. | |
| tick: Simulation tick when the error occurred. | |
| component: Component name (e.g. "SelfOrganizationRuntimeAdapter"). | |
| phase: Phase within the component (e.g. "build_signal", "analyze", "publish"). | |
| exception_type: Fully qualified exception type name. | |
| message: Human-readable error message. | |
| fatal: If True, the simulation should stop. If False, the error is | |
| isolated and the simulation can continue. | |
| traceback_hash: SHA-256 of the traceback text, for deduplication. | |
| """ | |
| timestamp: int | |
| tick: int | |
| component: str | |
| phase: str | |
| exception_type: str | |
| message: str | |
| fatal: bool = False | |
| traceback_hash: str = "" | |
| # ============================================================================ | |
| # Bounded error buffer | |
| # ============================================================================ | |
| class ErrorBuffer: | |
| """Thread-safe bounded buffer for RuntimeErrorEvents. | |
| The buffer holds the most recent N events and exposes them for | |
| dashboard/API visibility. All public methods are protected by | |
| ``threading.RLock`` for genuine thread safety. | |
| """ | |
| def __init__(self, max_size: int = 100) -> None: | |
| import threading | |
| self._lock = threading.RLock() | |
| self._max_size = max_size | |
| self._events: list[RuntimeErrorEvent] = [] | |
| def push(self, event: RuntimeErrorEvent) -> None: | |
| with self._lock: | |
| self._events.append(event) | |
| if len(self._events) > self._max_size: | |
| self._events.pop(0) | |
| def events(self) -> Sequence[RuntimeErrorEvent]: | |
| with self._lock: | |
| return tuple(self._events) | |
| def latest(self) -> RuntimeErrorEvent | None: | |
| with self._lock: | |
| return self._events[-1] if self._events else None | |
| def count(self) -> int: | |
| with self._lock: | |
| return len(self._events) | |
| def clear(self) -> None: | |
| with self._lock: | |
| self._events.clear() | |
| # ============================================================================ | |
| # Shared global error buffer (canonical owner) | |
| # ============================================================================ | |
| # The canonical error buffer for the runtime adapter. Accessed by the | |
| # adapter, the OperatorBridge, and the dashboard integration status. | |
| _runtime_error_buffer: ErrorBuffer = ErrorBuffer() | |
| def get_error_buffer() -> ErrorBuffer: | |
| """Return the canonical shared error buffer.""" | |
| return _runtime_error_buffer | |
| # ============================================================================ | |
| # Runtime Adapter | |
| # ============================================================================ | |
| class SelfOrganizationRuntimeAdapter: | |
| """Post-tick hook that feeds real HomeostasisSignals into the structural coordinator. | |
| The adapter is attached to the RuntimeController and runs after every | |
| tick. It builds a HomeostasisSignal from the live network, passes it | |
| through the SelfOrganizationPolicy, and publishes the resulting | |
| PolicyReport to the coordinator. | |
| Config-authoritative: | |
| ``interval_ticks`` and ``policy_config`` come from the YAML | |
| ``self_organization`` section. Hardcoded defaults are only used | |
| when no config is provided (e.g. in tests). | |
| Error visibility: | |
| Exceptions are captured as structured RuntimeErrorEvent and pushed | |
| to the canonical error buffer. They are observable through the | |
| dashboard and integration status. The simulation continues unless | |
| the error is marked fatal. | |
| """ | |
| def __init__( | |
| self, | |
| homeostasis_engine: HomeostasisEngine, | |
| coordinator: SelfOrganizationCoordinator, | |
| *, | |
| interval_ticks: int = 10, | |
| policy_config: SelfOrganizationPolicyConfig | None = None, | |
| error_buffer: ErrorBuffer | None = None, | |
| ) -> None: | |
| self.homeostasis_engine = homeostasis_engine | |
| self.coordinator = coordinator | |
| self.interval_ticks = interval_ticks | |
| self.policy = SelfOrganizationPolicy( | |
| policy_config or SelfOrganizationPolicyConfig(enabled=True, dry_run=True) | |
| ) | |
| self._last_tick: int = 0 | |
| self._error_buffer = error_buffer or _runtime_error_buffer | |
| def __call__(self, tick: int, _result: Any) -> None: # noqa: ARG001 | |
| """Post-tick hook: build signal, run policy, publish to coordinator. | |
| Runs every ``interval_ticks`` ticks to avoid excessive overhead. | |
| Errors are captured as structured RuntimeErrorEvent and pushed to | |
| the error buffer. The simulation continues — a hook error must | |
| never break the simulation. | |
| """ | |
| if tick - self._last_tick < self.interval_ticks: | |
| return | |
| self._last_tick = tick | |
| # --- Phase: build_signal --- | |
| try: | |
| signal = self.homeostasis_engine.build_signal(tick=tick) | |
| except Exception as exc: | |
| self._capture_error(tick, "build_signal", exc) | |
| return | |
| # --- Phase: analyze --- | |
| try: | |
| report = self.policy.analyze(signal) | |
| except Exception as exc: | |
| self._capture_error(tick, "analyze", exc) | |
| return | |
| # --- Phase: publish --- | |
| if report.proposals: | |
| try: | |
| self.coordinator.publish(report) | |
| except Exception as exc: | |
| self._capture_error(tick, "publish", exc) | |
| return | |
| def _capture_error(self, tick: int, phase: str, exc: Exception) -> None: | |
| """Capture a structured error event.""" | |
| import hashlib | |
| import time | |
| tb_text = traceback.format_exc() | |
| tb_hash = hashlib.sha256(tb_text.encode("utf-8")).hexdigest()[:16] | |
| event = RuntimeErrorEvent( | |
| timestamp=time.monotonic_ns(), | |
| tick=tick, | |
| component="SelfOrganizationRuntimeAdapter", | |
| phase=phase, | |
| exception_type=f"{type(exc).__module__}.{type(exc).__qualname__}", | |
| message=str(exc), | |
| fatal=False, | |
| traceback_hash=tb_hash, | |
| ) | |
| self._error_buffer.push(event) | |