Spaces:
Running
Running
File size: 5,409 Bytes
07fb183 91e2c8b 07fb183 91e2c8b 07fb183 91e2c8b 07fb183 91e2c8b 07fb183 | 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 | """CPU scalar reference backend for canonical Wave-4 parity."""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from copy import deepcopy
from typing import Any
from src.runtime.backend import BackendCapabilities, BackendState, RunResult, StepResult
from src.verification.parity import config_fingerprint, execution_fingerprint
from .backend import cpu_recurrent_reference
from .state import (
canonical_int,
prefix_inputs,
recurrent_inputs_from_mapping,
recurrent_inputs_to_mapping,
set_external_tick_config,
stable_digest,
step_payload,
validate_checkpoint,
)
class CPUReferenceBackend:
backend_name = "cpu-reference"
backend_version = "wave4-scalar-reference-v1"
def __init__(self) -> None:
self._config: dict[str, object] | None = None
self._seed = 0
self._tick = 0
def initialize(self, config: Mapping[str, object], seed: int) -> None:
inputs = recurrent_inputs_from_mapping(config, seed=seed)
self._config = recurrent_inputs_to_mapping(inputs)
self._seed = seed
self._tick = 0
def _require_config(self) -> dict[str, object]:
if self._config is None:
raise RuntimeError("backend is not initialized")
return self._config
def _execute_prefix(self, total_ticks: int) -> Mapping[str, object]:
inputs = recurrent_inputs_from_mapping(self._require_config(), seed=self._seed)
return cpu_recurrent_reference(prefix_inputs(inputs, total_ticks))
def _step_from_outputs(
self, outputs: Mapping[str, object], tick: int
) -> StepResult:
config = self._require_config()
payload = step_payload(
outputs,
tick=tick,
n_neurons=canonical_int(config["n_neurons"], field="n_neurons"),
)
spikes_raw = payload["spikes"]
if not isinstance(spikes_raw, list):
raise ValueError("spike payload must be a list")
spikes = tuple(
index for index, value in enumerate(spikes_raw) if int(value) == 1
)
return StepResult(
tick=tick + 1,
spikes=spikes,
state_digest=stable_digest(payload),
metrics={
"voltage": payload["voltage"],
"adaptation": payload["adaptation"],
"spike_count": len(spikes),
},
)
def set_external_tick(self, tick: int, currents: Sequence[float]) -> None:
if tick != self._tick:
raise ValueError("live external input tick must equal continuation cursor")
set_external_tick_config(self._require_config(), tick=tick, currents=currents)
def step(self, tick: int) -> StepResult:
if tick != self._tick:
raise ValueError("tick must equal the backend continuation cursor")
config = self._require_config()
if tick >= canonical_int(config["ticks"], field="ticks"):
raise ValueError("configured tick limit reached")
outputs = self._execute_prefix(tick + 1)
self._tick = tick + 1
return self._step_from_outputs(outputs, tick)
def run(self, ticks: int) -> RunResult:
if type(ticks) is not int or ticks < 1:
raise ValueError("ticks must be a positive int")
config = self._require_config()
total = self._tick + ticks
if total > canonical_int(config["ticks"], field="ticks"):
raise ValueError("requested run exceeds configured tick limit")
start = self._tick
outputs = self._execute_prefix(total)
steps = tuple(
self._step_from_outputs(outputs, tick) for tick in range(start, total)
)
self._tick = total
return RunResult(
ticks_requested=ticks,
steps=steps,
final_state=self.snapshot(),
execution_fingerprint=execution_fingerprint(
seed=self._seed,
config_hash=config_fingerprint(config),
backend_name=self.backend_name,
backend_version=self.backend_version,
ticks=total,
),
)
def snapshot(self) -> BackendState:
config = self._require_config()
payload: dict[str, Any] = {
"backend": self.backend_version,
"seed": self._seed,
"tick": self._tick,
"config": deepcopy(config),
"replay_required": True,
}
return BackendState(
tick=self._tick,
payload=payload,
state_digest=stable_digest(payload),
)
def restore(self, state: BackendState) -> None:
raw_config, raw_seed = validate_checkpoint(state, self.backend_version)
self.initialize(raw_config, raw_seed)
self._tick = state.tick
def capabilities(self) -> BackendCapabilities:
return BackendCapabilities(
supports_recurrent=True,
supports_plasticity=True,
supports_pan_hyperstate=False,
supports_structural_plasticity=False,
max_neurons=4096,
max_ticks=2000,
max_edges=65536,
deterministic=True,
plasticity_semantics="REFERENCE_MATCH_TARGET_DRAFT",
execution_mode="BOUNDED_REPLAY_REFERENCE",
supports_live_external_input=True,
)
|