Skip to content

Compiler Service Contract

sc_neurocore.compiler_service defines the local contract for a future compiler service that coordinates surrogate SC optimisation, digital-twin replay, and live FPGA update packages. It is a deterministic boundary layer: it does not open sockets, run vendor tools, or claim deployed service infrastructure.

Python
from sc_neurocore.compiler_service import (
    CompilerServiceRequest,
    DigitalTwinSyncContract,
    build_compiler_service_response,
)
from sc_neurocore.optimizer.sc_optimizer import HardwareBudget, LayerProfile
from sc_neurocore.optimizer.surrogate_sc_optimizer import TargetHardwareProfile

target = TargetHardwareProfile(
    name="pynq-z2",
    budget=HardwareBudget(max_luts=12_000, max_power_mw=2_500.0),
)
request = CompilerServiceRequest(
    request_id="req-001",
    target=target,
    network=(LayerProfile("hidden", mac_count=256),),
    changed_fields=("weights", "lfsr_seeds"),
    twin_sync=DigitalTwinSyncContract(session_id="twin-a"),
)
response = build_compiler_service_response(request)

Update Classes

  • hot_swap — register/configuration updates such as weights, thresholds, LFSR seeds, bitstream length, decorrelator, or routing table.
  • partial_reconfiguration — overlay-level changes such as tile mapping or AER route overlays.
  • full_resynthesis_required — HDL, clock, PDK, I/O shape, layer-count, or top-module changes.

Every package includes validation gates for digital-twin replay, telemetry bounds, and rollback packaging before live hardware application.

sc_neurocore.compiler_service

Compiler-service contracts for digital-twin and live-FPGA update loops.

LiveUpdateKind

Bases: str, Enum

Kinds of update package a compiler service may emit.

Source code in src/sc_neurocore/compiler_service.py
Python
25
26
27
28
29
30
class LiveUpdateKind(str, Enum):
    """Kinds of update package a compiler service may emit."""

    HOT_SWAP = "hot_swap"
    PARTIAL_RECONFIGURATION = "partial_reconfiguration"
    FULL_RESYNTHESIS_REQUIRED = "full_resynthesis_required"

DigitalTwinSyncContract dataclass

Digital-twin sync requirements for a compiler-service request.

Source code in src/sc_neurocore/compiler_service.py
Python
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
@dataclass(frozen=True)
class DigitalTwinSyncContract:
    """Digital-twin sync requirements for a compiler-service request."""

    session_id: str
    twin_nodes: int = 1
    checkpoint_interval_ns: int = 1_000_000
    max_drift_us: float = 100.0
    require_replay_verification: bool = True
    event_channels: tuple[str, ...] = (
        "synthesis_observation",
        "device_telemetry",
        "fault_event",
        "update_ack",
    )

    def __post_init__(self) -> None:
        if not self.session_id:
            raise ValueError("session_id must be non-empty")
        if self.twin_nodes <= 0:
            raise ValueError("twin_nodes must be positive")
        if self.checkpoint_interval_ns <= 0:
            raise ValueError("checkpoint_interval_ns must be positive")
        if self.max_drift_us < 0:
            raise ValueError("max_drift_us must be non-negative")
        if not self.event_channels:
            raise ValueError("event_channels must be non-empty")

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-compatible sync contract."""

        return {
            "checkpoint_interval_ns": self.checkpoint_interval_ns,
            "event_channels": list(self.event_channels),
            "max_drift_us": self.max_drift_us,
            "require_replay_verification": self.require_replay_verification,
            "session_id": self.session_id,
            "twin_nodes": self.twin_nodes,
        }

to_dict()

Return a JSON-compatible sync contract.

Source code in src/sc_neurocore/compiler_service.py
Python
61
62
63
64
65
66
67
68
69
70
71
def to_dict(self) -> dict[str, Any]:
    """Return a JSON-compatible sync contract."""

    return {
        "checkpoint_interval_ns": self.checkpoint_interval_ns,
        "event_channels": list(self.event_channels),
        "max_drift_us": self.max_drift_us,
        "require_replay_verification": self.require_replay_verification,
        "session_id": self.session_id,
        "twin_nodes": self.twin_nodes,
    }

LiveUpdatePolicy dataclass

Policy deciding whether a compiler update needs re-synthesis.

Source code in src/sc_neurocore/compiler_service.py
Python
 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
@dataclass(frozen=True)
class LiveUpdatePolicy:
    """Policy deciding whether a compiler update needs re-synthesis."""

    hot_swap_fields: tuple[str, ...] = (
        "weights",
        "thresholds",
        "lfsr_seeds",
        "bitstream_length",
        "decorrelator",
        "routing_table",
    )
    partial_reconfiguration_fields: tuple[str, ...] = (
        "tile_mapping",
        "aer_route_overlay",
        "layer_enable_mask",
    )
    full_resynthesis_fields: tuple[str, ...] = (
        "hdl",
        "clock",
        "pdk",
        "io_shape",
        "layer_count",
        "top_module",
    )
    validation_gates: tuple[str, ...] = (
        "digital_twin_replay",
        "telemetry_bounds",
        "rollback_package",
    )

    def classify(self, changed_fields: tuple[str, ...]) -> LiveUpdateKind:
        """Classify changed fields into the minimum required update kind."""

        if not changed_fields:
            raise ValueError("changed_fields must be non-empty")
        known = (
            set(self.hot_swap_fields)
            | set(self.partial_reconfiguration_fields)
            | set(self.full_resynthesis_fields)
        )
        unknown = sorted(set(changed_fields) - known)
        if unknown:
            raise ValueError(f"unknown update fields: {', '.join(unknown)}")
        if any(field in self.full_resynthesis_fields for field in changed_fields):
            return LiveUpdateKind.FULL_RESYNTHESIS_REQUIRED
        if any(field in self.partial_reconfiguration_fields for field in changed_fields):
            return LiveUpdateKind.PARTIAL_RECONFIGURATION
        return LiveUpdateKind.HOT_SWAP

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-compatible policy manifest."""

        return {
            "full_resynthesis_fields": list(self.full_resynthesis_fields),
            "hot_swap_fields": list(self.hot_swap_fields),
            "partial_reconfiguration_fields": list(self.partial_reconfiguration_fields),
            "validation_gates": list(self.validation_gates),
        }

classify(changed_fields)

Classify changed fields into the minimum required update kind.

Source code in src/sc_neurocore/compiler_service.py
Python
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def classify(self, changed_fields: tuple[str, ...]) -> LiveUpdateKind:
    """Classify changed fields into the minimum required update kind."""

    if not changed_fields:
        raise ValueError("changed_fields must be non-empty")
    known = (
        set(self.hot_swap_fields)
        | set(self.partial_reconfiguration_fields)
        | set(self.full_resynthesis_fields)
    )
    unknown = sorted(set(changed_fields) - known)
    if unknown:
        raise ValueError(f"unknown update fields: {', '.join(unknown)}")
    if any(field in self.full_resynthesis_fields for field in changed_fields):
        return LiveUpdateKind.FULL_RESYNTHESIS_REQUIRED
    if any(field in self.partial_reconfiguration_fields for field in changed_fields):
        return LiveUpdateKind.PARTIAL_RECONFIGURATION
    return LiveUpdateKind.HOT_SWAP

to_dict()

Return a JSON-compatible policy manifest.

Source code in src/sc_neurocore/compiler_service.py
Python
124
125
126
127
128
129
130
131
132
def to_dict(self) -> dict[str, Any]:
    """Return a JSON-compatible policy manifest."""

    return {
        "full_resynthesis_fields": list(self.full_resynthesis_fields),
        "hot_swap_fields": list(self.hot_swap_fields),
        "partial_reconfiguration_fields": list(self.partial_reconfiguration_fields),
        "validation_gates": list(self.validation_gates),
    }

CompilerServiceRequest dataclass

One request accepted by the compiler-service contract boundary.

Source code in src/sc_neurocore/compiler_service.py
Python
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
@dataclass(frozen=True)
class CompilerServiceRequest:
    """One request accepted by the compiler-service contract boundary."""

    request_id: str
    target: TargetHardwareProfile
    network: tuple[LayerProfile, ...]
    changed_fields: tuple[str, ...]
    twin_sync: DigitalTwinSyncContract
    evidence_payload: dict[str, Any] | None = None
    objective: str = "minimise_power_under_accuracy_budget"

    def __post_init__(self) -> None:
        if not self.request_id:
            raise ValueError("request_id must be non-empty")
        if not self.network:
            raise ValueError("network must contain at least one layer")
        if not self.changed_fields:
            raise ValueError("changed_fields must be non-empty")
        if not self.objective:
            raise ValueError("objective must be non-empty")

    def to_dict(self) -> dict[str, Any]:
        """Return a deterministic request manifest."""

        return {
            "changed_fields": list(self.changed_fields),
            "evidence_payload_present": self.evidence_payload is not None,
            "network": [_layer_to_dict(layer) for layer in self.network],
            "objective": self.objective,
            "request_id": self.request_id,
            "target": _target_to_dict(self.target),
            "twin_sync": self.twin_sync.to_dict(),
        }

to_dict()

Return a deterministic request manifest.

Source code in src/sc_neurocore/compiler_service.py
Python
157
158
159
160
161
162
163
164
165
166
167
168
def to_dict(self) -> dict[str, Any]:
    """Return a deterministic request manifest."""

    return {
        "changed_fields": list(self.changed_fields),
        "evidence_payload_present": self.evidence_payload is not None,
        "network": [_layer_to_dict(layer) for layer in self.network],
        "objective": self.objective,
        "request_id": self.request_id,
        "target": _target_to_dict(self.target),
        "twin_sync": self.twin_sync.to_dict(),
    }

LiveUpdatePackage dataclass

Update package planned for a live FPGA/digital-twin loop.

Source code in src/sc_neurocore/compiler_service.py
Python
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
@dataclass(frozen=True)
class LiveUpdatePackage:
    """Update package planned for a live FPGA/digital-twin loop."""

    package_id: str
    request_id: str
    target_name: str
    kind: LiveUpdateKind
    changed_fields: tuple[str, ...]
    requires_full_resynthesis: bool
    validation_gates: tuple[str, ...]
    rollback_required: bool = True

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-compatible package summary."""

        return {
            "changed_fields": list(self.changed_fields),
            "kind": self.kind.value,
            "package_id": self.package_id,
            "request_id": self.request_id,
            "requires_full_resynthesis": self.requires_full_resynthesis,
            "rollback_required": self.rollback_required,
            "target_name": self.target_name,
            "validation_gates": list(self.validation_gates),
        }

to_dict()

Return a JSON-compatible package summary.

Source code in src/sc_neurocore/compiler_service.py
Python
184
185
186
187
188
189
190
191
192
193
194
195
196
def to_dict(self) -> dict[str, Any]:
    """Return a JSON-compatible package summary."""

    return {
        "changed_fields": list(self.changed_fields),
        "kind": self.kind.value,
        "package_id": self.package_id,
        "request_id": self.request_id,
        "requires_full_resynthesis": self.requires_full_resynthesis,
        "rollback_required": self.rollback_required,
        "target_name": self.target_name,
        "validation_gates": list(self.validation_gates),
    }

CompilerServiceResponse dataclass

Deterministic response from the contract boundary.

Source code in src/sc_neurocore/compiler_service.py
Python
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
@dataclass(frozen=True)
class CompilerServiceResponse:
    """Deterministic response from the contract boundary."""

    request: CompilerServiceRequest
    update_package: LiveUpdatePackage
    optimiser_report: SurrogateOptimizerReport | None = None
    service_status: str = "contract_only"
    accepted: bool = True
    diagnostics: tuple[str, ...] = field(default_factory=tuple)

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-compatible compiler-service response."""

        return {
            "accepted": self.accepted,
            "diagnostics": list(self.diagnostics),
            "optimiser_report": _report_to_dict(self.optimiser_report),
            "request": self.request.to_dict(),
            "service_status": self.service_status,
            "update_package": self.update_package.to_dict(),
        }

to_dict()

Return a JSON-compatible compiler-service response.

Source code in src/sc_neurocore/compiler_service.py
Python
210
211
212
213
214
215
216
217
218
219
220
def to_dict(self) -> dict[str, Any]:
    """Return a JSON-compatible compiler-service response."""

    return {
        "accepted": self.accepted,
        "diagnostics": list(self.diagnostics),
        "optimiser_report": _report_to_dict(self.optimiser_report),
        "request": self.request.to_dict(),
        "service_status": self.service_status,
        "update_package": self.update_package.to_dict(),
    }

build_compiler_service_contract(*, targets, sync, policy=None)

Build a deterministic compiler-service boundary manifest.

Source code in src/sc_neurocore/compiler_service.py
Python
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def build_compiler_service_contract(
    *,
    targets: tuple[TargetHardwareProfile, ...],
    sync: DigitalTwinSyncContract,
    policy: LiveUpdatePolicy | None = None,
) -> dict[str, Any]:
    """Build a deterministic compiler-service boundary manifest."""

    if not targets:
        raise ValueError("targets must be non-empty")
    target_names = [target.name for target in targets]
    if len(set(target_names)) != len(target_names):
        raise ValueError("target names must be unique")
    active_policy = policy or LiveUpdatePolicy()
    return {
        "schema_version": "1.0",
        "service_status": "contract_only",
        "supported_targets": [
            _target_to_dict(target) for target in sorted(targets, key=_target_key)
        ],
        "sync_contract": sync.to_dict(),
        "update_policy": active_policy.to_dict(),
    }

plan_live_update(request, policy=None)

Classify a request into a live-update package.

Source code in src/sc_neurocore/compiler_service.py
Python
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def plan_live_update(
    request: CompilerServiceRequest,
    policy: LiveUpdatePolicy | None = None,
) -> LiveUpdatePackage:
    """Classify a request into a live-update package."""

    active_policy = policy or LiveUpdatePolicy()
    kind = active_policy.classify(request.changed_fields)
    return LiveUpdatePackage(
        package_id=f"{request.request_id}:{kind.value}",
        request_id=request.request_id,
        target_name=request.target.name,
        kind=kind,
        changed_fields=request.changed_fields,
        requires_full_resynthesis=kind is LiveUpdateKind.FULL_RESYNTHESIS_REQUIRED,
        validation_gates=active_policy.validation_gates,
    )

build_compiler_service_response(request, *, optimiser_report=None, policy=None)

Build a response without invoking a network service or toolchain.

Source code in src/sc_neurocore/compiler_service.py
Python
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def build_compiler_service_response(
    request: CompilerServiceRequest,
    *,
    optimiser_report: SurrogateOptimizerReport | None = None,
    policy: LiveUpdatePolicy | None = None,
) -> CompilerServiceResponse:
    """Build a response without invoking a network service or toolchain."""

    update_package = plan_live_update(request, policy=policy)
    diagnostics = _response_diagnostics(update_package)
    return CompilerServiceResponse(
        request=request,
        update_package=update_package,
        optimiser_report=optimiser_report,
        diagnostics=diagnostics,
    )