Skip to content

Binding System

The binding system connects domain-specific signals to SPO's universal oscillator framework. A binding specification (YAML file) declares the complete interface between a domain and the phase dynamics engine.

Pipeline position

The binding system is the configuration layer of the SPO pipeline. It is loaded once at startup and configures all downstream subsystems:

binding_spec.yaml
       │
       ↓
  load_binding_spec()
       │
       ↓
  validate_binding_spec()
       │
       ↓
  BindingSpec
  ├── layers[] ──→ Oscillator Extractors (P/I/S)
  ├── coupling  ──→ CouplingBuilder.build()
  ├── policy    ──→ PolicyEngine rules
  └── actuators ──→ ActuationMapper mappings

Without a valid binding spec, SPO cannot start. The spec declares what to observe, how to couple, when to intervene, and where to actuate.

Role in the Architecture

The binding system is the first stage of the SPO pipeline:

Domain Data ─► binding_spec.yaml ─► Loader ─► Validator ─► BindingSpec
                                                              │
                                                    Oscillator Extractors
                                                    Coupling Templates
                                                    Policy Rules
                                                    Actuator Mappings

Every domainpack ships a binding specification. When SPO starts, the loader reads the YAML, the validator checks it against the schema, and the resulting BindingSpec configures all downstream subsystems.

Specification Structure

A binding spec declares:

name: power_grid
version: "1.0"
layers:
  - name: generator_phase
    channel: P
    extractor: hilbert
    frequency_range: [49.5, 50.5]
  - name: load_demand
    channel: I
    extractor: event_rate
coupling:
  template: distance_decay
  K_base: 0.47
  decay_alpha: 0.25
policy:
  rules:
    - condition: R < 0.6
      action: boost_K(0.1)
actuators:
  - name: governor
    knob: K
    scope: layer_0
    limits: [0.0, 2.0]

The schema is defined in docs/specs/binding_spec.schema.json and enforced by the validator at load time.

Resolved Runtime Summary

validate/inspect/run commands rely on a resolved summary that is produced from the YAML and includes inferred defaults (for example control_interval_steps and engine_mode). The full contract is documented in Resolved Runtime Defaults and exposed as a CLI summary plus audit metadata. The summary now embeds channel_algebra, so audit consumers can read required channels, optional channels, derived channels, group membership, coupling participants, and missing required channel evidence from the same resolved configuration record.

resolved

Deterministic summaries of binding runtime choices.

Resolved binding records expose timing, engine mode, layers, families, channels, driver key names, objectives, actuators, and optional feature flags for CLI output and audit headers. Raw driver configuration values are deliberately omitted because production bindings may include endpoints or deployment-local identifiers that should not be copied into public logs.

Classes

Functions:

resolved_binding_config 

resolved_binding_config(
    spec: BindingSpec,
) -> dict[str, object]

Build a deterministic, JSON-safe summary of binding runtime choices.

The summary intentionally exposes structural choices, enabled features, and driver key names only. It does not copy raw driver configuration values into audit metadata because production driver blocks may contain endpoints or deployment-local identifiers.

Parameters

spec : BindingSpec The binding specification whose resolved runtime choices are summarised.

Returns

dict[str, object] Deterministic, JSON-safe mapping of structural choices, enabled features, and driver key names; raw driver values are excluded.

Source code in src/scpn_phase_orchestrator/binding/resolved.py 
 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
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
def resolved_binding_config(spec: BindingSpec) -> dict[str, object]:
    """Build a deterministic, JSON-safe summary of binding runtime choices.

    The summary intentionally exposes structural choices, enabled features, and
    driver key names only. It does not copy raw driver configuration values into
    audit metadata because production driver blocks may contain endpoints or
    deployment-local identifiers.

    Parameters
    ----------
    spec : BindingSpec
        The binding specification whose resolved runtime choices are summarised.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of structural choices, enabled
        features, and driver key names; raw driver values are excluded.
    """
    n_osc = sum(len(layer.oscillator_ids) for layer in spec.layers)
    control_interval_steps = max(1, round(spec.control_period_s / spec.sample_period_s))
    family_channels = {
        name: family.channel
        for name, family in sorted(spec.oscillator_families.items())
    }
    driver_configs = spec.drivers.all_channel_configs()
    channels = sorted(spec.used_channels())

    family_summaries: dict[str, dict[str, object]] = {}
    for name, family in sorted(spec.oscillator_families.items()):
        family_summaries[name] = {
            "channel": family.channel,
            "extractor_type": family.extractor_type,
            "resolved_extractor_type": resolve_extractor_type(family.extractor_type),
            "config_keys": sorted(family.config),
        }

    layer_summaries: list[dict[str, object]] = []
    for layer in sorted(spec.layers, key=lambda item: item.index):
        channel = family_channels.get(layer.family) if layer.family else None
        layer_summaries.append(
            {
                "name": layer.name,
                "index": layer.index,
                "family": layer.family,
                "channel": channel,
                "oscillator_count": len(layer.oscillator_ids),
            }
        )

    channel_summaries: dict[str, dict[str, object]] = {}
    for channel in channels:
        family_names = sorted(
            name
            for name, family_channel in family_channels.items()
            if family_channel == channel
        )
        channel_layers = [
            layer
            for layer in spec.layers
            if layer.family is not None and layer.family in family_names
        ]
        extractors = sorted(
            {
                resolve_extractor_type(spec.oscillator_families[name].extractor_type)
                for name in family_names
            }
        )
        driver_config = driver_configs.get(channel, {})
        channel_spec = spec.channels.get(channel)
        channel_summaries[channel] = {
            "families": family_names,
            "extractors": extractors,
            "driver_configured": bool(driver_config),
            "driver_keys": sorted(driver_config),
            "layer_count": len(channel_layers),
            "oscillator_count": sum(
                len(layer.oscillator_ids) for layer in channel_layers
            ),
            "declared": channel_spec is not None,
            "role": channel_spec.role if channel_spec is not None else None,
            "required": channel_spec.required if channel_spec is not None else None,
            "units": channel_spec.units if channel_spec is not None else None,
            "metric_semantics": (
                channel_spec.metric_semantics if channel_spec is not None else None
            ),
            "coupling_participation": (
                channel_spec.coupling_participation
                if channel_spec is not None
                else None
            ),
            "audit_serialisation": (
                channel_spec.audit_serialisation if channel_spec is not None else None
            ),
            "replay_semantics": (
                channel_spec.replay_semantics if channel_spec is not None else None
            ),
            "supervisor_visibility": (
                channel_spec.supervisor_visibility if channel_spec is not None else None
            ),
            "derived_from": (
                list(channel_spec.derived_from) if channel_spec is not None else []
            ),
            "derive_rule": channel_spec.derive_rule
            if channel_spec is not None
            else None,
        }

    features = {
        "amplitude": spec.amplitude is not None,
        "geometry_prior": spec.geometry_prior is not None,
        "imprint_model": spec.imprint_model is not None,
        "protocol_net": spec.protocol_net is not None,
    }
    channel_algebra = build_channel_algebra_report(spec).to_audit_record()

    return {
        "name": spec.name,
        "version": spec.version,
        "safety_tier": spec.safety_tier,
        "validation_tier": spec.validation_tier,
        "sample_period_s": spec.sample_period_s,
        "control_period_s": spec.control_period_s,
        "control_interval_steps": control_interval_steps,
        "engine_mode": "stuart_landau" if spec.amplitude is not None else "kuramoto",
        "layer_count": len(spec.layers),
        "oscillator_count": n_osc,
        "channels": channel_summaries,
        "channel_groups": {
            name: {
                "channels": list(group.channels),
                "required": group.required,
                "description": group.description,
            }
            for name, group in sorted(spec.channel_groups.items())
        },
        "cross_channel_couplings": [
            {
                "source": coupling.source,
                "target": coupling.target,
                "strength": coupling.strength,
                "mode": coupling.mode,
                "template": coupling.template,
            }
            for coupling in spec.cross_channel_couplings
        ],
        "channel_algebra": channel_algebra,
        "families": family_summaries,
        "layers": layer_summaries,
        "unassigned_layer_count": sum(
            1 for layer in spec.layers if layer.family is None
        ),
        "coupling": {
            "base_strength": spec.coupling.base_strength,
            "decay_alpha": spec.coupling.decay_alpha,
            "templates": sorted(spec.coupling.templates),
        },
        "objectives": {
            "good_layers": list(spec.objectives.good_layers),
            "bad_layers": list(spec.objectives.bad_layers),
            "good_weight": spec.objectives.good_weight,
            "bad_weight": spec.objectives.bad_weight,
        },
        "boundaries": [
            {
                "name": boundary.name,
                "variable": boundary.variable,
                "severity": boundary.severity,
            }
            for boundary in sorted(spec.boundaries, key=lambda item: item.name)
        ],
        "actuators": [
            {
                "name": actuator.name,
                "knob": actuator.knob,
                "scope": actuator.scope,
                "limits": list(actuator.limits),
                "rate_limit_per_step": actuator.rate_limit_per_step,
            }
            for actuator in sorted(spec.actuators, key=lambda item: item.name)
        ],
        "features": features,
    }

format_resolved_binding_config 

format_resolved_binding_config(
    summary: dict[str, object],
) -> list[str]

Render a compact, human-readable summary for CLI output.

Parameters

summary : dict[str, object] A mapping produced by :func:resolved_binding_config.

Returns

list[str] Formatted output lines suitable for printing to a terminal.

Source code in src/scpn_phase_orchestrator/binding/resolved.py 
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
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
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
310
311
312
313
314
315
def format_resolved_binding_config(summary: dict[str, object]) -> list[str]:
    """Render a compact, human-readable summary for CLI output.

    Parameters
    ----------
    summary : dict[str, object]
        A mapping produced by :func:`resolved_binding_config`.

    Returns
    -------
    list[str]
        Formatted output lines suitable for printing to a terminal.
    """
    channels = summary["channels"]
    assert isinstance(channels, dict)  # nosec B101
    features = summary["features"]
    assert isinstance(features, dict)  # nosec B101
    enabled_features = sorted(name for name, enabled in features.items() if enabled)
    feature_text = ", ".join(enabled_features) if enabled_features else "none"
    channel_names = ", ".join(sorted(str(channel) for channel in channels)) or "none"

    lines = [
        "Resolved configuration:",
        (
            f"  domain: {summary['name']} v{summary['version']} "
            f"({summary['safety_tier']})"
        ),
        (
            f"  timing: sample={summary['sample_period_s']}s "
            f"control={summary['control_period_s']}s "
            f"interval={summary['control_interval_steps']} steps"
        ),
        (
            f"  structure: layers={summary['layer_count']} "
            f"oscillators={summary['oscillator_count']} channels={channel_names}"
        ),
        f"  engine: {summary['engine_mode']} features={feature_text}",
    ]

    for channel, raw_info in sorted(channels.items(), key=lambda item: str(item[0])):
        assert isinstance(raw_info, dict)  # nosec B101
        families = _string_list(raw_info.get("families")) or ["none"]
        extractors = _string_list(raw_info.get("extractors")) or ["none"]
        driver_keys = _string_list(raw_info.get("driver_keys")) or ["none"]
        lines.append(
            f"  channel {channel}: families={','.join(families)} "
            f"extractors={','.join(extractors)} "
            f"driver_keys={','.join(driver_keys)} "
            f"layers={raw_info.get('layer_count', 0)} "
            f"oscillators={raw_info.get('oscillator_count', 0)}"
        )
        if raw_info.get("declared"):
            role = raw_info.get("role") or "domain"
            replay = raw_info.get("replay_semantics") or "phase"
            derived_from = _string_list(raw_info.get("derived_from"))
            derived = f" derived_from={','.join(derived_from)}" if derived_from else ""
            lines.append(
                f"    metadata: role={role} replay={replay} "
                f"supervisor={raw_info.get('supervisor_visibility')}{derived}"
            )

    groups = summary.get("channel_groups", {})
    if isinstance(groups, dict) and groups:
        group_names = ", ".join(sorted(str(name) for name in groups))
        lines.append(f"  channel_groups: {group_names}")

    couplings = summary.get("cross_channel_couplings", [])
    if isinstance(couplings, list) and couplings:
        lines.append(f"  cross_channel_couplings: {len(couplings)}")

    algebra = summary.get("channel_algebra", {})
    if isinstance(algebra, dict):
        required = _string_list(algebra.get("required_channels"))
        optional = _string_list(algebra.get("optional_channels"))
        derived_channels = _string_list(algebra.get("derived_channels"))
        missing = _string_list(algebra.get("missing_required_channels"))
        delayed = _string_list(algebra.get("delayed_channels"))
        uncertain = _string_list(algebra.get("uncertain_channels"))
        visible = _string_list(algebra.get("supervisor_visible_channels"))
        participating = _string_list(algebra.get("coupling_participating_channels"))
        lines.append(
            "  channel_algebra: "
            f"required={len(required)} optional={len(optional)} "
            f"derived={len(derived_channels)} visible={len(visible)} "
            f"coupling_participants={len(participating)} "
            f"delayed={len(delayed)} uncertain={len(uncertain)}"
        )
        if missing:
            lines.append(f"    missing_required_channels: {','.join(missing)}")

    unassigned = summary.get("unassigned_layer_count", 0)
    if unassigned:
        lines.append(
            f"  note: {unassigned} layer(s) have no explicit oscillator family binding"
        )
    return lines

N-Channel Algebra Summary

build_channel_algebra_report() produces a deterministic, JSON-safe view of declared channels, required/optional status, derived channels, group membership, supervisor visibility, coupling participation, and cross-channel edges. It is intended for audit, replay, and reporting surfaces that need a channel-count-agnostic view without re-parsing YAML.

The same report classifies delayed and uncertain channels from existing role, metric_semantics, and replay_semantics metadata. This lets audit and reporting surfaces expose delayed/uncertain policy evidence without changing the binding schema.

The report also emits runtime policy records for every declared channel. Delayed channels use hold_last_runtime_evidence, uncertain channels use confidence_weight_runtime_contribution, missing required channels use block_required_channel, and missing optional channels use drop_optional_channel. This gives supervisor/runtime callers deterministic handling semantics without adding new binding-schema fields.

ChannelRuntimeExecutor applies those delayed and uncertain policies during spo run. Delayed channels contribute the previous tick's layer evidence once available, with the first tick explicitly marked as current_tick_prime. Uncertain channels scale their layer R contribution by a named-channel driver confidence_weight or confidence value clamped to [0, 1]. The executed layer states are the states consumed by supervisor decisions and boundary observation, while the audit log records raw versus executed R and psi values under channel_runtime.

from scpn_phase_orchestrator.binding import (
    build_channel_algebra_report,
    load_binding_spec,
)

spec = load_binding_spec("domainpacks/power_safety_nchannel/binding_spec.yaml")
report = build_channel_algebra_report(spec)
audit_record = report.to_audit_record()

This report is read-only. It complements validate_binding_spec() rather than replacing validation gates.

channel_algebra

Deterministic N-channel algebra summaries for binding specs.

Classes

ChannelCouplingEdge dataclass 

ChannelCouplingEdge(
    source: str,
    target: str,
    strength: float,
    mode: str,
    template: str | None,
)

JSON-safe cross-channel coupling edge.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a serialisable coupling-edge record.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the ChannelCouplingEdge fields.

Source code in src/scpn_phase_orchestrator/binding/channel_algebra.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def to_audit_record(self) -> dict[str, object]:
    """Return a serialisable coupling-edge record.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the ChannelCouplingEdge fields.
    """
    return {
        "source": self.source,
        "target": self.target,
        "strength": self.strength,
        "mode": self.mode,
        "template": self.template,
    }

ChannelRuntimePolicy dataclass 

ChannelRuntimePolicy(
    channel: str,
    evidence_required: bool,
    delay_policy: str,
    uncertainty_policy: str,
    missing_policy: str,
)

Runtime handling policy derived from channel metadata.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a serialisable runtime-policy record.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the ChannelRuntimePolicy fields.

Source code in src/scpn_phase_orchestrator/binding/channel_algebra.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def to_audit_record(self) -> dict[str, object]:
    """Return a serialisable runtime-policy record.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the ChannelRuntimePolicy fields.
    """
    return {
        "channel": self.channel,
        "evidence_required": self.evidence_required,
        "delay_policy": self.delay_policy,
        "uncertainty_policy": self.uncertainty_policy,
        "missing_policy": self.missing_policy,
    }

ChannelAlgebraReport dataclass 

ChannelAlgebraReport(
    channels: tuple[str, ...],
    declared_channels: tuple[str, ...],
    required_channels: tuple[str, ...],
    optional_channels: tuple[str, ...],
    derived_channels: tuple[str, ...],
    delayed_channels: tuple[str, ...],
    uncertain_channels: tuple[str, ...],
    runtime_evidence_channels: tuple[str, ...],
    missing_required_channels: tuple[str, ...],
    supervisor_visible_channels: tuple[str, ...],
    coupling_participating_channels: tuple[str, ...],
    replay_semantics: dict[str, str],
    runtime_policies: dict[str, ChannelRuntimePolicy],
    channel_groups: dict[str, tuple[str, ...]],
    channel_membership: dict[str, tuple[str, ...]],
    coupling_edges: tuple[ChannelCouplingEdge, ...],
)

Deterministic channel algebra view for audit, replay, and reporting.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a serialisable channel algebra record.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the ChannelAlgebraReport fields.

Source code in src/scpn_phase_orchestrator/binding/channel_algebra.py 
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
def to_audit_record(self) -> dict[str, object]:
    """Return a serialisable channel algebra record.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the ChannelAlgebraReport fields.
    """
    return {
        "channels": list(self.channels),
        "declared_channels": list(self.declared_channels),
        "required_channels": list(self.required_channels),
        "optional_channels": list(self.optional_channels),
        "derived_channels": list(self.derived_channels),
        "delayed_channels": list(self.delayed_channels),
        "uncertain_channels": list(self.uncertain_channels),
        "runtime_evidence_channels": list(self.runtime_evidence_channels),
        "missing_required_channels": list(self.missing_required_channels),
        "supervisor_visible_channels": list(self.supervisor_visible_channels),
        "coupling_participating_channels": list(
            self.coupling_participating_channels
        ),
        "replay_semantics": dict(sorted(self.replay_semantics.items())),
        "runtime_policies": {
            channel: policy.to_audit_record()
            for channel, policy in sorted(self.runtime_policies.items())
        },
        "channel_groups": {
            name: list(channels)
            for name, channels in sorted(self.channel_groups.items())
        },
        "channel_membership": {
            channel: list(groups)
            for channel, groups in sorted(self.channel_membership.items())
        },
        "coupling_edges": [edge.to_audit_record() for edge in self.coupling_edges],
    }

Functions:

build_channel_algebra_report 

build_channel_algebra_report(
    spec: BindingSpec,
) -> ChannelAlgebraReport

Build a deterministic N-channel algebra report from a binding spec.

The report is a read-only structural view. It does not validate or mutate the binding; callers should still run validate_binding_spec() for gates.

Parameters

spec : BindingSpec The binding specification to analyse.

Returns

ChannelAlgebraReport A read-only report of channels, coupling edges, and runtime policies derived from the spec.

Source code in src/scpn_phase_orchestrator/binding/channel_algebra.py 
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
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def build_channel_algebra_report(spec: BindingSpec) -> ChannelAlgebraReport:
    """Build a deterministic N-channel algebra report from a binding spec.

    The report is a read-only structural view. It does not validate or mutate
    the binding; callers should still run `validate_binding_spec()` for gates.

    Parameters
    ----------
    spec : BindingSpec
        The binding specification to analyse.

    Returns
    -------
    ChannelAlgebraReport
        A read-only report of channels, coupling edges, and runtime policies
        derived from the spec.
    """
    channels = tuple(sorted(spec.used_channels()))
    declared_channels = tuple(sorted(spec.channels))
    runtime_evidence_channels = _runtime_evidence_channels(spec)
    required_channels = tuple(
        sorted(
            channel
            for channel, channel_spec in spec.channels.items()
            if channel_spec.required
        )
    )
    optional_channels = tuple(
        sorted(
            channel
            for channel, channel_spec in spec.channels.items()
            if not channel_spec.required
        )
    )
    derived_channels = tuple(
        sorted(
            channel
            for channel, channel_spec in spec.channels.items()
            if channel_spec.derived_from
            or channel_spec.replay_semantics == "derived"
            or channel_spec.derive_rule is not None
        )
    )
    delayed_channels = tuple(
        sorted(
            channel
            for channel, channel_spec in spec.channels.items()
            if _mentions_policy_marker(
                (
                    channel_spec.role,
                    channel_spec.metric_semantics,
                    channel_spec.replay_semantics,
                ),
                ("delayed", "delay", "lagged", "external"),
            )
        )
    )
    uncertain_channels = tuple(
        sorted(
            channel
            for channel, channel_spec in spec.channels.items()
            if _mentions_policy_marker(
                (
                    channel_spec.role,
                    channel_spec.metric_semantics,
                    channel_spec.replay_semantics,
                ),
                ("uncertain", "uncertainty", "probabilistic", "confidence"),
            )
        )
    )
    missing_required_channels = tuple(
        sorted(
            channel
            for channel in required_channels
            if channel not in runtime_evidence_channels
            and channel not in derived_channels
        )
    )
    supervisor_visible_channels = tuple(
        sorted(
            channel
            for channel in channels
            if spec.channels.get(channel) is None
            or spec.channels[channel].supervisor_visibility
        )
    )
    coupling_participating_channels = tuple(
        sorted(
            channel
            for channel in channels
            if spec.channels.get(channel) is None
            or spec.channels[channel].coupling_participation
        )
    )
    channel_groups = {
        name: tuple(group.channels)
        for name, group in sorted(spec.channel_groups.items())
    }
    channel_membership = _channel_membership(channels, channel_groups)
    replay_semantics = {
        channel: spec.channels[channel].replay_semantics
        for channel in declared_channels
    }
    runtime_policies = _runtime_policies(
        declared_channels=declared_channels,
        required_channels=required_channels,
        delayed_channels=delayed_channels,
        uncertain_channels=uncertain_channels,
    )
    coupling_edges = tuple(
        ChannelCouplingEdge(
            source=coupling.source,
            target=coupling.target,
            strength=coupling.strength,
            mode=coupling.mode,
            template=coupling.template,
        )
        for coupling in spec.cross_channel_couplings
    )
    return ChannelAlgebraReport(
        channels=channels,
        declared_channels=declared_channels,
        required_channels=required_channels,
        optional_channels=optional_channels,
        derived_channels=derived_channels,
        delayed_channels=delayed_channels,
        uncertain_channels=uncertain_channels,
        runtime_evidence_channels=runtime_evidence_channels,
        missing_required_channels=missing_required_channels,
        supervisor_visible_channels=supervisor_visible_channels,
        coupling_participating_channels=coupling_participating_channels,
        replay_semantics=replay_semantics,
        runtime_policies=runtime_policies,
        channel_groups=channel_groups,
        channel_membership=channel_membership,
        coupling_edges=coupling_edges,
    )

channel_runtime

Runtime execution of delayed and uncertain N-channel policies.

Classes

ChannelLayerRuntimeEvidence dataclass 

ChannelLayerRuntimeEvidence(
    layer_index: int,
    channel: str,
    raw_R: float,
    executed_R: float,
    raw_psi: float,
    executed_psi: float,
    delay_policy: str,
    uncertainty_policy: str,
    evidence_source: str,
    confidence_weight: float,
)

Per-layer evidence showing how a channel policy affected execution.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a serialisable layer runtime-evidence record.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the ChannelLayerRuntimeEvidence fields.

Source code in src/scpn_phase_orchestrator/binding/channel_runtime.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def to_audit_record(self) -> dict[str, object]:
    """Return a serialisable layer runtime-evidence record.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the ChannelLayerRuntimeEvidence
        fields.
    """
    return {
        "layer_index": self.layer_index,
        "channel": self.channel,
        "raw_R": self.raw_R,
        "executed_R": self.executed_R,
        "raw_psi": self.raw_psi,
        "executed_psi": self.executed_psi,
        "delay_policy": self.delay_policy,
        "uncertainty_policy": self.uncertainty_policy,
        "evidence_source": self.evidence_source,
        "confidence_weight": self.confidence_weight,
    }

ChannelRuntimeExecution dataclass 

ChannelRuntimeExecution(
    layers: tuple[LayerState, ...],
    evidence: tuple[ChannelLayerRuntimeEvidence, ...],
)

Executed layer states plus audit evidence for one runtime tick.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a serialisable runtime execution record.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the ChannelRuntimeExecution fields.

Source code in src/scpn_phase_orchestrator/binding/channel_runtime.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def to_audit_record(self) -> dict[str, object]:
    """Return a serialisable runtime execution record.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the ChannelRuntimeExecution
        fields.
    """
    return {
        "layers": [item.to_audit_record() for item in self.evidence],
        "delayed_layers": [
            item.layer_index
            for item in self.evidence
            if item.delay_policy == "hold_last_runtime_evidence"
        ],
        "uncertain_layers": [
            item.layer_index
            for item in self.evidence
            if item.uncertainty_policy == "confidence_weight_runtime_contribution"
        ],
    }

ChannelRuntimeExecutor 

ChannelRuntimeExecutor(
    *,
    layer_channels: tuple[str, ...],
    report: ChannelAlgebraReport,
    confidence_weights: dict[str, float],
)

Apply N-channel delay and uncertainty policies to layer diagnostics.

The executor is intentionally deterministic and non-actuating. It only transforms the layer diagnostics consumed by the supervisor and audit log: delayed channels contribute the previous tick's layer evidence when available, while uncertain channels scale their contribution by an explicit driver confidence weight.

Initialise an executor from resolved channel policy inputs.

Source code in src/scpn_phase_orchestrator/binding/channel_runtime.py 
109
110
111
112
113
114
115
116
117
118
119
120
def __init__(
    self,
    *,
    layer_channels: tuple[str, ...],
    report: ChannelAlgebraReport,
    confidence_weights: dict[str, float],
) -> None:
    """Initialise an executor from resolved channel policy inputs."""
    self._layer_channels = layer_channels
    self._report = report
    self._confidence_weights = confidence_weights
    self._last_raw_layers: tuple[LayerState, ...] | None = None
Methods:
from_spec classmethod 
from_spec(spec: BindingSpec) -> ChannelRuntimeExecutor

Build a runtime executor from binding channel metadata.

Parameters

spec : BindingSpec The binding specification supplying channel layer metadata.

Returns

ChannelRuntimeExecutor An executor configured with the spec's per-channel runtime policies.

Source code in src/scpn_phase_orchestrator/binding/channel_runtime.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@classmethod
def from_spec(cls, spec: BindingSpec) -> ChannelRuntimeExecutor:
    """Build a runtime executor from binding channel metadata.

    Parameters
    ----------
    spec : BindingSpec
        The binding specification supplying channel layer metadata.

    Returns
    -------
    ChannelRuntimeExecutor
        An executor configured with the spec's per-channel runtime policies.
    """
    return cls(
        layer_channels=_layer_channels(spec),
        report=build_channel_algebra_report(spec),
        confidence_weights=_confidence_weights(spec),
    )
execute 
execute(
    raw_layers: list[LayerState],
) -> ChannelRuntimeExecution

Apply delayed/uncertain channel policies to this tick's layer states.

Parameters

raw_layers : list[LayerState] Per-layer states observed for the current tick, one per binding layer.

Returns

ChannelRuntimeExecution The policy-adjusted layer states with per-layer runtime evidence.

Raises

ValueError If raw_layers does not have one entry per configured binding layer.

Source code in src/scpn_phase_orchestrator/binding/channel_runtime.py 
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
def execute(self, raw_layers: list[LayerState]) -> ChannelRuntimeExecution:
    """Apply delayed/uncertain channel policies to this tick's layer states.

    Parameters
    ----------
    raw_layers : list[LayerState]
        Per-layer states observed for the current tick, one per binding
        layer.

    Returns
    -------
    ChannelRuntimeExecution
        The policy-adjusted layer states with per-layer runtime evidence.

    Raises
    ------
    ValueError
        If ``raw_layers`` does not have one entry per configured binding
        layer.
    """
    if len(raw_layers) != len(self._layer_channels):
        msg = (
            "raw layer count must match binding layer count: "
            f"{len(raw_layers)} != {len(self._layer_channels)}"
        )
        raise ValueError(msg)

    executed: list[LayerState] = []
    evidence: list[ChannelLayerRuntimeEvidence] = []
    delayed = set(self._report.delayed_channels)
    uncertain = set(self._report.uncertain_channels)
    policies = self._report.runtime_policies

    for idx, raw_layer in enumerate(raw_layers):
        channel = self._layer_channels[idx]
        policy = policies.get(channel)
        delay_policy = (
            policy.delay_policy if policy else "use_current_tick_evidence"
        )
        uncertainty_policy = (
            policy.uncertainty_policy
            if policy
            else "deterministic_runtime_contribution"
        )
        if channel in delayed and self._last_raw_layers is not None:
            base_layer = self._last_raw_layers[idx]
            evidence_source = "held_previous_tick"
        elif channel in delayed:
            base_layer = raw_layer
            evidence_source = "current_tick_prime"
        else:
            base_layer = raw_layer
            evidence_source = "current_tick"

        confidence = self._confidence_weights.get(channel, 1.0)
        executed_r = (
            base_layer.R * confidence if channel in uncertain else base_layer.R
        )
        executed_layer = replace(base_layer, R=executed_r)
        executed.append(executed_layer)
        evidence.append(
            ChannelLayerRuntimeEvidence(
                layer_index=idx,
                channel=channel,
                raw_R=raw_layer.R,
                executed_R=executed_layer.R,
                raw_psi=raw_layer.psi,
                executed_psi=executed_layer.psi,
                delay_policy=delay_policy,
                uncertainty_policy=uncertainty_policy,
                evidence_source=evidence_source,
                confidence_weight=confidence,
            )
        )

    self._last_raw_layers = tuple(raw_layers)
    return ChannelRuntimeExecution(layers=tuple(executed), evidence=tuple(evidence))

Functions:

Digital-Twin Binding Contract

build_digital_twin_binding_contract() turns a validated BindingSpec into a versioned, bidirectional contract for simulators, services, and hardware twins. The contract is deterministic and transport-neutral: it describes timing, layers, actuators, N-channel algebra, and allowed sync payload classes without opening sockets or applying actuation.

from scpn_phase_orchestrator.binding import (
    build_digital_twin_binding_contract,
    load_binding_spec,
)

spec = load_binding_spec("domainpacks/digital_twin_nchannel/binding_spec.yaml")
contract = build_digital_twin_binding_contract(spec)

payload = contract.to_audit_record()
stable_json = contract.to_json()

The emitted contract_hash is computed over the contract payload before the hash field is added, so replay systems can compare contract compatibility without re-parsing YAML. Default sync capabilities cover state snapshots, phase observations, proposed control actions, and audit replay.

Transport adapters should wrap payloads in DigitalTwinSyncEnvelope and run validate_digital_twin_sync_envelope() before handing data to a runtime or external twin. The validator checks contract-hash compatibility, declared capability names, allowed directions, integer-only non-negative sequence numbers, non-empty string-keyed payloads, and strict JSON-safe finite payload values before serialization. It remains transport-neutral: REST, gRPC, Kafka, file, and hardware adapters can all use the same validation record without this module opening sockets.

from scpn_phase_orchestrator.binding import (
    build_digital_twin_sync_envelope,
    validate_digital_twin_sync_envelope,
)

envelope = build_digital_twin_sync_envelope(
    contract,
    capability="state_snapshot",
    direction="twin_to_spo",
    sequence=1,
    payload={"layer": "machine_cells", "R": 0.91},
)
validation = validate_digital_twin_sync_envelope(contract, envelope)

For file-based replay or adapter smoke tests, the JSONL adapter writes one validated envelope shape per line and reads it back through the same contract gate:

from scpn_phase_orchestrator.binding import (
    read_digital_twin_sync_jsonl,
    write_digital_twin_sync_jsonl,
)

write_report = write_digital_twin_sync_jsonl("sync.jsonl", [envelope])
read_report = read_digital_twin_sync_jsonl(contract, "sync.jsonl")

The read report separates accepted envelope validations from malformed JSON, invalid envelope shapes, and contract-validation rejections. This is the reference behaviour concrete REST, gRPC, Kafka, file, and hardware adapters can mirror.

For runtime-facing tests that should not touch disk, use DigitalTwinSyncMemoryAdapter. It validates submissions against the same contract, queues accepted envelopes in order, and drops rejected envelopes while returning the validation reason to the caller.

from scpn_phase_orchestrator.binding import DigitalTwinSyncMemoryAdapter

adapter = DigitalTwinSyncMemoryAdapter.for_contract(contract)
validation = adapter.submit(envelope)
accepted_batch = adapter.drain()

Adapter implementations can also publish a DigitalTwinAdapterManifest before any runtime code is enabled. build_digital_twin_adapter_manifest() checks that the adapter only claims contract-declared capabilities, that live transports declare authentication, and that offline transports support replay.

from scpn_phase_orchestrator.binding import build_digital_twin_adapter_manifest

compatibility = build_digital_twin_adapter_manifest(
    contract,
    name="grpc-live",
    transport="grpc",
    sync_capabilities=("state_snapshot", "audit_replay"),
    supports_replay=True,
    requires_auth=True,
)

DigitalTwinSyncRestAdapter is the first concrete live boundary. It stays dependency-free and does not open a socket; web frameworks call handle_post() with parsed JSON and request headers, then map the returned HTTP-style status and body to the framework response.

from scpn_phase_orchestrator.binding import DigitalTwinSyncRestAdapter

adapter = DigitalTwinSyncRestAdapter.for_contract(contract)
response = adapter.handle_post(
    envelope.to_audit_record(),
    headers={"authorization": "Bearer ..."},
)
accepted = adapter.drain()

DigitalTwinSyncGrpcAdapter follows the same pattern for decoded unary gRPC requests. It avoids generated protobuf imports in the binding layer; a servicer passes decoded fields and metadata into handle_unary() and maps the returned gRPC-style status name to framework-native status handling.

from scpn_phase_orchestrator.binding import DigitalTwinSyncGrpcAdapter

adapter = DigitalTwinSyncGrpcAdapter.for_contract(contract)
response = adapter.handle_unary(
    envelope.to_audit_record(),
    metadata={"authorization": "Bearer ..."},
)
accepted = adapter.drain()

DigitalTwinSyncKafkaAdapter accepts decoded broker message records. It checks the configured topic, auth header, decoded value envelope, contract hash, and capability direction without importing Kafka clients or committing offsets.

from scpn_phase_orchestrator.binding import DigitalTwinSyncKafkaAdapter

adapter = DigitalTwinSyncKafkaAdapter.for_contract(contract)
response = adapter.handle_message(
    {"topic": "spo.digital_twin.sync", "value": envelope.to_audit_record()},
    headers={"authorization": "Bearer ..."},
)
accepted = adapter.drain()

DigitalTwinSyncHardwareAdapter accepts decoded device frames from a separate hardware integration layer. It requires a registered device ID and explicit safety interlock, and it always reports hardware_write_permitted=False; the binding layer validates and queues envelopes but never writes to physical devices.

from scpn_phase_orchestrator.binding import DigitalTwinSyncHardwareAdapter

adapter = DigitalTwinSyncHardwareAdapter.for_contract(
    contract,
    device_ids=("pynq-loopback-0",),
)
response = adapter.handle_frame(
    {
        "device_id": "pynq-loopback-0",
        "safety_interlock": True,
        "value": envelope.to_audit_record(),
    },
    headers={"authorization": "Bearer ..."},
)
accepted = adapter.drain()

digital_twin

Transport-neutral digital-twin contracts derived from bindings.

This package turns a validated BindingSpec into deterministic contract hashes, adapter manifests, sync capabilities, and envelope validation records for simulators, services, and hardware twins, split into responsibility modules (contract, envelope, evidence, and per-transport adapters) behind a stable re-export surface. REST, gRPC, Kafka, JSONL, hardware, and in-memory helpers validate decoded payloads only; they do not open sockets, spawn servers, or apply live control actions.

Classes

DigitalTwinSyncGrpcAdapter dataclass 

DigitalTwinSyncGrpcAdapter(
    contract: DigitalTwinBindingContract,
    compatibility: DigitalTwinAdapterCompatibility,
    _queue: list[DigitalTwinSyncEnvelope],
)

Dependency-free gRPC boundary for digital-twin sync payloads.

The adapter does not start a gRPC server or import generated protobuf classes. A real servicer can pass decoded protobuf fields into :meth:handle_unary; this boundary then applies the same contract checks as other transports before queuing accepted envelopes.

Methods:
for_contract classmethod 
for_contract(
    contract: DigitalTwinBindingContract,
    *,
    name: str = "grpc-sync",
    sync_capabilities: Sequence[
        str
    ] = _DEFAULT_SYNC_CAPABILITIES,
    requires_auth: bool = True,
    supports_replay: bool = False,
) -> DigitalTwinSyncGrpcAdapter

Create a gRPC adapter boundary for a digital-twin contract.

Parameters

contract : DigitalTwinBindingContract The digital-twin binding contract the adapter serves. name : str, optional Human-readable adapter name. sync_capabilities : Sequence[str], optional Sync capabilities the adapter advertises. requires_auth : bool, optional Whether the adapter boundary requires authentication. supports_replay : bool, optional Whether the adapter supports replay of past envelopes.

Returns

DigitalTwinSyncGrpcAdapter A new gRPC adapter boundary bound to the contract.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_grpc.py 
 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
@classmethod
def for_contract(
    cls,
    contract: DigitalTwinBindingContract,
    *,
    name: str = "grpc-sync",
    sync_capabilities: Sequence[str] = _DEFAULT_SYNC_CAPABILITIES,
    requires_auth: bool = True,
    supports_replay: bool = False,
) -> DigitalTwinSyncGrpcAdapter:
    """Create a gRPC adapter boundary for a digital-twin contract.

    Parameters
    ----------
    contract : DigitalTwinBindingContract
        The digital-twin binding contract the adapter serves.
    name : str, optional
        Human-readable adapter name.
    sync_capabilities : Sequence[str], optional
        Sync capabilities the adapter advertises.
    requires_auth : bool, optional
        Whether the adapter boundary requires authentication.
    supports_replay : bool, optional
        Whether the adapter supports replay of past envelopes.

    Returns
    -------
    DigitalTwinSyncGrpcAdapter
        A new gRPC adapter boundary bound to the contract.
    """
    compatibility = build_digital_twin_adapter_manifest(
        contract,
        name=name,
        transport="grpc",
        sync_capabilities=sync_capabilities,
        supports_replay=supports_replay,
        requires_auth=requires_auth,
        notes="dependency-free gRPC boundary",
    )
    return cls(contract=contract, compatibility=compatibility, _queue=[])
handle_unary 
handle_unary(
    request: Mapping[str, object],
    *,
    metadata: Mapping[str, str] | None = None,
) -> DigitalTwinSyncGrpcResponse

Validate one unary gRPC request and queue accepted envelopes.

Parameters

request : Mapping[str, object] The decoded unary gRPC request body. metadata : Mapping[str, str] or None, optional Optional request metadata (e.g. auth tokens).

Returns

DigitalTwinSyncGrpcResponse The response; accepted envelopes are queued for :meth:drain.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_grpc.py 
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
def handle_unary(
    self,
    request: Mapping[str, object],
    *,
    metadata: Mapping[str, str] | None = None,
) -> DigitalTwinSyncGrpcResponse:
    """Validate one unary gRPC request and queue accepted envelopes.

    Parameters
    ----------
    request : Mapping[str, object]
        The decoded unary gRPC request body.
    metadata : Mapping[str, str] or None, optional
        Optional request metadata (e.g. auth tokens).

    Returns
    -------
    DigitalTwinSyncGrpcResponse
        The response; accepted envelopes are queued for :meth:`drain`.
    """
    if not self.compatibility.compatible:
        return _grpc_response(
            "FAILED_PRECONDITION",
            False,
            "adapter_incompatible",
            {
                "reasons": list(self.compatibility.reasons),
                "contract_hash": self.contract.contract_hash,
            },
        )
    if self.compatibility.manifest.requires_auth and not _has_authorization(
        metadata,
    ):
        return _grpc_response(
            "UNAUTHENTICATED",
            False,
            "auth_required",
            {"contract_hash": self.contract.contract_hash},
        )
    envelope = _envelope_from_record(dict(request))
    if envelope is None:
        return _grpc_response(
            "INVALID_ARGUMENT",
            False,
            "invalid_envelope",
            {"contract_hash": self.contract.contract_hash},
        )
    validation = validate_digital_twin_sync_envelope(self.contract, envelope)
    if not validation.accepted:
        return _grpc_response(
            "FAILED_PRECONDITION",
            False,
            validation.reason,
            {
                "capability": envelope.capability,
                "sequence": envelope.sequence,
                "contract_hash": self.contract.contract_hash,
            },
        )
    self._queue.append(envelope)
    return _grpc_response(
        "OK",
        True,
        "accepted",
        {
            "capability": envelope.capability,
            "sequence": envelope.sequence,
            "contract_hash": self.contract.contract_hash,
        },
    )
drain 
drain() -> tuple[DigitalTwinSyncEnvelope, ...]

Return accepted gRPC envelopes in arrival order and clear the queue.

Returns

tuple[DigitalTwinSyncEnvelope, ...] The queued sync envelopes in submission order; the internal queue is left empty.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_grpc.py 
182
183
184
185
186
187
188
189
190
191
192
193
def drain(self) -> tuple[DigitalTwinSyncEnvelope, ...]:
    """Return accepted gRPC envelopes in arrival order and clear the queue.

    Returns
    -------
    tuple[DigitalTwinSyncEnvelope, ...]
        The queued sync envelopes in submission order; the internal queue is left
        empty.
    """
    drained = tuple(self._queue)
    self._queue.clear()
    return drained
to_audit_record 
to_audit_record() -> dict[str, object]

Return gRPC adapter state without exposing payload contents.

Returns

dict[str, object] Deterministic, JSON-safe state of the DigitalTwinSyncGrpcAdapter (queue counters and status); no network surface or payload contents are exposed.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_grpc.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def to_audit_record(self) -> dict[str, object]:
    """Return gRPC adapter state without exposing payload contents.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe state of the DigitalTwinSyncGrpcAdapter (queue
        counters and status); no network surface or payload contents are exposed.
    """
    return {
        "contract_hash": self.contract.contract_hash,
        "manifest": self.compatibility.manifest.to_audit_record(),
        "compatible": self.compatibility.compatible,
        "queued_count": len(self._queue),
        "queued_sequences": [envelope.sequence for envelope in self._queue],
    }

DigitalTwinSyncGrpcResponse dataclass 

DigitalTwinSyncGrpcResponse(
    status_code: str,
    accepted: bool,
    reason: str,
    message: dict[str, object],
)

gRPC-style response for a digital-twin sync boundary.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe gRPC adapter response.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinSyncGrpcResponse fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_grpc.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe gRPC adapter response.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the DigitalTwinSyncGrpcResponse
        fields.
    """
    return {
        "status_code": self.status_code,
        "accepted": self.accepted,
        "reason": self.reason,
        "message": dict(self.message),
    }

DigitalTwinSyncHardwareAdapter dataclass 

DigitalTwinSyncHardwareAdapter(
    contract: DigitalTwinBindingContract,
    compatibility: DigitalTwinAdapterCompatibility,
    device_ids: tuple[str, ...],
    _queue: list[DigitalTwinSyncEnvelope],
)

No-I/O hardware boundary for digital-twin sync payloads.

The adapter validates decoded frames from a hardware integration layer. It never opens device files, writes registers, toggles GPIO, or applies actuation; accepted envelopes are only queued for caller-controlled review.

Methods:
for_contract classmethod 
for_contract(
    contract: DigitalTwinBindingContract,
    *,
    device_ids: Sequence[str],
    name: str = "hardware-sync",
    sync_capabilities: Sequence[
        str
    ] = _DEFAULT_SYNC_CAPABILITIES,
    requires_auth: bool = True,
    supports_replay: bool = True,
) -> DigitalTwinSyncHardwareAdapter

Create a no-I/O hardware boundary for a digital-twin contract.

Parameters

contract : DigitalTwinBindingContract The digital-twin binding contract the adapter serves. device_ids : Sequence[str] Identifiers of the hardware devices the boundary serves. name : str, optional Human-readable adapter name. sync_capabilities : Sequence[str], optional Sync capabilities the adapter advertises. requires_auth : bool, optional Whether the adapter boundary requires authentication. supports_replay : bool, optional Whether the adapter supports replay of past envelopes.

Returns

DigitalTwinSyncHardwareAdapter A new no-I/O hardware boundary bound to the contract.

Raises

ValueError If device_ids is empty.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_hardware.py 
 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
@classmethod
def for_contract(
    cls,
    contract: DigitalTwinBindingContract,
    *,
    device_ids: Sequence[str],
    name: str = "hardware-sync",
    sync_capabilities: Sequence[str] = _DEFAULT_SYNC_CAPABILITIES,
    requires_auth: bool = True,
    supports_replay: bool = True,
) -> DigitalTwinSyncHardwareAdapter:
    """Create a no-I/O hardware boundary for a digital-twin contract.

    Parameters
    ----------
    contract : DigitalTwinBindingContract
        The digital-twin binding contract the adapter serves.
    device_ids : Sequence[str]
        Identifiers of the hardware devices the boundary serves.
    name : str, optional
        Human-readable adapter name.
    sync_capabilities : Sequence[str], optional
        Sync capabilities the adapter advertises.
    requires_auth : bool, optional
        Whether the adapter boundary requires authentication.
    supports_replay : bool, optional
        Whether the adapter supports replay of past envelopes.

    Returns
    -------
    DigitalTwinSyncHardwareAdapter
        A new no-I/O hardware boundary bound to the contract.

    Raises
    ------
    ValueError
        If ``device_ids`` is empty.
    """
    if not device_ids:
        raise ValueError("hardware device_ids must not be empty")
    checked_device_ids = tuple(device_ids)
    for device_id in checked_device_ids:
        _require_non_empty(device_id, "hardware device_id")
    compatibility = build_digital_twin_adapter_manifest(
        contract,
        name=name,
        transport="hardware",
        sync_capabilities=sync_capabilities,
        supports_replay=supports_replay,
        requires_auth=requires_auth,
        notes="no-I/O hardware boundary",
    )
    return cls(
        contract=contract,
        compatibility=compatibility,
        device_ids=checked_device_ids,
        _queue=[],
    )
handle_frame 
handle_frame(
    frame: Mapping[str, object],
    *,
    headers: Mapping[str, str] | None = None,
) -> DigitalTwinSyncHardwareResponse

Validate one decoded hardware frame and queue accepted envelopes.

Parameters

frame : Mapping[str, object] The decoded hardware hardware frame. headers : Mapping[str, str] or None, optional Optional transport headers (e.g. auth tokens).

Returns

DigitalTwinSyncHardwareResponse The hardware response; accepted envelopes are queued for :meth:drain.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_hardware.py 
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
def handle_frame(
    self,
    frame: Mapping[str, object],
    *,
    headers: Mapping[str, str] | None = None,
) -> DigitalTwinSyncHardwareResponse:
    """Validate one decoded hardware frame and queue accepted envelopes.

    Parameters
    ----------
    frame : Mapping[str, object]
        The decoded hardware hardware frame.
    headers : Mapping[str, str] or None, optional
        Optional transport headers (e.g. auth tokens).

    Returns
    -------
    DigitalTwinSyncHardwareResponse
        The hardware response; accepted envelopes are queued for :meth:`drain`.
    """
    device_id = frame.get("device_id")
    if not isinstance(device_id, str) or device_id not in self.device_ids:
        return _hardware_response(
            False,
            "device_not_registered",
            {"device_id": device_id, "registered_devices": list(self.device_ids)},
        )
    if frame.get("safety_interlock") is not True:
        return _hardware_response(
            False,
            "safety_interlock_required",
            {"device_id": device_id},
        )
    if not self.compatibility.compatible:
        return _hardware_response(
            False,
            "adapter_incompatible",
            {
                "device_id": device_id,
                "reasons": list(self.compatibility.reasons),
                "contract_hash": self.contract.contract_hash,
            },
        )
    if self.compatibility.manifest.requires_auth and not _has_authorization(
        headers,
    ):
        return _hardware_response(
            False,
            "auth_required",
            {"device_id": device_id, "contract_hash": self.contract.contract_hash},
        )
    value = frame.get("value")
    if not isinstance(value, Mapping):
        return _hardware_response(
            False,
            "invalid_frame_value",
            {"device_id": device_id, "contract_hash": self.contract.contract_hash},
        )
    envelope = _envelope_from_record(dict(value))
    if envelope is None:
        return _hardware_response(
            False,
            "invalid_envelope",
            {"device_id": device_id, "contract_hash": self.contract.contract_hash},
        )
    validation = validate_digital_twin_sync_envelope(self.contract, envelope)
    if not validation.accepted:
        return _hardware_response(
            False,
            validation.reason,
            {
                "device_id": device_id,
                "capability": envelope.capability,
                "sequence": envelope.sequence,
                "contract_hash": self.contract.contract_hash,
            },
        )
    self._queue.append(envelope)
    return _hardware_response(
        True,
        "accepted",
        {
            "device_id": device_id,
            "capability": envelope.capability,
            "sequence": envelope.sequence,
            "contract_hash": self.contract.contract_hash,
        },
    )
drain 
drain() -> tuple[DigitalTwinSyncEnvelope, ...]

Return accepted hardware envelopes in arrival order and clear the queue.

Returns

tuple[DigitalTwinSyncEnvelope, ...] The queued sync envelopes in submission order; the internal queue is left empty.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_hardware.py 
218
219
220
221
222
223
224
225
226
227
228
229
def drain(self) -> tuple[DigitalTwinSyncEnvelope, ...]:
    """Return accepted hardware envelopes in arrival order and clear the queue.

    Returns
    -------
    tuple[DigitalTwinSyncEnvelope, ...]
        The queued sync envelopes in submission order; the internal queue is left
        empty.
    """
    drained = tuple(self._queue)
    self._queue.clear()
    return drained
to_audit_record 
to_audit_record() -> dict[str, object]

Return hardware adapter state without exposing payload contents.

Returns

dict[str, object] Deterministic, JSON-safe state of the DigitalTwinSyncHardwareAdapter (queue counters and status); no network surface or payload contents are exposed.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_hardware.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def to_audit_record(self) -> dict[str, object]:
    """Return hardware adapter state without exposing payload contents.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe state of the DigitalTwinSyncHardwareAdapter (queue
        counters and status); no network surface or payload contents are exposed.
    """
    return {
        "contract_hash": self.contract.contract_hash,
        "manifest": self.compatibility.manifest.to_audit_record(),
        "compatible": self.compatibility.compatible,
        "device_ids": list(self.device_ids),
        "queued_count": len(self._queue),
        "queued_sequences": [envelope.sequence for envelope in self._queue],
        "hardware_write_permitted": False,
    }

DigitalTwinSyncHardwareResponse dataclass 

DigitalTwinSyncHardwareResponse(
    accepted: bool,
    reason: str,
    hardware_write_permitted: bool,
    frame: dict[str, object],
)

No-I/O response for a hardware digital-twin sync boundary.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe hardware adapter response.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinSyncHardwareResponse fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_hardware.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe hardware adapter response.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the
        DigitalTwinSyncHardwareResponse fields.
    """
    return {
        "accepted": self.accepted,
        "reason": self.reason,
        "hardware_write_permitted": self.hardware_write_permitted,
        "frame": dict(self.frame),
    }

DigitalTwinSyncKafkaAdapter dataclass 

DigitalTwinSyncKafkaAdapter(
    contract: DigitalTwinBindingContract,
    compatibility: DigitalTwinAdapterCompatibility,
    topic: str,
    _queue: list[DigitalTwinSyncEnvelope],
)

Dependency-free Kafka boundary for digital-twin sync payloads.

The adapter expects a broker consumer to pass a decoded message dictionary. It does not import Kafka clients, open sockets, or commit offsets. Accepted envelopes are queued for caller-controlled runtime handoff.

Methods:
for_contract classmethod 
for_contract(
    contract: DigitalTwinBindingContract,
    *,
    topic: str = "spo.digital_twin.sync",
    name: str = "kafka-sync",
    sync_capabilities: Sequence[
        str
    ] = _DEFAULT_SYNC_CAPABILITIES,
    requires_auth: bool = True,
    supports_replay: bool = True,
) -> DigitalTwinSyncKafkaAdapter

Create a Kafka message-boundary adapter for a digital-twin contract.

Parameters

contract : DigitalTwinBindingContract The digital-twin binding contract the adapter serves. topic : str, optional Kafka topic the adapter binds to. name : str, optional Human-readable adapter name. sync_capabilities : Sequence[str], optional Sync capabilities the adapter advertises. requires_auth : bool, optional Whether the adapter boundary requires authentication. supports_replay : bool, optional Whether the adapter supports replay of past envelopes.

Returns

DigitalTwinSyncKafkaAdapter A new Kafka message-boundary adapter bound to the contract.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_kafka.py 
 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
@classmethod
def for_contract(
    cls,
    contract: DigitalTwinBindingContract,
    *,
    topic: str = "spo.digital_twin.sync",
    name: str = "kafka-sync",
    sync_capabilities: Sequence[str] = _DEFAULT_SYNC_CAPABILITIES,
    requires_auth: bool = True,
    supports_replay: bool = True,
) -> DigitalTwinSyncKafkaAdapter:
    """Create a Kafka message-boundary adapter for a digital-twin contract.

    Parameters
    ----------
    contract : DigitalTwinBindingContract
        The digital-twin binding contract the adapter serves.
    topic : str, optional
        Kafka topic the adapter binds to.
    name : str, optional
        Human-readable adapter name.
    sync_capabilities : Sequence[str], optional
        Sync capabilities the adapter advertises.
    requires_auth : bool, optional
        Whether the adapter boundary requires authentication.
    supports_replay : bool, optional
        Whether the adapter supports replay of past envelopes.

    Returns
    -------
    DigitalTwinSyncKafkaAdapter
        A new Kafka message-boundary adapter bound to the contract.
    """
    _require_non_empty(topic, "kafka topic")
    compatibility = build_digital_twin_adapter_manifest(
        contract,
        name=name,
        transport="kafka",
        sync_capabilities=sync_capabilities,
        supports_replay=supports_replay,
        requires_auth=requires_auth,
        notes="dependency-free Kafka boundary",
    )
    return cls(
        contract=contract,
        compatibility=compatibility,
        topic=topic,
        _queue=[],
    )
handle_message 
handle_message(
    message: Mapping[str, object],
    *,
    headers: Mapping[str, str] | None = None,
) -> DigitalTwinSyncKafkaResponse

Validate one decoded Kafka message and queue accepted envelopes.

Parameters

message : Mapping[str, object] The decoded Kafka message body. headers : Mapping[str, str] or None, optional Optional transport headers (e.g. auth tokens).

Returns

DigitalTwinSyncKafkaResponse The Kafka response; accepted envelopes are queued for :meth:drain.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_kafka.py 
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
def handle_message(
    self,
    message: Mapping[str, object],
    *,
    headers: Mapping[str, str] | None = None,
) -> DigitalTwinSyncKafkaResponse:
    """Validate one decoded Kafka message and queue accepted envelopes.

    Parameters
    ----------
    message : Mapping[str, object]
        The decoded Kafka message body.
    headers : Mapping[str, str] or None, optional
        Optional transport headers (e.g. auth tokens).

    Returns
    -------
    DigitalTwinSyncKafkaResponse
        The Kafka response; accepted envelopes are queued for :meth:`drain`.
    """
    message_topic = message.get("topic", self.topic)
    if not isinstance(message_topic, str) or message_topic != self.topic:
        return _kafka_response(
            False,
            "topic_mismatch",
            False,
            {"expected_topic": self.topic, "observed_topic": message_topic},
        )
    if not self.compatibility.compatible:
        return _kafka_response(
            False,
            "adapter_incompatible",
            True,
            {
                "reasons": list(self.compatibility.reasons),
                "contract_hash": self.contract.contract_hash,
            },
        )
    if self.compatibility.manifest.requires_auth and not _has_authorization(
        headers,
    ):
        return _kafka_response(
            False,
            "auth_required",
            True,
            {"contract_hash": self.contract.contract_hash},
        )
    value = message.get("value")
    if not isinstance(value, Mapping):
        return _kafka_response(
            False,
            "invalid_message_value",
            False,
            {"contract_hash": self.contract.contract_hash},
        )
    envelope = _envelope_from_record(dict(value))
    if envelope is None:
        return _kafka_response(
            False,
            "invalid_envelope",
            False,
            {"contract_hash": self.contract.contract_hash},
        )
    validation = validate_digital_twin_sync_envelope(self.contract, envelope)
    if not validation.accepted:
        return _kafka_response(
            False,
            validation.reason,
            False,
            {
                "capability": envelope.capability,
                "sequence": envelope.sequence,
                "contract_hash": self.contract.contract_hash,
            },
        )
    self._queue.append(envelope)
    return _kafka_response(
        True,
        "accepted",
        False,
        {
            "topic": self.topic,
            "capability": envelope.capability,
            "sequence": envelope.sequence,
            "contract_hash": self.contract.contract_hash,
        },
    )
drain 
drain() -> tuple[DigitalTwinSyncEnvelope, ...]

Return accepted Kafka envelopes in arrival order and clear the queue.

Returns

tuple[DigitalTwinSyncEnvelope, ...] The queued sync envelopes in submission order; the internal queue is left empty.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_kafka.py 
208
209
210
211
212
213
214
215
216
217
218
219
def drain(self) -> tuple[DigitalTwinSyncEnvelope, ...]:
    """Return accepted Kafka envelopes in arrival order and clear the queue.

    Returns
    -------
    tuple[DigitalTwinSyncEnvelope, ...]
        The queued sync envelopes in submission order; the internal queue is left
        empty.
    """
    drained = tuple(self._queue)
    self._queue.clear()
    return drained
to_audit_record 
to_audit_record() -> dict[str, object]

Return Kafka adapter state without exposing payload contents.

Returns

dict[str, object] Deterministic, JSON-safe state of the DigitalTwinSyncKafkaAdapter (queue counters and status); no network surface or payload contents are exposed.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_kafka.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def to_audit_record(self) -> dict[str, object]:
    """Return Kafka adapter state without exposing payload contents.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe state of the DigitalTwinSyncKafkaAdapter (queue
        counters and status); no network surface or payload contents are exposed.
    """
    return {
        "contract_hash": self.contract.contract_hash,
        "manifest": self.compatibility.manifest.to_audit_record(),
        "compatible": self.compatibility.compatible,
        "topic": self.topic,
        "queued_count": len(self._queue),
        "queued_sequences": [envelope.sequence for envelope in self._queue],
    }

DigitalTwinSyncKafkaResponse dataclass 

DigitalTwinSyncKafkaResponse(
    accepted: bool,
    reason: str,
    retryable: bool,
    message: dict[str, object],
)

Broker-style response for a Kafka digital-twin sync boundary.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe Kafka adapter response.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinSyncKafkaResponse fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_kafka.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe Kafka adapter response.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the DigitalTwinSyncKafkaResponse
        fields.
    """
    return {
        "accepted": self.accepted,
        "reason": self.reason,
        "retryable": self.retryable,
        "message": dict(self.message),
    }

DigitalTwinSyncMemoryAdapter dataclass 

DigitalTwinSyncMemoryAdapter(
    contract: DigitalTwinBindingContract,
    _queue: list[DigitalTwinSyncEnvelope],
)

In-memory reference adapter for validated digital-twin sync payloads.

Methods:
for_contract classmethod 
for_contract(
    contract: DigitalTwinBindingContract,
) -> DigitalTwinSyncMemoryAdapter

Create an empty adapter for a digital-twin binding contract.

Parameters

contract : DigitalTwinBindingContract The digital-twin binding contract the adapter serves.

Returns

DigitalTwinSyncMemoryAdapter An empty in-memory adapter bound to the contract.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_memory.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@classmethod
def for_contract(
    cls,
    contract: DigitalTwinBindingContract,
) -> DigitalTwinSyncMemoryAdapter:
    """Create an empty adapter for a digital-twin binding contract.

    Parameters
    ----------
    contract : DigitalTwinBindingContract
        The digital-twin binding contract the adapter serves.

    Returns
    -------
    DigitalTwinSyncMemoryAdapter
        An empty in-memory adapter bound to the contract.
    """
    return cls(contract=contract, _queue=[])
submit 
submit(
    envelope: DigitalTwinSyncEnvelope,
) -> DigitalTwinTransportValidation

Validate and queue one envelope when accepted.

Parameters

envelope : DigitalTwinSyncEnvelope The sync envelope to validate and queue.

Returns

DigitalTwinTransportValidation The validation result; the envelope is queued only when accepted.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_memory.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def submit(
    self,
    envelope: DigitalTwinSyncEnvelope,
) -> DigitalTwinTransportValidation:
    """Validate and queue one envelope when accepted.

    Parameters
    ----------
    envelope : DigitalTwinSyncEnvelope
        The sync envelope to validate and queue.

    Returns
    -------
    DigitalTwinTransportValidation
        The validation result; the envelope is queued only when accepted.
    """
    validation = validate_digital_twin_sync_envelope(self.contract, envelope)
    if validation.accepted:
        self._queue.append(envelope)
    return validation
drain 
drain() -> tuple[DigitalTwinSyncEnvelope, ...]

Return queued envelopes in submission order and clear the queue.

Returns

tuple[DigitalTwinSyncEnvelope, ...] The queued sync envelopes in submission order; the internal queue is left empty.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_memory.py 
70
71
72
73
74
75
76
77
78
79
80
81
def drain(self) -> tuple[DigitalTwinSyncEnvelope, ...]:
    """Return queued envelopes in submission order and clear the queue.

    Returns
    -------
    tuple[DigitalTwinSyncEnvelope, ...]
        The queued sync envelopes in submission order; the internal queue is left
        empty.
    """
    drained = tuple(self._queue)
    self._queue.clear()
    return drained
to_audit_record 
to_audit_record() -> dict[str, object]

Return adapter state without exposing any network surface.

Returns

dict[str, object] Deterministic, JSON-safe state of the DigitalTwinSyncMemoryAdapter (queue counters and status); no network surface or payload contents are exposed.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_memory.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def to_audit_record(self) -> dict[str, object]:
    """Return adapter state without exposing any network surface.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe state of the DigitalTwinSyncMemoryAdapter (queue
        counters and status); no network surface or payload contents are exposed.
    """
    return {
        "contract_hash": self.contract.contract_hash,
        "queued_count": len(self._queue),
        "queued_sequences": [envelope.sequence for envelope in self._queue],
    }

DigitalTwinSyncRestAdapter dataclass 

DigitalTwinSyncRestAdapter(
    contract: DigitalTwinBindingContract,
    compatibility: DigitalTwinAdapterCompatibility,
    _queue: list[DigitalTwinSyncEnvelope],
)

Dependency-free REST boundary for digital-twin sync payloads.

The adapter deliberately does not open sockets. Web frameworks can call :meth:handle_post from a route handler after parsing request JSON and headers; the adapter then enforces manifest compatibility, authentication posture, envelope shape, and contract validation before queuing payloads.

Methods:
for_contract classmethod 
for_contract(
    contract: DigitalTwinBindingContract,
    *,
    name: str = "rest-sync",
    sync_capabilities: Sequence[
        str
    ] = _DEFAULT_SYNC_CAPABILITIES,
    requires_auth: bool = True,
    supports_replay: bool = False,
) -> DigitalTwinSyncRestAdapter

Create a REST adapter boundary for a digital-twin contract.

Parameters

contract : DigitalTwinBindingContract The digital-twin binding contract the adapter serves. name : str, optional Human-readable adapter name. sync_capabilities : Sequence[str], optional Sync capabilities the adapter advertises. requires_auth : bool, optional Whether the adapter boundary requires authentication. supports_replay : bool, optional Whether the adapter supports replay of past envelopes.

Returns

DigitalTwinSyncRestAdapter A new REST adapter boundary bound to the contract.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_rest.py 
 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
@classmethod
def for_contract(
    cls,
    contract: DigitalTwinBindingContract,
    *,
    name: str = "rest-sync",
    sync_capabilities: Sequence[str] = _DEFAULT_SYNC_CAPABILITIES,
    requires_auth: bool = True,
    supports_replay: bool = False,
) -> DigitalTwinSyncRestAdapter:
    """Create a REST adapter boundary for a digital-twin contract.

    Parameters
    ----------
    contract : DigitalTwinBindingContract
        The digital-twin binding contract the adapter serves.
    name : str, optional
        Human-readable adapter name.
    sync_capabilities : Sequence[str], optional
        Sync capabilities the adapter advertises.
    requires_auth : bool, optional
        Whether the adapter boundary requires authentication.
    supports_replay : bool, optional
        Whether the adapter supports replay of past envelopes.

    Returns
    -------
    DigitalTwinSyncRestAdapter
        A new REST adapter boundary bound to the contract.
    """
    compatibility = build_digital_twin_adapter_manifest(
        contract,
        name=name,
        transport="rest",
        sync_capabilities=sync_capabilities,
        supports_replay=supports_replay,
        requires_auth=requires_auth,
        notes="dependency-free REST boundary",
    )
    return cls(contract=contract, compatibility=compatibility, _queue=[])
handle_post 
handle_post(
    body: Mapping[str, object],
    *,
    headers: Mapping[str, str] | None = None,
) -> DigitalTwinSyncRestResponse

Validate one HTTP POST body and queue accepted sync envelopes.

Parameters

body : Mapping[str, object] The decoded REST HTTP POST body. headers : Mapping[str, str] or None, optional Optional transport headers (e.g. auth tokens).

Returns

DigitalTwinSyncRestResponse The REST response; accepted envelopes are queued for :meth:drain.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_rest.py 
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
def handle_post(
    self,
    body: Mapping[str, object],
    *,
    headers: Mapping[str, str] | None = None,
) -> DigitalTwinSyncRestResponse:
    """Validate one HTTP POST body and queue accepted sync envelopes.

    Parameters
    ----------
    body : Mapping[str, object]
        The decoded REST HTTP POST body.
    headers : Mapping[str, str] or None, optional
        Optional transport headers (e.g. auth tokens).

    Returns
    -------
    DigitalTwinSyncRestResponse
        The REST response; accepted envelopes are queued for :meth:`drain`.
    """
    if not self.compatibility.compatible:
        return _rest_response(
            503,
            False,
            "adapter_incompatible",
            {
                "reasons": list(self.compatibility.reasons),
                "contract_hash": self.contract.contract_hash,
            },
        )
    if self.compatibility.manifest.requires_auth and not _has_authorization(
        headers,
    ):
        return _rest_response(
            401,
            False,
            "auth_required",
            {"contract_hash": self.contract.contract_hash},
        )
    envelope = _envelope_from_record(dict(body))
    if envelope is None:
        return _rest_response(
            400,
            False,
            "invalid_envelope",
            {"contract_hash": self.contract.contract_hash},
        )
    validation = validate_digital_twin_sync_envelope(self.contract, envelope)
    if not validation.accepted:
        return _rest_response(
            422,
            False,
            validation.reason,
            {
                "capability": envelope.capability,
                "sequence": envelope.sequence,
                "contract_hash": self.contract.contract_hash,
            },
        )
    self._queue.append(envelope)
    return _rest_response(
        202,
        True,
        "accepted",
        {
            "capability": envelope.capability,
            "sequence": envelope.sequence,
            "contract_hash": self.contract.contract_hash,
        },
    )
drain 
drain() -> tuple[DigitalTwinSyncEnvelope, ...]

Return accepted REST envelopes in arrival order and clear the queue.

Returns

tuple[DigitalTwinSyncEnvelope, ...] The queued sync envelopes in submission order; the internal queue is left empty.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_rest.py 
182
183
184
185
186
187
188
189
190
191
192
193
def drain(self) -> tuple[DigitalTwinSyncEnvelope, ...]:
    """Return accepted REST envelopes in arrival order and clear the queue.

    Returns
    -------
    tuple[DigitalTwinSyncEnvelope, ...]
        The queued sync envelopes in submission order; the internal queue is left
        empty.
    """
    drained = tuple(self._queue)
    self._queue.clear()
    return drained
to_audit_record 
to_audit_record() -> dict[str, object]

Return REST adapter state without exposing payload contents.

Returns

dict[str, object] Deterministic, JSON-safe state of the DigitalTwinSyncRestAdapter (queue counters and status); no network surface or payload contents are exposed.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_rest.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def to_audit_record(self) -> dict[str, object]:
    """Return REST adapter state without exposing payload contents.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe state of the DigitalTwinSyncRestAdapter (queue
        counters and status); no network surface or payload contents are exposed.
    """
    return {
        "contract_hash": self.contract.contract_hash,
        "manifest": self.compatibility.manifest.to_audit_record(),
        "compatible": self.compatibility.compatible,
        "queued_count": len(self._queue),
        "queued_sequences": [envelope.sequence for envelope in self._queue],
    }

DigitalTwinSyncRestResponse dataclass 

DigitalTwinSyncRestResponse(
    status_code: int,
    accepted: bool,
    reason: str,
    body: dict[str, object],
)

HTTP-style response for a REST digital-twin sync boundary.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe REST adapter response.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinSyncRestResponse fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/adapter_rest.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe REST adapter response.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the DigitalTwinSyncRestResponse
        fields.
    """
    return {
        "status_code": self.status_code,
        "accepted": self.accepted,
        "reason": self.reason,
        "body": dict(self.body),
    }

DigitalTwinAdapterCompatibility dataclass 

DigitalTwinAdapterCompatibility(
    compatible: bool,
    reasons: tuple[str, ...],
    manifest: DigitalTwinAdapterManifest,
    contract_hash: str,
)

Compatibility result for an adapter manifest and binding contract.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe adapter compatibility report.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinAdapterCompatibility fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/contract.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe adapter compatibility report.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the
        DigitalTwinAdapterCompatibility fields.
    """
    return {
        "compatible": self.compatible,
        "reasons": list(self.reasons),
        "manifest": self.manifest.to_audit_record(),
        "contract_hash": self.contract_hash,
    }

DigitalTwinAdapterManifest dataclass 

DigitalTwinAdapterManifest(
    name: str,
    transport: str,
    sync_capabilities: tuple[str, ...],
    supports_replay: bool,
    requires_auth: bool,
    notes: str = "",
)

Reviewable manifest for a concrete digital-twin transport adapter.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe adapter manifest.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinAdapterManifest fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/contract.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe adapter manifest.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the DigitalTwinAdapterManifest
        fields.
    """
    return {
        "name": self.name,
        "transport": self.transport,
        "sync_capabilities": list(self.sync_capabilities),
        "supports_replay": self.supports_replay,
        "requires_auth": self.requires_auth,
        "notes": self.notes,
    }

DigitalTwinBindingContract dataclass 

DigitalTwinBindingContract(
    contract_version: str,
    binding_name: str,
    binding_version: str,
    safety_tier: str,
    sample_period_s: float,
    control_period_s: float,
    layers: tuple[DigitalTwinLayerContract, ...],
    actuators: tuple[dict[str, object], ...],
    channel_algebra: ChannelAlgebraReport,
    sync_capabilities: tuple[
        DigitalTwinSyncCapability, ...
    ],
    contract_hash: str,
)

Versioned bidirectional contract derived from a binding spec.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a deterministic JSON-safe digital-twin contract.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinBindingContract fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/contract.py 
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
def to_audit_record(self) -> dict[str, object]:
    """Return a deterministic JSON-safe digital-twin contract.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the DigitalTwinBindingContract
        fields.
    """
    return {
        "contract_version": self.contract_version,
        "binding": {
            "name": self.binding_name,
            "version": self.binding_version,
            "safety_tier": self.safety_tier,
        },
        "timing": {
            "sample_period_s": self.sample_period_s,
            "control_period_s": self.control_period_s,
        },
        "layers": [layer.to_audit_record() for layer in self.layers],
        "actuators": list(self.actuators),
        "channel_algebra": self.channel_algebra.to_audit_record(),
        "sync_capabilities": [
            capability.to_audit_record() for capability in self.sync_capabilities
        ],
        "contract_hash": self.contract_hash,
    }
to_json 
to_json() -> str

Serialise the contract with deterministic key ordering.

Returns

str The contract serialised as a JSON string with deterministically sorted keys.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/contract.py 
207
208
209
210
211
212
213
214
215
def to_json(self) -> str:
    """Serialise the contract with deterministic key ordering.

    Returns
    -------
    str
        The contract serialised as a JSON string with deterministically sorted keys.
    """
    return json.dumps(self.to_audit_record(), sort_keys=True, separators=(",", ":"))

DigitalTwinLayerContract dataclass 

DigitalTwinLayerContract(
    name: str,
    index: int,
    oscillator_count: int,
    oscillator_ids: tuple[str, ...],
    family: str | None = None,
)

Reduced layer contract exposed to simulators and hardware twins.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe layer contract.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinLayerContract fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/contract.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe layer contract.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the DigitalTwinLayerContract
        fields.
    """
    return {
        "name": self.name,
        "index": self.index,
        "oscillator_count": self.oscillator_count,
        "oscillator_ids": list(self.oscillator_ids),
        "family": self.family,
    }

DigitalTwinSyncCapability dataclass 

DigitalTwinSyncCapability(
    name: str, direction: str, payload: str
)

Named live-sync capability declared by the binding contract.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe sync capability contract.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinSyncCapability fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/contract.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe sync capability contract.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the DigitalTwinSyncCapability
        fields.
    """
    return {
        "name": self.name,
        "direction": self.direction,
        "payload": self.payload,
    }

DigitalTwinSyncEnvelope dataclass 

DigitalTwinSyncEnvelope(
    contract_hash: str,
    capability: str,
    direction: str,
    sequence: int,
    payload: dict[str, object],
)

Transport-neutral live-sync payload envelope for digital twins.

Methods:
__post_init__ 
__post_init__() -> None

Validate envelope identity, sequence, and payload invariants.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/envelope.py 
32
33
34
35
36
37
38
39
40
41
def __post_init__(self) -> None:
    """Validate envelope identity, sequence, and payload invariants."""
    _require_non_empty(self.contract_hash, "contract_hash")
    _require_non_empty(self.capability, "capability")
    _require_non_empty(self.direction, "direction")
    if not isinstance(self.sequence, int) or isinstance(self.sequence, bool):
        raise ValueError("sequence must be a non-negative integer")
    if self.sequence < 0:
        raise ValueError("sequence must be >= 0")
    _validate_payload(self.payload)
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe sync envelope.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinSyncEnvelope fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/envelope.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe sync envelope.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the DigitalTwinSyncEnvelope
        fields.
    """
    return {
        "contract_hash": self.contract_hash,
        "capability": self.capability,
        "direction": self.direction,
        "sequence": self.sequence,
        "payload": dict(self.payload),
    }
to_json 
to_json() -> str

Serialise the envelope with deterministic key ordering.

Returns

str The envelope serialised as a JSON string with deterministically sorted keys.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/envelope.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def to_json(self) -> str:
    """Serialise the envelope with deterministic key ordering.

    Returns
    -------
    str
        The envelope serialised as a JSON string with deterministically sorted keys.
    """
    return json.dumps(
        self.to_audit_record(),
        allow_nan=False,
        sort_keys=True,
        separators=(",", ":"),
    )

DigitalTwinSyncJsonlReport dataclass 

DigitalTwinSyncJsonlReport(
    path: str,
    written: int,
    accepted: tuple[DigitalTwinTransportValidation, ...],
    rejected: tuple[dict[str, object], ...],
)

JSONL file-adapter replay report for digital-twin sync envelopes.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe file-adapter report.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinSyncJsonlReport fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/envelope.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe file-adapter report.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the DigitalTwinSyncJsonlReport
        fields.
    """
    return {
        "path": self.path,
        "written": self.written,
        "accepted_count": len(self.accepted),
        "rejected_count": len(self.rejected),
        "accepted": [validation.to_audit_record() for validation in self.accepted],
        "rejected": list(self.rejected),
    }

DigitalTwinTransportValidation dataclass 

DigitalTwinTransportValidation(
    accepted: bool,
    reason: str,
    envelope: DigitalTwinSyncEnvelope,
)

Validation result for one digital-twin sync envelope.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe validation record.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinTransportValidation fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/envelope.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe validation record.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the DigitalTwinTransportValidation
        fields.
    """
    return {
        "accepted": self.accepted,
        "reason": self.reason,
        "envelope": self.envelope.to_audit_record(),
    }

DigitalTwinOperatorEvidence dataclass 

DigitalTwinOperatorEvidence(
    contract_hash: str,
    accepted_count: int,
    rejected_count: int,
    adapter_count: int,
    unhealthy_adapter_count: int,
    latest_sequence: int | None,
    capability_counts: dict[str, int],
    direction_counts: dict[str, int],
    max_abs_twin_residual: float | None,
    mismatch_reasons: tuple[str, ...],
    status: str,
)

Transport-neutral operator summary for live or replayed twin sync.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe operator evidence record.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the DigitalTwinOperatorEvidence fields.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/evidence.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe operator evidence record.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the DigitalTwinOperatorEvidence
        fields.
    """
    return {
        "contract_hash": self.contract_hash,
        "accepted_count": self.accepted_count,
        "rejected_count": self.rejected_count,
        "adapter_count": self.adapter_count,
        "unhealthy_adapter_count": self.unhealthy_adapter_count,
        "latest_sequence": self.latest_sequence,
        "capability_counts": dict(sorted(self.capability_counts.items())),
        "direction_counts": dict(sorted(self.direction_counts.items())),
        "max_abs_twin_residual": self.max_abs_twin_residual,
        "mismatch_reasons": list(self.mismatch_reasons),
        "status": self.status,
    }

Functions:

build_digital_twin_adapter_manifest 

build_digital_twin_adapter_manifest(
    contract: DigitalTwinBindingContract,
    *,
    name: str,
    transport: str,
    sync_capabilities: Sequence[str],
    supports_replay: bool,
    requires_auth: bool,
    notes: str = "",
) -> DigitalTwinAdapterCompatibility

Build and validate a transport-adapter manifest against a contract.

Parameters

contract : DigitalTwinBindingContract The contract the adapter must satisfy. name : str Adapter name. transport : str Transport identifier (e.g. rest, grpc, kafka). sync_capabilities : Sequence[str] Capabilities the adapter implements. supports_replay : bool Whether the adapter supports replay. requires_auth : bool Whether the adapter requires authentication. notes : str, optional Free-form manifest notes.

Returns

DigitalTwinAdapterCompatibility The adapter compatibility report against the contract.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/contract.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
def build_digital_twin_adapter_manifest(
    contract: DigitalTwinBindingContract,
    *,
    name: str,
    transport: str,
    sync_capabilities: Sequence[str],
    supports_replay: bool,
    requires_auth: bool,
    notes: str = "",
) -> DigitalTwinAdapterCompatibility:
    """Build and validate a transport-adapter manifest against a contract.

    Parameters
    ----------
    contract : DigitalTwinBindingContract
        The contract the adapter must satisfy.
    name : str
        Adapter name.
    transport : str
        Transport identifier (e.g. ``rest``, ``grpc``, ``kafka``).
    sync_capabilities : Sequence[str]
        Capabilities the adapter implements.
    supports_replay : bool
        Whether the adapter supports replay.
    requires_auth : bool
        Whether the adapter requires authentication.
    notes : str, optional
        Free-form manifest notes.

    Returns
    -------
    DigitalTwinAdapterCompatibility
        The adapter compatibility report against the contract.
    """
    manifest = DigitalTwinAdapterManifest(
        name=name,
        transport=transport,
        sync_capabilities=tuple(sync_capabilities),
        supports_replay=supports_replay,
        requires_auth=requires_auth,
        notes=notes,
    )
    declared = {capability.name for capability in contract.sync_capabilities}
    reasons: list[str] = []
    missing = sorted(set(manifest.sync_capabilities) - declared)
    if missing:
        reasons.append(f"capability_not_declared:{','.join(missing)}")
    if (
        manifest.transport in {"rest", "grpc", "kafka", "hardware"}
        and not requires_auth
    ):
        reasons.append("live_transport_requires_auth")
    if manifest.transport in {"jsonl", "memory"} and not supports_replay:
        reasons.append("offline_transport_requires_replay")
    return DigitalTwinAdapterCompatibility(
        compatible=not reasons,
        reasons=tuple(reasons),
        manifest=manifest,
        contract_hash=contract.contract_hash,
    )

build_digital_twin_binding_contract 

build_digital_twin_binding_contract(
    spec: BindingSpec,
    *,
    contract_version: str = _DEFAULT_CONTRACT_VERSION,
    sync_capabilities: Sequence[
        str
    ] = _DEFAULT_SYNC_CAPABILITIES,
) -> DigitalTwinBindingContract

Build a versioned live-sync contract from a validated binding spec.

The contract is read-only and transport-neutral. It describes what a simulator, service twin, or hardware twin may exchange with SPO without opening network connections or applying actuation.

Parameters

spec : BindingSpec The validated binding specification. contract_version : str, optional Semantic version label for the emitted contract. sync_capabilities : Sequence[str], optional Capabilities the contract advertises.

Returns

DigitalTwinBindingContract A read-only, transport-neutral live-sync contract.

Raises

ValueError If the spec cannot form a valid live-sync contract.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/contract.py 
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
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
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
def build_digital_twin_binding_contract(
    spec: BindingSpec,
    *,
    contract_version: str = _DEFAULT_CONTRACT_VERSION,
    sync_capabilities: Sequence[str] = _DEFAULT_SYNC_CAPABILITIES,
) -> DigitalTwinBindingContract:
    """Build a versioned live-sync contract from a validated binding spec.

    The contract is read-only and transport-neutral. It describes what a
    simulator, service twin, or hardware twin may exchange with SPO without
    opening network connections or applying actuation.

    Parameters
    ----------
    spec : BindingSpec
        The validated binding specification.
    contract_version : str, optional
        Semantic version label for the emitted contract.
    sync_capabilities : Sequence[str], optional
        Capabilities the contract advertises.

    Returns
    -------
    DigitalTwinBindingContract
        A read-only, transport-neutral live-sync contract.

    Raises
    ------
    ValueError
        If the spec cannot form a valid live-sync contract.
    """
    _require_non_empty(contract_version, "contract_version")
    if not sync_capabilities:
        raise ValueError("sync_capabilities must contain at least one capability")
    capabilities = tuple(
        _capability_from_name(capability) for capability in sync_capabilities
    )
    layers = tuple(
        DigitalTwinLayerContract(
            name=layer.name,
            index=layer.index,
            oscillator_count=len(layer.oscillator_ids),
            oscillator_ids=tuple(layer.oscillator_ids),
            family=layer.family,
        )
        for layer in sorted(spec.layers, key=lambda item: item.index)
    )
    actuator_records: list[dict[str, object]] = [
        {
            "name": actuator.name,
            "knob": actuator.knob,
            "scope": actuator.scope,
            "limits": list(actuator.limits),
        }
        for actuator in sorted(spec.actuators, key=lambda item: item.name)
    ]
    actuators = tuple(actuator_records)
    channel_algebra = build_channel_algebra_report(spec)
    base_record: dict[str, object] = {
        "contract_version": contract_version,
        "binding": {
            "name": spec.name,
            "version": spec.version,
            "safety_tier": spec.safety_tier,
        },
        "timing": {
            "sample_period_s": spec.sample_period_s,
            "control_period_s": spec.control_period_s,
        },
        "layers": [layer.to_audit_record() for layer in layers],
        "actuators": list(actuators),
        "channel_algebra": channel_algebra.to_audit_record(),
        "sync_capabilities": [
            capability.to_audit_record() for capability in capabilities
        ],
    }
    contract_hash = _record_hash(base_record)
    return DigitalTwinBindingContract(
        contract_version=contract_version,
        binding_name=spec.name,
        binding_version=spec.version,
        safety_tier=spec.safety_tier,
        sample_period_s=spec.sample_period_s,
        control_period_s=spec.control_period_s,
        layers=layers,
        actuators=actuators,
        channel_algebra=channel_algebra,
        sync_capabilities=capabilities,
        contract_hash=contract_hash,
    )

build_digital_twin_sync_envelope 

build_digital_twin_sync_envelope(
    contract: DigitalTwinBindingContract,
    *,
    capability: str,
    direction: str,
    sequence: int,
    payload: dict[str, object],
) -> DigitalTwinSyncEnvelope

Build a transport-neutral sync payload envelope for a contract.

This helper does not send data. It creates the deterministic envelope that REST, gRPC, Kafka, file, or hardware adapters can validate before handing a payload to the runtime.

Parameters

contract : DigitalTwinBindingContract The contract the envelope conforms to. capability : str The sync capability the envelope exercises. direction : str Sync direction (e.g. inbound/outbound). sequence : int Monotonic envelope sequence number. payload : dict[str, object] The deterministic payload to wrap.

Returns

DigitalTwinSyncEnvelope A validated, transport-neutral sync envelope.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/envelope.py 
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
def build_digital_twin_sync_envelope(
    contract: DigitalTwinBindingContract,
    *,
    capability: str,
    direction: str,
    sequence: int,
    payload: dict[str, object],
) -> DigitalTwinSyncEnvelope:
    """Build a transport-neutral sync payload envelope for a contract.

    This helper does not send data. It creates the deterministic envelope that
    REST, gRPC, Kafka, file, or hardware adapters can validate before handing a
    payload to the runtime.

    Parameters
    ----------
    contract : DigitalTwinBindingContract
        The contract the envelope conforms to.
    capability : str
        The sync capability the envelope exercises.
    direction : str
        Sync direction (e.g. ``inbound``/``outbound``).
    sequence : int
        Monotonic envelope sequence number.
    payload : dict[str, object]
        The deterministic payload to wrap.

    Returns
    -------
    DigitalTwinSyncEnvelope
        A validated, transport-neutral sync envelope.
    """
    return DigitalTwinSyncEnvelope(
        contract_hash=contract.contract_hash,
        capability=capability,
        direction=direction,
        sequence=sequence,
        payload=payload,
    )

read_digital_twin_sync_jsonl 

read_digital_twin_sync_jsonl(
    contract: DigitalTwinBindingContract, path: str | Path
) -> DigitalTwinSyncJsonlReport

Read JSONL sync envelopes and validate them against a contract.

Parameters

contract : DigitalTwinBindingContract The contract to validate envelopes against. path : str or pathlib.Path JSONL file to read.

Returns

DigitalTwinSyncJsonlReport A report of read, accepted, and rejected envelopes.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/envelope.py 
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
def read_digital_twin_sync_jsonl(
    contract: DigitalTwinBindingContract,
    path: str | Path,
) -> DigitalTwinSyncJsonlReport:
    """Read JSONL sync envelopes and validate them against a contract.

    Parameters
    ----------
    contract : DigitalTwinBindingContract
        The contract to validate envelopes against.
    path : str or pathlib.Path
        JSONL file to read.

    Returns
    -------
    DigitalTwinSyncJsonlReport
        A report of read, accepted, and rejected envelopes.
    """
    source = Path(path)
    accepted: list[DigitalTwinTransportValidation] = []
    rejected: list[dict[str, object]] = []
    for line_number, raw_line in enumerate(
        source.read_text(encoding="utf-8").splitlines(),
        start=1,
    ):
        if not raw_line.strip():
            continue
        try:
            raw_record = json.loads(raw_line)
        except json.JSONDecodeError:
            rejected.append(_jsonl_rejection(line_number, "malformed_json"))
            continue
        envelope = _envelope_from_record(raw_record)
        if envelope is None:
            rejected.append(_jsonl_rejection(line_number, "invalid_envelope"))
            continue
        validation = validate_digital_twin_sync_envelope(contract, envelope)
        if validation.accepted:
            accepted.append(validation)
        else:
            rejected.append(_jsonl_rejection(line_number, validation.reason))
    return DigitalTwinSyncJsonlReport(
        path=str(source),
        written=0,
        accepted=tuple(accepted),
        rejected=tuple(rejected),
    )

validate_digital_twin_sync_envelope 

validate_digital_twin_sync_envelope(
    contract: DigitalTwinBindingContract,
    envelope: DigitalTwinSyncEnvelope,
) -> DigitalTwinTransportValidation

Validate a digital-twin sync envelope against a binding contract.

Parameters

contract : DigitalTwinBindingContract The binding contract to validate against. envelope : DigitalTwinSyncEnvelope The sync envelope to validate.

Returns

DigitalTwinTransportValidation The validation result (accepted, or rejected with reasons).

Source code in src/scpn_phase_orchestrator/binding/digital_twin/envelope.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def validate_digital_twin_sync_envelope(
    contract: DigitalTwinBindingContract,
    envelope: DigitalTwinSyncEnvelope,
) -> DigitalTwinTransportValidation:
    """Validate a digital-twin sync envelope against a binding contract.

    Parameters
    ----------
    contract : DigitalTwinBindingContract
        The binding contract to validate against.
    envelope : DigitalTwinSyncEnvelope
        The sync envelope to validate.

    Returns
    -------
    DigitalTwinTransportValidation
        The validation result (accepted, or rejected with reasons).
    """
    if envelope.contract_hash != contract.contract_hash:
        return _transport_validation(False, "contract_hash_mismatch", envelope)
    capability = _find_capability(contract, envelope.capability)
    if capability is None:
        return _transport_validation(False, "capability_not_declared", envelope)
    if not _direction_allowed(
        declared=capability.direction,
        observed=envelope.direction,
    ):
        return _transport_validation(False, "direction_not_allowed", envelope)
    if not envelope.payload:
        return _transport_validation(False, "payload_empty", envelope)
    return _transport_validation(True, "accepted", envelope)

write_digital_twin_sync_jsonl 

write_digital_twin_sync_jsonl(
    path: str | Path,
    envelopes: Sequence[DigitalTwinSyncEnvelope],
) -> DigitalTwinSyncJsonlReport

Write sync envelopes to deterministic JSONL for offline replay.

Parameters

path : str or pathlib.Path Destination JSONL file path. envelopes : Sequence[DigitalTwinSyncEnvelope] The envelopes to serialise, in order.

Returns

DigitalTwinSyncJsonlReport A report of the written file and envelope count.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/envelope.py 
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
def write_digital_twin_sync_jsonl(
    path: str | Path,
    envelopes: Sequence[DigitalTwinSyncEnvelope],
) -> DigitalTwinSyncJsonlReport:
    """Write sync envelopes to deterministic JSONL for offline replay.

    Parameters
    ----------
    path : str or pathlib.Path
        Destination JSONL file path.
    envelopes : Sequence[DigitalTwinSyncEnvelope]
        The envelopes to serialise, in order.

    Returns
    -------
    DigitalTwinSyncJsonlReport
        A report of the written file and envelope count.
    """
    target = Path(path)
    lines = [envelope.to_json() for envelope in envelopes]
    target.write_text("\n".join(lines) + ("\n" if lines else ""), encoding="utf-8")
    return DigitalTwinSyncJsonlReport(
        path=str(target),
        written=len(lines),
        accepted=(),
        rejected=(),
    )

build_digital_twin_operator_evidence 

build_digital_twin_operator_evidence(
    contract: DigitalTwinBindingContract,
    validations: Sequence[DigitalTwinTransportValidation],
    *,
    rejected: Sequence[Mapping[str, object]] = (),
    adapter_records: Sequence[Mapping[str, object]] = (),
    residual_warning_threshold: float = 0.05,
    residual_critical_threshold: float = 0.2,
) -> DigitalTwinOperatorEvidence

Summarise live or replayed digital-twin sync evidence for operators.

Accepted validations may come from REST, gRPC, Kafka, hardware, memory, or JSONL replay paths. Rejected JSONL lines and adapter audit records are folded into the same deterministic summary so dashboards can display live and replayed health with the same fields.

Parameters

contract : DigitalTwinBindingContract The binding contract under observation. validations : Sequence[DigitalTwinTransportValidation] Accepted transport validations from any sync path. rejected : Sequence[Mapping[str, object]], optional Rejected JSONL lines folded into the summary. adapter_records : Sequence[Mapping[str, object]], optional Adapter audit records to include. residual_warning_threshold : float, optional Residual fraction above which a warning status is raised. residual_critical_threshold : float, optional Residual fraction above which a critical status is raised.

Returns

DigitalTwinOperatorEvidence A deterministic operator-facing health summary.

Raises

ValueError If the residual warning/critical thresholds are inconsistent.

Source code in src/scpn_phase_orchestrator/binding/digital_twin/evidence.py 
 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
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
def build_digital_twin_operator_evidence(
    contract: DigitalTwinBindingContract,
    validations: Sequence[DigitalTwinTransportValidation],
    *,
    rejected: Sequence[Mapping[str, object]] = (),
    adapter_records: Sequence[Mapping[str, object]] = (),
    residual_warning_threshold: float = 0.05,
    residual_critical_threshold: float = 0.2,
) -> DigitalTwinOperatorEvidence:
    """Summarise live or replayed digital-twin sync evidence for operators.

    Accepted validations may come from REST, gRPC, Kafka, hardware, memory, or
    JSONL replay paths. Rejected JSONL lines and adapter audit records are
    folded into the same deterministic summary so dashboards can display live
    and replayed health with the same fields.

    Parameters
    ----------
    contract : DigitalTwinBindingContract
        The binding contract under observation.
    validations : Sequence[DigitalTwinTransportValidation]
        Accepted transport validations from any sync path.
    rejected : Sequence[Mapping[str, object]], optional
        Rejected JSONL lines folded into the summary.
    adapter_records : Sequence[Mapping[str, object]], optional
        Adapter audit records to include.
    residual_warning_threshold : float, optional
        Residual fraction above which a warning status is raised.
    residual_critical_threshold : float, optional
        Residual fraction above which a critical status is raised.

    Returns
    -------
    DigitalTwinOperatorEvidence
        A deterministic operator-facing health summary.

    Raises
    ------
    ValueError
        If the residual warning/critical thresholds are inconsistent.
    """
    warning_threshold = _validated_residual_threshold(
        residual_warning_threshold,
        "residual_warning_threshold",
    )
    critical_threshold = _validated_residual_threshold(
        residual_critical_threshold,
        "residual_critical_threshold",
    )
    if warning_threshold > critical_threshold:
        raise ValueError(
            "residual_warning_threshold must be <= residual_critical_threshold"
        )

    accepted: list[DigitalTwinTransportValidation] = []
    mismatch_reasons: list[str] = []
    capability_counts = {
        capability.name: 0 for capability in contract.sync_capabilities
    }
    direction_counts: dict[str, int] = {}
    latest_sequence: int | None = None
    residuals: list[float] = []

    for validation in validations:
        envelope = validation.envelope
        if envelope.contract_hash != contract.contract_hash:
            mismatch_reasons.append("contract_hash_mismatch")
            continue
        if not validation.accepted:
            mismatch_reasons.append(validation.reason)
            continue
        accepted.append(validation)
        capability_counts[envelope.capability] = (
            capability_counts.get(envelope.capability, 0) + 1
        )
        direction_counts[envelope.direction] = (
            direction_counts.get(
                envelope.direction,
                0,
            )
            + 1
        )
        latest_sequence = (
            envelope.sequence
            if latest_sequence is None
            else max(latest_sequence, envelope.sequence)
        )
        residual = _extract_twin_residual(envelope.payload)
        if residual is not None:
            residuals.append(abs(residual))

    for rejection in rejected:
        reason = rejection.get("reason")
        if isinstance(reason, str) and reason:
            mismatch_reasons.append(reason)
        else:
            mismatch_reasons.append("rejected")

    unhealthy_adapter_count = sum(
        1 for record in adapter_records if record.get("compatible") is False
    )
    max_abs_residual = max(residuals) if residuals else None
    rejected_count = len(validations) - len(accepted) + len(rejected)
    status = _operator_status(
        rejected_count=rejected_count,
        unhealthy_adapter_count=unhealthy_adapter_count,
        max_abs_residual=max_abs_residual,
        warning_threshold=warning_threshold,
        critical_threshold=critical_threshold,
    )
    return DigitalTwinOperatorEvidence(
        contract_hash=contract.contract_hash,
        accepted_count=len(accepted),
        rejected_count=rejected_count,
        adapter_count=len(adapter_records),
        unhealthy_adapter_count=unhealthy_adapter_count,
        latest_sequence=latest_sequence,
        capability_counts=capability_counts,
        direction_counts=direction_counts,
        max_abs_twin_residual=max_abs_residual,
        mismatch_reasons=tuple(sorted(mismatch_reasons)),
        status=status,
    )

Types

Core type definitions shared across the binding subsystem.

BindingSpec (dataclass)

Field Type Required Description
name str yes Domainpack name
version str yes Spec version
safety_tier str yes Safety classification
sample_period_s float yes Input sampling interval
control_period_s float yes Control loop interval
layers list[HierarchyLayer] yes Oscillator layers
oscillator_families dict[str, OscillatorFamily] yes P/I/S families
coupling CouplingSpec yes K_nm parameters
drivers DriverSpec yes External drive config
objectives ObjectivePartition yes Optimisation targets
boundaries list[BoundaryDef] yes Safety boundaries
actuators list[ActuatorMapping] yes Output actuators
imprint_model ImprintSpec \| None no Memory dynamics
geometry_prior GeometrySpec \| None no Spatial constraints
protocol_net ProtocolNetSpec \| None no Petri net FSM
amplitude AmplitudeSpec \| None no Stuart-Landau params

Other types

  • ActuatorMapping — maps a control knob to a named actuator with scope and limits
  • HierarchyLayer — declares a layer with channel, extractor, and frequency range. When family is set, it must reference a key under oscillator_families; omitting family uses the physical-channel default, but misspelled family names fail validation and direct runtime construction.
  • VALID_KNOBS — recognised control knobs: K, alpha, zeta, Psi

types

Typed dataclass model for SPO domain binding specifications.

These dataclasses are the in-memory contract produced by the YAML loader and consumed by validators, CLIs, engines, supervisors, audit summaries, and digital-twin exporters. Constructors keep lightweight invariant checks where local consistency is unambiguous; cross-field and deployment-policy checks live in binding.validator so error reporting can stay complete and actionable.

Classes

HierarchyLayer dataclass 

HierarchyLayer(
    name: str,
    index: int,
    oscillator_ids: list[str],
    omegas: list[float] | None = None,
    family: str | None = None,
)

Single layer in the SCPN oscillator hierarchy.

OscillatorFamily dataclass 

OscillatorFamily(
    channel: str,
    extractor_type: str,
    config: dict[str, Any],
)

Phase extraction configuration for one oscillator group.

CouplingSpec dataclass 

CouplingSpec(
    base_strength: float,
    decay_alpha: float,
    templates: dict[str, str],
)

Parameters for K_nm coupling matrix construction.

DriverSpec dataclass 

DriverSpec(
    physical: dict[str, Any],
    informational: dict[str, Any],
    symbolic: dict[str, Any],
    extra: dict[str, dict[str, Any]] | None = None,
)

Configuration for standard and named external driver channels.

Methods:
channel_config 
channel_config(channel: str) -> dict[str, Any]

Return driver config for a standard or named channel.

Parameters

channel : str A standard channel (P/physical, I/informational, S/symbolic) or a named extension channel id.

Returns

dict[str, Any] The driver configuration mapping for channel, or an empty mapping when the channel has no configured driver.

Source code in src/scpn_phase_orchestrator/binding/types.py 
 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
def channel_config(self, channel: str) -> dict[str, Any]:
    """Return driver config for a standard or named channel.

    Parameters
    ----------
    channel : str
        A standard channel (``P``/``physical``, ``I``/``informational``,
        ``S``/``symbolic``) or a named extension channel id.

    Returns
    -------
    dict[str, Any]
        The driver configuration mapping for *channel*, or an empty mapping
        when the channel has no configured driver.
    """
    standard = {
        "P": self.physical,
        "physical": self.physical,
        "I": self.informational,
        "informational": self.informational,
        "S": self.symbolic,
        "symbolic": self.symbolic,
    }
    if channel in standard:
        return standard[channel]
    return (self.extra or {}).get(channel, {})
all_channel_configs 
all_channel_configs() -> dict[str, dict[str, Any]]

Return standard driver configs plus named extension channels.

Returns

dict[str, dict[str, Any]] Mapping of channel id (P/I/S plus any named extension channels) to its driver configuration mapping.

Source code in src/scpn_phase_orchestrator/binding/types.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def all_channel_configs(self) -> dict[str, dict[str, Any]]:
    """Return standard driver configs plus named extension channels.

    Returns
    -------
    dict[str, dict[str, Any]]
        Mapping of channel id (``P``/``I``/``S`` plus any named extension
        channels) to its driver configuration mapping.
    """
    configs = {
        "P": self.physical,
        "I": self.informational,
        "S": self.symbolic,
    }
    configs.update(self.extra or {})
    return configs

ChannelSpec dataclass 

ChannelSpec(
    role: str,
    required: bool = True,
    units: str | None = None,
    metric_semantics: str | None = None,
    coupling_participation: bool = True,
    audit_serialisation: bool = True,
    replay_semantics: str = "phase",
    supervisor_visibility: bool = True,
    derived_from: list[str] = list(),
    derive_rule: str | None = None,
)

Typed binding channel metadata for N-channel domainpacks.

ChannelGroupSpec dataclass 

ChannelGroupSpec(
    channels: list[str],
    required: bool = True,
    description: str | None = None,
)

Named set of channels used for validation and supervisor summaries.

CrossChannelCouplingSpec dataclass 

CrossChannelCouplingSpec(
    source: str,
    target: str,
    strength: float,
    mode: str = "bidirectional",
    template: str | None = None,
)

Declared coupling relation between two binding channels.

ObjectivePartition dataclass 

ObjectivePartition(
    good_layers: list[int],
    bad_layers: list[int],
    good_weight: float = 1.0,
    bad_weight: float = 1.0,
)

Partition of layers into good (synchronise) and bad (desynchronise) subsets.

BoundaryDef dataclass 

BoundaryDef(
    name: str,
    variable: str,
    lower: float | None,
    upper: float | None,
    severity: str,
)

Defines a soft or hard boundary on a monitored variable.

ActuatorMapping dataclass 

ActuatorMapping(
    name: str,
    knob: str,
    scope: str,
    limits: tuple[float, float],
    rate_limit_per_step: float | None = None,
)

Maps a control knob to a named actuator with scope and limits.

ImprintSpec dataclass 

ImprintSpec(
    decay_rate: float,
    saturation: float,
    modulates: list[str],
)

Parameters for the L9 memory imprint model.

GeometrySpec dataclass 

GeometrySpec(constraint_type: str, params: dict[str, Any])

Geometry constraint type and parameters for K_nm projection.

ProtocolTransitionSpec dataclass 

ProtocolTransitionSpec(
    name: str,
    inputs: list[dict[str, Any]],
    outputs: list[dict[str, Any]],
    guard: str | None = None,
)

One transition in the Petri net protocol specification.

ProtocolNetSpec dataclass 

ProtocolNetSpec(
    places: list[str],
    initial: dict[str, int],
    place_regime: dict[str, str],
    transitions: list[ProtocolTransitionSpec],
)

Full Petri net specification: places, initial marking, and transitions.

AmplitudeSpec dataclass 

AmplitudeSpec(
    mu: float,
    epsilon: float,
    amp_coupling_strength: float = 0.0,
    amp_coupling_decay: float = 0.3,
)

Amplitude dynamics parameters (Stuart-Landau bifurcation).

BindingSpec dataclass 

BindingSpec(
    name: str,
    version: str,
    safety_tier: str,
    sample_period_s: float,
    control_period_s: float,
    layers: list[HierarchyLayer],
    oscillator_families: dict[str, OscillatorFamily],
    coupling: CouplingSpec,
    drivers: DriverSpec,
    objectives: ObjectivePartition,
    boundaries: list[BoundaryDef],
    actuators: list[ActuatorMapping],
    validation_tier: str = DEFAULT_VALIDATION_TIER,
    imprint_model: ImprintSpec | None = None,
    geometry_prior: GeometrySpec | None = None,
    protocol_net: ProtocolNetSpec | None = None,
    amplitude: AmplitudeSpec | None = None,
    channels: dict[str, ChannelSpec] = dict(),
    channel_groups: dict[str, ChannelGroupSpec] = dict(),
    cross_channel_couplings: list[
        CrossChannelCouplingSpec
    ] = list(),
    value_alignment: dict[str, Any] = dict(),
)

Complete domainpack binding: layers, coupling, drivers, and actuators.

Methods:
get_omegas 
get_omegas() -> list[float]

Collect natural frequencies from all layers.

Falls back to 1.0 rad/s per oscillator when a layer defines no omegas.

Returns

list[float] Natural frequencies in rad/s, concatenated in layer order, one per oscillator.

Raises

ValueError If a layer defines an omegas list whose length differs from its oscillator count.

Source code in src/scpn_phase_orchestrator/binding/types.py 
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
def get_omegas(self) -> list[float]:
    """Collect natural frequencies from all layers.

    Falls back to 1.0 rad/s per oscillator when a layer defines no omegas.

    Returns
    -------
    list[float]
        Natural frequencies in rad/s, concatenated in layer order, one per
        oscillator.

    Raises
    ------
    ValueError
        If a layer defines an ``omegas`` list whose length differs from its
        oscillator count.
    """
    result: list[float] = []
    for layer in self.layers:
        n = len(layer.oscillator_ids)
        if layer.omegas is not None:
            if len(layer.omegas) != n:
                msg = (
                    f"Layer {layer.name!r}: omegas length {len(layer.omegas)}"
                    f" != oscillator count {n}"
                )
                raise ValueError(msg)
            result.extend(layer.omegas)
        else:
            result.extend([1.0] * n)
    return result
used_channels 
used_channels() -> set[str]

Return channels referenced by families, drivers, and algebra.

Returns

set[str] The set of channel identifiers referenced by oscillator families, configured drivers, and cross-channel coupling declarations.

Source code in src/scpn_phase_orchestrator/binding/types.py 
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
def used_channels(self) -> set[str]:
    """Return channels referenced by families, drivers, and algebra.

    Returns
    -------
    set[str]
        The set of channel identifiers referenced by oscillator families,
        configured drivers, and cross-channel coupling declarations.
    """
    used = {family.channel for family in self.oscillator_families.values()}
    used.update(self.drivers.all_channel_configs())
    used.update(self.channels)
    for channel in self.channels.values():
        used.update(channel.derived_from)
    for group in self.channel_groups.values():
        used.update(group.channels)
    for coupling in self.cross_channel_couplings:
        used.add(coupling.source)
        used.add(coupling.target)
    return used

Functions:

is_valid_channel_id 

is_valid_channel_id(channel: str) -> bool

Return True when channel is a valid binding channel identifier.

Parameters

channel : str Candidate channel identifier to validate.

Returns

bool True if channel matches the binding channel-id grammar.

Source code in src/scpn_phase_orchestrator/binding/types.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
def is_valid_channel_id(channel: str) -> bool:
    """Return True when *channel* is a valid binding channel identifier.

    Parameters
    ----------
    channel : str
        Candidate channel identifier to validate.

    Returns
    -------
    bool
        ``True`` if *channel* matches the binding channel-id grammar.
    """
    return bool(_CHANNEL_ID_RE.fullmatch(channel))

resolve_extractor_type 

resolve_extractor_type(raw: str) -> str

Map alias to algorithm name; pass algorithm names through unchanged.

Parameters

raw : str An extractor alias or canonical algorithm name.

Returns

str The canonical extractor algorithm name; unknown values pass through unchanged.

Source code in src/scpn_phase_orchestrator/binding/types.py 
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
def resolve_extractor_type(raw: str) -> str:
    """Map alias to algorithm name; pass algorithm names through unchanged.

    Parameters
    ----------
    raw : str
        An extractor alias or canonical algorithm name.

    Returns
    -------
    str
        The canonical extractor algorithm name; unknown values pass through
        unchanged.
    """
    return EXTRACTOR_ALIASES.get(raw, raw)

Loader

Loads binding specifications from YAML files. Supports:

  • Single-file specs (most domainpacks)
  • Multi-file specs with $ref template references
  • Environment variable interpolation for credentials and endpoints
  • Default value injection for optional fields

The loader does not validate — that is the validator's job. This separation allows testing with deliberately invalid specs.

loader

Fail-closed YAML/JSON loader for domain binding specifications.

The loader converts untrusted mapping/list/scalar input into the typed BindingSpec dataclass graph. Every required field, optional field type, number pair, channel identifier, and nested section is checked during parsing so later runtime code receives structured values rather than raw YAML objects.

Classes

BindingLoadError

Bases: BindingError

Raised when a binding spec cannot be parsed.

Functions:

load_binding_spec 

load_binding_spec(path: str | Path) -> BindingSpec

Load a BindingSpec from a YAML or JSON file.

Parameters

path : str or pathlib.Path Filesystem path to the binding-spec .yaml/.yml/.json file.

Returns

BindingSpec The parsed, structurally typed binding specification.

Raises

BindingLoadError If the file cannot be read, is not valid finite YAML/JSON, contains duplicate mapping keys, or does not satisfy the binding-spec schema.

Source code in src/scpn_phase_orchestrator/binding/loader.py 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
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
464
465
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
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
def load_binding_spec(path: str | Path) -> BindingSpec:
    """Load a BindingSpec from a YAML or JSON file.

    Parameters
    ----------
    path : str or pathlib.Path
        Filesystem path to the binding-spec ``.yaml``/``.yml``/``.json`` file.

    Returns
    -------
    BindingSpec
        The parsed, structurally typed binding specification.

    Raises
    ------
    BindingLoadError
        If the file cannot be read, is not valid finite YAML/JSON, contains
        duplicate mapping keys, or does not satisfy the binding-spec schema.
    """
    path = Path(path)
    # Filename only in surfaced error messages — full filesystem paths must
    # not leak into logs or API clients.
    try:
        raw = path.read_text(encoding="utf-8")
    except (FileNotFoundError, PermissionError, IsADirectoryError) as exc:
        reason = exc.strerror or type(exc).__name__
        raise BindingLoadError(f"cannot read {path.name}: {reason}") from exc

    if path.suffix in (".yaml", ".yml"):
        import yaml

        try:
            # SafeLoader subclass only: blocks arbitrary Python constructors
            # while rejecting duplicate mapping keys.
            data = yaml.load(raw, Loader=_binding_spec_safe_loader(yaml))  # noqa: S506  # nosec B506
        except (RecursionError, yaml.YAMLError) as exc:
            raise BindingLoadError(f"YAML parse error in {path.name}: {exc}") from exc
    elif path.suffix == ".json":
        try:
            data = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise BindingLoadError(f"JSON parse error in {path.name}: {exc}") from exc
    else:
        raise BindingLoadError(f"Unsupported file extension: {path.suffix}")

    if not isinstance(data, dict):
        raise BindingLoadError(
            f"expected mapping at top level, got {type(data).__name__}"
        )

    layers_data = _require_list(_require(data, "layers", "root"), "layers")
    layers = []
    for i, raw_layer in enumerate(layers_data):
        lay = _require_mapping(raw_layer, f"layers[{i}]")
        layers.append(
            HierarchyLayer(
                name=_require_str(_require(lay, "name", "layers[]"), "layers[].name"),
                index=_require_int(
                    _require(lay, "index", "layers[]"), "layers[].index"
                ),
                oscillator_ids=_optional_str_list(
                    lay.get("oscillator_ids"), "layers[].oscillator_ids"
                ),
                omegas=_optional_number_list(lay.get("omegas"), "layers[].omegas"),
                family=_optional_str(lay.get("family"), "layers[].family"),
            )
        )

    families_data = _require_mapping(
        _require(data, "oscillator_families", "root"), "oscillator_families"
    )
    osc_families = {}
    for key, raw_family in families_data.items():
        family_name = _require_str(key, "oscillator_families key")
        family_data = _require_mapping(raw_family, f"oscillator_families.{family_name}")
        extractor_type = _require_str(
            _require(
                family_data, "extractor_type", f"oscillator_families.{family_name}"
            ),
            f"oscillator_families.{family_name}.extractor_type",
        )
        osc_families[family_name] = OscillatorFamily(
            channel=_require_str(
                _require(family_data, "channel", f"oscillator_families.{family_name}"),
                f"oscillator_families.{family_name}.channel",
            ),
            extractor_type=resolve_extractor_type(extractor_type),
            config=_optional_mapping(
                family_data.get("config"), f"oscillator_families.{family_name}.config"
            ),
        )

    coupling_data = _require_mapping(_require(data, "coupling", "root"), "coupling")
    coupling = CouplingSpec(
        base_strength=_require_number(
            _require(coupling_data, "base_strength", "coupling"),
            "coupling.base_strength",
        ),
        decay_alpha=_require_number(
            _require(coupling_data, "decay_alpha", "coupling"),
            "coupling.decay_alpha",
        ),
        templates=_optional_mapping(
            coupling_data.get("templates"), "coupling.templates"
        ),
    )

    drivers = _load_drivers(data)
    channels = _load_channels(data)
    channel_groups = _load_channel_groups(data)
    cross_channel_couplings = _load_cross_channel_couplings(data)

    obj = _require_mapping(_require(data, "objectives", "root"), "objectives")
    objectives = ObjectivePartition(
        good_layers=_require_list(
            _require(obj, "good_layers", "objectives"), "objectives.good_layers"
        ),
        bad_layers=_require_list(
            _require(obj, "bad_layers", "objectives"), "objectives.bad_layers"
        ),
        good_weight=_require_number(
            obj.get("good_weight", 1.0), "objectives.good_weight"
        ),
        bad_weight=_require_number(obj.get("bad_weight", 1.0), "objectives.bad_weight"),
    )

    boundaries = []
    for i, raw_boundary in enumerate(
        _optional_list(data.get("boundaries"), "boundaries")
    ):
        b = _require_mapping(raw_boundary, f"boundaries[{i}]")
        boundaries.append(
            BoundaryDef(
                name=_require_str(
                    _require(b, "name", "boundaries[]"), "boundaries[].name"
                ),
                variable=_require_str(
                    _require(b, "variable", "boundaries[]"), "boundaries[].variable"
                ),
                lower=_optional_number(b.get("lower"), "boundaries[].lower"),
                upper=_optional_number(b.get("upper"), "boundaries[].upper"),
                severity=_require_str(
                    _require(b, "severity", "boundaries[]"), "boundaries[].severity"
                ),
            )
        )

    actuators = []
    for i, raw_actuator in enumerate(
        _optional_list(data.get("actuators"), "actuators")
    ):
        a = _require_mapping(raw_actuator, f"actuators[{i}]")
        actuators.append(
            ActuatorMapping(
                name=_require_str(
                    _require(a, "name", "actuators[]"), "actuators[].name"
                ),
                knob=_require_str(
                    _require(a, "knob", "actuators[]"), "actuators[].knob"
                ),
                scope=_require_str(
                    _require(a, "scope", "actuators[]"), "actuators[].scope"
                ),
                limits=_require_number_pair(
                    _require(a, "limits", "actuators[]"), "actuators[].limits"
                ),
                rate_limit_per_step=_optional_number(
                    a.get("rate_limit_per_step"),
                    "actuators[].rate_limit_per_step",
                ),
            )
        )

    imprint_data = data.get("imprint_model") or data.get("imprint")
    imprint = None
    if imprint_data:
        imprint_map = _require_mapping(imprint_data, "imprint_model")
        imprint = ImprintSpec(
            decay_rate=_require_number(
                _require(imprint_map, "decay_rate", "imprint_model"),
                "imprint_model.decay_rate",
            ),
            saturation=_require_number(
                _require(imprint_map, "saturation", "imprint_model"),
                "imprint_model.saturation",
            ),
            modulates=_optional_list(
                imprint_map.get("modulates"), "imprint_model.modulates"
            ),
        )

    geo_data = data.get("geometry_prior")
    geometry = None
    if geo_data:
        geo_map = _require_mapping(geo_data, "geometry_prior")
        geometry = GeometrySpec(
            constraint_type=_require_str(
                _require(geo_map, "constraint_type", "geometry_prior"),
                "geometry_prior.constraint_type",
            ),
            params=_optional_mapping(geo_map.get("params"), "geometry_prior.params"),
        )

    pnet_data = data.get("protocol_net")
    protocol_net = None
    if pnet_data:
        pnet_map = _require_mapping(pnet_data, "protocol_net")
        pnet_transitions = []
        for i, raw_transition in enumerate(
            _require_list(
                _require(pnet_map, "transitions", "protocol_net"),
                "protocol_net.transitions",
            )
        ):
            t = _require_mapping(raw_transition, f"protocol_net.transitions[{i}]")
            pnet_transitions.append(
                ProtocolTransitionSpec(
                    name=_require_str(
                        _require(t, "name", "protocol_net.transitions[]"),
                        "protocol_net.transitions[].name",
                    ),
                    inputs=_optional_list(
                        t.get("inputs"), "protocol_net.transitions[].inputs"
                    ),
                    outputs=_optional_list(
                        t.get("outputs"), "protocol_net.transitions[].outputs"
                    ),
                    guard=t.get("guard"),
                )
            )
        protocol_net = ProtocolNetSpec(
            places=_require_list(
                _require(pnet_map, "places", "protocol_net"), "protocol_net.places"
            ),
            initial=_require_mapping(
                _require(pnet_map, "initial", "protocol_net"), "protocol_net.initial"
            ),
            place_regime=_optional_mapping(
                pnet_map.get("place_regime"), "protocol_net.place_regime"
            ),
            transitions=pnet_transitions,
        )

    amp_data = data.get("amplitude")
    amplitude = None
    if amp_data:
        amp_map = _require_mapping(amp_data, "amplitude")
        amplitude = AmplitudeSpec(
            mu=_require_number(_require(amp_map, "mu", "amplitude"), "amplitude.mu"),
            epsilon=_require_number(
                _require(amp_map, "epsilon", "amplitude"), "amplitude.epsilon"
            ),
            amp_coupling_strength=_require_number(
                amp_map.get("amp_coupling_strength", 0.0),
                "amplitude.amp_coupling_strength",
            ),
            amp_coupling_decay=_require_number(
                amp_map.get("amp_coupling_decay", 0.3),
                "amplitude.amp_coupling_decay",
            ),
        )

    return BindingSpec(
        name=_require_str(_require(data, "name", "root"), "name"),
        version=_require_str(_require(data, "version", "root"), "version"),
        safety_tier=_require_str(_require(data, "safety_tier", "root"), "safety_tier"),
        validation_tier=_str_or_default(
            data.get("validation_tier"), "validation_tier", DEFAULT_VALIDATION_TIER
        ),
        sample_period_s=_require_number(
            _require(data, "sample_period_s", "root"), "sample_period_s"
        ),
        control_period_s=_require_number(
            _require(data, "control_period_s", "root"), "control_period_s"
        ),
        layers=layers,
        oscillator_families=osc_families,
        coupling=coupling,
        drivers=drivers,
        objectives=objectives,
        boundaries=boundaries,
        actuators=actuators,
        imprint_model=imprint,
        geometry_prior=geometry,
        protocol_net=protocol_net,
        amplitude=amplitude,
        channels=channels,
        channel_groups=channel_groups,
        cross_channel_couplings=cross_channel_couplings,
        value_alignment=_optional_mapping(
            data.get("value_alignment"), "value_alignment"
        ),
    )

Validator

Schema validation for binding specifications. Checks:

  • Field types and required fields (against JSON schema)
  • Cross-references: actuator scopes must match declared layers, and explicit layer families must match oscillator_families
  • Frequency ranges: f_min < f_max, both positive
  • Channel constraints: at least one layer, no duplicate names
  • Template resolution: referenced templates must exist

Validation errors are collected (not raised on first failure) so that users see all problems at once.

validator

Cross-field validation for loaded binding specifications.

validate_binding_spec returns every actionable configuration error it can find instead of failing at the first issue. It checks version shape, safety tier, timing, layer/objective references, N-channel declarations, extractor aliases, boundary and actuator scopes, imprint, amplitude, geometry, and protocol-net consistency before a binding is used by runtime code.

Classes

Functions:

validate_binding_spec 

validate_binding_spec(spec: BindingSpec) -> list[str]

Validate a BindingSpec and return a list of error strings.

Parameters

spec : BindingSpec The binding specification to validate.

Returns

list[str] Human-readable validation error messages; an empty list means the spec is structurally and cross-field valid.

Source code in src/scpn_phase_orchestrator/binding/validator.py 
 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
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
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
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
def validate_binding_spec(spec: BindingSpec) -> list[str]:
    """Validate a BindingSpec and return a list of error strings.

    Parameters
    ----------
    spec : BindingSpec
        The binding specification to validate.

    Returns
    -------
    list[str]
        Human-readable validation error messages; an empty list means the
        spec is structurally and cross-field valid.
    """
    errors: list[str] = []

    if not spec.name:
        errors.append("name must be non-empty")

    parts = spec.version.split(".")
    if len(parts) != 3 or not all(p.isdigit() for p in parts):
        errors.append(f"version must be major.minor.patch, got {spec.version!r}")

    if spec.safety_tier not in VALID_SAFETY_TIERS:
        errors.append(
            f"safety_tier must be one of {VALID_SAFETY_TIERS}, got {spec.safety_tier!r}"
        )

    if spec.validation_tier not in VALID_VALIDATION_TIERS:
        errors.append(
            "validation_tier must be one of "
            f"{VALID_VALIDATION_TIERS}, got {spec.validation_tier!r}"
        )

    if not math.isfinite(spec.sample_period_s) or spec.sample_period_s <= 0:
        errors.append(
            f"sample_period_s must be finite and > 0, got {spec.sample_period_s}"
        )

    if not math.isfinite(spec.control_period_s) or spec.control_period_s <= 0:
        errors.append(
            f"control_period_s must be finite and > 0, got {spec.control_period_s}"
        )

    if spec.control_period_s < spec.sample_period_s:
        errors.append("control_period_s must be >= sample_period_s")

    if not spec.layers:
        errors.append("at least one layer is required")

    layer_indices = {lay.index for lay in spec.layers}
    oscillator_family_names = set(spec.oscillator_families)

    for layer in spec.layers:
        if layer.family is not None and layer.family not in oscillator_family_names:
            errors.append(
                f"layer {layer.name!r}: family {layer.family!r} is not defined "
                "in oscillator_families"
            )

    used_channels = spec.used_channels()
    for channel_id in used_channels:
        if not is_valid_channel_id(channel_id):
            errors.append(
                f"channel {channel_id!r}: must match [A-Za-z][A-Za-z0-9_-]{{0,63}}"
            )

    family_driver_channels = {
        family.channel for family in spec.oscillator_families.values()
    }
    family_driver_channels.update(spec.drivers.all_channel_configs())
    declared_or_used = family_driver_channels | set(spec.channels)
    undeclared_named_channels = sorted(
        channel
        for channel in family_driver_channels
        if channel not in STANDARD_CHANNELS and channel not in spec.channels
    )
    for channel in undeclared_named_channels:
        errors.append(
            f"channel {channel!r}: named N-channel drivers/families must be "
            "declared under channels"
        )

    for channel_name, channel_spec in spec.channels.items():
        if channel_spec.replay_semantics not in _VALID_REPLAY_SEMANTICS:
            errors.append(
                f"channel {channel_name!r}: replay_semantics must be one of "
                f"{sorted(_VALID_REPLAY_SEMANTICS)}, "
                f"got {channel_spec.replay_semantics!r}"
            )
        for source in channel_spec.derived_from:
            if source == channel_name:
                errors.append(
                    f"channel {channel_name!r}: derived_from must not include itself"
                )
            if source not in declared_or_used:
                errors.append(
                    f"channel {channel_name!r}: derived_from references unknown "
                    f"channel {source!r}"
                )
        if channel_spec.derived_from and not channel_spec.derive_rule:
            errors.append(
                f"channel {channel_name!r}: derive_rule is required when "
                "derived_from is set"
            )
        if channel_spec.derived_from and channel_spec.replay_semantics != "derived":
            errors.append(
                f"channel {channel_name!r}: derived_from channels must use "
                "replay_semantics='derived'"
            )
        if channel_spec.replay_semantics == "derived" and not channel_spec.derived_from:
            errors.append(
                f"channel {channel_name!r}: replay_semantics='derived' requires "
                "derived_from"
            )
        if channel_spec.derive_rule and not channel_spec.derived_from:
            errors.append(
                f"channel {channel_name!r}: derive_rule requires derived_from"
            )
        if (
            channel_spec.required
            and not channel_spec.derived_from
            and channel_name not in family_driver_channels
        ):
            errors.append(
                f"channel {channel_name!r}: required channel must be backed by "
                "an oscillator family or driver"
            )

    for group_name, group in spec.channel_groups.items():
        if not group.channels:
            errors.append(f"channel_group {group_name!r}: channels must not be empty")
        for channel in group.channels:
            if channel not in declared_or_used:
                errors.append(
                    f"channel_group {group_name!r}: references unknown channel "
                    f"{channel!r}"
                )

    for i, coupling in enumerate(spec.cross_channel_couplings):
        if coupling.source not in declared_or_used:
            errors.append(
                f"cross_channel_couplings[{i}]: source references unknown channel "
                f"{coupling.source!r}"
            )
        if coupling.target not in declared_or_used:
            errors.append(
                f"cross_channel_couplings[{i}]: target references unknown channel "
                f"{coupling.target!r}"
            )
        if coupling.source == coupling.target:
            errors.append(
                f"cross_channel_couplings[{i}]: source and target must differ"
            )
        if not math.isfinite(coupling.strength) or coupling.strength < 0.0:
            errors.append(
                f"cross_channel_couplings[{i}].strength must be finite and >= 0"
            )
        if coupling.mode not in _VALID_CROSS_CHANNEL_MODES:
            errors.append(
                f"cross_channel_couplings[{i}].mode must be one of "
                f"{sorted(_VALID_CROSS_CHANNEL_MODES)}, got {coupling.mode!r}"
            )

    for family_name, fam in spec.oscillator_families.items():
        if not is_valid_channel_id(fam.channel):
            errors.append(
                f"oscillator_family {family_name!r}: channel must be a non-empty "
                f"identifier matching [A-Za-z][A-Za-z0-9_-]{{0,63}}, got "
                f"{fam.channel!r}"
            )
        if fam.extractor_type not in VALID_EXTRACTORS:
            errors.append(
                f"oscillator_family {family_name!r}: extractor_type must be one of "
                f"{sorted(VALID_EXTRACTORS)}, got {fam.extractor_type!r}"
            )

    if not spec.objectives.good_layers and not spec.objectives.bad_layers:
        errors.append("objectives must define at least one good or bad layer")

    for ref in spec.objectives.good_layers + spec.objectives.bad_layers:
        if ref not in layer_indices:
            errors.append(f"objectives reference layer index {ref} not in layers")

    for bdef in spec.boundaries:
        if bdef.severity not in VALID_SEVERITIES:
            errors.append(
                f"boundary {bdef.name!r}: severity must be one of "
                f"{VALID_SEVERITIES}, got {bdef.severity!r}"
            )
        if (
            bdef.lower is not None
            and bdef.upper is not None
            and bdef.lower > bdef.upper
        ):
            errors.append(
                f"boundary {bdef.name!r}: lower ({bdef.lower}) "
                f"must be <= upper ({bdef.upper})"
            )

    valid_scopes = {"global"} | {f"layer_{lay.index}" for lay in spec.layers}
    for act in spec.actuators:
        if act.knob not in VALID_KNOBS:
            errors.append(
                f"actuator {act.name!r}: knob must be one of "
                f"{VALID_KNOBS}, got {act.knob!r}"
            )
        if (
            len(act.limits) != 2
            or not all(math.isfinite(limit) for limit in act.limits)
            or act.limits[0] > act.limits[1]
        ):
            errors.append(f"actuator {act.name!r}: limits must be finite and lo <= hi")
        if act.scope not in valid_scopes:
            errors.append(
                f"actuator {act.name!r}: scope {act.scope!r} does not match any "
                f"layer index; valid scopes: {sorted(valid_scopes)}"
            )

    if spec.imprint_model is not None:
        if (
            not math.isfinite(spec.imprint_model.decay_rate)
            or spec.imprint_model.decay_rate < 0.0
        ):
            errors.append(
                "imprint_model.decay_rate must be finite and >= 0, "
                f"got {spec.imprint_model.decay_rate}"
            )
        if (
            not math.isfinite(spec.imprint_model.saturation)
            or spec.imprint_model.saturation <= 0.0
        ):
            errors.append(
                "imprint_model.saturation must be finite and > 0, "
                f"got {spec.imprint_model.saturation}"
            )

    if spec.amplitude is not None:
        if not math.isfinite(spec.amplitude.mu):
            errors.append("amplitude.mu must be finite")
        if not math.isfinite(spec.amplitude.epsilon) or spec.amplitude.epsilon < 0.0:
            errors.append(
                "amplitude.epsilon must be finite and >= 0, "
                f"got {spec.amplitude.epsilon}"
            )

    return errors

validate_binding_spec_security 

validate_binding_spec_security(
    spec: BindingSpec,
) -> list[str]

Return security-review findings for a loaded binding spec.

Normal binding validation checks structure and cross-field consistency. This stricter pass is intended for spo validate --security and rejects executable-looking payloads in free-form configuration fields. Binding specs remain declarative data; they must not carry Python code, loader tags, import expressions, subprocess references, or deserialisation gadgets.

Parameters

spec : BindingSpec The loaded binding specification to security-review.

Returns

list[str] Security-review findings; an empty list means no executable-looking payloads were detected in free-form configuration fields.

Source code in src/scpn_phase_orchestrator/binding/validator.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
def validate_binding_spec_security(spec: BindingSpec) -> list[str]:
    """Return security-review findings for a loaded binding spec.

    Normal binding validation checks structure and cross-field consistency.
    This stricter pass is intended for ``spo validate --security`` and rejects
    executable-looking payloads in free-form configuration fields. Binding specs
    remain declarative data; they must not carry Python code, loader tags,
    import expressions, subprocess references, or deserialisation gadgets.

    Parameters
    ----------
    spec : BindingSpec
        The loaded binding specification to security-review.

    Returns
    -------
    list[str]
        Security-review findings; an empty list means no executable-looking
        payloads were detected in free-form configuration fields.
    """
    findings: list[str] = []
    for location, value in _walk_security_values(spec, "binding"):
        if isinstance(value, str):
            normalised = value.strip().lower()
            marker = next(
                (
                    marker
                    for marker in _EXECUTABLE_CONFIG_MARKERS
                    if marker in normalised
                ),
                None,
            )
            if marker is not None:
                findings.append(
                    f"{location}: executable-looking marker {marker!r} is not "
                    "allowed in binding specs"
                )
    return findings

Security

The binding loader enforces security constraints:

  1. Path traversal rejection — ../ sequences in file paths are rejected to prevent reading outside the domainpack directory
  2. Schema validation — all fields are checked against the JSON schema before any data is used
  3. Environment variable interpolation — only whitelisted env vars are substituted; arbitrary code execution is not possible
  4. Size limits — binding specs exceeding 1 MB are rejected

These protections are tested in tests/test_binding_loader_security.py with adversarial inputs including malicious YAML, oversized files, and path traversal attempts.

Hard scan (spo validate --security --hard)

spo validate --security rejects executable-looking payloads in the binding spec itself. The harder --security --hard pass additionally scans the files that ship beside the binding — the domainpack's Python scenarios and YAML configuration — for the patterns that let an untrusted domainpack run arbitrary code when it is loaded or executed: dynamic evaluation (eval / exec), insecure deserialisation (pickle.load), unsafe YAML deep loads (yaml.load without a safe loader, or !!python/ construction tags), and shell command execution. The scan is review-only — it reports each match with its file, line and category and never edits, executes, or imports the scanned files, so it is safe to run on a domainpack of unknown provenance before deciding to trust it.

security_scan

Scan a domainpack's user-facing files for dangerous code and config patterns.

spo validate --security rejects executable-looking payloads in the binding spec itself. The harder --security --hard pass goes one level further and statically scans the files that ship beside the binding — the domainpack's Python scenarios and YAML configuration — for the patterns that let an untrusted domainpack run arbitrary code when it is loaded or executed: dynamic evaluation (eval / exec), insecure deserialisation (pickle.load), unsafe YAML deep loads (yaml.load without a safe loader, or !!python/ construction tags), and shell command execution.

The scan is review-only: it reports every match with its file, line and category; it never edits, executes, or imports the scanned files, so it is safe to run on a domainpack of unknown provenance before deciding whether to trust it.

Classes

UnsafePatternFinding dataclass 

UnsafePatternFinding(
    path: str, line: int, category: str, snippet: str
)

One dangerous pattern located during a hard security scan.

Parameters

path : str The scanned file, relative to the scan root. line : int The one-based line number of the match. category : str The danger class, for example "dynamic-eval" or "unsafe-yaml". snippet : str The stripped source line containing the match.

Functions:

scan_unsafe_patterns 

scan_unsafe_patterns(
    root: Path,
) -> tuple[UnsafePatternFinding, ...]

Scan a directory tree for dangerous code and configuration patterns.

Parameters

root : pathlib.Path The directory (a domainpack) or single file to scan.

Returns

tuple[UnsafePatternFinding, ...] Every located pattern, ordered by path then line.

Raises

ValueError If root does not exist.

Source code in src/scpn_phase_orchestrator/binding/security_scan.py 
 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
def scan_unsafe_patterns(root: Path) -> tuple[UnsafePatternFinding, ...]:
    """Scan a directory tree for dangerous code and configuration patterns.

    Parameters
    ----------
    root : pathlib.Path
        The directory (a domainpack) or single file to scan.

    Returns
    -------
    tuple[UnsafePatternFinding, ...]
        Every located pattern, ordered by path then line.

    Raises
    ------
    ValueError
        If ``root`` does not exist.
    """
    if not root.exists():
        raise ValueError(f"scan root does not exist: {root}")
    base = root if root.is_dir() else root.parent
    candidates = [root] if root.is_file() else sorted(root.rglob("*"))
    findings: list[UnsafePatternFinding] = []
    for path in candidates:
        if not path.is_file() or path.suffix not in _SCANNED_SUFFIXES:
            continue
        relative = path.relative_to(base).as_posix() if path != base else path.name
        text = path.read_text(encoding="utf-8", errors="replace")
        for index, line in enumerate(text.splitlines(), start=1):
            category = (
                _scan_python_line(line)
                if path.suffix == ".py"
                else ("unsafe-yaml-tag" if _YAML_PYTHON_TAG.search(line) else None)
            )
            if category is not None:
                findings.append(
                    UnsafePatternFinding(relative, index, category, line.strip())
                )
    return tuple(findings)

Domainpacks

A domainpack is a directory containing a binding spec plus optional data files (coupling templates, calibration data, policy rules). SPO ships with built-in domainpacks for common domains:

Domainpack Layers Channels Description
power_grid generators, loads P, I AC power system sync
neural_eeg cortical regions P EEG phase dynamics
microservices API endpoints I IT infrastructure sync
tokamak plasma + magnetics P Fusion plasma control
smart_factory machines, queues P, I, S Manufacturing sync

Each domainpack is validated at load time against the schema. Invalid domainpacks produce detailed error messages listing all violations.

Performance: load_binding_spec() < 10 ms.

Symbolic Binding Compiler

The SemanticDomainCompiler is the first review-gated symbolic-to-binding path. It translates a domain intent string into a BindingSpec and can also emit a complete artefact bundle:

  • binding_spec.yaml for the domain interface
  • policy.yaml with a conservative low-coherence recovery rule
  • review_notebook.ipynb with validation and policy-review cells
  • audit.json with confidence factors, matched keywords, local retrieval evidence, validation status, dry-run coherence, and Petri-net review reachability metadata
  • README.md for the generated domainpack directory

The compiler remains deterministic and local. It extracts layer counts, domain-family keywords, oscillator counts, channel declarations, safe default actuator mappings, and a review transition in protocol_net. The generated binding is passed through validate_binding_spec() and a short UPDEEngine dry run before artefacts are returned.

Local retrieval scans existing domainpacks/*/binding_spec.yaml, domainpack README content, and long-form public docs under docs/. Each evidence record is tagged with source: domainpack or source: docs, records matched terms, and contributes the top score to generated confidence factors. Retrieval records now also carry a deterministic rank plus ranking_features such as matched-term count, prompt-term count, source priority, name/phrase match evidence, and term density. Domainpack retrieval can be disabled with retrieval_root=None; docs retrieval can be disabled with docs_root=None.

The generated review notebook also carries compiler-side execution evidence. Before returning artefacts, the compiler writes the generated binding and policy to a temporary review directory and runs the same binding-schema and policy-loader checks that the notebook asks the reviewer to execute. The result is recorded in audit.json and notebook metadata under notebook_execution.

CLI usage:

spo generate "A 3-layer cardiac rhythm suppression system" \
  --name cardiac_review \
  --output-dir domainpacks/cardiac_review
spo validate domainpacks/cardiac_review/binding_spec.yaml

semantic

Review-only symbolic compiler from natural-language intent to bindings.

The semantic compiler produces a candidate BindingSpec, policy YAML, review notebook, retrieval evidence, and audit record from local heuristics and domainpack/docs evidence. The implementation is split into responsibility modules (input coercion, retrieval evidence, review notebook, YAML serialisation, and the orchestrating compiler) behind a stable re-export surface. Generated artefacts are intentionally reviewable and fail validation before use; this package does not auto-accept live deployment bindings or actuate a system.

Classes

GeneratedBindingArtifacts dataclass 

GeneratedBindingArtifacts(
    binding_spec: BindingSpec,
    binding_yaml: str,
    policy_yaml: str,
    notebook_json: str,
    audit_record: dict[str, Any],
    retrieval_evidence: list[RetrievalEvidence],
    validation_errors: list[str],
    dry_run_order_parameter: float,
)

Reviewable outputs from symbolic domain intent compilation.

Attributes
schema_valid property 
schema_valid: bool

Return True when the generated binding passed validator checks.

Returns

bool True when the generated binding passed every validator check.

Methods:
write_domainpack 
write_domainpack(output_dir: str | Path) -> None

Write generated artefacts as a reviewable domainpack directory.

Parameters

output_dir : str or pathlib.Path Destination directory; the binding spec, policy, review notebook, audit record, and README are written beneath it.

Source code in src/scpn_phase_orchestrator/binding/semantic/compiler.py 
 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
def write_domainpack(self, output_dir: str | Path) -> None:
    """Write generated artefacts as a reviewable domainpack directory.

    Parameters
    ----------
    output_dir : str or pathlib.Path
        Destination directory; the binding spec, policy, review notebook,
        audit record, and README are written beneath it.
    """
    path = _coerce_output_dir(output_dir)
    path.mkdir(parents=True, exist_ok=True)
    (path / "binding_spec.yaml").write_text(self.binding_yaml, encoding="utf-8")
    (path / "policy.yaml").write_text(self.policy_yaml, encoding="utf-8")
    (path / "review_notebook.ipynb").write_text(
        self.notebook_json,
        encoding="utf-8",
    )
    (path / "audit.json").write_text(
        json.dumps(self.audit_record, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    readme = (
        f"# {self.binding_spec.name} domainpack\n\n"
        "Generated from symbolic intent. Review `binding_spec.yaml`, "
        "`policy.yaml`, `review_notebook.ipynb`, and `audit.json` before "
        "use with live systems.\n"
    )
    (path / "README.md").write_text(readme, encoding="utf-8")

SemanticDomainCompiler

Semantic Compiler Bridge for natural language domain modeling.

Translates plain-English system descriptions into formal BindingSpec configurations. It extracts hierarchical structures, typical frequencies, and coupling constraints from text.

Methods:
compile 
compile(
    prompt: str,
    *,
    name: str = "semantically_generated_domain",
    oscillators_per_layer: int = 8,
) -> BindingSpec

Translate a symbolic domain prompt into a BindingSpec.

Parameters

prompt : str Natural-language description of the target domain. name : str, optional Name for the generated binding spec. oscillators_per_layer : int, optional Number of oscillators to allocate per generated layer.

Returns

BindingSpec The compiled, structurally typed binding specification.

Source code in src/scpn_phase_orchestrator/binding/semantic/compiler.py 
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
def compile(
    self,
    prompt: str,
    *,
    name: str = "semantically_generated_domain",
    oscillators_per_layer: int = 8,
) -> BindingSpec:
    """Translate a symbolic domain prompt into a BindingSpec.

    Parameters
    ----------
    prompt : str
        Natural-language description of the target domain.
    name : str, optional
        Name for the generated binding spec.
    oscillators_per_layer : int, optional
        Number of oscillators to allocate per generated layer.

    Returns
    -------
    BindingSpec
        The compiled, structurally typed binding specification.
    """
    return self.compile_artifacts(
        prompt,
        name=name,
        oscillators_per_layer=oscillators_per_layer,
        dry_run_steps=3,
    ).binding_spec
compile_artifacts 
compile_artifacts(
    prompt: str,
    *,
    name: str = "semantically_generated_domain",
    oscillators_per_layer: int = 8,
    dry_run_steps: int = 8,
    retrieval_root: str | Path | None = "domainpacks",
    docs_root: str | Path | None = "docs",
) -> GeneratedBindingArtifacts

Compile domain intent into binding, policy, audit, and dry-run artefacts.

Parameters

prompt : str Natural-language description of the target domain. name : str, optional Name for the generated binding spec. oscillators_per_layer : int, optional Number of oscillators to allocate per generated layer. dry_run_steps : int, optional Number of integration steps for the embedded dry-run check. retrieval_root : str or pathlib.Path or None, optional Root directory searched for retrieval grounding evidence. docs_root : str or pathlib.Path or None, optional Root directory searched for documentation grounding evidence.

Returns

GeneratedBindingArtifacts The binding, policy, audit record, retrieval evidence, and dry-run result bundle.

Raises

ValueError If the generated binding fails validation or the embedded dry run.

Source code in src/scpn_phase_orchestrator/binding/semantic/compiler.py 
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
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
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
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
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
def compile_artifacts(
    self,
    prompt: str,
    *,
    name: str = "semantically_generated_domain",
    oscillators_per_layer: int = 8,
    dry_run_steps: int = 8,
    retrieval_root: str | Path | None = "domainpacks",
    docs_root: str | Path | None = "docs",
) -> GeneratedBindingArtifacts:
    """Compile domain intent into binding, policy, audit, and dry-run artefacts.

    Parameters
    ----------
    prompt : str
        Natural-language description of the target domain.
    name : str, optional
        Name for the generated binding spec.
    oscillators_per_layer : int, optional
        Number of oscillators to allocate per generated layer.
    dry_run_steps : int, optional
        Number of integration steps for the embedded dry-run check.
    retrieval_root : str or pathlib.Path or None, optional
        Root directory searched for retrieval grounding evidence.
    docs_root : str or pathlib.Path or None, optional
        Root directory searched for documentation grounding evidence.

    Returns
    -------
    GeneratedBindingArtifacts
        The binding, policy, audit record, retrieval evidence, and dry-run
        result bundle.

    Raises
    ------
    ValueError
        If the generated binding fails validation or the embedded dry run.
    """
    (
        prompt,
        name,
        oscillators_per_layer,
        dry_run_steps,
        retrieval_root,
        docs_root,
    ) = _validate_compilation_inputs(
        prompt=prompt,
        name=name,
        oscillators_per_layer=oscillators_per_layer,
        dry_run_steps=dry_run_steps,
        retrieval_root=retrieval_root,
        docs_root=docs_root,
    )

    # Heuristic: Layer detection
    layer_match = _LAYER_PATTERN.search(prompt)
    num_layers = int(layer_match.group(1)) if layer_match else 2
    if num_layers < 1:
        raise ValueError("layer count must be >= 1")
    if num_layers > _MAX_LAYERS:
        raise ValueError(f"layer count must be <= {_MAX_LAYERS}")

    # Heuristic: Discipline detection
    lowered = prompt.lower()
    matched_keywords = sorted(
        {
            word
            for word in (
                "bio",
                "brain",
                "cardiac",
                "cell",
                "finance",
                "fusion",
                "grid",
                "plasma",
                "power",
                "traffic",
            )
            if word in lowered
        }
    )
    if any(word in lowered for word in ["bio", "cell", "brain", "cardiac"]):
        base_freq = 10.0
        domain_family = "biological"
    elif any(word in lowered for word in ["power", "grid", "fusion", "plasma"]):
        base_freq = 50.0
        domain_family = "physical"
    elif any(word in lowered for word in ["finance", "traffic"]):
        base_freq = 1.0
        domain_family = "network"
    else:
        base_freq = 1.0
        domain_family = "generic"

    layers = []
    for i in range(num_layers):
        layers.append(
            HierarchyLayer(
                name=f"layer_{i}",
                index=i,
                oscillator_ids=[
                    f"osc_{i}_{j}" for j in range(oscillators_per_layer)
                ],
                omegas=[base_freq * (10**i)] * oscillators_per_layer,
                family="default",
            )
        )

    osc_families = {
        "default": OscillatorFamily(
            channel="P", extractor_type="hilbert", config={}
        )
    }

    coupling = CouplingSpec(base_strength=0.5, decay_alpha=0.3, templates={})
    drivers = DriverSpec(physical={}, informational={}, symbolic={})
    objectives = ObjectivePartition(
        good_layers=list(range(num_layers)), bad_layers=[]
    )

    spec = BindingSpec(
        name=name,
        version="1.0.0",
        safety_tier="research",
        sample_period_s=0.01,
        control_period_s=0.1,
        layers=layers,
        oscillator_families=osc_families,
        coupling=coupling,
        drivers=drivers,
        objectives=objectives,
        boundaries=[
            BoundaryDef(
                name="low_global_coherence",
                variable="R_good",
                lower=0.0,
                upper=1.0,
                severity="soft",
            )
        ],
        actuators=[
            ActuatorMapping("global_coupling", "K", "global", (0.0, 2.0)),
            ActuatorMapping("global_drive", "zeta", "global", (0.0, 1.0)),
        ],
        channels={
            "P": ChannelSpec(
                role=domain_family,
                units="rad",
                metric_semantics="phase",
                replay_semantics="phase",
            ),
        },
        channel_groups={
            "primary": ChannelGroupSpec(
                channels=["P"],
                description="Primary phase-observation channel",
            ),
        },
        protocol_net=ProtocolNetSpec(
            places=["draft", "validated"],
            initial={"draft": 1},
            place_regime={"draft": "NOMINAL", "validated": "NOMINAL"},
            transitions=[
                ProtocolTransitionSpec(
                    name="accept_after_review",
                    inputs=[{"place": "draft"}],
                    outputs=[{"place": "validated"}],
                    guard="stability_proxy > 0.0",
                )
            ],
        ),
    )
    validation_errors = validate_binding_spec(spec)
    dry_run_r = _dry_run_order_parameter(spec, dry_run_steps)
    binding_yaml = _binding_spec_to_yaml(spec)
    policy_yaml = _policy_yaml_for(spec)
    retrieval_evidence = _retrieve_local_evidence(
        prompt,
        domainpack_root=retrieval_root,
        docs_root=docs_root,
    )
    retrieval_records = [
        evidence.to_audit_record() for evidence in retrieval_evidence
    ]
    retrieval_score = (
        max(evidence.score for evidence in retrieval_evidence)
        if retrieval_evidence
        else 0.0
    )
    notebook_execution = _review_notebook_execution_evidence(
        binding_yaml=binding_yaml,
        policy_yaml=policy_yaml,
        expected_name=spec.name,
    )
    confidence = _confidence(
        matched_keywords=matched_keywords,
        has_layer_count=layer_match is not None,
        domain_family=domain_family,
        retrieval_score=retrieval_score,
    )
    notebook_json = _review_notebook_for(
        spec,
        confidence=confidence,
        retrieval_records=retrieval_records,
        notebook_execution=notebook_execution,
    )
    audit_record = {
        "compiler": "symbolic_binding_v0",
        "schema_valid": not validation_errors,
        "validation_errors": validation_errors,
        "intent_boundary": {
            "sanitised": True,
            "max_chars": _MAX_PROMPT_CHARS,
            "llm_execution": False,
        },
        "review_gate": _review_gate_record(),
        "confidence": confidence,
        "confidence_factors": {
            "domain_keywords": len(matched_keywords),
            "explicit_layer_count": layer_match is not None,
            "domain_family": domain_family,
            "retrieval_score": retrieval_score,
        },
        "domain_family": domain_family,
        "matched_keywords": matched_keywords,
        "retrieval_evidence": retrieval_records,
        "notebook_execution": notebook_execution,
        "layers": num_layers,
        "oscillators_per_layer": oscillators_per_layer,
        "dry_run_steps": dry_run_steps,
        "dry_run_order_parameter": dry_run_r,
        "petri_reachability": {
            "initial_place": "draft",
            "review_transition": "accept_after_review",
            "target_place": "validated",
        },
    }
    _validate_generated_audit_schema(audit_record)
    return GeneratedBindingArtifacts(
        binding_spec=spec,
        binding_yaml=binding_yaml,
        policy_yaml=policy_yaml,
        notebook_json=notebook_json,
        audit_record=audit_record,
        retrieval_evidence=retrieval_evidence,
        validation_errors=validation_errors,
        dry_run_order_parameter=dry_run_r,
    )

RetrievalEvidence dataclass 

RetrievalEvidence(
    domainpack: str,
    path: str,
    score: float,
    matched_terms: list[str],
    summary: str,
    source: str = "domainpack",
    rank: int = 0,
    ranking_features: dict[str, float] = dict(),
)

Local domainpack evidence used during symbolic binding generation.

Methods:
to_audit_record 
to_audit_record() -> dict[str, Any]

Return a JSON-safe retrieval evidence record.

Returns

dict[str, Any] Deterministic, JSON-safe audit mapping of the RetrievalEvidence fields.

Source code in src/scpn_phase_orchestrator/binding/semantic/retrieval.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def to_audit_record(self) -> dict[str, Any]:
    """Return a JSON-safe retrieval evidence record.

    Returns
    -------
    dict[str, Any]
        Deterministic, JSON-safe audit mapping of the RetrievalEvidence fields.
    """
    return {
        "domainpack": self.domainpack,
        "path": self.path,
        "rank": self.rank,
        "score": self.score,
        "matched_terms": self.matched_terms,
        "summary": self.summary,
        "source": self.source,
        "ranking_features": dict(sorted(self.ranking_features.items())),
    }

Functions:

compile_symbolic_binding 

compile_symbolic_binding(
    prompt: str,
    *,
    name: str = "semantically_generated_domain",
    oscillators_per_layer: int = 8,
    dry_run_steps: int = 8,
    retrieval_root: str | Path | None = "domainpacks",
    docs_root: str | Path | None = "docs",
) -> GeneratedBindingArtifacts

Compile domain intent into a reviewable generated domainpack.

Parameters

prompt : str Natural-language description of the target domain. name : str, optional Name for the generated binding spec. oscillators_per_layer : int, optional Number of oscillators to allocate per generated layer. dry_run_steps : int, optional Number of integration steps for the embedded dry-run check. retrieval_root : str or pathlib.Path or None, optional Root directory searched for retrieval grounding evidence. docs_root : str or pathlib.Path or None, optional Root directory searched for documentation grounding evidence.

Returns

GeneratedBindingArtifacts The generated domainpack artefact bundle.

Raises

ValueError If the compilation inputs are invalid or the generated binding fails validation or its dry run.

Source code in src/scpn_phase_orchestrator/binding/semantic/compiler.py 
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
def compile_symbolic_binding(
    prompt: str,
    *,
    name: str = "semantically_generated_domain",
    oscillators_per_layer: int = 8,
    dry_run_steps: int = 8,
    retrieval_root: str | Path | None = "domainpacks",
    docs_root: str | Path | None = "docs",
) -> GeneratedBindingArtifacts:
    """Compile domain intent into a reviewable generated domainpack.

    Parameters
    ----------
    prompt : str
        Natural-language description of the target domain.
    name : str, optional
        Name for the generated binding spec.
    oscillators_per_layer : int, optional
        Number of oscillators to allocate per generated layer.
    dry_run_steps : int, optional
        Number of integration steps for the embedded dry-run check.
    retrieval_root : str or pathlib.Path or None, optional
        Root directory searched for retrieval grounding evidence.
    docs_root : str or pathlib.Path or None, optional
        Root directory searched for documentation grounding evidence.

    Returns
    -------
    GeneratedBindingArtifacts
        The generated domainpack artefact bundle.

    Raises
    ------
    ValueError
        If the compilation inputs are invalid or the generated binding fails
        validation or its dry run.
    """
    _validate_compilation_inputs(
        prompt=prompt,
        name=name,
        oscillators_per_layer=oscillators_per_layer,
        dry_run_steps=dry_run_steps,
        retrieval_root=retrieval_root,
        docs_root=docs_root,
    )
    return SemanticDomainCompiler().compile_artifacts(
        prompt,
        name=name,
        oscillators_per_layer=oscillators_per_layer,
        dry_run_steps=dry_run_steps,
        retrieval_root=retrieval_root,
        docs_root=docs_root,
    )

Topos Binding Examples

Domain-level topos obligation fixtures and semantic validation examples used by the public roadmap and direct test-linkage gates.

topos_examples

Deterministic topos obligation examples for binding review surfaces.

Classes

ToposProofObligation dataclass 

ToposProofObligation(
    name: str, description: str, passed: bool = True
)

Single proof obligation attached to one domain example.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe obligation audit record.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the ToposProofObligation fields.

Source code in src/scpn_phase_orchestrator/binding/topos_examples.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe obligation audit record.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the ToposProofObligation fields.
    """
    return {
        "name": str(self.name).strip(),
        "description": str(self.description).strip(),
        "passed": bool(self.passed),
    }

ToposDomainObligation dataclass 

ToposDomainObligation(
    domain: str,
    symbolic_prompt: str,
    binding_spec: BindingSpec,
    policy_rules: tuple[PolicyRule, ...],
    obligations: tuple[ToposProofObligation, ...],
    binding_object_count: int,
    policy_object_count: int,
    non_actuating: bool,
    proof_boundary: str,
    passed: bool,
)

Concrete domain obligation example record.

The object keeps live repository objects for compilation correctness and converts them into deterministic audit material through to_audit_record.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Convert this example into a deterministic JSON-safe audit record.

Returns

dict[str, object] Deterministic, JSON-safe audit mapping of the ToposDomainObligation fields.

Source code in src/scpn_phase_orchestrator/binding/topos_examples.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def to_audit_record(self) -> dict[str, object]:
    """Convert this example into a deterministic JSON-safe audit record.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe audit mapping of the ToposDomainObligation fields.
    """
    self._validate()
    obligations = [obligation.to_audit_record() for obligation in self.obligations]
    obligation_names = [entry["name"] for entry in obligations]
    record: dict[str, object] = {
        "domain": self.domain,
        "symbolic_prompt": self.symbolic_prompt,
        "binding_object_count": self.binding_object_count,
        "policy_object_count": self.policy_object_count,
        "obligation_names": obligation_names,
        "passed": self.passed,
        "non_actuating": self.non_actuating,
        "proof_boundary": self.proof_boundary,
    }
    record["example_hash"] = self._example_hash(record)
    return record

Functions:

build_topos_domain_obligation_examples 

build_topos_domain_obligation_examples() -> tuple[
    dict[str, object], ...
]

Build deterministic topos obligation examples for benchmark consumption.

Returns

tuple[dict[str, object], ...] Deterministic, JSON-safe obligation example records.

Raises

ValueError If an internally constructed obligation example fails its own consistency checks.

Source code in src/scpn_phase_orchestrator/binding/topos_examples.py 
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
464
465
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
def build_topos_domain_obligation_examples() -> tuple[dict[str, object], ...]:
    """Build deterministic topos obligation examples for benchmark consumption.

    Returns
    -------
    tuple[dict[str, object], ...]
        Deterministic, JSON-safe obligation example records.

    Raises
    ------
    ValueError
        If an internally constructed obligation example fails its own
        consistency checks.
    """
    examples = (
        _build_domain_example(
            domain="power_grid",
            symbolic_prompt=(
                "3-layer power grid synchronization with oscillatory frequency"
                " coherence and voltage phase balancing under load shifts"
            ),
            compilation_name="topos_power_grid",
            oscillators_per_layer=2,
            dry_run_steps=2,
            policy_rules=_power_grid_policy_rules(),
            obligations=(
                (
                    "power_grid_coherence_guard",
                    "Maintain categorical coherence boundaries under "
                    "load perturbations.",
                ),
                (
                    "grid_frequency_protective_limit",
                    "Prove stability under stepped frequency excursions.",
                ),
            ),
        ),
        _build_domain_example(
            domain="cardiac_rhythm",
            symbolic_prompt=(
                "2-layer cardiac rhythm monitoring for atrial arrhythmia"
                " synchrony and phase reset timing."
            ),
            compilation_name="topos_cardiac_rhythm",
            oscillators_per_layer=3,
            dry_run_steps=2,
            policy_rules=_cardiac_policy_rules(),
            obligations=(
                (
                    "cardiac_rhythm_variability_guard",
                    "Track rhythm-domain invariants across coupled cardiac layers.",
                ),
                (
                    "cardiac_synchrony_cat_proof",
                    "Preserve rhythm recovery envelope under symbolic perturbations.",
                ),
            ),
        ),
        _build_domain_example(
            domain="cyber_industrial",
            symbolic_prompt=(
                "4-layer cyber industrial control pilot with defensive"
                " synchronization, segmentation, and actuator isolation rules"
            ),
            compilation_name="topos_cyber_industrial",
            oscillators_per_layer=2,
            dry_run_steps=3,
            policy_rules=_cyber_industrial_policy_rules(),
            obligations=(
                (
                    "cyber_industrial_boundary_containment",
                    "Ensure categorical separation between operational "
                    "and threat modes.",
                ),
                (
                    "industrial_attack_mitigation_guard",
                    "Prevent policy drift during incident escalation windows.",
                ),
            ),
        ),
    )

    records = [example.to_audit_record() for example in examples]

    for record in records:
        if not isinstance(record, dict):
            raise ValueError("example manifest must be a dict")
        example_hash = record.get("example_hash")
        if not isinstance(example_hash, str) or not example_hash:
            raise ValueError("each example must have a stable example_hash")

    return tuple(records)

Topos Semantic Binding

Topos-oriented semantic binding helpers for categorical validation surfaces.

topos_semantic

Deterministic audit/proof-obligation validation for symbolic bindings.

Classes

SymbolicBindingObligation dataclass 

SymbolicBindingObligation(
    name: str, status: str, evidence: str
)

Single proof obligation outcome for the symbolic-binding functor.

Methods:
to_audit_record 
to_audit_record() -> dict[str, str]

Return a deterministic JSON-safe audit record.

Returns

dict[str, str] Deterministic, JSON-safe audit mapping of the SymbolicBindingObligation fields.

Source code in src/scpn_phase_orchestrator/binding/topos_semantic.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def to_audit_record(self) -> dict[str, str]:
    """Return a deterministic JSON-safe audit record.

    Returns
    -------
    dict[str, str]
        Deterministic, JSON-safe audit mapping of the SymbolicBindingObligation
        fields.
    """
    return {
        "name": self.name,
        "status": self.status,
        "evidence": self.evidence,
    }

SymbolicBindingObject dataclass 

SymbolicBindingObject(name: str, kind: str, detail: str)

Category object used in the symbolic-binding proof sketch.

Methods:
to_audit_record 
to_audit_record() -> dict[str, str]

Return a deterministic JSON-safe audit record.

Returns

dict[str, str] Deterministic, JSON-safe audit mapping of the SymbolicBindingObject fields.

Source code in src/scpn_phase_orchestrator/binding/topos_semantic.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
def to_audit_record(self) -> dict[str, str]:
    """Return a deterministic JSON-safe audit record.

    Returns
    -------
    dict[str, str]
        Deterministic, JSON-safe audit mapping of the SymbolicBindingObject fields.
    """
    return {
        "name": self.name,
        "kind": self.kind,
        "detail": self.detail,
    }

SymbolicBindingMorphism dataclass 

SymbolicBindingMorphism(
    source: str,
    target: str,
    label: str,
    deterministic: bool = True,
)

Deterministic relation between symbolic-binding validation objects.

Methods:
to_audit_record 
to_audit_record() -> dict[str, Any]

Return a deterministic JSON-safe audit record.

Returns

dict[str, Any] Deterministic, JSON-safe audit mapping of the SymbolicBindingMorphism fields.

Source code in src/scpn_phase_orchestrator/binding/topos_semantic.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def to_audit_record(self) -> dict[str, Any]:
    """Return a deterministic JSON-safe audit record.

    Returns
    -------
    dict[str, Any]
        Deterministic, JSON-safe audit mapping of the SymbolicBindingMorphism
        fields.
    """
    return {
        "source": self.source,
        "target": self.target,
        "label": self.label,
        "deterministic": self.deterministic,
    }

SymbolicBindingValidationReport dataclass 

SymbolicBindingValidationReport(
    schema_name: str,
    schema_version: str,
    object_count: int,
    morphism_count: int,
    obligation_records: tuple[
        SymbolicBindingObligation, ...
    ],
    objects: tuple[SymbolicBindingObject, ...],
    morphisms: tuple[SymbolicBindingMorphism, ...],
    passed: bool,
    report_hash: str,
    proof_boundary: str,
    non_actuating: bool = True,
)

JSON-safe deterministic report for symbolic-binding validation.

Methods:
to_audit_record 
to_audit_record() -> dict[str, Any]

Return a deterministic JSON-safe audit record.

Returns

dict[str, Any] Deterministic, JSON-safe audit mapping of the SymbolicBindingValidationReport fields.

Source code in src/scpn_phase_orchestrator/binding/topos_semantic.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def to_audit_record(self) -> dict[str, Any]:
    """Return a deterministic JSON-safe audit record.

    Returns
    -------
    dict[str, Any]
        Deterministic, JSON-safe audit mapping of the
        SymbolicBindingValidationReport fields.
    """
    return {
        "schema_name": self.schema_name,
        "schema_version": self.schema_version,
        "object_count": self.object_count,
        "morphism_count": self.morphism_count,
        "obligation_records": [
            obligation.to_audit_record() for obligation in self.obligation_records
        ],
        "objects": [obj.to_audit_record() for obj in self.objects],
        "morphisms": [morphism.to_audit_record() for morphism in self.morphisms],
        "passed": self.passed,
        "report_hash": self.report_hash,
        "proof_boundary": self.proof_boundary,
        "non_actuating": self.non_actuating,
    }

Functions:

validate_symbolic_binding_functor 

validate_symbolic_binding_functor(
    artifacts: GeneratedBindingArtifacts,
) -> SymbolicBindingValidationReport

Validate symbolic compiler output as a source-to-binding functor.

Parameters

artifacts : GeneratedBindingArtifacts The generated binding artefacts to check for functorial consistency.

Returns

SymbolicBindingValidationReport The validation report: objects, morphisms, and any obligation failures.

Source code in src/scpn_phase_orchestrator/binding/topos_semantic.py 
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
592
def validate_symbolic_binding_functor(
    artifacts: GeneratedBindingArtifacts,
) -> SymbolicBindingValidationReport:
    """Validate symbolic compiler output as a source-to-binding functor.

    Parameters
    ----------
    artifacts : GeneratedBindingArtifacts
        The generated binding artefacts to check for functorial consistency.

    Returns
    -------
    SymbolicBindingValidationReport
        The validation report: objects, morphisms, and any obligation failures.
    """
    artifacts = _validate_binding_artefact_inputs(artifacts)
    obligations: list[SymbolicBindingObligation] = []

    _add_obligation(
        obligations,
        name="artifacts_input_type",
        passed=True,
        evidence="artifacts instance is GeneratedBindingArtifacts",
    )

    schema_errors = validate_binding_spec(artifacts.binding_spec)
    schema_valid = len(schema_errors) == 0
    _add_obligation(
        obligations,
        name="schema_validation_has_no_errors",
        passed=schema_valid,
        evidence=(
            "binding schema has no validation errors"
            if schema_valid
            else f"schema errors: {', '.join(schema_errors)}"
        ),
    )

    layer_objects, layer_morphisms, _ = _collect_layer_objects_and_morphisms(
        artifacts,
        obligations,
    )
    evidence_objects, evidence_morphisms = _collect_evidence_objects_and_morphisms(
        artifacts,
        obligations,
    )

    _add_obligation(
        obligations,
        name="binding_layer_and_family_presence",
        passed=bool(
            artifacts.binding_spec.layers and artifacts.binding_spec.oscillator_families
        ),
        evidence=(
            "binding has non-empty layers and oscillator families"
            if artifacts.binding_spec.layers
            and artifacts.binding_spec.oscillator_families
            else "binding must define both layers and oscillator families"
        ),
    )

    _check_audit_boundary_preserved(
        artifacts,
        obligations,
        schema_valid=schema_valid,
    )

    objects = tuple(
        sorted((*layer_objects, *evidence_objects), key=lambda item: item.name)
    )
    morphisms = tuple(
        sorted(
            (*layer_morphisms, *evidence_morphisms),
            key=lambda item: (item.source, item.target, item.label),
        )
    )

    obligations = sorted(obligations, key=lambda item: item.name)
    report = SymbolicBindingValidationReport(
        schema_name=_SCHEMA_NAME,
        schema_version=_SCHEMA_VERSION,
        object_count=len(objects),
        morphism_count=len(morphisms),
        obligation_records=tuple(obligations),
        objects=objects,
        morphisms=morphisms,
        passed=all(item.status == "passed" for item in obligations),
        report_hash="",
        proof_boundary=_PROOF_BOUNDARY,
        non_actuating=True,
    )

    return SymbolicBindingValidationReport(
        schema_name=report.schema_name,
        schema_version=report.schema_version,
        object_count=report.object_count,
        morphism_count=report.morphism_count,
        obligation_records=report.obligation_records,
        objects=report.objects,
        morphisms=report.morphisms,
        passed=report.passed,
        report_hash=_build_report_hash(report.to_audit_record()),
        proof_boundary=report.proof_boundary,
        non_actuating=report.non_actuating,
    )

Every binding spec carries a validation_tier (scaffold, partial, or externally_validated; see VALID_VALIDATION_TIERS in binding.types) recording how much external evidence the scaffold carries. A binding is a reusable scaffold, not a validated detector, so scaffold is the honest default; a pack is promoted only with a citable evidence trail. These helpers let a Studio Hub select a single tier or group every pack by tier for a tiered gallery, keeping a broad gallery from reading as a broad set of validated solutions. See the Domainpack validation tiers guide.

gallery

Filter and group binding specs by their validation posture for a Hub gallery.

A Studio Hub that lists SPO's domainpacks needs to keep a broad gallery from reading as a broad set of validated solutions. Every :class:BindingSpec carries a validation_tier — one of :data:~scpn_phase_orchestrator.binding.types.VALID_VALIDATION_TIERS. These helpers let a gallery select a single tier (for example show only externally-validated packs) or group every pack by tier for a tiered display. The grouping always covers every tier — including empty ones — so the gallery shape is stable as packs are promoted.

Classes

Functions:

select_specs_by_validation_tier 

select_specs_by_validation_tier(
    specs: Iterable[BindingSpec], tier: str
) -> tuple[BindingSpec, ...]

Return the specs at one validation tier, preserving input order.

Parameters

specs: The binding specs to filter. tier: The validation tier to keep, one of :data:~scpn_phase_orchestrator.binding.types.VALID_VALIDATION_TIERS.

Returns

tuple[BindingSpec, ...] The specs whose validation_tier equals tier (possibly empty).

Raises

ValueError If tier is not a known validation tier.

Source code in src/scpn_phase_orchestrator/binding/gallery.py 
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
def select_specs_by_validation_tier(
    specs: Iterable[BindingSpec], tier: str
) -> tuple[BindingSpec, ...]:
    """Return the specs at one validation tier, preserving input order.

    Parameters
    ----------
    specs:
        The binding specs to filter.
    tier:
        The validation tier to keep, one of
        :data:`~scpn_phase_orchestrator.binding.types.VALID_VALIDATION_TIERS`.

    Returns
    -------
    tuple[BindingSpec, ...]
        The specs whose ``validation_tier`` equals ``tier`` (possibly empty).

    Raises
    ------
    ValueError
        If ``tier`` is not a known validation tier.
    """
    _require_known_tier(tier)
    return tuple(spec for spec in specs if spec.validation_tier == tier)

group_specs_by_validation_tier 

group_specs_by_validation_tier(
    specs: Iterable[BindingSpec],
) -> dict[str, tuple[BindingSpec, ...]]

Group specs by validation tier, covering every tier for a stable shape.

Parameters

specs: The binding specs to group.

Returns

dict[str, tuple[BindingSpec, ...]] A mapping from every tier in :data:~scpn_phase_orchestrator.binding.types.VALID_VALIDATION_TIERS (in sorted order) to the specs at that tier, preserving input order within a tier. Tiers with no specs map to an empty tuple. A spec whose validation_tier is not a known tier is ignored, since the validator is the gate that rejects such specs.

Source code in src/scpn_phase_orchestrator/binding/gallery.py 
 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
def group_specs_by_validation_tier(
    specs: Iterable[BindingSpec],
) -> dict[str, tuple[BindingSpec, ...]]:
    """Group specs by validation tier, covering every tier for a stable shape.

    Parameters
    ----------
    specs:
        The binding specs to group.

    Returns
    -------
    dict[str, tuple[BindingSpec, ...]]
        A mapping from every tier in
        :data:`~scpn_phase_orchestrator.binding.types.VALID_VALIDATION_TIERS`
        (in sorted order) to the specs at that tier, preserving input order
        within a tier. Tiers with no specs map to an empty tuple. A spec whose
        ``validation_tier`` is not a known tier is ignored, since the validator
        is the gate that rejects such specs.
    """
    grouped: dict[str, list[BindingSpec]] = {
        tier: [] for tier in sorted(VALID_VALIDATION_TIERS)
    }
    for spec in specs:
        bucket = grouped.get(spec.validation_tier)
        if bucket is not None:
            bucket.append(spec)
    return {tier: tuple(members) for tier, members in grouped.items()}