Spaces:
Running
Running
File size: 5,333 Bytes
76af1af | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | """Backend-neutral execution contract for canonical MHRN runtimes."""
from __future__ import annotations
import json
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Protocol, runtime_checkable
def _assert_json_serializable(value: object, *, field: str) -> None:
try:
json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False)
except (TypeError, ValueError) as exc:
raise ValueError(f"{field} must be canonical JSON-serializable") from exc
@dataclass(frozen=True, slots=True)
class BackendCapabilities:
"""Declared execution features and bounded limits of one backend."""
supports_recurrent: bool
supports_plasticity: bool
supports_pan_hyperstate: bool
supports_structural_plasticity: bool
max_neurons: int
max_ticks: int
deterministic: bool
max_edges: int | None = None
plasticity_semantics: str = "NOT_SUPPORTED"
execution_mode: str = "GENERAL"
supports_live_external_input: bool = False
def __post_init__(self) -> None:
if self.max_neurons < 1:
raise ValueError("max_neurons must be >= 1")
if self.max_ticks < 1:
raise ValueError("max_ticks must be >= 1")
if self.max_edges is not None and self.max_edges < 0:
raise ValueError("max_edges must be >= 0 when declared")
if not self.plasticity_semantics:
raise ValueError("plasticity_semantics must not be empty")
if not self.execution_mode:
raise ValueError("execution_mode must not be empty")
def to_mapping(self) -> dict[str, object]:
return {
"supports_recurrent": self.supports_recurrent,
"supports_plasticity": self.supports_plasticity,
"supports_pan_hyperstate": self.supports_pan_hyperstate,
"supports_structural_plasticity": self.supports_structural_plasticity,
"max_neurons": self.max_neurons,
"max_ticks": self.max_ticks,
"max_edges": self.max_edges,
"deterministic": self.deterministic,
"plasticity_semantics": self.plasticity_semantics,
"execution_mode": self.execution_mode,
"supports_live_external_input": self.supports_live_external_input,
}
@dataclass(frozen=True, slots=True)
class BackendState:
"""Backend-neutral continuation state.
The payload must be data-only and canonical-JSON serializable. Opaque
device pointers, CUDA handles and process-local object identities are
forbidden.
"""
tick: int
payload: Mapping[str, object]
state_digest: str
def __post_init__(self) -> None:
if self.tick < 0:
raise ValueError("tick must be >= 0")
if not self.state_digest:
raise ValueError("state_digest must not be empty")
_assert_json_serializable(dict(self.payload), field="payload")
@dataclass(frozen=True, slots=True)
class StepResult:
"""One canonical backend step result."""
tick: int
spikes: tuple[int, ...]
state_digest: str
metrics: Mapping[str, object]
def __post_init__(self) -> None:
if self.tick < 0:
raise ValueError("tick must be >= 0")
if not self.state_digest:
raise ValueError("state_digest must not be empty")
if any(spike < 0 for spike in self.spikes):
raise ValueError("spike neuron ids must be >= 0")
_assert_json_serializable(dict(self.metrics), field="metrics")
@dataclass(frozen=True, slots=True)
class RunResult:
"""Bounded backend run result."""
ticks_requested: int
steps: tuple[StepResult, ...]
final_state: BackendState
execution_fingerprint: str
def __post_init__(self) -> None:
if self.ticks_requested < 1:
raise ValueError("ticks_requested must be >= 1")
if len(self.steps) > self.ticks_requested:
raise ValueError("steps cannot exceed ticks_requested")
if not self.execution_fingerprint:
raise ValueError("execution_fingerprint must not be empty")
@runtime_checkable
class ExecutionBackend(Protocol):
"""Backend-neutral execution surface consumed by MHRN runtime control."""
def initialize(self, config: Mapping[str, object], seed: int) -> None:
"""Initialize from a data-only configuration and deterministic seed."""
...
def step(self, tick: int) -> StepResult:
"""Advance exactly one canonical tick."""
...
def run(self, ticks: int) -> RunResult:
"""Advance a bounded number of ticks."""
...
def snapshot(self) -> BackendState:
"""Capture backend-neutral continuation state."""
...
def restore(self, state: BackendState) -> None:
"""Restore a previously captured backend-neutral state."""
...
def capabilities(self) -> BackendCapabilities:
"""Return declared capabilities and bounded limits."""
...
@runtime_checkable
class LiveInputExecutionBackend(ExecutionBackend, Protocol):
"""ExecutionBackend extension for deterministic per-tick external input."""
def set_external_tick(self, tick: int, currents: Sequence[float]) -> None:
"""Set external currents for exactly the current continuation tick."""
...
|