"""Typed bridge used by HTTP handlers to control Brain-5D safely. This module provides the OperatorBridge class, which translates dashboard HTTP requests into typed controller operations with proper error handling and consistent response formatting. """ from __future__ import annotations from dataclasses import replace from typing import Any, TypeAlias, cast from src.controller.runtime import ControllerCommand, RuntimeController from src.dashboard.live_projection import ( LiveProjectionService, NetworkAccess, TelemetryFrameStore, ) from src.self_organization.approval import ( ProposalApprovalPolicy, StructuralPlasticityConfig, ) from src.self_organization.coordinator import SelfOrganizationCoordinator from src.self_organization.plasticity import StructuralPlasticityEngine from src.self_organization.runtime_adapter import get_error_buffer from src.visualization.structural_heatmap import ( StructuralHeatmapKind, StructuralHeatmapSource, ) from .models import ( JSONValue, StructuralMetrics, SystemMetrics, to_json_serializable, ) from .structural_api import StructuralCommandResult # Type aliases for JSON responses JSONMapping: TypeAlias = dict[str, JSONValue] # Allowed heatmap kinds HEATMAP_KINDS = { "neuron_additions", "neuron_removals", "synapse_additions", "synapse_removals", "total_structural_activity", } class OperatorBridge: """Translate dashboard requests into typed controller operations. This bridge provides a clean, type-safe interface between the HTTP dashboard endpoints and the Brain-5D runtime controller. It handles structural plasticity coordination, heatmap generation, and all controller commands. All methods that return JSONMapping are guaranteed to produce JSON-serializable responses with consistent error handling. """ def __init__( self, controller: RuntimeController, coordinator: SelfOrganizationCoordinator | None = None, plasticity: StructuralPlasticityEngine | None = None, approval_policy: ProposalApprovalPolicy | None = None, structural_heatmaps: StructuralHeatmapSource | None = None, live_projection: LiveProjectionService | None = None, telemetry_store: TelemetryFrameStore | None = None, ) -> None: """Initialize the operator bridge. Args: controller: The runtime controller instance (required). coordinator: Optional self-organization coordinator. plasticity: Optional structural plasticity engine. approval_policy: Optional proposal approval policy. structural_heatmaps: Optional heatmap source for visualizations. live_projection: Optional live projection service for real-time visualization. Created automatically from the controller's network if not provided. telemetry_store: Optional TelemetryFrameStore for post-tick frame capture. If provided, the live projection service reads from this store instead of capturing directly. """ # controller is required by type annotation, so no None check needed self.controller = controller self.coordinator = coordinator self.plasticity = plasticity self.approval_policy = approval_policy self.structural_heatmaps = structural_heatmaps self.telemetry_store = telemetry_store self.live_projection = live_projection or LiveProjectionService( cast("NetworkAccess", controller.network), frame_store=telemetry_store, ) # ========================================================================= # Status and Telemetry # ========================================================================= def status(self) -> JSONMapping: """Get the current system status as a JSON-serializable dictionary. Returns: Complete system status including telemetry, self-organization reports, and structural history. """ telemetry = self.controller.telemetry payload: JSONMapping = { "state": telemetry.controller_state.value, "tick": telemetry.tick, "ticks_per_second": telemetry.ticks_per_second, "batch_duration_ms": telemetry.batch_duration_ms, "spikes_this_batch": telemetry.spikes_this_batch, "neurons": telemetry.neurons, "synapses": telemetry.synapses, "queue_depth": telemetry.queue_depth, "requested_ticks": telemetry.requested_ticks, "completed_ticks": telemetry.completed_ticks, "last_error": telemetry.last_error, "target_hz": telemetry.target_hz, "simulation_speed_ratio": telemetry.simulation_speed_ratio, "tick_latency_ms": telemetry.tick_latency_ms, "jitter_ms": telemetry.jitter_ms, "compute_saturation": telemetry.compute_saturation, "runtime_mode": telemetry.runtime_mode, "tick_profile": cast(JSONValue, telemetry.tick_profile), "max_possible_hz": telemetry.max_possible_hz, } # Add self-organization report if available if self.coordinator is not None: report = self.coordinator.latest() if report is not None: payload["self_organization"] = { "tick": report.tick, "neurogenesis_pressure": report.neurogenesis_pressure, "pruning_pressure": report.pruning_pressure, "synapse_sprouting_pressure": report.synapse_sprouting_pressure, "synapse_pruning_pressure": report.synapse_pruning_pressure, "proposals": [ { "proposal_id": p.proposal_id, "kind": p.kind.value, "neuron_id": p.neuron_id, "target_id": p.target_id, "reason": p.reason, "confidence": p.confidence, } for p in report.proposals ], } # Add structural history if available - cast to JSONValue to satisfy type checker if self.plasticity is not None: payload["structural_history"] = cast( list[JSONValue], self.structural_history(100) ) return payload def structural_status(self) -> JSONMapping: """Get structural plasticity subsystem status. Returns: Dictionary with configuration and proposal statistics. """ report = self.coordinator.latest() if self.coordinator is not None else None return { "configured": self.coordinator is not None and self.plasticity is not None, "proposal_tick": report.tick if report is not None else None, "proposal_count": len(report.proposals) if report is not None else 0, "decision_count": ( len(self.coordinator.decisions()) if self.coordinator is not None else 0 ), "history_count": ( len(self.plasticity.history()) if self.plasticity is not None else 0 ), } def structural_config(self) -> JSONMapping: """Get the current structural plasticity configuration. Returns: Dictionary with all configuration parameters, or a minimal response if not configured. """ policy = self.approval_policy if policy is None: return {"configured": False} config = policy.config return { "configured": True, "enabled": config.enabled, "dry_run": config.dry_run, "auto_approval": config.auto_approval, "auto_approval_threshold": config.auto_approval_threshold, "max_changes_per_tick": config.max_changes_per_tick, "max_neuron_additions_per_tick": config.max_neuron_additions_per_tick, "max_neuron_removals_per_tick": config.max_neuron_removals_per_tick, "max_synapse_additions_per_tick": config.max_synapse_additions_per_tick, "max_synapse_removals_per_tick": config.max_synapse_removals_per_tick, "min_neurons": config.min_neurons, "max_neurons": config.max_neurons, "allow_neuron_pruning": config.allow_neuron_pruning, "allow_synapse_pruning": config.allow_synapse_pruning, "cooldown_ticks": config.cooldown_ticks, } # ========================================================================= # Self-Organization Proposals # ========================================================================= def structural_proposals(self) -> list[JSONMapping]: """Get all pending structural plasticity proposals. Returns: List of proposal dictionaries, each with ID, kind, and metadata. """ report = self.coordinator.latest() if self.coordinator is not None else None if report is None: return [] return [ { "proposal_id": proposal.proposal_id, "kind": proposal.kind.value, "neuron_id": proposal.neuron_id, "target_id": proposal.target_id, "reason": proposal.reason, "confidence": proposal.confidence, } for proposal in report.proposals ] def structural_history(self, limit: int = 100) -> list[JSONMapping]: """Get structural plasticity history records. Args: limit: Maximum number of records to return. Returns: List of history record dictionaries. """ if self.plasticity is None: return [] return [ { "sequence": record.sequence, "tick": record.tick, "kind": record.kind.value, "proposal_id": record.proposal_id, "neuron_id": record.neuron_id, "source_id": record.source_id, "target_id": record.target_id, "reason": record.reason, "approved_by": record.approved_by, "automatic": record.automatic, "undo_of_sequence": record.undo_of_sequence, } for record in self.plasticity.history(limit) ] # ========================================================================= # Heatmap Generation # ========================================================================= def structural_heatmap(self, kind: str) -> JSONMapping: """Generate a structural plasticity heatmap. Args: kind: Heatmap type (neuron_additions, neuron_removals, synapse_additions, synapse_removals, total_structural_activity). Returns: Heatmap data including values, range, and metadata. Raises: RuntimeError: If heatmap source or plasticity engine is not configured. ValueError: If the heatmap kind is unknown. """ source = self.structural_heatmaps if source is None: raise RuntimeError("structural heatmap source is not configured") if self.plasticity is None: raise RuntimeError("structural plasticity engine is not configured") if kind not in HEATMAP_KINDS: raise ValueError(f"unknown structural heatmap kind: {kind}") result = source.build( self.plasticity.history(), cast(StructuralHeatmapKind, kind), ) return { "kind": result.kind, "values": result.values.tolist(), "min_value": result.min_value, "max_value": result.max_value, "populated_cells": result.populated_cells, "sequence_from": result.sequence_from, "sequence_to": result.sequence_to, } # ========================================================================= # Controller Commands # ========================================================================= def command(self, command: str, *, ticks: int | None = None) -> JSONMapping: """Execute a controller command. Args: command: Command name (start, pause, resume, stop, step, run_ticks, snapshot). ticks: Number of ticks to run (for run_ticks command). Returns: JSON response with success status and current status. """ try: cmd = ControllerCommand(command) except ValueError: return {"ok": False, "error": f"unknown command: {command}"} try: if cmd == ControllerCommand.START: self.controller.start() elif cmd == ControllerCommand.PAUSE: self.controller.pause() elif cmd == ControllerCommand.RESUME: self.controller.resume() elif cmd == ControllerCommand.STOP: self.controller.stop() elif cmd == ControllerCommand.STEP: self.controller.step_once() elif cmd == ControllerCommand.RUN_TICKS: self.controller.run_ticks(ticks or 1) elif cmd == ControllerCommand.SNAPSHOT: self.controller.request_snapshot() except Exception as e: return {"ok": False, "error": str(e)} return {"ok": True, "status": self.status()} # ========================================================================= # Proposal Management # ========================================================================= def apply_proposal(self, proposal_id: str, *, approved: bool) -> JSONMapping: """Apply or reject a structural proposal. Args: proposal_id: ID of the proposal to process. approved: Whether to approve (True) or reject (False). Returns: JSON response with success status and change details. """ if self.coordinator is None: return {"ok": False, "error": "structural coordinator not configured"} if self.plasticity is None: return {"ok": False, "error": "structural plasticity not configured"} proposal = self.coordinator.find(proposal_id) if proposal is None: return {"ok": False, "error": f"proposal not found: {proposal_id}"} # Record the decision decision_recorded = ( self.coordinator.approve(proposal_id) if approved else self.coordinator.reject(proposal_id) ) if not decision_recorded: return { "ok": False, "error": f"failed to record decision for: {proposal_id}", } # Apply the change if approved if approved: change = self.plasticity.apply_proposal( self.controller.telemetry.tick, proposal, approved=True, ) if change is None: return {"ok": False, "error": "failed to apply approved proposal"} return { "ok": True, "change": { "tick": change.tick, "proposal_id": change.proposal_id, "kind": change.kind.value, "description": change.description, }, } return {"ok": True, "change": None} def approve_structural(self, proposal_id: str) -> StructuralCommandResult: """Approve a structural plasticity proposal. Args: proposal_id: ID of the proposal to approve. Returns: StructuralCommandResult with success status and message. """ if self.coordinator is None: return StructuralCommandResult( False, "structural coordinator not configured" ) if self.plasticity is None: return StructuralCommandResult( False, "structural plasticity not configured" ) proposal = self.coordinator.find(proposal_id) if proposal is None: return StructuralCommandResult(False, f"proposal not found: {proposal_id}") if not self.coordinator.approve(proposal_id): return StructuralCommandResult(False, "failed to approve proposal") change = self.plasticity.apply_proposal( self.controller.telemetry.tick, proposal, approved=True, ) if change is None: return StructuralCommandResult(False, "failed to apply approved proposal") return StructuralCommandResult( True, f"proposal {proposal_id} approved and applied" ) def reject_structural(self, proposal_id: str) -> StructuralCommandResult: """Reject a structural plasticity proposal. Args: proposal_id: ID of the proposal to reject. Returns: StructuralCommandResult with success status and message. """ coordinator = self.coordinator if coordinator is None: return StructuralCommandResult( False, "structural coordinator not configured" ) if not coordinator.reject(proposal_id): return StructuralCommandResult(False, f"proposal not found: {proposal_id}") return StructuralCommandResult(True, f"proposal {proposal_id} rejected") def undo_structural(self) -> StructuralCommandResult: """Undo the last structural plasticity change. Returns: StructuralCommandResult with success status and message. """ if self.plasticity is None: return StructuralCommandResult( False, "structural plasticity not configured" ) undone = self.plasticity.undo_last_change(tick=self.controller.telemetry.tick) if undone: return StructuralCommandResult(True, "last change undone") return StructuralCommandResult(False, "no changes to undo") # ========================================================================= # Configuration Management # ========================================================================= def set_auto_approval(self, enabled: bool) -> StructuralCommandResult: """Enable or disable automatic approval of proposals. Args: enabled: Whether to enable auto-approval. Returns: StructuralCommandResult with success status and message. """ if self.approval_policy is None: # Create a new policy with default config self.approval_policy = ProposalApprovalPolicy( StructuralPlasticityConfig(auto_approval=enabled) ) else: # Update existing policy self.approval_policy = ProposalApprovalPolicy( replace(self.approval_policy.config, auto_approval=enabled) ) return StructuralCommandResult( True, f"auto-approval {'enabled' if enabled else 'disabled'}" ) def update_structural_config(self, **kwargs: Any) -> StructuralCommandResult: """Update structural plasticity configuration parameters. Args: **kwargs: Configuration parameters to update. Returns: StructuralCommandResult with success status and message. """ if self.approval_policy is None: return StructuralCommandResult( False, "structural plasticity not configured" ) # Get current config and update with new values current = self.approval_policy.config config_dict = { k: v for k, v in current.__dict__.items() if not k.startswith("_") } for key, value in kwargs.items(): if key in config_dict: config_dict[key] = value else: return StructuralCommandResult( False, f"unknown config parameter: {key}" ) # Create new config and policy new_config = StructuralPlasticityConfig(**config_dict) self.approval_policy = ProposalApprovalPolicy(new_config) return StructuralCommandResult(True, "configuration updated") # ========================================================================= # Convenience Methods # ========================================================================= def run_ticks(self, count: int) -> StructuralCommandResult: """Run a specified number of ticks. Args: count: Number of ticks to execute. Returns: StructuralCommandResult with success status and message. """ if count <= 0: return StructuralCommandResult(False, "tick count must be positive") try: self.controller.run_ticks(count) return StructuralCommandResult(True, f"executed {count} ticks") except Exception as e: return StructuralCommandResult(False, f"failed to run ticks: {e}") def single_step(self) -> StructuralCommandResult: """Execute a single tick. Returns: StructuralCommandResult with success status and message. """ try: self.controller.step_once() return StructuralCommandResult(True, "executed one tick") except Exception as e: return StructuralCommandResult(False, f"failed to execute tick: {e}") def request_snapshot(self) -> StructuralCommandResult: """Request a snapshot of the current system state. Returns: StructuralCommandResult with success status and message. """ try: self.controller.request_snapshot() return StructuralCommandResult(True, "snapshot requested") except Exception as e: return StructuralCommandResult(False, f"failed to request snapshot: {e}") # ========================================================================= # Metrics Extraction # ========================================================================= def get_system_metrics(self) -> SystemMetrics: """Extract system metrics from the current telemetry. Returns: SystemMetrics object with current values. """ telemetry = self.controller.telemetry return SystemMetrics( tick=telemetry.tick, neurons=telemetry.neurons, synapses=telemetry.synapses, spikes_total=0, # Not available in telemetry; use spikes_this_batch instead spikes_last_tick=telemetry.spikes_this_batch, core_step_ms=telemetry.batch_duration_ms, mean_energy=0.0, # Not available in telemetry; placeholder ) def get_structural_metrics(self) -> StructuralMetrics: """Extract structural metrics from the current state. Returns: StructuralMetrics object with current values. """ if self.coordinator is None or self.plasticity is None: return StructuralMetrics() report = self.coordinator.latest() if report is None: return StructuralMetrics() return StructuralMetrics( neuron_count=self.controller.telemetry.neurons, synapse_count=self.controller.telemetry.synapses, # Convert float pressures to int for the model new_neurons=int(report.neurogenesis_pressure), pruned_neurons=int(report.pruning_pressure), new_synapses=int(report.synapse_sprouting_pressure), pruned_synapses=int(report.synapse_pruning_pressure), growth_budget=1.0, # Placeholder - actual value from config used_budget=0.0, # Placeholder - actual value from engine structural_changes=len(report.proposals), ) def runtime_errors(self) -> list[dict[str, JSONValue]]: """Return structured runtime error events for dashboard visibility. Returns: List of error event dictionaries with timestamp, tick, component, phase, exception_type, message, fatal, and traceback_hash. """ buffer = get_error_buffer() return [ { "timestamp": e.timestamp, "tick": e.tick, "component": e.component, "phase": e.phase, "exception_type": e.exception_type, "message": e.message, "fatal": e.fatal, "traceback_hash": e.traceback_hash, } for e in buffer.events ] def to_json(self, obj: Any) -> JSONValue: """Convert any object to a JSON-serializable value. Delegates to the global to_json_serializable function. """ return to_json_serializable(obj)