Skip to content

Solvers

Combinatorial optimization via SC-native Ising machines.

  • StochasticIsingGraph — Quantum-inspired Ising solver. Spins S_i in {-1, +1} (mapped to 0/1 for SC). Energy: E = -Sum(J_ij * S_i * S_j) - Sum(h_i * S_i). Finds minimum-energy configuration via simulated annealing with SC arithmetic.

Maps to SC hardware: spin products = AND gates, energy accumulation = popcount.

Python
import numpy as np
from sc_neurocore.solvers import StochasticIsingGraph

J = np.random.randn(10, 10)
J = (J + J.T) / 2
np.fill_diagonal(J, 0)
solver = StochasticIsingGraph(num_spins=10, J=J)
solution = solver.solve(n_steps=1000)

sc_neurocore.solvers.ising

Quantum-inspired Ising machine solver with stochastic Metropolis annealing.

StochasticIsingGraph dataclass

Quantum-Inspired Ising Machine Solver.

Spins S_i in {-1, 1} (mapped to 0, 1 for SC). Energy E = -Sum(J_ij * S_i * S_j) - Sum(h_i * S_i). Goal: Find configuration that minimizes E.

Source code in src/sc_neurocore/solvers/ising.py
Python
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
@dataclass
class StochasticIsingGraph:
    """
    Quantum-Inspired Ising Machine Solver.

    Spins S_i in {-1, 1} (mapped to 0, 1 for SC).
    Energy E = -Sum(J_ij * S_i * S_j) - Sum(h_i * S_i).
    Goal: Find configuration that minimizes E.
    """

    num_spins: int
    J: np.ndarray[Any, Any]  # Coupling matrix (symmetric, zero diagonal)
    h: np.ndarray[Any, Any]  # Bias vector
    temperature: float = 1.0
    anneal_rate: float = 0.99

    def __post_init__(self) -> None:
        """Initialise random spins and their bipolar ``{-1, 1}`` representation."""
        # Initialize spins randomly
        self.spins = np.random.randint(0, 2, self.num_spins).astype(np.int8)
        # Convert 0/1 to -1/1 for physics calc
        self.bipolar_spins = 2 * self.spins - 1

    def step(self) -> float:
        """Perform one parallel Metropolis-Hastings update and return the energy.

        Returns
        -------
        float
            The global Ising energy after applying the accepted spin flips.
        """
        # Calculate local field H_i = Sum(J_ij * S_j) + h_i
        # Using matrix multiplication
        local_field = np.dot(self.J, self.bipolar_spins) + self.h

        # Calculate Energy Difference Delta_E if we flip S_i
        # Delta_E = 2 * S_i * H_i
        # (Physics convention)

        delta_E = 2 * self.bipolar_spins * local_field

        # Probability of flipping: P = min(1, exp(-Delta_E / T))
        # If Delta_E < 0 (flip reduces energy), P=1 (always flip, greedy)
        # If Delta_E > 0 (flip increases energy), P = exp(...)

        # Vectorized probability calculation
        flip_prob = np.exp(-delta_E / self.temperature)
        flip_prob = np.minimum(1.0, flip_prob)

        # Determine flips
        random_draws = np.random.random(self.num_spins)
        should_flip = random_draws < flip_prob

        # Apply flips
        # Flip -1 to 1 and 1 to -1: S_new = -S_old
        self.bipolar_spins[should_flip] *= -1

        # Update 0/1 representation
        self.spins = (self.bipolar_spins + 1) // 2

        # Anneal
        self.temperature *= self.anneal_rate

        return self.get_energy()

    def get_energy(self) -> float:
        """Calculate global energy."""
        # E = -0.5 * S^T * J * S - h^T * S
        # Factor 0.5 because J_ij is counted twice in full matrix sum
        interaction = -0.5 * np.dot(self.bipolar_spins, np.dot(self.J, self.bipolar_spins))
        bias = -np.dot(self.h, self.bipolar_spins)
        return float(interaction + bias)

    def get_config(self) -> np.ndarray[Any, Any]:
        """Return the current spin configuration as a ``0/1`` int8 array."""
        return self.spins

__post_init__()

Initialise random spins and their bipolar {-1, 1} representation.

Source code in src/sc_neurocore/solvers/ising.py
Python
33
34
35
36
37
38
def __post_init__(self) -> None:
    """Initialise random spins and their bipolar ``{-1, 1}`` representation."""
    # Initialize spins randomly
    self.spins = np.random.randint(0, 2, self.num_spins).astype(np.int8)
    # Convert 0/1 to -1/1 for physics calc
    self.bipolar_spins = 2 * self.spins - 1

step()

Perform one parallel Metropolis-Hastings update and return the energy.

Returns

float The global Ising energy after applying the accepted spin flips.

Source code in src/sc_neurocore/solvers/ising.py
Python
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
def step(self) -> float:
    """Perform one parallel Metropolis-Hastings update and return the energy.

    Returns
    -------
    float
        The global Ising energy after applying the accepted spin flips.
    """
    # Calculate local field H_i = Sum(J_ij * S_j) + h_i
    # Using matrix multiplication
    local_field = np.dot(self.J, self.bipolar_spins) + self.h

    # Calculate Energy Difference Delta_E if we flip S_i
    # Delta_E = 2 * S_i * H_i
    # (Physics convention)

    delta_E = 2 * self.bipolar_spins * local_field

    # Probability of flipping: P = min(1, exp(-Delta_E / T))
    # If Delta_E < 0 (flip reduces energy), P=1 (always flip, greedy)
    # If Delta_E > 0 (flip increases energy), P = exp(...)

    # Vectorized probability calculation
    flip_prob = np.exp(-delta_E / self.temperature)
    flip_prob = np.minimum(1.0, flip_prob)

    # Determine flips
    random_draws = np.random.random(self.num_spins)
    should_flip = random_draws < flip_prob

    # Apply flips
    # Flip -1 to 1 and 1 to -1: S_new = -S_old
    self.bipolar_spins[should_flip] *= -1

    # Update 0/1 representation
    self.spins = (self.bipolar_spins + 1) // 2

    # Anneal
    self.temperature *= self.anneal_rate

    return self.get_energy()

get_energy()

Calculate global energy.

Source code in src/sc_neurocore/solvers/ising.py
Python
82
83
84
85
86
87
88
def get_energy(self) -> float:
    """Calculate global energy."""
    # E = -0.5 * S^T * J * S - h^T * S
    # Factor 0.5 because J_ij is counted twice in full matrix sum
    interaction = -0.5 * np.dot(self.bipolar_spins, np.dot(self.J, self.bipolar_spins))
    bias = -np.dot(self.h, self.bipolar_spins)
    return float(interaction + bias)

get_config()

Return the current spin configuration as a 0/1 int8 array.

Source code in src/sc_neurocore/solvers/ising.py
Python
90
91
92
def get_config(self) -> np.ndarray[Any, Any]:
    """Return the current spin configuration as a ``0/1`` int8 array."""
    return self.spins

Exact-current LIF profile

ExactCurrentLIFProfile is the versioned semantic contract for the existing ExactLIFSolver exact-flow, hard-reset current-based LIF model. Its canonical JSON and SHA-256 bind the equation, model-source bytes, parameters, normalized units, piecewise-constant input family, analytical crossing solver, inclusive threshold, hard reset, zero refractory interval, explicit shot reset, binary64 behavior, and absence of stochastic state. Unknown fields, schema versions, unit changes, source drift, and digest mismatches fail closed.

ExactCurrentLIFSession preserves voltage and shot-relative time across calls. Each call returns an immutable packet containing the exact producer commit, profile digest, input ticks, initial and final state, exact spike events, and an ordered state trace. Threshold and reset samples share the analytical crossing timestamp, so downstream consumers do not have to infer event ordering from a grid-aligned spike count.

Python
from sc_neurocore.solvers import (
    CurrentDriveTick,
    ExactCurrentLIFProfile,
    ExactCurrentLIFSession,
)

profile = ExactCurrentLIFProfile(tau_ms=10.0)
session = ExactCurrentLIFSession(
    profile,
    producer_commit="0123456789abcdef0123456789abcdef01234567",
    shot_id="mif-shot-17",
)
packet = session.execute(
    [
        CurrentDriveTick(duration_ms=2.0, currents=(10.0, 20.0)),
        CurrentDriveTick(duration_ms=3.0, currents=(30.0,)),
    ]
)
assert packet.profile_digest == profile.digest
checkpoint = session.serialize_state()
validated = type(packet).from_json(
    packet.to_json(),
    profile=profile,
    expected_producer_commit="0123456789abcdef0123456789abcdef01234567",
)
assert validated == packet

The canonical default profile and a four-tick complete-state receipt are shipped as exact_current_lif_profile_v1.json and exact_current_lif_multitick_v1.json under src/sc_neurocore/neurons/reference_trace_data/. The receipt binds its profile digest and the exact implementation commit and includes three off-grid events.

The profile is a precise software/reference contract. It does not by itself claim fixed-point equivalence, generated RTL parity, synthesis timing, PPA, board/HIL evidence, or biological fidelity. Those remain separate gates.

sc_neurocore.solvers.exact_lif_profile

Digest-bound, stateful execution for the exact-current SC LIF profile.

The profile binds the existing :class:ExactLIFSolver implementation rather than introducing another neuron model. It makes the solver, units, event ordering, reset boundary and numerical behaviour explicit for consumers.

ExactCurrentLIFProfile dataclass

Immutable parameters and semantics for one exact-current LIF instance.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
@dataclass(frozen=True)
class ExactCurrentLIFProfile:
    """Immutable parameters and semantics for one exact-current LIF instance."""

    tau_ms: float = 20.0
    v_rest: float = -65.0
    v_threshold: float = -50.0
    v_reset: float = -65.0
    resistance: float = 1.0

    def __post_init__(self) -> None:
        object.__setattr__(self, "tau_ms", _positive("tau_ms", self.tau_ms))
        object.__setattr__(self, "v_rest", _finite("v_rest", self.v_rest))
        object.__setattr__(self, "v_threshold", _finite("v_threshold", self.v_threshold))
        object.__setattr__(self, "v_reset", _finite("v_reset", self.v_reset))
        object.__setattr__(self, "resistance", _positive("resistance", self.resistance))
        if self.v_threshold <= self.v_rest:
            raise ValueError("v_threshold must be greater than v_rest")
        if self.v_reset >= self.v_threshold:
            raise ValueError("v_reset must be below v_threshold")

    def to_payload(self) -> dict[str, Any]:
        """Return the complete canonical profile payload."""
        return {
            "schema": PROFILE_SCHEMA,
            "profile": PROFILE_NAME,
            "model": {
                "family": "current_based_leaky_integrate_and_fire",
                "equation": "tau_ms*dV/dt=-(V-v_rest)+resistance*I",
                "source_identity": MODEL_SOURCE,
                "source_path": MODEL_SOURCE_PATH,
                "source_sha256": MODEL_SOURCE_SHA256,
                "source_scope": "Rotter-Diesmann exact integration of the SC current-based hard-reset LIF equation",
            },
            "parameters": {
                "tau_ms": self.tau_ms,
                "v_rest": self.v_rest,
                "v_threshold": self.v_threshold,
                "v_reset": self.v_reset,
                "resistance": self.resistance,
            },
            "domains": {
                "tau_ms": "finite > 0",
                "resistance": "finite > 0",
                "voltage": "finite binary64; v_threshold > v_rest and v_reset",
                "current_contribution": "finite binary64",
                "summed_current": "finite binary64",
                "tick_duration_ms": "finite > 0",
                "shot_time_ms": "finite >= 0",
            },
            "units": {
                "time": "ms",
                "voltage": "normalized_voltage",
                "current": "normalized_current",
                "resistance": "normalized_resistance",
            },
            "input": {
                "family": "piecewise_constant_current",
                "delivery": "all simultaneous contributions summed at tick start",
                "delay_ms": 0.0,
                "timestamp_domain": "float64_ms_relative_to_shot",
            },
            "solver": {
                "identity": "closed_form_piecewise_constant_event_driven",
                "threshold_crossing": "analytical_within_tick",
                "absolute_tolerance": 0.0,
                "relative_tolerance": 0.0,
                "singular_policy": "tau_ms and resistance must be finite and positive",
            },
            "events": {
                "order": [
                    "sum_inputs",
                    "evolve",
                    "detect_threshold_ge",
                    "emit",
                    "hard_reset",
                    "continue_tick",
                ],
                "threshold_comparison": "greater_than_or_equal",
                "timestamp": "analytical_crossing_time",
                "tie_break": "single neuron; input contributions are commutatively summed before evolution",
            },
            "state": {
                "variables": ["voltage", "time_ms", "shot_id", "reset_epoch"],
                "serialization_schema": STATE_SCHEMA,
                "trace_phases": ["initial", "threshold", "reset", "tick_end"],
                "initial_voltage": self.v_rest,
                "persistence": "across execute calls",
                "reset_boundary": "explicit shot reset only",
                "refractory_ms": 0.0,
                "refractory_input_policy": "not_applicable_zero_duration",
            },
            "numeric": {
                "representation": "IEEE-754 binary64",
                "rounding": "round_to_nearest_ties_to_even",
                "overflow": "fail_closed_non_finite",
                "saturation": "none",
                "backend": "python_reference",
            },
            "backend_capabilities": {
                "python_reference": "required_exact",
                "native_and_fixed_point": "separate_digest_bound_profiles_required",
            },
            "rng": {"algorithm": "none", "seed": None, "state": None},
            "compatibility": {
                "unknown_fields": "reject",
                "unknown_schema": "reject",
                "digest_mismatch": "reject",
            },
        }

    @property
    def digest(self) -> str:
        """Return the SHA-256 of canonical profile JSON."""
        return hashlib.sha256(_canonical_json(self.to_payload()).encode()).hexdigest()

    def to_json(self) -> str:
        """Serialize the profile canonically."""
        return _canonical_json(self.to_payload())

    def verify_source_binding(self) -> None:
        """Fail closed if the shipped model source differs from this profile."""
        source_path = Path(__file__).resolve().parents[1] / MODEL_SOURCE_PATH
        observed = hashlib.sha256(source_path.read_bytes()).hexdigest()
        if observed != MODEL_SOURCE_SHA256:
            raise ValueError(
                f"model source digest mismatch: expected {MODEL_SOURCE_SHA256}, observed {observed}"
            )

    @classmethod
    def from_json(cls, serialized: str | bytes) -> ExactCurrentLIFProfile:
        """Parse strict canonical semantics; reject versions, fields and units."""
        raw = _load_json("profile", serialized)
        payload = _mapping("profile", raw)
        expected = set(cls().to_payload())
        _exact_keys("profile", payload, expected)
        if payload["schema"] != PROFILE_SCHEMA or payload["profile"] != PROFILE_NAME:
            raise ValueError("unsupported profile schema or name")
        parameters = _mapping("parameters", payload["parameters"])
        _exact_keys(
            "parameters", parameters, {"tau_ms", "v_rest", "v_threshold", "v_reset", "resistance"}
        )
        parsed = cls(**parameters)
        expected_payload = parsed.to_payload()
        for field in expected - {"parameters"}:
            if payload[field] != expected_payload[field]:
                raise ValueError(f"unsupported or altered profile field: {field}")
        return parsed

digest property

Return the SHA-256 of canonical profile JSON.

to_payload()

Return the complete canonical profile payload.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def to_payload(self) -> dict[str, Any]:
    """Return the complete canonical profile payload."""
    return {
        "schema": PROFILE_SCHEMA,
        "profile": PROFILE_NAME,
        "model": {
            "family": "current_based_leaky_integrate_and_fire",
            "equation": "tau_ms*dV/dt=-(V-v_rest)+resistance*I",
            "source_identity": MODEL_SOURCE,
            "source_path": MODEL_SOURCE_PATH,
            "source_sha256": MODEL_SOURCE_SHA256,
            "source_scope": "Rotter-Diesmann exact integration of the SC current-based hard-reset LIF equation",
        },
        "parameters": {
            "tau_ms": self.tau_ms,
            "v_rest": self.v_rest,
            "v_threshold": self.v_threshold,
            "v_reset": self.v_reset,
            "resistance": self.resistance,
        },
        "domains": {
            "tau_ms": "finite > 0",
            "resistance": "finite > 0",
            "voltage": "finite binary64; v_threshold > v_rest and v_reset",
            "current_contribution": "finite binary64",
            "summed_current": "finite binary64",
            "tick_duration_ms": "finite > 0",
            "shot_time_ms": "finite >= 0",
        },
        "units": {
            "time": "ms",
            "voltage": "normalized_voltage",
            "current": "normalized_current",
            "resistance": "normalized_resistance",
        },
        "input": {
            "family": "piecewise_constant_current",
            "delivery": "all simultaneous contributions summed at tick start",
            "delay_ms": 0.0,
            "timestamp_domain": "float64_ms_relative_to_shot",
        },
        "solver": {
            "identity": "closed_form_piecewise_constant_event_driven",
            "threshold_crossing": "analytical_within_tick",
            "absolute_tolerance": 0.0,
            "relative_tolerance": 0.0,
            "singular_policy": "tau_ms and resistance must be finite and positive",
        },
        "events": {
            "order": [
                "sum_inputs",
                "evolve",
                "detect_threshold_ge",
                "emit",
                "hard_reset",
                "continue_tick",
            ],
            "threshold_comparison": "greater_than_or_equal",
            "timestamp": "analytical_crossing_time",
            "tie_break": "single neuron; input contributions are commutatively summed before evolution",
        },
        "state": {
            "variables": ["voltage", "time_ms", "shot_id", "reset_epoch"],
            "serialization_schema": STATE_SCHEMA,
            "trace_phases": ["initial", "threshold", "reset", "tick_end"],
            "initial_voltage": self.v_rest,
            "persistence": "across execute calls",
            "reset_boundary": "explicit shot reset only",
            "refractory_ms": 0.0,
            "refractory_input_policy": "not_applicable_zero_duration",
        },
        "numeric": {
            "representation": "IEEE-754 binary64",
            "rounding": "round_to_nearest_ties_to_even",
            "overflow": "fail_closed_non_finite",
            "saturation": "none",
            "backend": "python_reference",
        },
        "backend_capabilities": {
            "python_reference": "required_exact",
            "native_and_fixed_point": "separate_digest_bound_profiles_required",
        },
        "rng": {"algorithm": "none", "seed": None, "state": None},
        "compatibility": {
            "unknown_fields": "reject",
            "unknown_schema": "reject",
            "digest_mismatch": "reject",
        },
    }

to_json()

Serialize the profile canonically.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
219
220
221
def to_json(self) -> str:
    """Serialize the profile canonically."""
    return _canonical_json(self.to_payload())

verify_source_binding()

Fail closed if the shipped model source differs from this profile.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
223
224
225
226
227
228
229
230
def verify_source_binding(self) -> None:
    """Fail closed if the shipped model source differs from this profile."""
    source_path = Path(__file__).resolve().parents[1] / MODEL_SOURCE_PATH
    observed = hashlib.sha256(source_path.read_bytes()).hexdigest()
    if observed != MODEL_SOURCE_SHA256:
        raise ValueError(
            f"model source digest mismatch: expected {MODEL_SOURCE_SHA256}, observed {observed}"
        )

from_json(serialized) classmethod

Parse strict canonical semantics; reject versions, fields and units.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
@classmethod
def from_json(cls, serialized: str | bytes) -> ExactCurrentLIFProfile:
    """Parse strict canonical semantics; reject versions, fields and units."""
    raw = _load_json("profile", serialized)
    payload = _mapping("profile", raw)
    expected = set(cls().to_payload())
    _exact_keys("profile", payload, expected)
    if payload["schema"] != PROFILE_SCHEMA or payload["profile"] != PROFILE_NAME:
        raise ValueError("unsupported profile schema or name")
    parameters = _mapping("parameters", payload["parameters"])
    _exact_keys(
        "parameters", parameters, {"tau_ms", "v_rest", "v_threshold", "v_reset", "resistance"}
    )
    parsed = cls(**parameters)
    expected_payload = parsed.to_payload()
    for field in expected - {"parameters"}:
        if payload[field] != expected_payload[field]:
            raise ValueError(f"unsupported or altered profile field: {field}")
    return parsed

CurrentDriveTick dataclass

One duration with simultaneous piecewise-constant current inputs.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
@dataclass(frozen=True)
class CurrentDriveTick:
    """One duration with simultaneous piecewise-constant current inputs."""

    duration_ms: float
    currents: tuple[float, ...]

    def __post_init__(self) -> None:
        object.__setattr__(self, "duration_ms", _positive("duration_ms", self.duration_ms))
        currents = tuple(_finite("current", value) for value in self.currents)
        object.__setattr__(self, "currents", currents)
        _sum_currents(currents)

    @property
    def total_current(self) -> float:
        """Return the order-independent sum delivered at tick start."""
        return _sum_currents(self.currents)

    def to_payload(self) -> dict[str, Any]:
        return {
            "duration_ms": self.duration_ms,
            "currents": list(self.currents),
            "total_current": self.total_current,
        }

total_current property

Return the order-independent sum delivered at tick start.

ExactLIFState dataclass

Complete persistent runtime state at a shot-relative instant.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
@dataclass(frozen=True)
class ExactLIFState:
    """Complete persistent runtime state at a shot-relative instant."""

    voltage: float
    time_ms: float
    shot_id: str
    reset_epoch: int

    def __post_init__(self) -> None:
        object.__setattr__(self, "voltage", _finite("voltage", self.voltage))
        time_ms = _finite("time_ms", self.time_ms)
        if time_ms < 0.0:
            raise ValueError("time_ms must be non-negative")
        object.__setattr__(self, "time_ms", time_ms)
        if not isinstance(self.shot_id, str) or not self.shot_id or len(self.shot_id) > 128:
            raise ValueError("shot_id must be a non-empty string of at most 128 characters")
        if (
            isinstance(self.reset_epoch, bool)
            or not isinstance(self.reset_epoch, int)
            or self.reset_epoch < 0
        ):
            raise ValueError("reset_epoch must be a non-negative integer")

    def to_payload(self) -> dict[str, Any]:
        return {
            "voltage": self.voltage,
            "time_ms": self.time_ms,
            "shot_id": self.shot_id,
            "reset_epoch": self.reset_epoch,
        }

ExactLIFStateSample dataclass

One ordered point in the complete execution state trace.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
@dataclass(frozen=True)
class ExactLIFStateSample:
    """One ordered point in the complete execution state trace."""

    sequence: int
    tick: int
    phase: Literal["initial", "threshold", "reset", "tick_end"]
    time_ms: float
    voltage: float

    def to_payload(self) -> dict[str, Any]:
        return {
            "sequence": self.sequence,
            "tick": self.tick,
            "phase": self.phase,
            "time_ms": self.time_ms,
            "voltage": self.voltage,
        }

ExactLIFEvent dataclass

One exact threshold-crossing event.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
@dataclass(frozen=True)
class ExactLIFEvent:
    """One exact threshold-crossing event."""

    sequence: int
    tick: int
    time_ms: float
    voltage_before_reset: float

    def to_payload(self) -> dict[str, Any]:
        return {
            "sequence": self.sequence,
            "tick": self.tick,
            "time_ms": self.time_ms,
            "voltage_before_reset": self.voltage_before_reset,
        }

ExactLIFExecutionPacket dataclass

Immutable complete trace and provenance packet for one execute call.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
@dataclass(frozen=True)
class ExactLIFExecutionPacket:
    """Immutable complete trace and provenance packet for one execute call."""

    producer_commit: str
    profile_digest: str
    initial_state: ExactLIFState
    ticks: tuple[CurrentDriveTick, ...]
    state_trace: tuple[ExactLIFStateSample, ...]
    events: tuple[ExactLIFEvent, ...]
    final_state: ExactLIFState

    def to_payload(self) -> dict[str, Any]:
        return {
            "schema": PACKET_SCHEMA,
            "producer_commit": self.producer_commit,
            "profile": {
                "name": PROFILE_NAME,
                "schema": PROFILE_SCHEMA,
                "sha256": self.profile_digest,
            },
            "solver": "closed_form_piecewise_constant_event_driven",
            "numeric": "IEEE-754 binary64/round_to_nearest_ties_to_even/fail_closed_non_finite",
            "rng": "none",
            "reset_boundary": "explicit_shot_reset_only",
            "initial_state": self.initial_state.to_payload(),
            "ticks": [tick.to_payload() for tick in self.ticks],
            "state_trace": [sample.to_payload() for sample in self.state_trace],
            "events": [event.to_payload() for event in self.events],
            "final_state": self.final_state.to_payload(),
        }

    def to_json(self) -> str:
        """Serialize the packet canonically for deterministic evidence."""
        return _canonical_json(self.to_payload())

    @classmethod
    def from_json(
        cls,
        serialized: str | bytes,
        *,
        profile: ExactCurrentLIFProfile,
        expected_producer_commit: str,
    ) -> ExactLIFExecutionPacket:
        """Validate a packet by strict parsing and deterministic replay."""
        payload = _mapping("packet", _load_json("packet", serialized))
        _exact_keys(
            "packet",
            payload,
            {
                "schema",
                "producer_commit",
                "profile",
                "solver",
                "numeric",
                "rng",
                "reset_boundary",
                "initial_state",
                "ticks",
                "state_trace",
                "events",
                "final_state",
            },
        )
        if payload["schema"] != PACKET_SCHEMA:
            raise ValueError("unsupported packet schema")
        if payload["producer_commit"] != expected_producer_commit:
            raise ValueError("packet producer commit mismatch")
        profile_binding = _mapping("packet profile", payload["profile"])
        _exact_keys("packet profile", profile_binding, {"name", "schema", "sha256"})
        expected_binding = {
            "name": PROFILE_NAME,
            "schema": PROFILE_SCHEMA,
            "sha256": profile.digest,
        }
        if profile_binding != expected_binding:
            raise ValueError("packet profile binding mismatch")
        initial_payload = _mapping("initial_state", payload["initial_state"])
        _exact_keys(
            "initial_state", initial_payload, {"voltage", "time_ms", "shot_id", "reset_epoch"}
        )
        initial_state = ExactLIFState(**initial_payload)
        raw_ticks = payload["ticks"]
        if not isinstance(raw_ticks, list):
            raise ValueError("ticks must be an array")
        ticks: list[CurrentDriveTick] = []
        for raw_tick in raw_ticks:
            tick_payload = _mapping("tick", raw_tick)
            _exact_keys("tick", tick_payload, {"duration_ms", "currents", "total_current"})
            currents = tick_payload["currents"]
            if not isinstance(currents, list):
                raise ValueError("tick currents must be an array")
            tick = CurrentDriveTick(tick_payload["duration_ms"], tuple(currents))
            if tick_payload["total_current"] != tick.total_current:
                raise ValueError("tick total_current mismatch")
            ticks.append(tick)
        verifier = ExactCurrentLIFSession(
            profile,
            producer_commit=expected_producer_commit,
            shot_id=initial_state.shot_id,
        )
        verifier.restore_state(
            _canonical_json(
                {
                    "schema": STATE_SCHEMA,
                    "profile_sha256": profile.digest,
                    "state": initial_state.to_payload(),
                }
            )
        )
        replay = verifier.execute(ticks)
        if replay.to_payload() != payload:
            raise ValueError("packet content failed deterministic replay")
        return replay

to_json()

Serialize the packet canonically for deterministic evidence.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
382
383
384
def to_json(self) -> str:
    """Serialize the packet canonically for deterministic evidence."""
    return _canonical_json(self.to_payload())

from_json(serialized, *, profile, expected_producer_commit) classmethod

Validate a packet by strict parsing and deterministic replay.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
@classmethod
def from_json(
    cls,
    serialized: str | bytes,
    *,
    profile: ExactCurrentLIFProfile,
    expected_producer_commit: str,
) -> ExactLIFExecutionPacket:
    """Validate a packet by strict parsing and deterministic replay."""
    payload = _mapping("packet", _load_json("packet", serialized))
    _exact_keys(
        "packet",
        payload,
        {
            "schema",
            "producer_commit",
            "profile",
            "solver",
            "numeric",
            "rng",
            "reset_boundary",
            "initial_state",
            "ticks",
            "state_trace",
            "events",
            "final_state",
        },
    )
    if payload["schema"] != PACKET_SCHEMA:
        raise ValueError("unsupported packet schema")
    if payload["producer_commit"] != expected_producer_commit:
        raise ValueError("packet producer commit mismatch")
    profile_binding = _mapping("packet profile", payload["profile"])
    _exact_keys("packet profile", profile_binding, {"name", "schema", "sha256"})
    expected_binding = {
        "name": PROFILE_NAME,
        "schema": PROFILE_SCHEMA,
        "sha256": profile.digest,
    }
    if profile_binding != expected_binding:
        raise ValueError("packet profile binding mismatch")
    initial_payload = _mapping("initial_state", payload["initial_state"])
    _exact_keys(
        "initial_state", initial_payload, {"voltage", "time_ms", "shot_id", "reset_epoch"}
    )
    initial_state = ExactLIFState(**initial_payload)
    raw_ticks = payload["ticks"]
    if not isinstance(raw_ticks, list):
        raise ValueError("ticks must be an array")
    ticks: list[CurrentDriveTick] = []
    for raw_tick in raw_ticks:
        tick_payload = _mapping("tick", raw_tick)
        _exact_keys("tick", tick_payload, {"duration_ms", "currents", "total_current"})
        currents = tick_payload["currents"]
        if not isinstance(currents, list):
            raise ValueError("tick currents must be an array")
        tick = CurrentDriveTick(tick_payload["duration_ms"], tuple(currents))
        if tick_payload["total_current"] != tick.total_current:
            raise ValueError("tick total_current mismatch")
        ticks.append(tick)
    verifier = ExactCurrentLIFSession(
        profile,
        producer_commit=expected_producer_commit,
        shot_id=initial_state.shot_id,
    )
    verifier.restore_state(
        _canonical_json(
            {
                "schema": STATE_SCHEMA,
                "profile_sha256": profile.digest,
                "state": initial_state.to_payload(),
            }
        )
    )
    replay = verifier.execute(ticks)
    if replay.to_payload() != payload:
        raise ValueError("packet content failed deterministic replay")
    return replay

ExactCurrentLIFSession

Stateful, failure-atomic executor for :class:ExactCurrentLIFProfile.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
class ExactCurrentLIFSession:
    """Stateful, failure-atomic executor for :class:`ExactCurrentLIFProfile`."""

    def __init__(
        self,
        profile: ExactCurrentLIFProfile,
        *,
        producer_commit: str,
        shot_id: str = "shot-0",
    ) -> None:
        if not isinstance(profile, ExactCurrentLIFProfile):
            raise TypeError("profile must be an ExactCurrentLIFProfile")
        if not isinstance(producer_commit, str) or _COMMIT_RE.fullmatch(producer_commit) is None:
            raise ValueError("producer_commit must be a lowercase 40-character Git SHA-1")
        profile.verify_source_binding()
        self.profile = profile
        self.producer_commit = producer_commit
        self._state = ExactLIFState(profile.v_rest, 0.0, shot_id, 0)

    @property
    def state(self) -> ExactLIFState:
        """Return the current immutable persistent state."""
        return self._state

    def reset_shot(self, shot_id: str) -> ExactLIFState:
        """Reset voltage and time explicitly at a new shot boundary."""
        candidate = ExactLIFState(self.profile.v_rest, 0.0, shot_id, self._state.reset_epoch + 1)
        self._state = candidate
        return candidate

    def serialize_state(self) -> str:
        """Serialize state with the exact profile digest."""
        return _canonical_json(
            {
                "schema": STATE_SCHEMA,
                "profile_sha256": self.profile.digest,
                "state": self._state.to_payload(),
            }
        )

    def restore_state(self, serialized: str | bytes) -> ExactLIFState:
        """Restore compatible state atomically; reject drift and unknown fields."""
        raw = _load_json("state", serialized)
        payload = _mapping("state envelope", raw)
        _exact_keys("state envelope", payload, {"schema", "profile_sha256", "state"})
        if payload["schema"] != STATE_SCHEMA:
            raise ValueError("unsupported state schema")
        if payload["profile_sha256"] != self.profile.digest:
            raise ValueError("state profile digest mismatch")
        state_payload = _mapping("state", payload["state"])
        _exact_keys("state", state_payload, {"voltage", "time_ms", "shot_id", "reset_epoch"})
        candidate = ExactLIFState(**state_payload)
        if candidate.voltage >= self.profile.v_threshold:
            raise ValueError("restored voltage must be below threshold")
        self._state = candidate
        return candidate

    def execute(self, ticks: Sequence[CurrentDriveTick]) -> ExactLIFExecutionPacket:
        """Execute a complete current sequence and commit state only on success."""
        frozen_ticks = tuple(ticks)
        if any(not isinstance(tick, CurrentDriveTick) for tick in frozen_ticks):
            raise TypeError("ticks must contain only CurrentDriveTick values")
        initial = self._state
        voltage = initial.voltage
        time_ms = initial.time_ms
        state_trace = [ExactLIFStateSample(0, -1, "initial", time_ms, voltage)]
        events: list[ExactLIFEvent] = []
        solver = ExactLIFSolver(
            tau=self.profile.tau_ms,
            v_rest=self.profile.v_rest,
            v_thresh=self.profile.v_threshold,
            v_reset=self.profile.v_reset,
            r_m=self.profile.resistance,
        )

        for tick_index, tick in enumerate(frozen_ticks):
            end_ms = time_ms + tick.duration_ms
            if not math.isfinite(end_ms):
                raise ValueError("execution time overflowed binary64")
            emitted_this_tick = 0
            while time_ms < end_ms:
                remaining = end_ms - time_ms
                crossing = solver.next_spike_time(voltage, tick.total_current)
                if crossing is not None and not math.isfinite(crossing):
                    raise FloatingPointError("non-finite threshold-crossing interval")
                if crossing is None or crossing > remaining:
                    voltage = solver.evolve_to_time(voltage, remaining, tick.total_current)
                    if not math.isfinite(voltage):
                        raise FloatingPointError("membrane evolution produced a non-finite voltage")
                    time_ms = end_ms
                    break
                if crossing <= 0.0:
                    raise FloatingPointError("non-positive threshold-crossing interval")
                next_time_ms = time_ms + crossing
                if not math.isfinite(next_time_ms) or next_time_ms <= time_ms:
                    raise FloatingPointError("threshold-crossing time made no finite progress")
                time_ms = next_time_ms
                voltage = self.profile.v_threshold
                state_trace.append(
                    ExactLIFStateSample(len(state_trace), tick_index, "threshold", time_ms, voltage)
                )
                events.append(ExactLIFEvent(len(events), tick_index, time_ms, voltage))
                voltage = self.profile.v_reset
                state_trace.append(
                    ExactLIFStateSample(len(state_trace), tick_index, "reset", time_ms, voltage)
                )
                emitted_this_tick += 1
                if emitted_this_tick > _MAX_EVENTS_PER_TICK:
                    raise ValueError("event count exceeded the bounded execution contract")
            state_trace.append(
                ExactLIFStateSample(len(state_trace), tick_index, "tick_end", time_ms, voltage)
            )

        final = ExactLIFState(voltage, time_ms, initial.shot_id, initial.reset_epoch)
        packet = ExactLIFExecutionPacket(
            producer_commit=self.producer_commit,
            profile_digest=self.profile.digest,
            initial_state=initial,
            ticks=frozen_ticks,
            state_trace=tuple(state_trace),
            events=tuple(events),
            final_state=final,
        )
        packet.to_json()
        self._state = final
        return packet

state property

Return the current immutable persistent state.

reset_shot(shot_id)

Reset voltage and time explicitly at a new shot boundary.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
490
491
492
493
494
def reset_shot(self, shot_id: str) -> ExactLIFState:
    """Reset voltage and time explicitly at a new shot boundary."""
    candidate = ExactLIFState(self.profile.v_rest, 0.0, shot_id, self._state.reset_epoch + 1)
    self._state = candidate
    return candidate

serialize_state()

Serialize state with the exact profile digest.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
496
497
498
499
500
501
502
503
504
def serialize_state(self) -> str:
    """Serialize state with the exact profile digest."""
    return _canonical_json(
        {
            "schema": STATE_SCHEMA,
            "profile_sha256": self.profile.digest,
            "state": self._state.to_payload(),
        }
    )

restore_state(serialized)

Restore compatible state atomically; reject drift and unknown fields.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
def restore_state(self, serialized: str | bytes) -> ExactLIFState:
    """Restore compatible state atomically; reject drift and unknown fields."""
    raw = _load_json("state", serialized)
    payload = _mapping("state envelope", raw)
    _exact_keys("state envelope", payload, {"schema", "profile_sha256", "state"})
    if payload["schema"] != STATE_SCHEMA:
        raise ValueError("unsupported state schema")
    if payload["profile_sha256"] != self.profile.digest:
        raise ValueError("state profile digest mismatch")
    state_payload = _mapping("state", payload["state"])
    _exact_keys("state", state_payload, {"voltage", "time_ms", "shot_id", "reset_epoch"})
    candidate = ExactLIFState(**state_payload)
    if candidate.voltage >= self.profile.v_threshold:
        raise ValueError("restored voltage must be below threshold")
    self._state = candidate
    return candidate

execute(ticks)

Execute a complete current sequence and commit state only on success.

Source code in src/sc_neurocore/solvers/exact_lif_profile.py
Python
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
def execute(self, ticks: Sequence[CurrentDriveTick]) -> ExactLIFExecutionPacket:
    """Execute a complete current sequence and commit state only on success."""
    frozen_ticks = tuple(ticks)
    if any(not isinstance(tick, CurrentDriveTick) for tick in frozen_ticks):
        raise TypeError("ticks must contain only CurrentDriveTick values")
    initial = self._state
    voltage = initial.voltage
    time_ms = initial.time_ms
    state_trace = [ExactLIFStateSample(0, -1, "initial", time_ms, voltage)]
    events: list[ExactLIFEvent] = []
    solver = ExactLIFSolver(
        tau=self.profile.tau_ms,
        v_rest=self.profile.v_rest,
        v_thresh=self.profile.v_threshold,
        v_reset=self.profile.v_reset,
        r_m=self.profile.resistance,
    )

    for tick_index, tick in enumerate(frozen_ticks):
        end_ms = time_ms + tick.duration_ms
        if not math.isfinite(end_ms):
            raise ValueError("execution time overflowed binary64")
        emitted_this_tick = 0
        while time_ms < end_ms:
            remaining = end_ms - time_ms
            crossing = solver.next_spike_time(voltage, tick.total_current)
            if crossing is not None and not math.isfinite(crossing):
                raise FloatingPointError("non-finite threshold-crossing interval")
            if crossing is None or crossing > remaining:
                voltage = solver.evolve_to_time(voltage, remaining, tick.total_current)
                if not math.isfinite(voltage):
                    raise FloatingPointError("membrane evolution produced a non-finite voltage")
                time_ms = end_ms
                break
            if crossing <= 0.0:
                raise FloatingPointError("non-positive threshold-crossing interval")
            next_time_ms = time_ms + crossing
            if not math.isfinite(next_time_ms) or next_time_ms <= time_ms:
                raise FloatingPointError("threshold-crossing time made no finite progress")
            time_ms = next_time_ms
            voltage = self.profile.v_threshold
            state_trace.append(
                ExactLIFStateSample(len(state_trace), tick_index, "threshold", time_ms, voltage)
            )
            events.append(ExactLIFEvent(len(events), tick_index, time_ms, voltage))
            voltage = self.profile.v_reset
            state_trace.append(
                ExactLIFStateSample(len(state_trace), tick_index, "reset", time_ms, voltage)
            )
            emitted_this_tick += 1
            if emitted_this_tick > _MAX_EVENTS_PER_TICK:
                raise ValueError("event count exceeded the bounded execution contract")
        state_trace.append(
            ExactLIFStateSample(len(state_trace), tick_index, "tick_end", time_ms, voltage)
        )

    final = ExactLIFState(voltage, time_ms, initial.shot_id, initial.reset_epoch)
    packet = ExactLIFExecutionPacket(
        producer_commit=self.producer_commit,
        profile_digest=self.profile.digest,
        initial_state=initial,
        ticks=frozen_ticks,
        state_trace=tuple(state_trace),
        events=tuple(events),
        final_state=final,
    )
    packet.to_json()
    self._state = final
    return packet