Skip to content

Studio API

Studio helper functions are pure payload builders used by the Streamlit operator surface and by tests that validate browser-facing contracts without opening transports or writing hardware.

The public scpn_phase_orchestrator.studio facade exports the passive physics review panels below alongside the lower-level helper module. Operator code can therefore depend on the Studio package surface without importing private helper paths or bypassing the review-only safety gates.

The workflow dataclasses (ImportedSourceSummary, BindingProposal, RuntimeSnapshot, ExportManifest, and StudioProjectState) fail closed on malformed audit payloads: identifiers must be non-empty strings, hashes must be valid SHA-256 hex digests, telemetry and confidence values must be finite non-boolean numbers, metadata/provenance must be JSON-safe with string keys, exports must be explicit ExportManifest sequences, and export safety posture must be either review_artifact or deployable.

The deployment helpers keep package materialisation operator-invoked. Validation warnings block Docker, WASM, and hardware targets with explicit blocked_reasons; malformed readiness rows or command tables raise ValueError instead of emitting partial deployment artefacts. Command tables include only reviewable non-blocked targets, and materialisation plans mark which commands write artefacts without opening a network or permitting hardware writes.

build_studio_product_manifest() returns the metadata-only standalone Studio product manifest used by the packaged tools/spo_studio.py shell. The manifest lists every passive physics review panel, the public builder name, the claim/proof boundary, required evidence, disabled execution gates, and a deterministic manifest_sha256 over the canonical manifest payload while keeping network_opened, hardware_write_permitted, qpu_execution_permitted, and actuation_permitted false. It does not execute panel builders, import optional runtimes, open transports, or touch hardware. Custom panel registries fail closed when panel IDs are missing or duplicated, or when a panel enables actuation, live merge, hot patching, execution, or bypasses operator review.

build_studio_control_feed(snapshot, studio_version=...) emits the live studio.control-feed.v1 envelope used by the runtime server at /api/studio-feed. The envelope mirrors the sibling scpn-control feed shape (feed_schema, studio, studio_version, content_digest, verbs, claims) and adds a SPO-specific runtime payload with runtime.schema=spo.studio-runtime-snapshot.v1. It is read-only Studio ingestion data, not a command channel.

build_studio_evidence_emissions(snapshot, studio_version=..., activity_timestamp=...) turns that same validated snapshot into concrete studio.runtime-state.v1, studio.phase-coherence.v1, and studio.regime-state.v1 schema-B evidence. Each result binds immutable canonical artifact bytes to a Platform EvidenceBundle, passes the era-v2 federation gate, and remains a bounded-model boundary claim on the numerical-model substrate. Runtime and coherence observations are measured; the unverified regime label is explicitly producer-asserted. Seal-bound numbers are decimal strings, and a recursive guard rejects any JSON float before canonicalization. This is evidence emission only: it does not sign, publish, actuate, or upgrade a claim to reference validation.

build_integrated_information_panel(records) renders passive integrated_information monitor audit records into an operator payload with latest Phi proxy values, normalised-Phi and total-integration ranges, minimum-partition review cards, and explicit actuation_permitted: false plus consciousness_claim_permitted: false gates. The helper requires the engineering_proxy_not_theoretical_iit claim boundary, finite real-valued non-negative information metrics, Phi/log-bin normalisation consistency, integer-only minimum partitions, and symmetric bounded pairwise-MI matrices when matrix evidence is supplied.

build_strange_loop_studio_panel(records) renders offline StrangeLoopSupervisor drift-scenario result records into a Studio payload with scenario pass counts, triggered-mode summaries, maximum drift/oscillation/ over-control scores, minimum control coherence, and failed-scenario IDs. It requires the strange_loop_drift_review_not_live_actuation claim boundary, non_actuating: true, execution_disabled: true, supported expected triggers, finite non-negative metrics, unit-interval coherence, and SHA-256 scenario/ result hashes before Studio may display the evidence. The payload remains review-only and sets actuation_permitted: false.

build_twin_confidence_studio_panel(score_records, summary_record) renders digital-twin confidence audit evidence into a Studio payload. It consumes real TwinConfidenceScore.to_audit_record() rows plus the matching TwinConfidenceSummary.to_audit_record() aggregate, validates every score and summary SHA-256 hash, cross-checks status counts and confidence aggregates against the supplied score sequence, exposes latest/worst status cards and backend summaries, and keeps actuation_permitted, live_merge_permitted, hot_patch_permitted, and execution_disabled locked to review-only values.

build_morphogenetic_field_studio_panel(svg_artifact) renders a passive render_morphogenetic_field_svg() artefact into a Studio payload with SVG metadata, fixed-width ASCII heatmap rows, field-energy statistics, strongest off-diagonal topology edges, and explicit actuation_permitted: false. The helper requires a complete SVG document, square snapshot shape, bounded unit-interval field statistics, shape-compatible heatmap rows, sorted off-diagonal top-edge records, and finite non-negative L2 energy before Studio may expose the field rendering to an operator.

build_multiverse_counterfactual_studio_panel(manifest, risk_report) joins a passive multiverse rollout manifest with the matching branch-risk report and renders branch comparison rows, approval counts, rejected-branch IDs, coherence ranges, safest-branch metadata, and immutable manifest/report hashes. The helper requires non-actuating execution-disabled rollout and risk claim boundaries, supported NumPy or JAX vectorised backends, unit-interval R metrics with ordered intervals, matching branch hashes across rollout and risk records, finite topology pressure metrics, approval/rejection counts that match the per-branch risk decisions, and explicit actuation_permitted: false output.

build_hybrid_order_studio_panel(records, scenarios=...) renders passive hybrid classical-quantum order-parameter audit records and deterministic scenario fixtures into a Studio payload. It exposes classical R/Psi, bipartition Von Neumann entropy, normalised entropy, participation-ratio ranges, strongest-entanglement review cards, simulator backend summaries, and scenario candidate rows. The helper requires the quantum_cosimulation_monitor_not_qpu_execution claim boundary, true non_actuating and execution_disabled flags, finite unit-interval coherence and normalised-entropy metrics, positive qubit counts, valid two-group bipartitions, supported local NumPy simulator backends, and matching record SHA-256 hashes before Studio may display the evidence. The payload sets both actuation_permitted: false and qpu_execution_permitted: false.

build_information_geometry_studio_panel(records, scenarios=...) renders passive information-geometry control proposal audit records and deterministic domain scenarios into a Studio payload. It exposes Fisher-Rao and Wasserstein distance ranges, metric-tensor and positive metric-diagonal ranges, natural-gradient norm ranges, curvature/geodesic metrics, backend summaries, latest review proposal cards, and scenario candidate rows. The helper requires the information_geometry_control_not_live_actuation claim boundary, true non_actuating and execution_disabled flags, supported NumPy/JAX information geometry backends, SHA-256 proposal/scenario hashes, unit-mass simplex coordinates, matching target/tangent dimensions, positive finite metric diagonals, symmetric metric tensors, matching geodesic/Fisher-Rao metrics, and strictly positive review TTL/max-step values before Studio may display the evidence. The payload sets actuation_permitted: false and never exposes executable control actions.

build_sheaf_cohomology_studio_panel(records, summaries, control_proposals) renders passive sheaf-Laplacian obstruction evidence into an operator payload. It joins sheaf_coherence() audit records, obstruction-summary residual rows, and review-only propose_sheaf_obstruction_control() records under the sheaf_cohomology_review_not_live_actuation claim boundary. The helper requires finite non-negative obstruction/energy metrics, shape-compatible Laplacian and residual dimensions, valid cohomology-dimension evidence, explicit residual-edge triage rows, true non_actuating, execution_disabled, and operator_review_required flags, bounded update norms, and monotone accepted projections before Studio may display the evidence. The payload sets actuation_permitted: false, hot_patch_permitted: false, and live_merge_permitted: false; it is an operator review surface, not a live sheaf-control channel.

build_topos_semantic_binding_studio_panel(symbolic_reports, policy_reports, examples=...) renders passive Topos-theoretic semantic-binding evidence into an operator payload. It joins symbolic-binding functor validation reports, policy-composition category reports, and deterministic domain obligation examples while preserving the prototype proof boundary. The helper requires the categorical_validation_prototype_not_formal_topos_proof proof boundary, true non_actuating flags, supported validation schemas, SHA-256 report and example hashes, matching object/morphism counts, deterministic morphisms, well-formed obligation records, and passed domain examples before Studio may display the evidence. The payload sets actuation_permitted: false and formal_proof_claim_permitted: false; it is a review surface, not a machine-checked proof or executable policy channel.

build_evolutionary_supervisor_policy_search_studio_panel(reports, examples=..., dsl_reports=...) renders passive evolutionary supervisor policy search evidence into an operator payload. It joins offline replay-search reports, enriched deterministic domain examples, and optional policy-DSL mutation reports while preserving review-only gates. The helper requires the offline_evolutionary_supervisor_review_not_live_actuation claim boundary for core search reports, evolutionary_supervisor_search_not_live_actuation for domain examples, true non_actuating and execution_disabled flags where the source schema carries them, disabled live merge, disabled hot patching, operator review requirements, SHA-256 report/candidate/scenario hashes, matching candidate and accepted/rejected counts, finite replay/STL metrics, JSON-safe STL scalar or sequence diagnostics, non-empty blocked reasons for rejected candidates, no blocked reasons for accepted candidates, and candidate-level actuation denial before Studio may display the evidence. The payload sets actuation_permitted: false, hot_patch_permitted: false, and live_merge_permitted: false.

build_autopoietic_lineage_studio_panel(manifests) renders passive autopoietic lineage sandbox evidence into an operator payload. It preserves offline lineage manifests, domain-labelled replay corpus rows, child-policy SHA-256 hashes, accepted child rows, rejected child rows, and blocked-reason evidence under the autopoietic_lineage_sandbox_review_not_live_merge claim boundary. The helper requires the lineage sandbox schema, true review and execution-disabled flags, disabled live merge, disabled hot patching, disabled actuation, canonical SHA-256 hashes, matching child/replay/domain counts, finite replay metrics, and candidate-level policy diffs before Studio may display the evidence. The payload sets actuation_permitted: false, hot_patch_permitted: false, live_merge_permitted: false, and operator_review_required: true; it is a review surface for comparing resource-bounded child-policy proposals, not a self-modifying runtime.

build_intergenerational_inheritance_studio_panel(histories) renders passive intergenerational policy-inheritance history evidence into an operator payload. It preserves signed inheritance-history packages, inheritance SHA-256 hashes, child-policy hashes, HMAC signature metadata, replay-domain coverage, and multi-objective replay-fitness rows under the intergenerational_inheritance_review_not_direct_hot_patch claim boundary. The helper requires deterministic history hashes, matching history/signed-count rows, contiguous generation indices, finite fitness metrics, reviewed-hot-patch merge semantics, disabled direct hot patching, and disabled actuation before Studio may display the evidence. The payload sets direct_hot_patch_permitted: false, hot_patch_permitted: false, live_merge_permitted: false, actuation_permitted: false, and operator_review_required: true; it is a signed review surface, not an executable patch channel.

Operational role

The Studio API is the fixed review boundary between interactive analysis and runtime actuation logic. This module collects deterministic payloads from validated evidence sources and intentionally keeps execution-disabled behavior visible in each panel.

Use these helpers when you need to:

  • prepare human-readable summaries for a control meeting,
  • export reviewer-facing context from replay outputs,
  • compare model outputs under the same binding assumptions across runs,
  • keep policy review evidence separated from any command that can mutate system state.

All panel builders preserve claim markers and hash references so each display can be reconciled against its source artifacts.

workflow

Pure-Python workflow state objects for SPO Studio audit artefacts.

Classes

ImportedSourceSummary dataclass 

ImportedSourceSummary(
    source_kind: str,
    sha256: str,
    byte_count: int,
    channel_count: int,
    sample_count: int,
)

Audit-ready summary of a source imported into SPO Studio.

Methods:
from_payload classmethod 
from_payload(
    *,
    source_kind: str,
    payload: bytes,
    channel_count: int,
    sample_count: int,
) -> ImportedSourceSummary

Create a stable source summary from raw imported bytes.

Parameters

source_kind : str Kind of imported source data. payload : bytes The payload mapping or bytes. channel_count : int Number of channels in the source. sample_count : int Number of samples in the source.

Returns

ImportedSourceSummary A stable source summary from raw imported bytes.

Source code in src/scpn_phase_orchestrator/studio/workflow.py 
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
@classmethod
def from_payload(
    cls,
    *,
    source_kind: str,
    payload: bytes,
    channel_count: int,
    sample_count: int,
) -> ImportedSourceSummary:
    """Create a stable source summary from raw imported bytes.

    Parameters
    ----------
    source_kind : str
        Kind of imported source data.
    payload : bytes
        The payload mapping or bytes.
    channel_count : int
        Number of channels in the source.
    sample_count : int
        Number of samples in the source.

    Returns
    -------
    ImportedSourceSummary
        A stable source summary from raw imported bytes.
    """
    return cls(
        source_kind=source_kind,
        sha256=_stable_sha256(payload),
        byte_count=len(payload),
        channel_count=channel_count,
        sample_count=sample_count,
    )
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-ready audit record.

Returns

dict[str, object] A JSON-ready audit record.

Source code in src/scpn_phase_orchestrator/studio/workflow.py 
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-ready audit record.

    Returns
    -------
    dict[str, object]
        A JSON-ready audit record.
    """
    return {
        "source_kind": self.source_kind,
        "sha256": self.sha256,
        "byte_count": self.byte_count,
        "channel_count": self.channel_count,
        "sample_count": self.sample_count,
    }

BindingProposal dataclass 

BindingProposal(
    yaml_text: str,
    validation_errors: tuple[str, ...] = (),
    inferred_channels: tuple[str, ...] = (),
    confidence_factors: dict[str, float] = dict(),
    provenance: Mapping[str, JsonValue] = dict(),
)

Candidate binding specification and its validation context.

Attributes
yaml_sha256 property 
yaml_sha256: str

Stable digest for the proposed binding YAML.

Returns

str Stable digest for the proposed binding YAML.

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

Return a JSON-ready audit record.

Returns

dict[str, object] A JSON-ready audit record.

Source code in src/scpn_phase_orchestrator/studio/workflow.py 
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-ready audit record.

    Returns
    -------
    dict[str, object]
        A JSON-ready audit record.
    """
    return {
        "yaml_sha256": self.yaml_sha256,
        "validation_errors": list(self.validation_errors),
        "inferred_channels": list(self.inferred_channels),
        "confidence_factors": _finite_float_mapping(
            self.confidence_factors,
            field_name="confidence_factors",
        ),
        "provenance": _json_safe_mapping(
            self.provenance,
            field_name="provenance",
        ),
    }

RuntimeSnapshot dataclass 

RuntimeSnapshot(
    R: float,
    Psi: float,
    K: float,
    alpha: float,
    zeta: float,
    regime: str,
    layer_metrics: tuple[tuple[str, float], ...] = (),
    hierarchy_watermarks: dict[str, int] = dict(),
    replay_status: str = "not_started",
)

Current runtime telemetry exposed to Studio review workflows.

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

Return a JSON-ready audit record.

Returns

dict[str, object] A JSON-ready audit record.

Source code in src/scpn_phase_orchestrator/studio/workflow.py 
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
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-ready audit record.

    Returns
    -------
    dict[str, object]
        A JSON-ready audit record.
    """
    return {
        "R": _finite_float(self.R, field_name="R"),
        "Psi": _finite_float(self.Psi, field_name="Psi"),
        "K": _finite_float(self.K, field_name="K"),
        "alpha": _finite_float(self.alpha, field_name="alpha"),
        "zeta": _finite_float(self.zeta, field_name="zeta"),
        "regime": self.regime,
        "layer_metrics": [
            {
                "layer": _layer_metric_name(layer),
                "value": _finite_float(value, field_name="layer_metrics"),
            }
            for layer, value in self.layer_metrics
        ],
        "hierarchy_watermarks": _hierarchy_watermarks(self.hierarchy_watermarks),
        "replay_status": self.replay_status,
    }

ExportManifest dataclass 

ExportManifest(
    target_kind: str,
    file_name: str,
    payload: str,
    command: str,
    safety_posture: SafetyPosture,
    warnings: tuple[str, ...] = (),
)

Manifest for an exportable Studio artefact.

Attributes
payload_sha256 property 
payload_sha256: str

Stable digest for the exported payload.

Returns

str Stable digest for the exported payload.

Methods:
review_artifact classmethod 
review_artifact(
    *,
    target_kind: str,
    file_name: str,
    payload: str,
    command: str,
    warnings: tuple[str, ...] = (),
) -> ExportManifest

Build a manifest for review-only artefacts.

Parameters

target_kind : str Target export kind. file_name : str Destination file name. payload : str The payload mapping or bytes. command : str The command string. warnings : tuple[str, ...] Warning messages.

Returns

ExportManifest A manifest for review-only artefacts.

Source code in src/scpn_phase_orchestrator/studio/workflow.py 
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
@classmethod
def review_artifact(
    cls,
    *,
    target_kind: str,
    file_name: str,
    payload: str,
    command: str,
    warnings: tuple[str, ...] = (),
) -> ExportManifest:
    """Build a manifest for review-only artefacts.

    Parameters
    ----------
    target_kind : str
        Target export kind.
    file_name : str
        Destination file name.
    payload : str
        The payload mapping or bytes.
    command : str
        The command string.
    warnings : tuple[str, ...]
        Warning messages.

    Returns
    -------
    ExportManifest
        A manifest for review-only artefacts.
    """
    return cls(
        target_kind=target_kind,
        file_name=file_name,
        payload=payload,
        command=command,
        safety_posture="review_artifact",
        warnings=warnings,
    )
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-ready audit record.

Returns

dict[str, object] A JSON-ready audit record.

Source code in src/scpn_phase_orchestrator/studio/workflow.py 
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-ready audit record.

    Returns
    -------
    dict[str, object]
        A JSON-ready audit record.
    """
    return {
        "target_kind": self.target_kind,
        "file_name": self.file_name,
        "payload_sha256": self.payload_sha256,
        "command": self.command,
        "safety_posture": self.safety_posture,
        "warnings": list(self.warnings),
    }

StudioProjectState dataclass 

StudioProjectState(
    project_name: str,
    source: ImportedSourceSummary,
    binding: BindingProposal,
    runtime: RuntimeSnapshot,
    exports: tuple[ExportManifest, ...] = (),
    metadata: Mapping[str, JsonValue] = dict(),
)

Complete SPO Studio state for serialisation into audit logs.

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

Return a JSON-ready project-state audit record.

Returns

dict[str, object] A JSON-ready project-state audit record.

Source code in src/scpn_phase_orchestrator/studio/workflow.py 
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-ready project-state audit record.

    Returns
    -------
    dict[str, object]
        A JSON-ready project-state audit record.
    """
    return {
        "project_name": self.project_name,
        "source": self.source.to_audit_record(),
        "binding": self.binding.to_audit_record(),
        "runtime": self.runtime.to_audit_record(),
        "exports": [manifest.to_audit_record() for manifest in self.exports],
        "metadata": _json_safe_mapping(self.metadata, field_name="metadata"),
    }

product

Standalone Studio product manifest for passive physics review panels.

Functions:

build_studio_product_manifest 

build_studio_product_manifest(
    *,
    panel_registry: Sequence[
        PanelRecord
    ] = STUDIO_REVIEW_PANEL_REGISTRY,
) -> dict[str, object]

Return the standalone Studio product manifest.

The manifest is intentionally metadata-only. It lets a packaged Studio shell discover passive panel contracts and required evidence without importing optional runtimes, executing panel builders, opening transports, or touching hardware.

Parameters

panel_registry : Sequence[PanelRecord] The Studio panel records.

Returns

dict[str, object] The standalone Studio product manifest.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/studio/product.py 
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
def build_studio_product_manifest(
    *,
    panel_registry: Sequence[PanelRecord] = STUDIO_REVIEW_PANEL_REGISTRY,
) -> dict[str, object]:
    """Return the standalone Studio product manifest.

    The manifest is intentionally metadata-only. It lets a packaged Studio shell
    discover passive panel contracts and required evidence without importing
    optional runtimes, executing panel builders, opening transports, or touching
    hardware.

    Parameters
    ----------
    panel_registry : Sequence[PanelRecord]
        The Studio panel records.

    Returns
    -------
    dict[str, object]
        The standalone Studio product manifest.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    panels = tuple(dict(panel) for panel in panel_registry)
    panel_ids = [panel.get("panel_id") for panel in panels]
    if any(not isinstance(panel_id, str) or not panel_id for panel_id in panel_ids):
        raise ValueError("panel_registry entries require non-empty panel_id values")
    if len(set(panel_ids)) != len(panel_ids):
        raise ValueError("panel_registry panel_id values must be unique")
    for panel in panels:
        _require_disabled_panel_gates(panel)
    manifest = {
        "manifest_kind": "studio_product_manifest",
        "product": "spo_studio",
        "standalone_shell": "tools/spo_studio.py",
        "launch_command": "streamlit run tools/spo_studio.py",
        "review_panel_count": len(panels),
        "review_panels": panels,
        "operator_review_required": True,
        "actuation_permitted": False,
        "live_merge_permitted": False,
        "hot_patch_permitted": False,
        "network_opened": False,
        "hardware_write_permitted": False,
        "qpu_execution_permitted": False,
    }
    manifest["manifest_sha256"] = _manifest_sha256(manifest)
    return manifest

live_feed

Build the SPO live Studio feed using the shared control-feed envelope.

Classes

StudioFeedVerb dataclass 

StudioFeedVerb(
    name: str,
    safety_tier: str,
    side_effect: str,
    timing_class: str,
    domain_distinctive: bool,
)

A verb summary in the studio.control-feed.v1 wire envelope.

Attributes

name : str Verb name rendered by Studio. safety_tier : str Studio safety tier: research, certified, or production. side_effect : str Studio side-effect class: read-only, simulated, or live-hardware. timing_class : str Studio timing class: batch, interactive, or realtime. domain_distinctive : bool Whether the verb is domain-specific rather than core-spine.

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

Return the verb as a JSON-safe feed record.

Returns

dict[str, object] The verb record emitted in the feed's verbs list.

Source code in src/scpn_phase_orchestrator/studio/live_feed.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def to_record(self) -> dict[str, object]:
    """Return the verb as a JSON-safe feed record.

    Returns
    -------
    dict[str, object]
        The verb record emitted in the feed's ``verbs`` list.
    """
    return {
        "name": self.name,
        "safety_tier": self.safety_tier,
        "side_effect": self.side_effect,
        "timing_class": self.timing_class,
        "domain_distinctive": self.domain_distinctive,
    }

StudioFeedClaim dataclass 

StudioFeedClaim(
    schema: str, status: str, admission: str, kind: str
)

A claim summary in the studio.control-feed.v1 wire envelope.

Attributes

schema : str Evidence schema identifier. status : str Claim-boundary status from the shared Studio lattice. admission : str Runtime admission decision, admitted or rejected. kind : str Evidence modality, such as measured or curated.

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

Return the claim as a JSON-safe feed record.

Returns

dict[str, str] The claim record emitted in the feed's claims list.

Source code in src/scpn_phase_orchestrator/studio/live_feed.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def to_record(self) -> dict[str, str]:
    """Return the claim as a JSON-safe feed record.

    Returns
    -------
    dict[str, str]
        The claim record emitted in the feed's ``claims`` list.
    """
    return {
        "schema": self.schema,
        "status": self.status,
        "admission": self.admission,
        "kind": self.kind,
    }

Functions:

build_studio_control_feed 

build_studio_control_feed(
    snapshot: Mapping[str, object], *, studio_version: str
) -> dict[str, object]

Build a live Studio feed from a runtime server snapshot.

The envelope mirrors the sibling CONTROL studio.control-feed.v1 shape (feed_schema, studio, studio_version, content_digest, verbs, claims) and adds a SPO-specific runtime payload for live state ingestion. The extra field is additive: existing Studio loaders that only consume verbs and claims can ignore it.

Parameters

snapshot : Mapping[str, object] JSON-safe snapshot from runtime.server.SimulationState.snapshot. studio_version : str SPO package version stamped into the feed.

Returns

dict[str, object] JSON-safe Studio feed document.

Source code in src/scpn_phase_orchestrator/studio/live_feed.py 
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
def build_studio_control_feed(
    snapshot: Mapping[str, object],
    *,
    studio_version: str,
) -> dict[str, object]:
    """Build a live Studio feed from a runtime server snapshot.

    The envelope mirrors the sibling CONTROL ``studio.control-feed.v1`` shape
    (``feed_schema``, ``studio``, ``studio_version``, ``content_digest``,
    ``verbs``, ``claims``) and adds a SPO-specific ``runtime`` payload for live
    state ingestion. The extra field is additive: existing Studio loaders that
    only consume verbs and claims can ignore it.

    Parameters
    ----------
    snapshot : Mapping[str, object]
        JSON-safe snapshot from ``runtime.server.SimulationState.snapshot``.
    studio_version : str
        SPO package version stamped into the feed.

    Returns
    -------
    dict[str, object]
        JSON-safe Studio feed document.
    """
    runtime = runtime_summary(snapshot)
    claims = claim_summaries(runtime)
    verbs = [verb.to_record() for verb in _VERBS]
    claim_records = [claim.to_record() for claim in claims]
    return {
        "feed_schema": FEED_SCHEMA,
        "studio": STUDIO_ID,
        "studio_version": studio_version,
        "content_digest": _content_digest(verbs, claim_records),
        "verbs": verbs,
        "claims": claim_records,
        "runtime": runtime,
    }

runtime_summary 

runtime_summary(
    snapshot: Mapping[str, object],
) -> dict[str, object]

Return the SPO runtime payload embedded in the live Studio feed.

Parameters

snapshot : Mapping[str, object] Server snapshot to validate and reduce.

Returns

dict[str, object] JSON-safe runtime summary with finite numeric fields.

Source code in src/scpn_phase_orchestrator/studio/live_feed.py 
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
def runtime_summary(snapshot: Mapping[str, object]) -> dict[str, object]:
    """Return the SPO runtime payload embedded in the live Studio feed.

    Parameters
    ----------
    snapshot : Mapping[str, object]
        Server snapshot to validate and reduce.

    Returns
    -------
    dict[str, object]
        JSON-safe runtime summary with finite numeric fields.
    """
    result: dict[str, object] = {
        "schema": RUNTIME_SCHEMA,
        "step": _required_int(snapshot, "step"),
        "r_global": _required_unit_float(snapshot, "R_global"),
        "regime": _required_text(snapshot, "regime"),
        "n_oscillators": _required_int(snapshot, "n_oscillators"),
        "amplitude_mode": _required_bool(snapshot, "amplitude_mode"),
        "layers": _layer_records(snapshot.get("layers")),
    }
    if snapshot.get("mean_amplitude") is not None:
        result["mean_amplitude"] = _required_float(snapshot, "mean_amplitude")
    return result

claim_summaries 

claim_summaries(
    runtime: Mapping[str, object],
) -> tuple[StudioFeedClaim, ...]

Return claim summaries derived from a validated runtime payload.

Parameters

runtime : Mapping[str, object] Validated spo.studio-runtime-snapshot.v1 payload.

Returns

tuple[StudioFeedClaim, ...] Claim-boundary summaries for Studio honesty rendering.

Raises

ValueError If the runtime payload has malformed coherence or layer fields.

Source code in src/scpn_phase_orchestrator/studio/live_feed.py 
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
def claim_summaries(runtime: Mapping[str, object]) -> tuple[StudioFeedClaim, ...]:
    """Return claim summaries derived from a validated runtime payload.

    Parameters
    ----------
    runtime : Mapping[str, object]
        Validated ``spo.studio-runtime-snapshot.v1`` payload.

    Returns
    -------
    tuple[StudioFeedClaim, ...]
        Claim-boundary summaries for Studio honesty rendering.

    Raises
    ------
    ValueError
        If the runtime payload has malformed coherence or layer fields.
    """
    r_global = _required_unit_float(runtime, "r_global")
    layers = runtime.get("layers")
    if not isinstance(layers, Sequence) or isinstance(layers, (str, bytes)):
        raise ValueError("runtime layers must be a sequence")
    admission = "admitted" if layers else "rejected"
    coherence_status = "bounded-model" if 0.0 <= r_global <= 1.0 else "validation-gap"
    return (
        StudioFeedClaim(RUNTIME_STATE_SCHEMA, "bounded-model", admission, "measured"),
        StudioFeedClaim(
            PHASE_COHERENCE_SCHEMA,
            coherence_status,
            admission,
            "measured",
        ),
        StudioFeedClaim(
            REGIME_STATE_SCHEMA,
            "bounded-model",
            admission,
            "producer-asserted",
        ),
    )

render_studio_control_feed_json 

render_studio_control_feed_json(
    snapshot: Mapping[str, object], *, studio_version: str
) -> str

Render a live Studio feed as deterministic JSON.

Parameters

snapshot : Mapping[str, object] Runtime server snapshot. studio_version : str SPO package version stamped into the feed.

Returns

str Sorted, indented JSON feed with a trailing newline.

Source code in src/scpn_phase_orchestrator/studio/live_feed.py 
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
def render_studio_control_feed_json(
    snapshot: Mapping[str, object],
    *,
    studio_version: str,
) -> str:
    """Render a live Studio feed as deterministic JSON.

    Parameters
    ----------
    snapshot : Mapping[str, object]
        Runtime server snapshot.
    studio_version : str
        SPO package version stamped into the feed.

    Returns
    -------
    str
        Sorted, indented JSON feed with a trailing newline.
    """
    return (
        json.dumps(
            build_studio_control_feed(snapshot, studio_version=studio_version),
            indent=2,
            sort_keys=True,
        )
        + "\n"
    )

evidence_bundles

Emit float-free Studio schema-B evidence from a SPO runtime snapshot.

The live Studio feed advertises three evidence families: runtime state, phase coherence, and regime state. This module makes those declarations load-bearing by reducing a validated runtime snapshot to an immutable canonical artifact, binding its digest into :class:scpn_studio_platform.evidence.EvidenceBundle, and passing the wire bundle through the Platform federation gate.

The producer never promotes a simulator observation to reference validation. Every bundle is admitted only as a boundary claim on the numerical-model substrate. Seal-bound artifacts and bundle wires contain no JSON floats: bounded numeric observations cross the wire as shortest-round-trip strings, so Python and JavaScript cannot disagree over integer-looking float encodings.

Classes

StudioEvidenceEmission dataclass 

StudioEvidenceEmission(
    bundle: EvidenceBundle,
    verdict: FederationVerdict,
    _artifact_bytes: bytes,
)

One immutable artifact, its schema-B bundle, and federation verdict.

Parameters

bundle : EvidenceBundle Platform schema-B bundle whose entity digest pins the artifact bytes. verdict : FederationVerdict Era-v2 Platform admission result for the bundle wire.

Notes

Canonical artifact bytes are held privately so callers cannot mutate a dictionary after its digest has been bound into bundle.entity.digest. The :attr:artifact property returns a fresh decoded mapping.

Attributes
artifact property 
artifact: dict[str, JsonValue]

Return a fresh copy of the canonical float-free evidence artifact.

Returns

dict[str, JsonValue] Decoded artifact data. Mutating it cannot invalidate the stored content digest because the canonical source bytes remain private.

Methods:
__post_init__ 
__post_init__() -> None

Reject inconsistent or non-canonical emission components.

Source code in src/scpn_phase_orchestrator/studio/evidence_bundles.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def __post_init__(self) -> None:
    """Reject inconsistent or non-canonical emission components."""
    artifact = self.artifact
    _assert_float_free(artifact)
    if canonicalize(artifact) != self._artifact_bytes:
        raise ValueError("artifact bytes must use Platform canonical serialization")
    if artifact.get("schema") != self.bundle.schema:
        raise ValueError("artifact and evidence bundle schemas must match")
    if content_digest(artifact) != self.bundle.entity.digest:
        raise ValueError("artifact digest does not match bundle entity digest")
    if not self.verdict.admitted or self.verdict.rejections:
        raise ValueError("evidence bundle must pass the Platform federation gate")
    if self.verdict.mode != "boundary":
        raise ValueError("SPO runtime evidence must remain a boundary claim")
    _assert_float_free(self.bundle.to_dict())
to_dict 
to_dict() -> dict[str, object]

Return the artifact and schema-B bundle as a seal-safe wire record.

Returns

dict[str, object] Mapping with artifact and evidence_bundle fields and no JSON floating-point values.

Source code in src/scpn_phase_orchestrator/studio/evidence_bundles.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def to_dict(self) -> dict[str, object]:
    """Return the artifact and schema-B bundle as a seal-safe wire record.

    Returns
    -------
    dict[str, object]
        Mapping with ``artifact`` and ``evidence_bundle`` fields and no
        JSON floating-point values.
    """
    record: dict[str, object] = {
        "artifact": self.artifact,
        "evidence_bundle": self.bundle.to_dict(),
    }
    _assert_float_free(record)
    return record

Functions:

build_studio_evidence_emissions 

build_studio_evidence_emissions(
    snapshot: Mapping[str, object],
    *,
    studio_version: str,
    activity_timestamp: str,
    operator: str = "local-operator",
) -> tuple[StudioEvidenceEmission, ...]

Build and federation-gate the three live SPO evidence emissions.

Parameters

snapshot : Mapping[str, object] Raw mapping from runtime.server.SimulationState.snapshot. studio_version : str Exact SPO package version responsible for the emission. activity_timestamp : str ISO-8601 timestamp recorded as both start and end of the snapshot reduction. It is explicit so deterministic replays remain deterministic. operator : str, optional Opaque local operator or tenant identity for the PROV agent.

Returns

tuple[StudioEvidenceEmission, ...] Runtime-state, phase-coherence, and regime-state emissions, in the same stable order as the schema-A manifest's evidence types.

Raises

ValueError If snapshot validation fails, metadata is empty, a seal-bound float is found, or the Platform gate does not admit a boundary-only bundle.

Source code in src/scpn_phase_orchestrator/studio/evidence_bundles.py 
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
def build_studio_evidence_emissions(
    snapshot: Mapping[str, object],
    *,
    studio_version: str,
    activity_timestamp: str,
    operator: str = "local-operator",
) -> tuple[StudioEvidenceEmission, ...]:
    """Build and federation-gate the three live SPO evidence emissions.

    Parameters
    ----------
    snapshot : Mapping[str, object]
        Raw mapping from ``runtime.server.SimulationState.snapshot``.
    studio_version : str
        Exact SPO package version responsible for the emission.
    activity_timestamp : str
        ISO-8601 timestamp recorded as both start and end of the snapshot
        reduction. It is explicit so deterministic replays remain deterministic.
    operator : str, optional
        Opaque local operator or tenant identity for the PROV agent.

    Returns
    -------
    tuple[StudioEvidenceEmission, ...]
        Runtime-state, phase-coherence, and regime-state emissions, in the same
        stable order as the schema-A manifest's evidence types.

    Raises
    ------
    ValueError
        If snapshot validation fails, metadata is empty, a seal-bound float is
        found, or the Platform gate does not admit a boundary-only bundle.
    """
    _require_text(studio_version, "studio_version")
    _require_text(activity_timestamp, "activity_timestamp")
    _require_text(operator, "operator")
    runtime = runtime_summary(snapshot)
    artifacts = _artifacts(runtime)
    specifications = (
        (
            "simulate",
            EvidenceKind.MEASURED,
            artifacts[0],
            "One validated SPO numerical-model runtime snapshot only; no "
            "temporal stability, safety, hardware, or actuation claim.",
        ),
        (
            "analyse",
            EvidenceKind.MEASURED,
            artifacts[1],
            "Finite order parameters from one validated SPO numerical-model "
            "snapshot; no reference-validation, stability, safety, hardware, "
            "or actuation claim.",
        ),
        (
            "supervise",
            EvidenceKind.PRODUCER_ASSERTED,
            artifacts[2],
            "Producer-reported regime label from one SPO numerical-model "
            "snapshot; no independent validation, forecast, safety, hardware, "
            "or actuation claim.",
        ),
    )
    return tuple(
        _build_emission(
            artifact=artifact,
            verb=verb,
            evidence_kind=kind,
            validity_note=note,
            studio_version=studio_version,
            activity_timestamp=activity_timestamp,
            operator=operator,
        )
        for verb, kind, artifact, note in specifications
    )

render_studio_evidence_emissions_json 

render_studio_evidence_emissions_json(
    snapshot: Mapping[str, object],
    *,
    studio_version: str,
    activity_timestamp: str,
    operator: str = "local-operator",
) -> str

Render canonical float-free JSON for all live evidence emissions.

Parameters

snapshot : Mapping[str, object] Raw mapping from runtime.server.SimulationState.snapshot. studio_version : str Exact SPO package version responsible for the emission. activity_timestamp : str ISO-8601 timestamp recorded in every PROV activity. operator : str, optional Opaque local operator or tenant identity for the PROV agent.

Returns

str Platform-canonical compact JSON array followed by one newline.

Source code in src/scpn_phase_orchestrator/studio/evidence_bundles.py 
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
def render_studio_evidence_emissions_json(
    snapshot: Mapping[str, object],
    *,
    studio_version: str,
    activity_timestamp: str,
    operator: str = "local-operator",
) -> str:
    """Render canonical float-free JSON for all live evidence emissions.

    Parameters
    ----------
    snapshot : Mapping[str, object]
        Raw mapping from ``runtime.server.SimulationState.snapshot``.
    studio_version : str
        Exact SPO package version responsible for the emission.
    activity_timestamp : str
        ISO-8601 timestamp recorded in every PROV activity.
    operator : str, optional
        Opaque local operator or tenant identity for the PROV agent.

    Returns
    -------
    str
        Platform-canonical compact JSON array followed by one newline.
    """
    emissions = build_studio_evidence_emissions(
        snapshot,
        studio_version=studio_version,
        activity_timestamp=activity_timestamp,
        operator=operator,
    )
    records = [emission.to_dict() for emission in emissions]
    _assert_float_free(records)
    canonical = cast("bytes", canonicalize(records))
    return canonical.decode("utf-8") + "\n"

ui_helpers

Pure helper layer for the SPO Studio Streamlit surface.

Classes

StudioKnobState dataclass 

StudioKnobState(
    K: float = 1.0,
    alpha: float = 0.0,
    zeta: float = 0.0,
    Psi: float = 0.0,
)

Review-only knob state used by Studio replay controls.

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

Return a JSON-safe knob record.

Returns

dict[str, float] A JSON-safe knob record.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/_state.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def to_audit_record(self) -> dict[str, float]:
    """Return a JSON-safe knob record.

    Returns
    -------
    dict[str, float]
        A JSON-safe knob record.
    """
    return {
        "K": float(self.K),
        "alpha": float(self.alpha),
        "zeta": float(self.zeta),
        "Psi": float(self.Psi),
    }

StudioReplayResult dataclass 

StudioReplayResult(
    project_state: StudioProjectState,
    r_history: tuple[float, ...],
    regime_history: tuple[str, ...],
    layer_table: tuple[dict[str, object], ...],
    oscillator_table: tuple[dict[str, object], ...],
    canvas_graph: Mapping[str, object],
    connector_plan: Mapping[str, object],
    export_manifests: tuple[ExportManifest, ...],
)

Replay output rendered by SPO Studio.

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

Return a JSON-safe replay audit record.

Returns

dict[str, object] A JSON-safe replay audit record.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/_state.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe replay audit record.

    Returns
    -------
    dict[str, object]
        A JSON-safe replay audit record.
    """
    return {
        "project": self.project_state.to_audit_record(),
        "r_history": list(self.r_history),
        "regime_history": list(self.regime_history),
        "layer_table": list(self.layer_table),
        "oscillator_table": list(self.oscillator_table),
        "canvas_graph": dict(self.canvas_graph),
        "connector_plan": dict(self.connector_plan),
        "exports": [
            manifest.to_audit_record() for manifest in self.export_manifests
        ],
    }

Functions:

apply_knob_update 

apply_knob_update(
    knobs: StudioKnobState,
    *,
    K: float | None = None,
    alpha: float | None = None,
    zeta: float | None = None,
    Psi: float | None = None,
) -> StudioKnobState

Return validated knobs after a UI edit.

Parameters

knobs : StudioKnobState The Studio knob state. K : float | None Coupling-strength knob value, or None. alpha : float | None Phase-lag knob value, or None. zeta : float | None Drive-strength knob value, or None. Psi : float | None Drive reference-phase knob value, or None.

Returns

StudioKnobState Validated knobs after a UI edit.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/_state.py 
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
def apply_knob_update(
    knobs: StudioKnobState,
    *,
    K: float | None = None,
    alpha: float | None = None,
    zeta: float | None = None,
    Psi: float | None = None,
) -> StudioKnobState:
    """Return validated knobs after a UI edit.

    Parameters
    ----------
    knobs : StudioKnobState
        The Studio knob state.
    K : float | None
        Coupling-strength knob value, or ``None``.
    alpha : float | None
        Phase-lag knob value, or ``None``.
    zeta : float | None
        Drive-strength knob value, or ``None``.
    Psi : float | None
        Drive reference-phase knob value, or ``None``.

    Returns
    -------
    StudioKnobState
        Validated knobs after a UI edit.
    """
    return StudioKnobState(
        K=knobs.K if K is None else K,
        alpha=knobs.alpha if alpha is None else alpha,
        zeta=knobs.zeta if zeta is None else zeta,
        Psi=knobs.Psi if Psi is None else Psi,
    )

discover_domainpacks 

discover_domainpacks(
    domainpack_dir: Path,
) -> tuple[str, ...]

Return domainpack names containing a binding spec.

Parameters

domainpack_dir : Path Directory containing domainpacks.

Returns

tuple[str, ...] Domainpack names containing a binding spec.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/_state.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def discover_domainpacks(domainpack_dir: Path) -> tuple[str, ...]:
    """Return domainpack names containing a binding spec.

    Parameters
    ----------
    domainpack_dir : Path
        Directory containing domainpacks.

    Returns
    -------
    tuple[str, ...]
        Domainpack names containing a binding spec.
    """
    if not domainpack_dir.exists():
        return ()
    return tuple(
        sorted(
            path.name
            for path in domainpack_dir.iterdir()
            if path.is_dir() and (path / "binding_spec.yaml").exists()
        )
    )

apply_canvas_binding_rewrite_candidate 

apply_canvas_binding_rewrite_candidate(
    candidate: Mapping[str, object],
    *,
    binding_spec_path: str | Path,
    operator_signoff: bool,
    create_backup: bool = True,
) -> dict[str, object]

Apply a reviewed canvas binding candidate with hash and validation gates.

Parameters

candidate : Mapping[str, object] The binding rewrite candidate mapping. binding_spec_path : str | Path Path to the binding-spec file. operator_signoff : bool Whether the operator has signed off. create_backup : bool Whether to write a backup before applying.

Returns

dict[str, object] A reviewed canvas binding candidate with hash and validation gates.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/canvas.py 
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
def apply_canvas_binding_rewrite_candidate(
    candidate: Mapping[str, object],
    *,
    binding_spec_path: str | Path,
    operator_signoff: bool,
    create_backup: bool = True,
) -> dict[str, object]:
    """Apply a reviewed canvas binding candidate with hash and validation gates.

    Parameters
    ----------
    candidate : Mapping[str, object]
        The binding rewrite candidate mapping.
    binding_spec_path : str | Path
        Path to the binding-spec file.
    operator_signoff : bool
        Whether the operator has signed off.
    create_backup : bool
        Whether to write a backup before applying.

    Returns
    -------
    dict[str, object]
        A reviewed canvas binding candidate with hash and validation gates.
    """
    path = Path(binding_spec_path)
    candidate_yaml = _require_non_empty_payload(
        candidate.get("candidate_yaml"),
        "candidate_yaml",
    )
    before_digest = _require_sha256_digest(
        candidate.get("before_yaml_sha256"),
        "before_yaml_sha256",
    )
    candidate_digest = _require_sha256_digest(
        candidate.get("candidate_yaml_sha256"),
        "candidate_yaml_sha256",
    )
    blocked_reasons = _binding_apply_blocked_reasons(
        candidate,
        path,
        candidate_yaml,
        before_digest,
        candidate_digest,
        operator_signoff=operator_signoff,
    )
    if blocked_reasons:
        return _binding_apply_record(
            candidate,
            path,
            status="blocked",
            before_digest=before_digest,
            after_digest="",
            backup_path="",
            blocked_reasons=blocked_reasons,
        )

    current_yaml = path.read_text(encoding="utf-8")
    backup_path = ""
    if create_backup:
        backup = _next_binding_backup_path(path, before_digest)
        backup.write_text(current_yaml, encoding="utf-8")
        backup_path = str(backup)
    _atomic_write_text(path, candidate_yaml)
    after_digest = sha256(path.read_text(encoding="utf-8").encode("utf-8")).hexdigest()
    return _binding_apply_record(
        candidate,
        path,
        status="applied",
        before_digest=before_digest,
        after_digest=after_digest,
        backup_path=backup_path,
        blocked_reasons=[],
    )

build_canvas_binding_rewrite_candidate 

build_canvas_binding_rewrite_candidate(
    result: StudioReplayResult,
    *,
    after_graph: Mapping[str, object],
) -> dict[str, object]

Build validated binding YAML candidate from reviewed canvas edits.

Parameters

result : StudioReplayResult The Studio replay result. after_graph : Mapping[str, object] The edited canvas graph after the change.

Returns

dict[str, object] Validated binding YAML candidate from reviewed canvas edits.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/canvas.py 
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
def build_canvas_binding_rewrite_candidate(
    result: StudioReplayResult,
    *,
    after_graph: Mapping[str, object],
) -> dict[str, object]:
    """Build validated binding YAML candidate from reviewed canvas edits.

    Parameters
    ----------
    result : StudioReplayResult
        The Studio replay result.
    after_graph : Mapping[str, object]
        The edited canvas graph after the change.

    Returns
    -------
    dict[str, object]
        Validated binding YAML candidate from reviewed canvas edits.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if not isinstance(result, StudioReplayResult):
        raise ValueError("replay result must be a StudioReplayResult")
    _, after_edges = _normalise_canvas_graph(after_graph, "after_graph")
    before_yaml = result.project_state.binding.yaml_text
    before_digest = sha256(before_yaml.encode("utf-8")).hexdigest()
    unsupported = [
        _require_non_empty_text(edge.get("id"), "canvas edge id")
        for edge in after_edges
        if edge.get("kind") != "cross_channel_coupling"
    ]
    if unsupported:
        return _blocked_binding_rewrite_candidate(
            result,
            before_digest,
            ["only cross_channel_coupling edges can rewrite binding YAML"],
        )

    try:
        candidate_yaml = _rewrite_binding_cross_channel_couplings(
            before_yaml,
            after_edges,
        )
    except ValueError as exc:
        return _blocked_binding_rewrite_candidate(result, before_digest, [str(exc)])

    validation_errors = _validate_candidate_binding_yaml(candidate_yaml)
    return {
        "candidate_kind": "canvas_binding_rewrite_candidate",
        "project_name": result.project_state.project_name,
        "status": "blocked" if validation_errors else "review_ready",
        "binding_spec_rewritten": False,
        "actuation_permitted": False,
        "network_opened": False,
        "before_yaml_sha256": before_digest,
        "candidate_yaml_sha256": sha256(candidate_yaml.encode("utf-8")).hexdigest(),
        "coupling_count_before": _canvas_graph_count(result, "edge_count"),
        "coupling_count_after": len(after_edges),
        "validation_errors": validation_errors,
        "candidate_yaml": candidate_yaml,
    }

build_canvas_edit_artifact 

build_canvas_edit_artifact(
    before_graph: Mapping[str, object],
    after_graph: Mapping[str, object],
) -> ExportManifest

Build a review artefact from edited Studio canvas graph rows.

Parameters

before_graph : Mapping[str, object] The canvas graph before the change. after_graph : Mapping[str, object] The edited canvas graph after the change.

Returns

ExportManifest A review artefact from edited Studio canvas graph rows.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/canvas.py 
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
def build_canvas_edit_artifact(
    before_graph: Mapping[str, object],
    after_graph: Mapping[str, object],
) -> ExportManifest:
    """Build a review artefact from edited Studio canvas graph rows.

    Parameters
    ----------
    before_graph : Mapping[str, object]
        The canvas graph before the change.
    after_graph : Mapping[str, object]
        The edited canvas graph after the change.

    Returns
    -------
    ExportManifest
        A review artefact from edited Studio canvas graph rows.
    """
    before_nodes, before_edges = _normalise_canvas_graph(before_graph, "before_graph")
    after_nodes, after_edges = _normalise_canvas_graph(after_graph, "after_graph")
    payload = json.dumps(
        {
            "artifact": "canvas_edit_review",
            "changed": (before_nodes, before_edges) != (after_nodes, after_edges),
            "node_count_before": len(before_nodes),
            "node_count_after": len(after_nodes),
            "edge_count_before": len(before_edges),
            "edge_count_after": len(after_edges),
            "nodes_before": before_nodes,
            "nodes_after": after_nodes,
            "edges_before": before_edges,
            "edges_after": after_edges,
        },
        sort_keys=True,
        indent=2,
    )
    return ExportManifest.review_artifact(
        target_kind="canvas_edit_review",
        file_name="canvas_edit_review.json",
        payload=payload,
        command="review canvas_edit_review.json before updating binding_spec.yaml",
    )

build_canvas_graph 

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

Return a deterministic layer/coupling graph for Studio canvas review.

Parameters

spec : BindingSpec The binding specification.

Returns

dict[str, object] A deterministic layer/coupling graph for Studio canvas review.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/canvas.py 
 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
def build_canvas_graph(spec: BindingSpec) -> dict[str, object]:
    """Return a deterministic layer/coupling graph for Studio canvas review.

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

    Returns
    -------
    dict[str, object]
        A deterministic layer/coupling graph for Studio canvas review.
    """
    family_channels = {
        family_name: family.channel
        for family_name, family in spec.oscillator_families.items()
    }
    channel_order = {
        channel: index for index, channel in enumerate(sorted(spec.used_channels()))
    }
    channels = tuple(sorted(channel_order))
    nodes: list[dict[str, object]] = []
    for layer in sorted(spec.layers, key=lambda item: item.index):
        family = layer.family or ""
        channel = family_channels.get(family, "")
        nodes.append(
            {
                "id": f"layer_{layer.index}",
                "label": layer.name,
                "kind": "layer",
                "layer_index": int(layer.index),
                "family": family,
                "channel": channel,
                "oscillator_count": len(layer.oscillator_ids),
                "x": float(layer.index) * 220.0,
                "y": float(channel_order.get(channel, 0)) * 140.0,
            }
        )
    for index, channel in enumerate(channels):
        nodes.append(
            {
                "id": _canvas_channel_id(channel),
                "label": channel,
                "kind": "channel",
                "channel": channel,
                "layer_index": -1,
                "family": "",
                "oscillator_count": 0,
                "x": float(index) * 220.0,
                "y": 420.0,
            }
        )

    edges = [
        {
            "id": f"cross_channel_{index}",
            "source": _canvas_channel_id(coupling.source),
            "target": _canvas_channel_id(coupling.target),
            "kind": "cross_channel_coupling",
            "source_channel": coupling.source,
            "target_channel": coupling.target,
            "strength": float(coupling.strength),
            "mode": coupling.mode,
            "template": coupling.template or "",
        }
        for index, coupling in enumerate(spec.cross_channel_couplings, 1)
        if coupling.source in channel_order and coupling.target in channel_order
    ]
    return {
        "canvas_kind": "layer_coupling_graph",
        "node_count": len(nodes),
        "layer_count": len(spec.layers),
        "channel_count": len(channels),
        "edge_count": len(edges),
        "nodes": nodes,
        "edges": edges,
    }

build_canvas_interaction_state 

build_canvas_interaction_state(
    *,
    canvas_artifact: ExportManifest,
    canvas_layout: ExportManifest,
    canvas_patch: ExportManifest,
    canvas_rewrite: Mapping[str, object],
    operator_signoff: bool,
) -> dict[str, object]

Summarise Canvas browser controls for deterministic operator feedback.

Parameters

canvas_artifact : ExportManifest The canvas edit artefact manifest. canvas_layout : ExportManifest The canvas layout manifest. canvas_patch : ExportManifest The canvas topology patch manifest. canvas_rewrite : Mapping[str, object] The canvas binding-rewrite candidate. operator_signoff : bool Whether the operator has signed off.

Returns

dict[str, object] Canvas browser controls for deterministic operator feedback.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/canvas.py 
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
def build_canvas_interaction_state(
    *,
    canvas_artifact: ExportManifest,
    canvas_layout: ExportManifest,
    canvas_patch: ExportManifest,
    canvas_rewrite: Mapping[str, object],
    operator_signoff: bool,
) -> dict[str, object]:
    """Summarise Canvas browser controls for deterministic operator feedback.

    Parameters
    ----------
    canvas_artifact : ExportManifest
        The canvas edit artefact manifest.
    canvas_layout : ExportManifest
        The canvas layout manifest.
    canvas_patch : ExportManifest
        The canvas topology patch manifest.
    canvas_rewrite : Mapping[str, object]
        The canvas binding-rewrite candidate.
    operator_signoff : bool
        Whether the operator has signed off.

    Returns
    -------
    dict[str, object]
        Canvas browser controls for deterministic operator feedback.
    """
    record = json.loads(canvas_artifact.payload)
    changed = bool(record.get("changed"))
    rewrite_status = _require_non_empty_text(
        canvas_rewrite.get("status"),
        "rewrite_status",
    )
    validation_errors = _string_list(
        canvas_rewrite.get("validation_errors", ()),
        "validation_errors",
    )
    disabled_reasons: list[str] = []
    if rewrite_status != "review_ready":
        disabled_reasons.append("binding rewrite candidate is blocked")
    disabled_reasons.extend(validation_errors)
    if not operator_signoff:
        disabled_reasons.append("operator sign-off required")
    apply_enabled = not disabled_reasons
    return {
        "state_kind": "studio_canvas_interaction_state",
        "changed": changed,
        "rewrite_status": rewrite_status,
        "apply_enabled": apply_enabled,
        "disabled_reasons": disabled_reasons,
        "next_action": _canvas_next_action(
            changed=changed,
            rewrite_status=rewrite_status,
            operator_signoff=operator_signoff,
            apply_enabled=apply_enabled,
        ),
        "status_message": (
            "Canvas edits need review before apply."
            if changed
            else "Canvas graph matches the current binding."
        ),
        "download_manifest": [
            canvas_artifact.file_name,
            canvas_layout.file_name,
            canvas_patch.file_name,
            "binding_rewrite_candidate.yaml",
        ],
        "candidate_yaml_sha256": canvas_rewrite.get("candidate_yaml_sha256", ""),
    }

build_canvas_layout_manifest 

build_canvas_layout_manifest(
    *, project_name: str, graph: Mapping[str, object]
) -> ExportManifest

Build a deterministic canvas layout manifest from node positions.

Parameters

project_name : str Name of the project. graph : Mapping[str, object] The canvas graph mapping.

Returns

ExportManifest A deterministic canvas layout manifest from node positions.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/canvas.py 
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
def build_canvas_layout_manifest(
    *,
    project_name: str,
    graph: Mapping[str, object],
) -> ExportManifest:
    """Build a deterministic canvas layout manifest from node positions.

    Parameters
    ----------
    project_name : str
        Name of the project.
    graph : Mapping[str, object]
        The canvas graph mapping.

    Returns
    -------
    ExportManifest
        A deterministic canvas layout manifest from node positions.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    nodes, edges = _normalise_canvas_graph(graph, "canvas_layout")
    positions = []
    for node in sorted(nodes, key=lambda item: str(item.get("id", ""))):
        try:
            node_id = _require_non_empty_text(node.get("id"), "canvas layout id")
            kind = _require_non_empty_text(node.get("kind"), "canvas layout kind")
            label = _require_non_empty_text(node.get("label"), "canvas layout label")
            x = _finite_number(node.get("x"), "canvas layout x")
            y = _finite_number(node.get("y"), "canvas layout y")
        except ValueError as exc:
            raise ValueError(f"canvas layout node is invalid: {exc}") from exc
        positions.append({"id": node_id, "kind": kind, "label": label, "x": x, "y": y})
    payload = json.dumps(
        {
            "manifest_kind": "canvas_layout_manifest",
            "project_name": _require_non_empty_text(project_name, "project_name"),
            "node_count": len(nodes),
            "edge_count": len(edges),
            "positions": positions,
        },
        sort_keys=True,
        indent=2,
    )
    return ExportManifest.review_artifact(
        target_kind="canvas_layout_manifest",
        file_name="canvas_layout_manifest.json",
        payload=payload,
        command="review canvas_layout_manifest.json before restoring Studio layout",
    )

build_canvas_topology_patch 

build_canvas_topology_patch(
    *,
    project_name: str,
    before_graph: Mapping[str, object],
    after_graph: Mapping[str, object],
) -> ExportManifest

Build a review patch for persistent Studio topology edits.

Parameters

project_name : str Name of the project. before_graph : Mapping[str, object] The canvas graph before the change. after_graph : Mapping[str, object] The edited canvas graph after the change.

Returns

ExportManifest A review patch for persistent Studio topology edits.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/canvas.py 
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
def build_canvas_topology_patch(
    *,
    project_name: str,
    before_graph: Mapping[str, object],
    after_graph: Mapping[str, object],
) -> ExportManifest:
    """Build a review patch for persistent Studio topology edits.

    Parameters
    ----------
    project_name : str
        Name of the project.
    before_graph : Mapping[str, object]
        The canvas graph before the change.
    after_graph : Mapping[str, object]
        The edited canvas graph after the change.

    Returns
    -------
    ExportManifest
        A review patch for persistent Studio topology edits.
    """
    before_nodes, before_edges = _normalise_canvas_graph(before_graph, "before_graph")
    after_nodes, after_edges = _normalise_canvas_graph(after_graph, "after_graph")
    _validate_canvas_edge_endpoints(after_nodes, after_edges)

    node_changes = _canvas_item_changes(
        before_nodes,
        after_nodes,
        fields=("id", "kind", "label", "x", "y"),
    )
    edge_changes = _canvas_item_changes(
        before_edges,
        after_edges,
        fields=("id", "kind", "source", "target"),
    )
    changed = any(node_changes[key] or edge_changes[key] for key in node_changes)
    payload = json.dumps(
        {
            "patch_kind": "canvas_topology_patch",
            "project_name": _require_non_empty_text(project_name, "project_name"),
            "status": "review_required",
            "changed": changed,
            "node_count_before": len(before_nodes),
            "node_count_after": len(after_nodes),
            "edge_count_before": len(before_edges),
            "edge_count_after": len(after_edges),
            "node_changes": node_changes,
            "edge_changes": edge_changes,
            "safety": {
                "binding_spec_rewritten": False,
                "actuation_permitted": False,
            },
        },
        sort_keys=True,
        indent=2,
    )
    return ExportManifest.review_artifact(
        target_kind="canvas_topology_patch",
        file_name="canvas_topology_patch.json",
        payload=payload,
        command="review canvas_topology_patch.json before rewriting binding_spec.yaml",
    )

build_oscillator_edit_artifact 

build_oscillator_edit_artifact(
    before_rows: Sequence[Mapping[str, object]],
    after_rows: Sequence[Mapping[str, object]],
) -> ExportManifest

Build a review artefact from edited oscillator table rows.

Parameters

before_rows : Sequence[Mapping[str, object]] The table rows before the change. after_rows : Sequence[Mapping[str, object]] The edited table rows after the change.

Returns

ExportManifest A review artefact from edited oscillator table rows.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/canvas.py 
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
def build_oscillator_edit_artifact(
    before_rows: Sequence[Mapping[str, object]],
    after_rows: Sequence[Mapping[str, object]],
) -> ExportManifest:
    """Build a review artefact from edited oscillator table rows.

    Parameters
    ----------
    before_rows : Sequence[Mapping[str, object]]
        The table rows before the change.
    after_rows : Sequence[Mapping[str, object]]
        The edited table rows after the change.

    Returns
    -------
    ExportManifest
        A review artefact from edited oscillator table rows.
    """
    before = _normalise_table_rows(before_rows, "before_rows")
    after = _normalise_table_rows(after_rows, "after_rows")
    payload = json.dumps(
        {
            "artifact": "oscillator_edit_review",
            "changed": before != after,
            "row_count_before": len(before),
            "row_count_after": len(after),
            "rows_before": before,
            "rows_after": after,
        },
        sort_keys=True,
        indent=2,
    )
    return ExportManifest.review_artifact(
        target_kind="oscillator_edit_review",
        file_name="oscillator_edit_review.json",
        payload=payload,
        command="review oscillator_edit_review.json before updating binding_spec.yaml",
    )

build_integrated_information_panel 

build_integrated_information_panel(
    records: Sequence[Mapping[str, object]],
) -> dict[str, object]

Return a Studio panel payload for integrated-information audit records.

The panel is deliberately pure and non-actuating: it converts validated monitor audit records into a deterministic operator payload suitable for rendering charts, latest-value tiles, and partition review cards. The input must preserve the monitor's explicit claim boundary so Studio cannot display the Phi proxy as a theoretical IIT or consciousness claim.

Parameters

records : Sequence[Mapping[str, object]] The records to summarise.

Returns

dict[str, object] A Studio panel payload for integrated-information audit records.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/charts.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
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
def build_integrated_information_panel(
    records: Sequence[Mapping[str, object]],
) -> dict[str, object]:
    """Return a Studio panel payload for integrated-information audit records.

    The panel is deliberately pure and non-actuating: it converts validated
    monitor audit records into a deterministic operator payload suitable for
    rendering charts, latest-value tiles, and partition review cards. The input
    must preserve the monitor's explicit claim boundary so Studio cannot display
    the Phi proxy as a theoretical IIT or consciousness claim.

    Parameters
    ----------
    records : Sequence[Mapping[str, object]]
        The records to summarise.

    Returns
    -------
    dict[str, object]
        A Studio panel payload for integrated-information audit records.
    """
    normalised_records = _normalise_integrated_information_records(records)
    latest = normalised_records[-1]
    strongest = max(
        normalised_records,
        key=lambda item: cast("float", item["phi"]),
    )
    phi_values = [cast("float", item["phi"]) for item in normalised_records]
    normalised_phi_values = [
        cast("float", item["normalised_phi"]) for item in normalised_records
    ]
    total_integration_values = [
        cast("float", item["total_integration"]) for item in normalised_records
    ]
    return {
        "panel_kind": "studio_integrated_information_panel",
        "monitor": "integrated_information",
        "record_count": len(normalised_records),
        "claim_boundary": "engineering_proxy_not_theoretical_iit",
        "latest": latest,
        "strongest_partition": strongest,
        "series": normalised_records,
        "phi_range": {
            "min": min(phi_values),
            "max": max(phi_values),
        },
        "normalised_phi_range": {
            "min": min(normalised_phi_values),
            "max": max(normalised_phi_values),
        },
        "total_integration_range": {
            "min": min(total_integration_values),
            "max": max(total_integration_values),
        },
        "operator_summary": (
            "latest Phi proxy "
            f"{cast('float', latest['phi']):.6g}; latest normalised Phi "
            f"{cast('float', latest['normalised_phi']):.6g}; records "
            f"{len(normalised_records)}"
        ),
        "operator_action": (
            "render as an engineering integration proxy; preserve the claim "
            "boundary and review the minimum partition before operational use"
        ),
        "actuation_permitted": False,
        "consciousness_claim_permitted": False,
    }

build_regime_chart_payload 

build_regime_chart_payload(
    regimes: Sequence[str],
) -> list[dict[str, object]]

Return deterministic chart rows for regime timelines.

Parameters

regimes : Sequence[str] Per-step regime labels.

Returns

list[dict[str, object]] Deterministic chart rows for regime timelines.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/charts.py 
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
def build_regime_chart_payload(regimes: Sequence[str]) -> list[dict[str, object]]:
    """Return deterministic chart rows for regime timelines.

    Parameters
    ----------
    regimes : Sequence[str]
        Per-step regime labels.

    Returns
    -------
    list[dict[str, object]]
        Deterministic chart rows for regime timelines.
    """
    regime_levels = {
        "critical": 0.0,
        "degraded": 1.0,
        "recovery": 1.5,
        "nominal": 2.0,
    }
    rows: list[dict[str, object]] = []
    for index, regime in enumerate(regimes, 1):
        regime_text = _require_non_empty_text(regime, "regime")
        rows.append(
            {
                "step": index,
                "regime": regime_text,
                "regime_level": regime_levels.get(regime_text, 0.0),
            }
        )
    return rows

build_series_chart_payload 

build_series_chart_payload(
    label: str, values: Sequence[float]
) -> list[dict[str, float | int]]

Return dense chart rows for a scalar time-series.

Parameters

label : str Series or chart label. values : Sequence[float] Scalar time-series values.

Returns

list[dict[str, float | int]] Dense chart rows for a scalar time-series.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/charts.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def build_series_chart_payload(
    label: str,
    values: Sequence[float],
) -> list[dict[str, float | int]]:
    """Return dense chart rows for a scalar time-series.

    Parameters
    ----------
    label : str
        Series or chart label.
    values : Sequence[float]
        Scalar time-series values.

    Returns
    -------
    list[dict[str, float | int]]
        Dense chart rows for a scalar time-series.
    """
    _require_non_empty_text(label, "label")
    return [
        {"step": index, label: _finite_number(value, label)}
        for index, value in enumerate(values, 1)
    ]

build_live_connector_plan 

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

Return non-opening connector ownership guidance for Studio.

Parameters

spec : BindingSpec The binding specification.

Returns

dict[str, object] Non-opening connector ownership guidance for Studio.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/connectors.py 
 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
def build_live_connector_plan(spec: BindingSpec) -> dict[str, object]:
    """Return non-opening connector ownership guidance for Studio.

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

    Returns
    -------
    dict[str, object]
        Non-opening connector ownership guidance for Studio.
    """
    contract = build_digital_twin_binding_contract(spec)
    connector_specs = (
        ("memory", True, False, "review offline memory connector"),
        ("jsonl", True, False, "review JSONL replay connector"),
        ("rest", False, True, "assign connector owner and auth policy"),
        ("grpc", False, True, "assign connector owner and auth policy"),
        ("kafka", False, True, "assign connector owner and auth policy"),
        ("hardware", False, True, "assign connector owner and auth policy"),
    )
    connectors: list[dict[str, object]] = []
    for transport, supports_replay, requires_auth, action in connector_specs:
        compatibility = build_digital_twin_adapter_manifest(
            contract,
            name=f"studio-{transport}",
            transport=transport,
            sync_capabilities=[
                capability.name for capability in contract.sync_capabilities
            ],
            supports_replay=supports_replay,
            requires_auth=requires_auth,
            notes="SPO Studio connector review",
        )
        manifest = compatibility.manifest
        owner_required = transport in {"rest", "grpc", "kafka", "hardware"}
        connectors.append(
            {
                "name": manifest.name,
                "transport": manifest.transport,
                "status": "owner_required" if owner_required else "review_ready",
                "compatible": compatibility.compatible,
                "reasons": list(compatibility.reasons),
                "sync_capabilities": list(manifest.sync_capabilities),
                "supports_replay": manifest.supports_replay,
                "requires_auth": manifest.requires_auth,
                "operator_action": action,
                "network_opened": False,
                "hardware_write_permitted": False,
            }
        )
    return {
        "plan_kind": "studio_live_connector_plan",
        "project_name": spec.name,
        "contract_hash": contract.contract_hash,
        "network_opened": False,
        "actuation_permitted": False,
        "connectors": connectors,
    }

build_live_connector_run_record 

build_live_connector_run_record(
    connector_plan: Mapping[str, object],
    *,
    transport: str,
    payload: Mapping[str, object],
    dry_run: bool = True,
) -> dict[str, object]

Return a gated live-connector execution record without opening transport.

Parameters

connector_plan : Mapping[str, object] The live-connector plan mapping. transport : str Transport identifier. payload : Mapping[str, object] The payload mapping or bytes. dry_run : bool Whether to run without opening transport.

Returns

dict[str, object] A gated live-connector execution record without opening transport.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/connectors.py 
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
def build_live_connector_run_record(
    connector_plan: Mapping[str, object],
    *,
    transport: str,
    payload: Mapping[str, object],
    dry_run: bool = True,
) -> dict[str, object]:
    """Return a gated live-connector execution record without opening transport.

    Parameters
    ----------
    connector_plan : Mapping[str, object]
        The live-connector plan mapping.
    transport : str
        Transport identifier.
    payload : Mapping[str, object]
        The payload mapping or bytes.
    dry_run : bool
        Whether to run without opening transport.

    Returns
    -------
    dict[str, object]
        A gated live-connector execution record without opening transport.
    """
    connector = _connector_by_transport(
        connector_plan,
        _require_non_empty_text(transport, "transport"),
    )
    payload_json = _stable_json_payload(payload, "payload")
    connector_status = _require_non_empty_text(connector.get("status"), "status")
    blocked_reasons: list[str] = []
    if connector_status != "review_ready":
        blocked_reasons.append("connector owner and auth policy required")
    if not dry_run:
        blocked_reasons.append("Studio live execution uses dry-run records only")

    status = "blocked" if blocked_reasons else "accepted"
    return {
        "record_kind": "studio_live_connector_run",
        "project_name": _require_non_empty_text(
            connector_plan.get("project_name"),
            "project_name",
        ),
        "transport": connector["transport"],
        "connector_name": connector["name"],
        "status": status,
        "dry_run": bool(dry_run),
        "payload_sha256": sha256(payload_json.encode("utf-8")).hexdigest(),
        "blocked_reasons": blocked_reasons,
        "operator_action": (
            "review dry-run connector payload"
            if status == "accepted"
            else _require_non_empty_text(
                connector.get("operator_action"),
                "operator_action",
            )
        ),
        "network_opened": False,
        "actuation_permitted": False,
        "hardware_write_permitted": False,
    }

build_owned_live_connector_runtime_record 

build_owned_live_connector_runtime_record(
    result: StudioReplayResult,
    *,
    transport: str,
    owner: str,
    auth_policy: Mapping[str, object],
    payload: Mapping[str, object],
    sequence: int = 1,
    capability: str = "audit_replay",
    direction: str = "twin_to_spo",
) -> dict[str, object]

Validate an owned live connector boundary without opening transport.

Parameters

result : StudioReplayResult The Studio replay result. transport : str Transport identifier. owner : str Owner of the connector boundary. auth_policy : Mapping[str, object] The connector authentication policy. payload : Mapping[str, object] The payload mapping or bytes. sequence : int Monotonic sequence number. capability : str The sync capability identifier. direction : str Sync direction (e.g. inbound/outbound).

Returns

dict[str, object] An owned live connector boundary without opening transport.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/connectors.py 
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
def build_owned_live_connector_runtime_record(
    result: StudioReplayResult,
    *,
    transport: str,
    owner: str,
    auth_policy: Mapping[str, object],
    payload: Mapping[str, object],
    sequence: int = 1,
    capability: str = "audit_replay",
    direction: str = "twin_to_spo",
) -> dict[str, object]:
    """Validate an owned live connector boundary without opening transport.

    Parameters
    ----------
    result : StudioReplayResult
        The Studio replay result.
    transport : str
        Transport identifier.
    owner : str
        Owner of the connector boundary.
    auth_policy : Mapping[str, object]
        The connector authentication policy.
    payload : Mapping[str, object]
        The payload mapping or bytes.
    sequence : int
        Monotonic sequence number.
    capability : str
        The sync capability identifier.
    direction : str
        Sync direction (e.g. ``inbound``/``outbound``).

    Returns
    -------
    dict[str, object]
        An owned live connector boundary without opening transport.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if not isinstance(result, StudioReplayResult):
        raise ValueError("replay result must be a StudioReplayResult")
    checked_transport = _require_non_empty_text(transport, "transport")
    checked_payload = _normalise_json_mapping(
        cast("Mapping[object, object]", payload),
        "payload",
    )
    payload_json = _stable_json_payload(checked_payload, "payload")
    blocked_reasons = _owned_runtime_blocked_reasons(
        result.connector_plan,
        checked_transport,
        owner,
        auth_policy,
    )
    base = _owned_runtime_base_record(
        result,
        transport=checked_transport,
        owner=owner,
        payload_sha256=sha256(payload_json.encode("utf-8")).hexdigest(),
        sequence=sequence,
        capability=capability,
        direction=direction,
    )
    if blocked_reasons:
        return {
            **base,
            "status": "blocked",
            "blocked_reasons": blocked_reasons,
            "response": {},
            "adapter": {},
            "queued_count": 0,
        }

    spec_path = _result_binding_spec_path(result)
    spec = load_binding_spec(spec_path)
    contract = build_digital_twin_binding_contract(spec)
    envelope = build_digital_twin_sync_envelope(
        contract,
        capability=_require_non_empty_text(capability, "capability"),
        direction=_require_non_empty_text(direction, "direction"),
        sequence=_non_negative_int(sequence, "sequence"),
        payload=checked_payload,
    )
    response, adapter_record = _run_owned_live_adapter(
        contract,
        transport=checked_transport,
        envelope_record=envelope.to_audit_record(),
    )
    return {
        **base,
        "status": "accepted" if response.get("accepted") is True else "blocked",
        "blocked_reasons": (
            [] if response.get("accepted") is True else [str(response["reason"])]
        ),
        "response": response,
        "adapter": adapter_record,
        "queued_count": _mapping_count(adapter_record, "queued_count"),
    }

build_command_table 

build_command_table(
    project_state: StudioProjectState,
) -> tuple[dict[str, object], ...]

Return copyable deployment-review commands for ready targets.

Parameters

project_state : StudioProjectState The Studio project state.

Returns

tuple[dict[str, object], ...] Copyable deployment-review commands for ready targets.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/deployment.py 
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
def build_command_table(
    project_state: StudioProjectState,
) -> tuple[dict[str, object], ...]:
    """Return copyable deployment-review commands for ready targets.

    Parameters
    ----------
    project_state : StudioProjectState
        The Studio project state.

    Returns
    -------
    tuple[dict[str, object], ...]
        Copyable deployment-review commands for ready targets.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    readiness = build_deployment_readiness(project_state)
    rows: list[dict[str, object]] = []
    for target in _require_sequence(readiness.get("targets"), "targets"):
        if not isinstance(target, Mapping):
            raise ValueError("readiness targets must be mappings")
        status = _require_non_empty_text(target.get("status"), "status")
        if status == "blocked":
            continue
        target_name = _require_non_empty_text(target.get("target"), "target")
        commands = target.get("commands", ())
        if isinstance(commands, str | bytes) or not isinstance(commands, Sequence):
            raise ValueError("target commands must be a sequence of strings")
        for index, command in enumerate(commands, 1):
            rows.append(
                {
                    "target": target_name,
                    "command_index": index,
                    "command": _require_non_empty_text(command, "command"),
                    "status": status,
                }
            )
    return tuple(rows)

build_deployment_package 

build_deployment_package(
    project_state: StudioProjectState,
) -> dict[str, object]

Return a deterministic deployment package manifest for Studio.

Parameters

project_state : StudioProjectState The Studio project state.

Returns

dict[str, object] A deterministic deployment package manifest for Studio.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/deployment.py 
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
def build_deployment_package(
    project_state: StudioProjectState,
) -> dict[str, object]:
    """Return a deterministic deployment package manifest for Studio.

    Parameters
    ----------
    project_state : StudioProjectState
        The Studio project state.

    Returns
    -------
    dict[str, object]
        A deterministic deployment package manifest for Studio.
    """
    readiness = build_deployment_readiness(project_state)
    targets = _readiness_targets(readiness)
    blocked_reasons = _deployment_blocked_reasons(project_state.exports)
    return {
        "package_kind": "studio_deployment_package",
        "project_name": project_state.project_name,
        "overall_status": readiness["overall_status"],
        "ready_targets": [
            target["target"] for target in targets if target["status"] == "ready"
        ],
        "postponed_targets": [
            target["target"] for target in targets if target["status"] == "postponed"
        ],
        "blocked_targets": [
            target["target"] for target in targets if target["status"] == "blocked"
        ],
        "blocked_reasons": list(blocked_reasons),
        "required_artifacts": _unique_artifacts(targets),
        "export_artifacts": [
            {
                "target_kind": manifest.target_kind,
                "file_name": manifest.file_name,
                "payload_sha256": manifest.payload_sha256,
                "safety_posture": manifest.safety_posture,
                "warnings": list(manifest.warnings),
            }
            for manifest in project_state.exports
        ],
        "commands": list(build_command_table(project_state)),
        "safety_gates": [
            "local replay completed",
            (
                "binding validation blocked"
                if blocked_reasons
                else "binding validation passed"
            ),
            "live actuation disabled",
            "hardware output requires verified evidence",
        ],
        "readiness": readiness,
    }

build_deployment_readiness 

build_deployment_readiness(
    project_state: StudioProjectState,
) -> dict[str, object]

Return target-specific deployment readiness guidance for Studio.

Parameters

project_state : StudioProjectState The Studio project state.

Returns

dict[str, object] Target-specific deployment readiness guidance for Studio.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/deployment.py 
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
def build_deployment_readiness(
    project_state: StudioProjectState,
) -> dict[str, object]:
    """Return target-specific deployment readiness guidance for Studio.

    Parameters
    ----------
    project_state : StudioProjectState
        The Studio project state.

    Returns
    -------
    dict[str, object]
        Target-specific deployment readiness guidance for Studio.
    """
    blocked_reasons = _deployment_blocked_reasons(project_state.exports)
    if blocked_reasons:
        return {
            "project_name": project_state.project_name,
            "overall_status": "blocked",
            "operator_next_step": "fix binding validation errors",
            "targets": [
                _blocked_target("docker", blocked_reasons),
                _blocked_target("wasm", blocked_reasons),
                _blocked_target("hardware", blocked_reasons),
            ],
        }

    return {
        "project_name": project_state.project_name,
        "overall_status": "review_ready",
        "operator_next_step": "review target-specific packaging",
        "targets": [
            {
                "target": "docker",
                "status": "ready",
                "required_artifacts": [
                    "binding_spec.yaml",
                    "spo_studio_audit.json",
                    "docker_manifest.json",
                ],
                "commands": [
                    "docker compose config",
                    "docker build -t scpn-phase-orchestrator:local .",
                    "docker run --rm -v $PWD:/workspace "
                    "scpn-phase-orchestrator:local "
                    "spo run binding_spec.yaml --audit audit.jsonl",
                ],
                "operator_action": "run docker manifest review before packaging",
            },
            {
                "target": "wasm",
                "status": "ready",
                "required_artifacts": [
                    "binding_spec.yaml",
                    "spo_studio_audit.json",
                    "wasm_manifest.json",
                ],
                "commands": [
                    "cd spo-kernel && wasm-pack build crates/spo-wasm "
                    "--target web --out-dir ../../../docs/wasm-pkg",
                ],
                "operator_action": "review browser-safe replay constraints",
            },
            {
                "target": "hardware",
                "status": "postponed",
                "required_artifacts": [
                    "binding_spec.yaml",
                    "spo_studio_audit.json",
                    "verified_hardware_target_evidence",
                ],
                "commands": [],
                "operator_action": "attach verified hardware-target evidence",
            },
        ],
    }

build_export_manifests 

build_export_manifests(
    *,
    project_name: str,
    binding_yaml: str,
    audit_payload: Mapping[str, object],
    validation_errors: Sequence[str],
) -> tuple[ExportManifest, ...]

Build review-only export manifests for Studio.

Parameters

project_name : str Name of the project. binding_yaml : str The binding spec serialised as YAML. audit_payload : Mapping[str, object] The audit payload mapping. validation_errors : Sequence[str] Binding validation error messages.

Returns

tuple[ExportManifest, ...] Review-only export manifests for Studio.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/deployment.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def build_export_manifests(
    *,
    project_name: str,
    binding_yaml: str,
    audit_payload: Mapping[str, object],
    validation_errors: Sequence[str],
) -> tuple[ExportManifest, ...]:
    """Build review-only export manifests for Studio.

    Parameters
    ----------
    project_name : str
        Name of the project.
    binding_yaml : str
        The binding spec serialised as YAML.
    audit_payload : Mapping[str, object]
        The audit payload mapping.
    validation_errors : Sequence[str]
        Binding validation error messages.

    Returns
    -------
    tuple[ExportManifest, ...]
        Review-only export manifests for Studio.
    """
    deploy_warnings = disabled_export_reasons(validation_errors)
    audit_export_payload = {
        **dict(audit_payload),
        "enabled": not deploy_warnings,
        "disabled_reasons": list(deploy_warnings),
    }
    audit_json = json.dumps(audit_export_payload, sort_keys=True, indent=2)
    docker_payload = json.dumps(
        {
            "project_name": project_name,
            "image": "scpn-phase-orchestrator:local",
            "command": "spo run binding_spec.yaml --audit audit.jsonl",
            "enabled": not deploy_warnings,
            "disabled_reasons": list(deploy_warnings),
        },
        sort_keys=True,
        indent=2,
    )
    wasm_payload = json.dumps(
        {
            "project_name": project_name,
            "target": "wasm_review_manifest",
            "enabled": not deploy_warnings,
            "disabled_reasons": list(deploy_warnings),
        },
        sort_keys=True,
        indent=2,
    )
    return (
        ExportManifest.review_artifact(
            target_kind="binding_spec",
            file_name="binding_spec.yaml",
            payload=binding_yaml,
            command="spo run binding_spec.yaml --audit audit.jsonl",
            warnings=deploy_warnings,
        ),
        ExportManifest.review_artifact(
            target_kind="audit_summary",
            file_name="spo_studio_audit.json",
            payload=audit_json,
            command="spo audit summary spo_studio_audit.json",
            warnings=deploy_warnings,
        ),
        ExportManifest.review_artifact(
            target_kind="docker_manifest",
            file_name="docker_manifest.json",
            payload=docker_payload,
            command="docker compose config",
            warnings=deploy_warnings,
        ),
        ExportManifest.review_artifact(
            target_kind="wasm_manifest",
            file_name="wasm_manifest.json",
            payload=wasm_payload,
            command="spo export wasm --manifest wasm_manifest.json",
            warnings=deploy_warnings,
        ),
    )

build_operator_checklist 

build_operator_checklist(
    project_state: StudioProjectState,
) -> tuple[dict[str, object], ...]

Return beginner-friendly ordered deployment steps for Studio.

Parameters

project_state : StudioProjectState The Studio project state.

Returns

tuple[dict[str, object], ...] Beginner-friendly ordered deployment steps for Studio.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/deployment.py 
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
def build_operator_checklist(
    project_state: StudioProjectState,
) -> tuple[dict[str, object], ...]:
    """Return beginner-friendly ordered deployment steps for Studio.

    Parameters
    ----------
    project_state : StudioProjectState
        The Studio project state.

    Returns
    -------
    tuple[dict[str, object], ...]
        Beginner-friendly ordered deployment steps for Studio.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    readiness = build_deployment_readiness(project_state)
    validation_blocked = readiness["overall_status"] == "blocked"
    steps: list[dict[str, object]] = [
        {
            "step": 1,
            "title": "Run local replay",
            "status": (
                "complete"
                if project_state.runtime.replay_status == "completed"
                else "blocked"
            ),
            "detail": project_state.runtime.replay_status,
        },
        {
            "step": 2,
            "title": "Validate binding",
            "status": "blocked" if validation_blocked else "complete",
            "detail": (
                "; ".join(_deployment_blocked_reasons(project_state.exports))
                if validation_blocked
                else "binding validation passed"
            ),
        },
    ]
    for target in _require_sequence(readiness.get("targets"), "targets"):
        if not isinstance(target, Mapping):
            raise ValueError("readiness targets must be mappings")
        target_name = _require_non_empty_text(target.get("target"), "target")
        status = _require_non_empty_text(target.get("status"), "status")
        operator_action = _require_non_empty_text(
            target.get("operator_action"),
            "operator_action",
        )
        blocked_detail = "; ".join(
            str(reason)
            for reason in _require_sequence(
                target.get("blocked_reasons", ()),
                "blocked_reasons",
            )
        )
        steps.append(
            {
                "step": len(steps) + 1,
                "title": f"Review {target_name} packaging",
                "target": target_name,
                "status": status,
                "detail": blocked_detail or operator_action,
            }
        )
    return tuple(steps)

build_package_materialisation_plan 

build_package_materialisation_plan(
    project_state: StudioProjectState,
) -> dict[str, object]

Return ordered, operator-invoked package materialisation commands.

Parameters

project_state : StudioProjectState The Studio project state.

Returns

dict[str, object] Ordered, operator-invoked package materialisation commands.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/deployment.py 
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
def build_package_materialisation_plan(
    project_state: StudioProjectState,
) -> dict[str, object]:
    """Return ordered, operator-invoked package materialisation commands.

    Parameters
    ----------
    project_state : StudioProjectState
        The Studio project state.

    Returns
    -------
    dict[str, object]
        Ordered, operator-invoked package materialisation commands.
    """
    package = build_deployment_package(project_state)
    command_rows = build_command_table(project_state)
    commands = [
        {
            "step": index,
            "target": _require_non_empty_text(row.get("target"), "target"),
            "command": _require_non_empty_text(row.get("command"), "command"),
            "status": _require_non_empty_text(row.get("status"), "status"),
            "requires_operator": True,
            "writes_artifact": _materialisation_command_writes_artifact(
                row.get("command")
            ),
        }
        for index, row in enumerate(command_rows, 1)
    ]
    readiness = build_deployment_readiness(project_state)
    targets = _readiness_targets(readiness)
    return {
        "plan_kind": "studio_package_materialisation_plan",
        "project_name": project_state.project_name,
        "overall_status": package["overall_status"],
        "execution_mode": "operator_invoked",
        "network_opened": False,
        "hardware_write_permitted": False,
        "commands": commands,
        "blocked_targets": list(
            _require_sequence(package.get("blocked_targets"), "blocked_targets")
        ),
        "blocked_reasons": list(
            _require_sequence(package.get("blocked_reasons"), "blocked_reasons")
        ),
        "postponed_targets": [
            {
                "target": target["target"],
                "reason": _require_non_empty_text(
                    target.get("operator_action"),
                    "operator_action",
                ),
            }
            for target in targets
            if target["status"] == "postponed"
        ],
        "required_artifacts": list(
            _require_sequence(package.get("required_artifacts"), "required_artifacts")
        ),
        "safety_gates": list(
            _require_sequence(package.get("safety_gates"), "safety_gates")
        ),
    }

build_service_process_manifest 

build_service_process_manifest(
    project_state: StudioProjectState,
) -> dict[str, object]

Return localhost-only service process packaging for Studio deployment.

Parameters

project_state : StudioProjectState The Studio project state.

Returns

dict[str, object] Localhost-only service process packaging for Studio deployment.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/deployment.py 
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
def build_service_process_manifest(
    project_state: StudioProjectState,
) -> dict[str, object]:
    """Return localhost-only service process packaging for Studio deployment.

    Parameters
    ----------
    project_state : StudioProjectState
        The Studio project state.

    Returns
    -------
    dict[str, object]
        Localhost-only service process packaging for Studio deployment.
    """
    blocked_reasons = _deployment_blocked_reasons(project_state.exports)
    if blocked_reasons:
        return {
            "manifest_kind": "studio_service_process_manifest",
            "project_name": project_state.project_name,
            "overall_status": "blocked",
            "execution_mode": "operator_invoked",
            "network_opened": False,
            "actuation_permitted": False,
            "hardware_write_permitted": False,
            "host_bind": "127.0.0.1",
            "compose_file": "spo_studio_services.compose.yaml",
            "services": [],
            "blocked_reasons": list(blocked_reasons),
            "required_artifacts": [],
            "compose_yaml": "",
            "compose_yaml_sha256": "",
        }

    services = _studio_service_processes()
    compose_yaml = _render_service_compose_yaml(services)
    return {
        "manifest_kind": "studio_service_process_manifest",
        "project_name": project_state.project_name,
        "overall_status": "operator_ready",
        "execution_mode": "operator_invoked",
        "network_opened": False,
        "actuation_permitted": False,
        "hardware_write_permitted": False,
        "host_bind": "127.0.0.1",
        "compose_file": "spo_studio_services.compose.yaml",
        "services": services,
        "blocked_reasons": [],
        "required_artifacts": [
            "binding_spec.yaml",
            "spo_studio_audit.json",
            "docker_manifest.json",
            "owned_connector_runtime.json",
        ],
        "operator_commands": [
            "docker compose -f spo_studio_services.compose.yaml config",
            "docker compose -f spo_studio_services.compose.yaml up spo-studio-ui",
        ],
        "compose_yaml": compose_yaml,
        "compose_yaml_sha256": sha256(compose_yaml.encode("utf-8")).hexdigest(),
    }

disabled_export_reasons 

disabled_export_reasons(
    validation_errors: Sequence[str],
) -> tuple[str, ...]

Return reasons deploy-like exports must stay review-only.

Parameters

validation_errors : Sequence[str] Binding validation error messages.

Returns

tuple[str, ...] Reasons deploy-like exports must stay review-only.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/deployment.py 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
def disabled_export_reasons(validation_errors: Sequence[str]) -> tuple[str, ...]:
    """Return reasons deploy-like exports must stay review-only.

    Parameters
    ----------
    validation_errors : Sequence[str]
        Binding validation error messages.

    Returns
    -------
    tuple[str, ...]
        Reasons deploy-like exports must stay review-only.
    """
    errors = tuple(str(error) for error in validation_errors)
    if not errors:
        return ()
    return (
        "binding validation must pass before deploy manifests are enabled",
        *errors,
    )

binding_spec_project_state 

binding_spec_project_state(
    *,
    project_name: str,
    spec_path: Path,
    knobs: StudioKnobState,
    runtime: RuntimeSnapshot,
) -> StudioProjectState

Create a Studio project state from an existing binding spec file.

Parameters

project_name : str Name of the project. spec_path : Path Filesystem path to the binding-spec file. knobs : StudioKnobState The Studio knob state. runtime : RuntimeSnapshot The workflow runtime snapshot.

Returns

StudioProjectState A Studio project state from an existing binding spec file.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/guidance.py 
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
def binding_spec_project_state(
    *,
    project_name: str,
    spec_path: Path,
    knobs: StudioKnobState,
    runtime: RuntimeSnapshot,
) -> StudioProjectState:
    """Create a Studio project state from an existing binding spec file.

    Parameters
    ----------
    project_name : str
        Name of the project.
    spec_path : Path
        Filesystem path to the binding-spec file.
    knobs : StudioKnobState
        The Studio knob state.
    runtime : RuntimeSnapshot
        The workflow runtime snapshot.

    Returns
    -------
    StudioProjectState
        A Studio project state from an existing binding spec file.
    """
    yaml_text = spec_path.read_text(encoding="utf-8")
    spec = load_binding_spec(spec_path)
    validation_errors = tuple(validate_binding_spec(spec))
    source = ImportedSourceSummary.from_payload(
        source_kind="binding_spec_yaml",
        payload=yaml_text.encode("utf-8"),
        channel_count=max(1, len(spec.used_channels())),
        sample_count=sum(len(layer.oscillator_ids) for layer in spec.layers),
    )
    provenance: dict[str, JsonValue] = {
        "source_path": str(spec_path),
        "knobs": dict(knobs.to_audit_record()),
        "validator": "validate_binding_spec",
    }
    binding = BindingProposal(
        yaml_text=yaml_text,
        validation_errors=validation_errors,
        inferred_channels=tuple(sorted(spec.used_channels())),
        confidence_factors={
            "validator_acceptance": 1.0 if not validation_errors else 0.0,
            "layer_coverage": 1.0 if spec.layers else 0.0,
        },
        provenance=provenance,
    )
    exports = build_export_manifests(
        project_name=project_name,
        binding_yaml=yaml_text,
        audit_payload={
            "project_name": project_name,
            "runtime": runtime.to_audit_record(),
        },
        validation_errors=validation_errors,
    )
    return StudioProjectState(
        project_name=project_name,
        source=source,
        binding=binding,
        runtime=runtime,
        exports=exports,
        metadata={
            "domainpack": project_name,
            "safety": "local_replay_only",
        },
    )

build_beginner_guidance 

build_beginner_guidance(
    result: StudioReplayResult,
) -> dict[str, object]

Return domain-term guidance for first-time Studio operators.

Parameters

result : StudioReplayResult The Studio replay result.

Returns

dict[str, object] Domain-term guidance for first-time Studio operators.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/guidance.py 
 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
def build_beginner_guidance(result: StudioReplayResult) -> dict[str, object]:
    """Return domain-term guidance for first-time Studio operators.

    Parameters
    ----------
    result : StudioReplayResult
        The Studio replay result.

    Returns
    -------
    dict[str, object]
        Domain-term guidance for first-time Studio operators.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if not isinstance(result, StudioReplayResult):
        raise ValueError("replay result must be a StudioReplayResult")
    project = result.project_state
    runtime = project.runtime
    layers = [
        _require_non_empty_text(row.get("name"), "layer")
        for row in result.layer_table
        if isinstance(row, Mapping)
    ]
    channels = [
        _require_non_empty_text(node.get("channel"), "channel")
        for node in _require_sequence(result.canvas_graph.get("nodes", ()), "nodes")
        if isinstance(node, Mapping) and node.get("kind") == "channel"
    ]
    validation_errors = list(project.binding.validation_errors)
    canvas_evidence = {
        "layers": _canvas_graph_count(result, "layer_count"),
        "channels": _canvas_graph_count(result, "channel_count"),
        "couplings": _canvas_graph_count(result, "edge_count"),
    }
    return {
        "guide_kind": "beginner_mode",
        "project_name": project.project_name,
        "actuation_permitted": False,
        "runtime_summary": {
            "replay_status": runtime.replay_status,
            "regime": runtime.regime,
            "R": float(runtime.R),
            "domain_signal": (
                "R summarises how closely the reviewed domain signals move together."
            ),
        },
        "concept_cards": [
            {
                "title": "Signals",
                "plain_language": (
                    "Each layer groups domain measurements that Studio reviews as "
                    "oscillators."
                ),
                "evidence": {
                    "layers": layers,
                    "channels": sorted(channels),
                    "source_kind": project.source.source_kind,
                },
            },
            {
                "title": "Coupling",
                "plain_language": (
                    "K raises or lowers how much the reviewed signals influence "
                    "one another during replay."
                ),
                "evidence": {
                    "K": float(runtime.K),
                    "alpha": float(runtime.alpha),
                    "zeta": float(runtime.zeta),
                    "Psi": float(runtime.Psi),
                    "cross_channel_edges": _canvas_graph_count(result, "edge_count"),
                },
            },
            {
                "title": "Objectives",
                "plain_language": (
                    "The objective is to keep reviewed good layers coherent while "
                    "validation errors block packaging."
                ),
                "evidence": {
                    "validation_errors": validation_errors,
                    "binding_ready": not validation_errors,
                },
            },
            {
                "title": "Supervisor",
                "plain_language": (
                    "The supervisor reads the replay regime and emits review "
                    "evidence only; live actuation stays disabled."
                ),
                "evidence": {
                    "regime": runtime.regime,
                    "hierarchy_watermarks": dict(runtime.hierarchy_watermarks),
                },
            },
        ],
        "next_actions": (
            ["review binding validation"]
            + (["fix validation errors"] if validation_errors else ["review exports"])
            + ["download project_state.json"]
        ),
        "walkthrough_steps": [
            {
                "step": 1,
                "title": "Load project",
                "status": "complete",
                "operator_action": "review source summary",
                "evidence": {"source_kind": project.source.source_kind},
            },
            {
                "step": 2,
                "title": "Run replay",
                "status": (
                    "complete" if runtime.replay_status == "completed" else "blocked"
                ),
                "operator_action": "run local replay",
                "evidence": {"replay_status": runtime.replay_status},
            },
            {
                "step": 3,
                "title": "Review binding",
                "status": "blocked" if validation_errors else "complete",
                "operator_action": (
                    "fix validation errors"
                    if validation_errors
                    else "review binding and continue"
                ),
                "evidence": {"validation_errors": validation_errors},
            },
            {
                "step": 4,
                "title": "Inspect canvas",
                "status": "complete",
                "operator_action": "inspect layer, channel, and coupling graph",
                "evidence": canvas_evidence,
            },
            {
                "step": 5,
                "title": "Prepare exports",
                "status": "blocked" if validation_errors else "ready",
                "operator_action": (
                    "fix validation errors"
                    if validation_errors
                    else "download review artefacts"
                ),
                "evidence": {
                    "export_count": len(result.export_manifests),
                    "connector_count": len(
                        _require_sequence(
                            result.connector_plan.get("connectors", ()),
                            "connectors",
                        )
                    ),
                },
            },
        ],
    }

build_error_report 

build_error_report(
    *,
    operation: str,
    error: Exception,
    project_name: str = "unknown",
) -> dict[str, object]

Return a path-safe operator report for failed Studio actions.

Parameters

operation : str The operation label. error : Exception The exception that was raised. project_name : str Name of the project.

Returns

dict[str, object] A path-safe operator report for failed Studio actions.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/guidance.py 
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
def build_error_report(
    *,
    operation: str,
    error: Exception,
    project_name: str = "unknown",
) -> dict[str, object]:
    """Return a path-safe operator report for failed Studio actions.

    Parameters
    ----------
    operation : str
        The operation label.
    error : Exception
        The exception that was raised.
    project_name : str
        Name of the project.

    Returns
    -------
    dict[str, object]
        A path-safe operator report for failed Studio actions.
    """
    return {
        "project_name": _require_non_empty_text(project_name, "project_name"),
        "operation": _require_non_empty_text(operation, "operation"),
        "status": "blocked",
        "error_type": type(error).__name__,
        "operator_action": "review input artefacts and rerun",
    }

build_runtime_snapshot 

build_runtime_snapshot(
    *,
    final_state: Mapping[str, object],
    knobs: StudioKnobState,
    hierarchy_watermarks: Mapping[str, int] | None = None,
    replay_status: str = "not_started",
) -> RuntimeSnapshot

Build a workflow runtime snapshot from a simulation state dict.

Parameters

final_state : Mapping[str, object] The final simulation state mapping. knobs : StudioKnobState The Studio knob state. hierarchy_watermarks : Mapping[str, int] | None Per-source hierarchy watermarks, or None. replay_status : str Replay status label.

Returns

RuntimeSnapshot A workflow runtime snapshot from a simulation state dict.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/guidance.py 
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
def build_runtime_snapshot(
    *,
    final_state: Mapping[str, object],
    knobs: StudioKnobState,
    hierarchy_watermarks: Mapping[str, int] | None = None,
    replay_status: str = "not_started",
) -> RuntimeSnapshot:
    """Build a workflow runtime snapshot from a simulation state dict.

    Parameters
    ----------
    final_state : Mapping[str, object]
        The final simulation state mapping.
    knobs : StudioKnobState
        The Studio knob state.
    hierarchy_watermarks : Mapping[str, int] | None
        Per-source hierarchy watermarks, or ``None``.
    replay_status : str
        Replay status label.

    Returns
    -------
    RuntimeSnapshot
        A workflow runtime snapshot from a simulation state dict.
    """
    layers = _layer_metrics(final_state.get("layers", ()))
    return RuntimeSnapshot(
        R=_finite_number(final_state.get("R_global", 0.0), "R_global"),
        Psi=knobs.Psi,
        K=knobs.K,
        alpha=knobs.alpha,
        zeta=knobs.zeta,
        regime=_require_non_empty_text(final_state.get("regime", "unknown"), "regime"),
        layer_metrics=layers,
        hierarchy_watermarks=dict(hierarchy_watermarks or {}),
        replay_status=replay_status,
    )

build_hardware_target_package 

build_hardware_target_package(
    result: StudioReplayResult,
) -> dict[str, object]

Return a review-only hardware target package for Studio.

Parameters

result : StudioReplayResult The Studio replay result.

Returns

dict[str, object] A review-only hardware target package for Studio.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/hardware.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def build_hardware_target_package(result: StudioReplayResult) -> dict[str, object]:
    """Return a review-only hardware target package for Studio.

    Parameters
    ----------
    result : StudioReplayResult
        The Studio replay result.

    Returns
    -------
    dict[str, object]
        A review-only hardware target package for Studio.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if not isinstance(result, StudioReplayResult):
        raise ValueError("replay result must be a StudioReplayResult")
    connector_plan = result.connector_plan
    hardware_connector = _connector_by_transport(connector_plan, "hardware")
    return {
        "package_kind": "studio_hardware_target_package",
        "project_name": result.project_state.project_name,
        "overall_status": "evidence_required",
        "contract_hash": _require_non_empty_text(
            connector_plan.get("contract_hash"),
            "contract_hash",
        ),
        "hardware_write_permitted": False,
        "network_opened": False,
        "targets": ["fpga_verilog", "neuromorphic_schedule"],
        "required_evidence": [
            "generated hardware artefact path",
            "simulator parity report",
            "target toolchain version",
            "operator sign-off",
        ],
        "commands": [
            "review connector_plan.json",
            "generate FPGA Verilog with KuramotoVerilogCompiler",
            "run simulator parity before hardware handoff",
        ],
        "connector": hardware_connector,
        "export_artifacts": [
            manifest.to_audit_record() for manifest in result.export_manifests
        ],
    }

build_verified_hardware_target_package 

build_verified_hardware_target_package(
    result: StudioReplayResult,
    *,
    evidence: Mapping[str, object],
) -> dict[str, object]

Return a verified hardware package only when evidence is complete.

Parameters

result : StudioReplayResult The Studio replay result. evidence : Mapping[str, object] Verification evidence mapping.

Returns

dict[str, object] A verified hardware package only when evidence is complete.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/hardware.py 
 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
def build_verified_hardware_target_package(
    result: StudioReplayResult,
    *,
    evidence: Mapping[str, object],
) -> dict[str, object]:
    """Return a verified hardware package only when evidence is complete.

    Parameters
    ----------
    result : StudioReplayResult
        The Studio replay result.
    evidence : Mapping[str, object]
        Verification evidence mapping.

    Returns
    -------
    dict[str, object]
        A verified hardware package only when evidence is complete.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if not isinstance(result, StudioReplayResult):
        raise ValueError("replay result must be a StudioReplayResult")
    if not isinstance(evidence, Mapping):
        raise ValueError("hardware evidence must be a mapping")

    base_package = build_hardware_target_package(result)
    normalised, invalid_evidence = _normalise_hardware_evidence(evidence)
    verified = not invalid_evidence
    return {
        "package_kind": "studio_verified_hardware_target_package",
        "project_name": result.project_state.project_name,
        "overall_status": "review_ready" if verified else "evidence_required",
        "evidence_status": "verified" if verified else "blocked",
        "contract_hash": base_package["contract_hash"],
        "hardware_write_permitted": False,
        "network_opened": False,
        "targets": list(_require_sequence(base_package.get("targets"), "targets")),
        "required_evidence": list(
            _require_sequence(
                base_package.get("required_evidence"),
                "required_evidence",
            )
        ),
        "invalid_evidence": invalid_evidence,
        "evidence": normalised,
        "connector": base_package["connector"],
        "commands": (
            [
                "review verified_hardware_target_package.json",
                "compare generated artefact hash before handoff",
                "archive simulator parity report with package",
            ]
            if verified
            else []
        ),
        "safety_gates": [
            "local replay completed",
            "binding validation passed",
            "hardware evidence verified" if verified else "hardware evidence blocked",
            "hardware output remains operator-controlled",
        ],
        "export_artifacts": list(
            _require_sequence(
                base_package.get("export_artifacts"),
                "export_artifacts",
            )
        ),
    }

build_evolutionary_supervisor_policy_search_studio_panel 

build_evolutionary_supervisor_policy_search_studio_panel(
    reports: Sequence[Mapping[str, object]],
    *,
    examples: Sequence[Mapping[str, object]] = (),
    dsl_reports: Sequence[Mapping[str, object]] = (),
) -> dict[str, object]

Return a Studio panel for offline evolutionary policy-search evidence.

The panel renders deterministic replay search reports, optional enriched domain examples, and optional policy-DSL mutation reports for operator review. It validates hashes, candidate counts, replay/STL summaries, DSL mutation records, and all disabled execution gates before exposing data to Studio. No returned field permits live merge, hot patching, or actuation.

Parameters

reports : Sequence[Mapping[str, object]] The report records. examples : Sequence[Mapping[str, object]] Example records. dsl_reports : Sequence[Mapping[str, object]] Policy-DSL search reports.

Returns

dict[str, object] A Studio panel for offline evolutionary policy-search evidence.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/panel_evolutionary.py 
 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
def build_evolutionary_supervisor_policy_search_studio_panel(
    reports: Sequence[Mapping[str, object]],
    *,
    examples: Sequence[Mapping[str, object]] = (),
    dsl_reports: Sequence[Mapping[str, object]] = (),
) -> dict[str, object]:
    """Return a Studio panel for offline evolutionary policy-search evidence.

    The panel renders deterministic replay search reports, optional enriched
    domain examples, and optional policy-DSL mutation reports for operator
    review. It validates hashes, candidate counts, replay/STL summaries, DSL
    mutation records, and all disabled execution gates before exposing data to
    Studio. No returned field permits live merge, hot patching, or actuation.

    Parameters
    ----------
    reports : Sequence[Mapping[str, object]]
        The report records.
    examples : Sequence[Mapping[str, object]]
        Example records.
    dsl_reports : Sequence[Mapping[str, object]]
        Policy-DSL search reports.

    Returns
    -------
    dict[str, object]
        A Studio panel for offline evolutionary policy-search evidence.
    """
    normalised_reports = _normalise_evolutionary_search_reports(reports)
    normalised_examples, example_rows = _normalise_evolutionary_examples(examples)
    normalised_dsl_reports = _normalise_evolutionary_dsl_reports(dsl_reports)
    candidate_counts = [
        cast("int", report["candidate_count"]) for report in normalised_reports
    ] + [cast("int", report["candidate_count"]) for report in normalised_dsl_reports]
    accepted_total = sum(
        cast("int", report["accepted_count"]) for report in normalised_reports
    ) + sum(cast("int", report["accepted_count"]) for report in normalised_dsl_reports)
    rejected_total = sum(
        cast("int", report["rejected_count"]) for report in normalised_reports
    ) + sum(cast("int", report["rejected_count"]) for report in normalised_dsl_reports)
    best_rows = [
        report["best_candidate"]
        for report in normalised_reports
        if report["best_candidate"] is not None
    ]
    replay_reward_values = [
        cast(
            "float",
            cast("Mapping[str, object]", report["replay_summary"])["mean_reward"],
        )
        for report in normalised_reports
    ]
    return {
        "panel_kind": "studio_evolutionary_supervisor_policy_search_panel",
        "supervisor": "evolutionary_policy_search",
        "search_report_count": len(normalised_reports),
        "dsl_report_count": len(normalised_dsl_reports),
        "example_count": len(normalised_examples),
        "claim_boundary": _EVOLUTIONARY_SEARCH_BOUNDARY,
        "non_actuating": True,
        "execution_disabled": True,
        "operator_review_required": True,
        "hot_patch_permitted": False,
        "live_merge_permitted": False,
        "actuation_permitted": False,
        "search_reports": normalised_reports,
        "dsl_reports": normalised_dsl_reports,
        "example_rows": example_rows,
        "example_domains": tuple(
            sorted({cast("str", example["domain"]) for example in normalised_examples})
        ),
        "best_candidate_rows": tuple(best_rows),
        "candidate_count_range": {
            "minimum": min(candidate_counts),
            "maximum": max(candidate_counts),
        },
        "accepted_candidate_total": accepted_total,
        "rejected_candidate_total": rejected_total,
        "replay_reward_range": {
            "minimum": min(replay_reward_values),
            "maximum": max(replay_reward_values),
        },
        "operator_summary": (
            "evolutionary policy-search review: "
            f"{len(normalised_reports)} replay report(s), "
            f"{len(normalised_dsl_reports)} DSL report(s), "
            f"{len(normalised_examples)} domain example(s), "
            f"{accepted_total} accepted candidate(s)"
        ),
        "operator_action": (
            "render as offline evolutionary review evidence only; compare replay "
            "reward, STL robustness, candidate rejection reasons, and DSL mutation "
            "rows before any separately reviewed policy merge workflow"
        ),
    }

build_hybrid_order_studio_panel 

build_hybrid_order_studio_panel(
    records: Sequence[Mapping[str, object]],
    *,
    scenarios: Sequence[Mapping[str, object]] = (),
) -> dict[str, object]

Return a Studio panel for hybrid classical-quantum order evidence.

The helper renders local simulator evidence only. It preserves the explicit quantum co-simulation claim boundary, validates deterministic record hashes, summarises statevector/density-matrix simulator backends, and folds deterministic scenario fixtures into candidate review rows. Nothing in the payload permits live QPU execution or actuation.

Parameters

records : Sequence[Mapping[str, object]] The records to summarise. scenarios : Sequence[Mapping[str, object]] Scenario records.

Returns

dict[str, object] A Studio panel for hybrid classical-quantum order evidence.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/panel_hybrid_order.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def build_hybrid_order_studio_panel(
    records: Sequence[Mapping[str, object]],
    *,
    scenarios: Sequence[Mapping[str, object]] = (),
) -> dict[str, object]:
    """Return a Studio panel for hybrid classical-quantum order evidence.

    The helper renders local simulator evidence only. It preserves the
    explicit quantum co-simulation claim boundary, validates deterministic
    record hashes, summarises statevector/density-matrix simulator backends,
    and folds deterministic scenario fixtures into candidate review rows.
    Nothing in the payload permits live QPU execution or actuation.

    Parameters
    ----------
    records : Sequence[Mapping[str, object]]
        The records to summarise.
    scenarios : Sequence[Mapping[str, object]]
        Scenario records.

    Returns
    -------
    dict[str, object]
        A Studio panel for hybrid classical-quantum order evidence.
    """
    normalised_records = _normalise_hybrid_order_records(records)
    normalised_scenarios, candidate_rows = _normalise_hybrid_order_scenarios(scenarios)
    entropies = [
        cast("float", record["entanglement_entropy"]) for record in normalised_records
    ]
    normalised_entropies = [
        cast("float", record["normalised_entanglement_entropy"])
        for record in normalised_records
    ]
    participation_ratios = [
        cast("float", record["participation_ratio"]) for record in normalised_records
    ]
    strongest = max(
        normalised_records,
        key=lambda record: cast("float", record["entanglement_entropy"]),
    )
    backends = tuple(
        sorted({cast("str", record["backend"]) for record in normalised_records})
    )
    scenario_domains = tuple(
        sorted({cast("str", scenario["domain"]) for scenario in normalised_scenarios})
    )
    return {
        "panel_kind": "studio_hybrid_order_panel",
        "monitor": "hybrid_entanglement_order_parameter",
        "record_count": len(normalised_records),
        "scenario_count": len(normalised_scenarios),
        "claim_boundary": _HYBRID_ORDER_CLAIM_BOUNDARY,
        "non_actuating": True,
        "execution_disabled": True,
        "actuation_permitted": False,
        "qpu_execution_permitted": False,
        "simulator_backends": backends,
        "scenario_domains": scenario_domains,
        "latest": normalised_records[-1],
        "strongest_entanglement": strongest,
        "series": normalised_records,
        "candidate_rows": candidate_rows,
        "entropy_range": {
            "minimum": min(entropies),
            "maximum": max(entropies),
        },
        "normalised_entanglement_range": {
            "minimum": min(normalised_entropies),
            "maximum": max(normalised_entropies),
        },
        "participation_ratio_range": {
            "minimum": min(participation_ratios),
            "maximum": max(participation_ratios),
        },
        "operator_summary": (
            "hybrid order review: "
            f"{len(normalised_records)} monitor records across "
            f"{len(backends)} local simulator backend(s); "
            f"max entropy {max(entropies):.6g}"
        ),
        "operator_action": (
            "render as local quantum co-simulation evidence only; compare "
            "classical R/Psi with entanglement entropy and keep QPU execution, "
            "actuation, and backend promotion behind separate evidence gates"
        ),
    }

build_information_geometry_studio_panel 

build_information_geometry_studio_panel(
    records: Sequence[Mapping[str, object]],
    *,
    scenarios: Sequence[Mapping[str, object]] = (),
) -> dict[str, object]

Return a Studio panel for information-geometry control evidence.

The panel renders already-computed Fisher-Rao/Wasserstein review proposals and deterministic scenario fixtures. It validates metric tensors, simplex coordinates, natural-gradient tangents, geodesic/curvature metrics, hash fields, and disabled-execution boundaries before exposing anything to the operator surface. No returned field is an executable control channel.

Parameters

records : Sequence[Mapping[str, object]] The records to summarise. scenarios : Sequence[Mapping[str, object]] Scenario records.

Returns

dict[str, object] A Studio panel for information-geometry control evidence.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/panel_information_geometry.py 
 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
def build_information_geometry_studio_panel(
    records: Sequence[Mapping[str, object]],
    *,
    scenarios: Sequence[Mapping[str, object]] = (),
) -> dict[str, object]:
    """Return a Studio panel for information-geometry control evidence.

    The panel renders already-computed Fisher-Rao/Wasserstein review proposals
    and deterministic scenario fixtures. It validates metric tensors, simplex
    coordinates, natural-gradient tangents, geodesic/curvature metrics, hash
    fields, and disabled-execution boundaries before exposing anything to the
    operator surface. No returned field is an executable control channel.

    Parameters
    ----------
    records : Sequence[Mapping[str, object]]
        The records to summarise.
    scenarios : Sequence[Mapping[str, object]]
        Scenario records.

    Returns
    -------
    dict[str, object]
        A Studio panel for information-geometry control evidence.
    """
    normalised_records = _normalise_information_geometry_records(records)
    normalised_scenarios, candidate_rows = _normalise_information_geometry_scenarios(
        scenarios
    )
    fisher_values = [
        cast("float", record["fisher_rao_distance"]) for record in normalised_records
    ]
    wasserstein_values = [
        cast("float", record["wasserstein_distance"]) for record in normalised_records
    ]
    gradient_values = [
        cast("float", record["natural_gradient_norm"]) for record in normalised_records
    ]
    curvature_values = [
        cast("float", record["curvature_proxy"]) for record in normalised_records
    ]
    metric_values = [
        value
        for record in normalised_records
        for row in cast("tuple[tuple[float, ...], ...]", record["metric_tensor"])
        for value in row
    ]
    metric_diagonal_values = [
        row[index]
        for record in normalised_records
        for index, row in enumerate(
            cast("tuple[tuple[float, ...], ...]", record["metric_tensor"])
        )
    ]
    backends = tuple(
        sorted({cast("str", record["backend"]) for record in normalised_records})
    )
    scenario_domains = tuple(
        sorted({cast("str", scenario["domain"]) for scenario in normalised_scenarios})
    )
    return {
        "panel_kind": "studio_information_geometry_panel",
        "supervisor": "information_geometry_control",
        "proposal_count": len(normalised_records),
        "scenario_count": len(normalised_scenarios),
        "claim_boundary": _INFORMATION_GEOMETRY_CLAIM_BOUNDARY,
        "non_actuating": True,
        "execution_disabled": True,
        "actuation_permitted": False,
        "backends": backends,
        "scenario_domains": scenario_domains,
        "latest": normalised_records[-1],
        "series": normalised_records,
        "candidate_rows": candidate_rows,
        "fisher_rao_range": {
            "minimum": min(fisher_values),
            "maximum": max(fisher_values),
        },
        "wasserstein_range": {
            "minimum": min(wasserstein_values),
            "maximum": max(wasserstein_values),
        },
        "natural_gradient_range": {
            "minimum": min(gradient_values),
            "maximum": max(gradient_values),
        },
        "curvature_range": {
            "minimum": min(curvature_values),
            "maximum": max(curvature_values),
        },
        "metric_tensor_range": {
            "minimum": min(metric_values),
            "maximum": max(metric_values),
        },
        "metric_diagonal_range": {
            "minimum": min(metric_diagonal_values),
            "maximum": max(metric_diagonal_values),
        },
        "operator_summary": (
            "information-geometry review: "
            f"{len(normalised_records)} proposal record(s) across "
            f"{len(backends)} backend(s); max Fisher-Rao distance "
            f"{max(fisher_values):.6g}"
        ),
        "operator_action": (
            "render as non-actuating geometry-aware control evidence; compare "
            "Fisher-Rao/Wasserstein distances, metric conditioning, and "
            "natural-gradient magnitude before any separately gated policy use"
        ),
    }

build_autopoietic_lineage_studio_panel 

build_autopoietic_lineage_studio_panel(
    manifests: Sequence[Mapping[str, object]],
) -> dict[str, object]

Build a passive Studio panel for autopoietic lineage sandbox review.

Parameters

manifests : Sequence[Mapping[str, object]] The manifest records.

Returns

dict[str, object] A passive Studio panel for autopoietic lineage sandbox review.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/panel_lineage.py 
 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
def build_autopoietic_lineage_studio_panel(
    manifests: Sequence[Mapping[str, object]],
) -> dict[str, object]:
    """Build a passive Studio panel for autopoietic lineage sandbox review.

    Parameters
    ----------
    manifests : Sequence[Mapping[str, object]]
        The manifest records.

    Returns
    -------
    dict[str, object]
        A passive Studio panel for autopoietic lineage sandbox review.
    """
    normalised_manifests = _normalise_autopoietic_lineage_manifests(manifests)
    replay_corpus_rows = tuple(
        row
        for manifest in normalised_manifests
        for row in cast(tuple[dict[str, object], ...], manifest["replay_corpus"])
    )
    replay_domains = tuple(sorted({str(row["domain"]) for row in replay_corpus_rows}))
    child_rows = tuple(
        child
        for manifest in normalised_manifests
        for child in cast(tuple[dict[str, object], ...], manifest["child_candidates"])
    )
    accepted_child_rows = tuple(
        child for child in child_rows if child["status"] == "accepted_for_review"
    )
    rejected_child_rows = tuple(
        child for child in child_rows if child["status"] == "rejected"
    )

    return {
        "panel_kind": "studio_autopoietic_lineage_panel",
        "supervisor": "autopoietic_lineage_sandbox",
        "manifest_count": len(normalised_manifests),
        "claim_boundary": _AUTOPOIETIC_LINEAGE_BOUNDARY,
        "non_actuating": True,
        "execution_disabled": True,
        "operator_review_required": True,
        "hot_patch_permitted": False,
        "live_merge_permitted": False,
        "actuation_permitted": False,
        "lineage_manifests": normalised_manifests,
        "replay_corpus_rows": replay_corpus_rows,
        "replay_domains": replay_domains,
        "replay_domain_count": len(replay_domains),
        "child_candidate_total": len(child_rows),
        "accepted_child_total": len(accepted_child_rows),
        "rejected_child_total": len(rejected_child_rows),
        "accepted_child_rows": accepted_child_rows,
        "rejected_child_rows": rejected_child_rows,
        "operator_summary": (
            "autopoietic lineage review: "
            f"{len(normalised_manifests)} manifest(s), "
            f"{len(replay_domains)} replay domain(s), "
            f"{len(accepted_child_rows)} accepted child candidate(s)"
        ),
        "operator_action": (
            "render as offline lineage sandbox evidence only; compare replay "
            "domains, policy diffs, and blocked reasons before a separately "
            "reviewed inheritance workflow"
        ),
    }

build_intergenerational_inheritance_studio_panel 

build_intergenerational_inheritance_studio_panel(
    histories: Sequence[Mapping[str, object]],
) -> dict[str, object]

Build a passive Studio panel for inheritance-history review.

Parameters

histories : Sequence[Mapping[str, object]] Inheritance-history records.

Returns

dict[str, object] A passive Studio panel for inheritance-history review.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/panel_lineage.py 
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
def build_intergenerational_inheritance_studio_panel(
    histories: Sequence[Mapping[str, object]],
) -> dict[str, object]:
    """Build a passive Studio panel for inheritance-history review.

    Parameters
    ----------
    histories : Sequence[Mapping[str, object]]
        Inheritance-history records.

    Returns
    -------
    dict[str, object]
        A passive Studio panel for inheritance-history review.
    """
    normalised_histories = _normalise_intergenerational_inheritance_histories(histories)
    child_rows = tuple(
        child
        for history in normalised_histories
        for child in cast(tuple[dict[str, object], ...], history["child_rows"])
    )
    replay_domains = tuple(
        sorted(
            {
                str(domain)
                for history in normalised_histories
                for domain in cast(tuple[str, ...], history["replay_domains"])
            }
        )
    )
    fitness_scores = tuple(
        float(cast(float, row["fitness_score"])) for row in child_rows
    )

    return {
        "panel_kind": "studio_intergenerational_inheritance_panel",
        "supervisor": "intergenerational_policy_inheritance",
        "history_count": len(normalised_histories),
        "claim_boundary": _INTERGENERATIONAL_HISTORY_BOUNDARY,
        "non_actuating": True,
        "execution_disabled": True,
        "operator_review_required": True,
        "direct_hot_patch_permitted": False,
        "hot_patch_permitted": False,
        "live_merge_permitted": False,
        "actuation_permitted": False,
        "histories": normalised_histories,
        "inheritance_child_rows": child_rows,
        "history_record_total": len(child_rows),
        "signed_metadata_total": sum(
            int(cast(int, history["signed_metadata_count"]))
            for history in normalised_histories
        ),
        "replay_domains": replay_domains,
        "replay_domain_count": len(replay_domains),
        "fitness_range": {
            "minimum": min(fitness_scores),
            "maximum": max(fitness_scores),
        },
        "operator_summary": (
            "intergenerational inheritance review: "
            f"{len(normalised_histories)} history package(s), "
            f"{len(child_rows)} signed child record(s), "
            f"{len(replay_domains)} replay domain(s)"
        ),
        "operator_action": (
            "render as signed inheritance-history evidence only; require "
            "separate operator approval before any reviewed hot-patch workflow"
        ),
    }

build_morphogenetic_field_studio_panel 

build_morphogenetic_field_studio_panel(
    svg_artifact: Mapping[str, object],
) -> dict[str, object]

Return a Studio panel payload for morphogenetic field SVG artefacts.

The helper renders already-computed topology-field evidence only. It validates the dependency-free SVG artefact, preserves the snapshot statistics and strongest off-diagonal field edges, and keeps actuation disabled for operator review.

Parameters

svg_artifact : Mapping[str, object] The morphogenetic field SVG artefact.

Returns

dict[str, object] A Studio panel payload for morphogenetic field SVG artefacts.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/panel_morphogenetic.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def build_morphogenetic_field_studio_panel(
    svg_artifact: Mapping[str, object],
) -> dict[str, object]:
    """Return a Studio panel payload for morphogenetic field SVG artefacts.

    The helper renders already-computed topology-field evidence only. It
    validates the dependency-free SVG artefact, preserves the snapshot
    statistics and strongest off-diagonal field edges, and keeps actuation
    disabled for operator review.

    Parameters
    ----------
    svg_artifact : Mapping[str, object]
        The morphogenetic field SVG artefact.

    Returns
    -------
    dict[str, object]
        A Studio panel payload for morphogenetic field SVG artefacts.
    """
    record = _normalise_morphogenetic_field_svg_artifact(svg_artifact)
    snapshot = cast("dict[str, object]", record["snapshot"])
    top_edges = cast("tuple[dict[str, object], ...]", snapshot["top_edges"])
    strongest_edge = top_edges[0] if top_edges else {}
    return {
        "panel_kind": "studio_morphogenetic_field_panel",
        "renderer": "morphogenetic_field_svg",
        "format": "svg",
        "width": record["width"],
        "height": record["height"],
        "shape": snapshot["shape"],
        "snapshot": snapshot,
        "top_edge_count": len(top_edges),
        "strongest_edge": strongest_edge,
        "field_energy": {
            "mean": snapshot["mean"],
            "minimum": snapshot["minimum"],
            "maximum": snapshot["maximum"],
            "l2_norm": snapshot["l2_norm"],
        },
        "svg": record["svg"],
        "actuation_permitted": False,
        "operator_action": (
            "render as passive topology-field evidence; review strongest "
            "off-diagonal edges before any downstream policy action"
        ),
    }

build_multiverse_counterfactual_studio_panel 

build_multiverse_counterfactual_studio_panel(
    manifest: Mapping[str, object],
    risk_report: Mapping[str, object],
) -> dict[str, object]

Return a Studio panel payload for multiverse branch review evidence.

The panel joins a non-actuating rollout manifest with a non-actuating risk gate report. It validates both audit artefacts before rendering branch comparison rows, safest-branch metadata, and coherence ranges for operator review. The helper never emits executable actions.

Parameters

manifest : Mapping[str, object] The manifest object. risk_report : Mapping[str, object] The multiverse risk report mapping.

Returns

dict[str, object] A Studio panel payload for multiverse branch review evidence.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/panel_multiverse.py 
 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
def build_multiverse_counterfactual_studio_panel(
    manifest: Mapping[str, object],
    risk_report: Mapping[str, object],
) -> dict[str, object]:
    """Return a Studio panel payload for multiverse branch review evidence.

    The panel joins a non-actuating rollout manifest with a non-actuating risk
    gate report. It validates both audit artefacts before rendering branch
    comparison rows, safest-branch metadata, and coherence ranges for operator
    review. The helper never emits executable actions.

    Parameters
    ----------
    manifest : Mapping[str, object]
        The manifest object.
    risk_report : Mapping[str, object]
        The multiverse risk report mapping.

    Returns
    -------
    dict[str, object]
        A Studio panel payload for multiverse branch review evidence.
    """
    rollout = _normalise_multiverse_manifest(manifest)
    risk = _normalise_multiverse_risk_report(risk_report)
    branch_rows = _join_multiverse_branch_rows(rollout, risk)
    final_values = [cast("float", row["final_R"]) for row in branch_rows]
    mean_values = [cast("float", row["mean_R"]) for row in branch_rows]
    min_values = [cast("float", row["min_R"]) for row in branch_rows]
    max_values = [cast("float", row["max_R"]) for row in branch_rows]
    rejected_ids = [
        cast("str", row["branch_id"])
        for row in branch_rows
        if row["risk_approved"] is not True
    ]
    return {
        "panel_kind": "studio_multiverse_counterfactual_panel",
        "simulator": "multiverse_counterfactual",
        "risk_gate": "multiverse_branch_risk_gate",
        "schema_version": rollout["schema_version"],
        "risk_schema_version": risk["schema_version"],
        "backend": rollout["backend"],
        "horizon": rollout["horizon"],
        "branch_count": rollout["branch_count"],
        "approved_count": risk["approved_count"],
        "rejected_count": risk["rejected_count"],
        "safest_branch_id": risk["safest_branch_id"],
        "safest_branch_hash": risk["safest_branch_hash"],
        "rejected_branch_ids": rejected_ids,
        "rejection_reasons": risk["rejection_reasons"],
        "manifest_hash": rollout["manifest_hash"],
        "risk_report_hash": risk["report_hash"],
        "claim_boundary": _MULTIVERSE_ROLLOUT_CLAIM_BOUNDARY,
        "risk_claim_boundary": _MULTIVERSE_RISK_CLAIM_BOUNDARY,
        "non_actuating": True,
        "execution_disabled": True,
        "actuation_permitted": False,
        "branch_rows": branch_rows,
        "coherence_range": {
            "minimum": min(min_values),
            "maximum": max(max_values),
            "final_minimum": min(final_values),
            "final_maximum": max(final_values),
            "mean_minimum": min(mean_values),
            "mean_maximum": max(mean_values),
        },
        "operator_summary": (
            "multiverse branch review: "
            f"{risk['approved_count']}/{rollout['branch_count']} branches approved"
        ),
        "operator_action": (
            "render as counterfactual review evidence only; no branch action "
            "may be applied without a separate safety-gated control workflow"
        ),
    }

build_sheaf_cohomology_studio_panel 

build_sheaf_cohomology_studio_panel(
    records: Sequence[Mapping[str, object]],
    *,
    summaries: Sequence[Mapping[str, object]],
    control_proposals: Sequence[Mapping[str, object]],
) -> dict[str, object]

Return a Studio panel for sheaf-cohomology review evidence.

The panel renders already-computed sheaf-Laplacian obstruction records, residual triage summaries, and review-only control proposals. It validates cohomology dimensions, finite obstruction/energy metrics, residual rows, disabled execution gates, and monotone accepted projections before exposing evidence to Studio. No returned field is an executable control channel.

Parameters

records : Sequence[Mapping[str, object]] The records to summarise. summaries : Sequence[Mapping[str, object]] Summary records. control_proposals : Sequence[Mapping[str, object]] Sheaf control-proposal records.

Returns

dict[str, object] A Studio panel for sheaf-cohomology review evidence.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/panel_sheaf.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
def build_sheaf_cohomology_studio_panel(
    records: Sequence[Mapping[str, object]],
    *,
    summaries: Sequence[Mapping[str, object]],
    control_proposals: Sequence[Mapping[str, object]],
) -> dict[str, object]:
    """Return a Studio panel for sheaf-cohomology review evidence.

    The panel renders already-computed sheaf-Laplacian obstruction records,
    residual triage summaries, and review-only control proposals. It validates
    cohomology dimensions, finite obstruction/energy metrics, residual rows,
    disabled execution gates, and monotone accepted projections before exposing
    evidence to Studio. No returned field is an executable control channel.

    Parameters
    ----------
    records : Sequence[Mapping[str, object]]
        The records to summarise.
    summaries : Sequence[Mapping[str, object]]
        Summary records.
    control_proposals : Sequence[Mapping[str, object]]
        Sheaf control-proposal records.

    Returns
    -------
    dict[str, object]
        A Studio panel for sheaf-cohomology review evidence.
    """
    normalised_records = _normalise_sheaf_cohomology_records(records)
    normalised_summaries, residual_rows = _normalise_sheaf_obstruction_summaries(
        summaries
    )
    normalised_proposals = _normalise_sheaf_control_proposals(control_proposals)
    obstruction_scores = [
        cast("float", record["obstruction_score"]) for record in normalised_records
    ]
    consistency_energies = [
        cast("float", record["consistency_energy"]) for record in normalised_records
    ]
    kernel_dimensions = [
        cast("int", record["kernel_dimension"]) for record in normalised_records
    ]
    obstruction_dimensions = [
        cast("int", record["obstruction_dimension"]) for record in normalised_records
    ]
    accepted_count = sum(
        1
        for proposal in normalised_proposals
        if proposal["accepted_for_review"] is True
    )
    critical_count = sum(
        1 for summary in normalised_summaries if summary["severity"] == "critical"
    )
    return {
        "panel_kind": "studio_sheaf_cohomology_panel",
        "supervisor": "sheaf_cohomology_control",
        "claim_boundary": _SHEAF_COHOMOLOGY_CLAIM_BOUNDARY,
        "non_actuating": True,
        "execution_disabled": True,
        "operator_review_required": True,
        "actuation_permitted": False,
        "live_merge_permitted": False,
        "hot_patch_permitted": False,
        "record_count": len(normalised_records),
        "summary_count": len(normalised_summaries),
        "control_proposal_count": len(normalised_proposals),
        "accepted_control_proposal_count": accepted_count,
        "critical_summary_count": critical_count,
        "records": normalised_records,
        "summaries": normalised_summaries,
        "control_proposals": normalised_proposals,
        "top_residual_rows": residual_rows,
        "obstruction_range": {
            "minimum": min(obstruction_scores),
            "maximum": max(obstruction_scores),
        },
        "consistency_energy_range": {
            "minimum": min(consistency_energies),
            "maximum": max(consistency_energies),
        },
        "cohomology_dimension_range": {
            "kernel_minimum": min(kernel_dimensions),
            "kernel_maximum": max(kernel_dimensions),
            "obstruction_minimum": min(obstruction_dimensions),
            "obstruction_maximum": max(obstruction_dimensions),
        },
        "operator_summary": (
            "sheaf-cohomology review: "
            f"{len(normalised_records)} obstruction record(s), "
            f"{len(residual_rows)} residual edge row(s), "
            f"{accepted_count}/{len(normalised_proposals)} accepted proposal(s)"
        ),
        "operator_action": (
            "render as non-actuating sheaf-Laplacian obstruction evidence; "
            "review residual edges and cohomology-dimension changes before any "
            "separately approved operator workflow"
        ),
    }

build_strange_loop_studio_panel 

build_strange_loop_studio_panel(
    records: Sequence[Mapping[str, object]],
) -> dict[str, object]

Return a Studio panel payload for strange-loop drift scenario records.

The panel renders precomputed StrangeLoopSupervisor review evidence. It does not observe live actions, execute recommendations, or apply control changes. All scenario records must keep the supervisor's non-actuating claim boundary and disabled-execution flags intact.

Parameters

records : Sequence[Mapping[str, object]] The records to summarise.

Returns

dict[str, object] A Studio panel payload for strange-loop drift scenario records.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/panel_strange_loop.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def build_strange_loop_studio_panel(
    records: Sequence[Mapping[str, object]],
) -> dict[str, object]:
    """Return a Studio panel payload for strange-loop drift scenario records.

    The panel renders precomputed ``StrangeLoopSupervisor`` review evidence.
    It does not observe live actions, execute recommendations, or apply control
    changes. All scenario records must keep the supervisor's non-actuating
    claim boundary and disabled-execution flags intact.

    Parameters
    ----------
    records : Sequence[Mapping[str, object]]
        The records to summarise.

    Returns
    -------
    dict[str, object]
        A Studio panel payload for strange-loop drift scenario records.
    """
    normalised_records = _normalise_strange_loop_records(records)
    drift_scores = [
        cast("float", record["max_drift_score"]) for record in normalised_records
    ]
    oscillation_scores = [
        cast("float", record["max_oscillation_score"]) for record in normalised_records
    ]
    overcontrol_scores = [
        cast("float", record["max_overcontrol_score"]) for record in normalised_records
    ]
    coherence_scores = [
        cast("float", record["min_control_coherence"]) for record in normalised_records
    ]
    failed_ids = [
        cast("str", record["scenario_id"])
        for record in normalised_records
        if record["passed_expected_trigger"] is not True
    ]
    triggered_modes = tuple(
        sorted(
            {cast("str", record["expected_trigger"]) for record in normalised_records}
        )
    )
    return {
        "panel_kind": "studio_strange_loop_panel",
        "supervisor": "strange_loop",
        "scenario_count": len(normalised_records),
        "passed_count": len(normalised_records) - len(failed_ids),
        "failed_scenario_ids": failed_ids,
        "triggered_modes": triggered_modes,
        "claim_boundary": _STRANGE_LOOP_CLAIM_BOUNDARY,
        "non_actuating": True,
        "execution_disabled": True,
        "actuation_permitted": False,
        "series": normalised_records,
        "maxima": {
            "drift_score": max(drift_scores),
            "oscillation_score": max(oscillation_scores),
            "overcontrol_score": max(overcontrol_scores),
        },
        "minima": {
            "control_coherence": min(coherence_scores),
        },
        "operator_summary": (
            "strange-loop scenario review: "
            f"{len(normalised_records) - len(failed_ids)}/"
            f"{len(normalised_records)} expected triggers passed"
        ),
        "operator_action": (
            "render as offline supervisor self-control evidence; keep all "
            "recommendations behind the normal review and safety gate"
        ),
    }

build_topos_semantic_binding_studio_panel 

build_topos_semantic_binding_studio_panel(
    symbolic_reports: Sequence[Mapping[str, object]],
    policy_reports: Sequence[Mapping[str, object]],
    *,
    examples: Sequence[Mapping[str, object]] = (),
) -> dict[str, object]

Return a Studio panel for Topos semantic-binding evidence.

The helper renders categorical validation reports and deterministic domain obligations as review evidence only. It validates schema names, proof boundaries, report hashes, obligation/object/morphism counts, non-actuation flags, and example hashes before exposing a compact payload to Studio. The payload intentionally makes no formal Topos proof claim and emits no executable policy actions.

Parameters

symbolic_reports : Sequence[Mapping[str, object]] Symbolic-binding reports. policy_reports : Sequence[Mapping[str, object]] Policy-composition reports. examples : Sequence[Mapping[str, object]] Example records.

Returns

dict[str, object] A Studio panel for Topos semantic-binding evidence.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/panel_topos.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def build_topos_semantic_binding_studio_panel(
    symbolic_reports: Sequence[Mapping[str, object]],
    policy_reports: Sequence[Mapping[str, object]],
    *,
    examples: Sequence[Mapping[str, object]] = (),
) -> dict[str, object]:
    """Return a Studio panel for Topos semantic-binding evidence.

    The helper renders categorical validation reports and deterministic domain
    obligations as review evidence only. It validates schema names, proof
    boundaries, report hashes, obligation/object/morphism counts, non-actuation
    flags, and example hashes before exposing a compact payload to Studio. The
    payload intentionally makes no formal Topos proof claim and emits no
    executable policy actions.

    Parameters
    ----------
    symbolic_reports : Sequence[Mapping[str, object]]
        Symbolic-binding reports.
    policy_reports : Sequence[Mapping[str, object]]
        Policy-composition reports.
    examples : Sequence[Mapping[str, object]]
        Example records.

    Returns
    -------
    dict[str, object]
        A Studio panel for Topos semantic-binding evidence.
    """
    normalised_symbolic = _normalise_topos_validation_reports(
        symbolic_reports,
        schema_name="symbolic_binding_functor",
        label="symbolic binding report",
    )
    normalised_policy = _normalise_topos_validation_reports(
        policy_reports,
        schema_name="policy_composition_category",
        label="policy composition report",
    )
    normalised_examples, example_rows = _normalise_topos_domain_examples(examples)
    all_reports = (*normalised_symbolic, *normalised_policy)
    object_counts = [cast("int", report["object_count"]) for report in all_reports]
    morphism_counts = [cast("int", report["morphism_count"]) for report in all_reports]
    failed_symbolic = [
        cast("str", report["report_hash"])
        for report in normalised_symbolic
        if report["passed"] is not True
    ]
    failed_policy = [
        cast("str", report["report_hash"])
        for report in normalised_policy
        if report["passed"] is not True
    ]
    return {
        "panel_kind": "studio_topos_semantic_binding_panel",
        "proof_surface": "topos_semantic_binding",
        "symbolic_report_count": len(normalised_symbolic),
        "policy_report_count": len(normalised_policy),
        "example_count": len(normalised_examples),
        "passed_symbolic_report_count": len(normalised_symbolic) - len(failed_symbolic),
        "passed_policy_report_count": len(normalised_policy) - len(failed_policy),
        "failed_symbolic_report_hashes": failed_symbolic,
        "failed_policy_report_hashes": failed_policy,
        "proof_boundary": _TOPOS_PROOF_BOUNDARY,
        "non_actuating": True,
        "actuation_permitted": False,
        "formal_proof_claim_permitted": False,
        "symbolic_reports": normalised_symbolic,
        "policy_reports": normalised_policy,
        "example_rows": example_rows,
        "example_domains": tuple(
            sorted({cast("str", example["domain"]) for example in normalised_examples})
        ),
        "object_count_range": {
            "minimum": min(object_counts),
            "maximum": max(object_counts),
        },
        "morphism_count_range": {
            "minimum": min(morphism_counts),
            "maximum": max(morphism_counts),
        },
        "operator_summary": (
            "Topos semantic-binding review: "
            f"{len(normalised_symbolic)} symbolic report(s), "
            f"{len(normalised_policy)} policy report(s), "
            f"{len(normalised_examples)} domain example(s)"
        ),
        "operator_action": (
            "render as categorical validation prototype evidence only; preserve "
            "the proof boundary and require a separate formal-methods gate "
            "before claiming machine-checked Topos proofs or applying policy"
        ),
    }

build_twin_confidence_studio_panel 

build_twin_confidence_studio_panel(
    score_records: Sequence[Mapping[str, object]],
    summary_record: Mapping[str, object],
) -> dict[str, object]

Return a Studio panel for digital-twin confidence audit evidence.

The panel consumes the real audit records emitted by monitor.twin_confidence. It validates score and summary hashes, cross-checks summary aggregates against the supplied tick records, preserves calibrated confidence/status evidence, and keeps the operator surface observability-only: no actuation, hot patch, live merge, or execution gate is enabled by this payload.

Parameters

score_records : Sequence[Mapping[str, object]] Chronological TwinConfidenceScore.to_audit_record() mappings. summary_record : Mapping[str, object] TwinConfidenceSummary.to_audit_record() mapping for the same score sequence.

Returns

dict[str, object] A review-only Studio panel payload for twin-confidence evidence.

Raises

ValueError If a record is malformed, a hash does not match, or the summary does not describe the supplied score sequence.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/panel_twin_confidence.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def build_twin_confidence_studio_panel(
    score_records: Sequence[Mapping[str, object]],
    summary_record: Mapping[str, object],
) -> dict[str, object]:
    """Return a Studio panel for digital-twin confidence audit evidence.

    The panel consumes the real audit records emitted by
    ``monitor.twin_confidence``. It validates score and summary hashes,
    cross-checks summary aggregates against the supplied tick records, preserves
    calibrated confidence/status evidence, and keeps the operator surface
    observability-only: no actuation, hot patch, live merge, or execution gate is
    enabled by this payload.

    Parameters
    ----------
    score_records : Sequence[Mapping[str, object]]
        Chronological ``TwinConfidenceScore.to_audit_record()`` mappings.
    summary_record : Mapping[str, object]
        ``TwinConfidenceSummary.to_audit_record()`` mapping for the same score
        sequence.

    Returns
    -------
    dict[str, object]
        A review-only Studio panel payload for twin-confidence evidence.

    Raises
    ------
    ValueError
        If a record is malformed, a hash does not match, or the summary does not
        describe the supplied score sequence.
    """
    scores = _normalise_twin_confidence_scores(score_records)
    summary = _normalise_twin_confidence_summary(summary_record)
    _require_matching_summary(scores, summary)

    latest = scores[-1]
    worst = max(
        scores,
        key=lambda score: (
            _STATUS_LEVELS[cast("str", score["status"])],
            cast("float", score["composite_z"]),
        ),
    )
    backends = tuple(sorted({cast("str", score["backend"]) for score in scores}))
    status_counts = {
        "healthy": summary["healthy_count"],
        "warning": summary["warning_count"],
        "critical": summary["critical_count"],
    }
    return {
        "panel_kind": "studio_twin_confidence_panel",
        "monitor": "digital_twin_confidence",
        "claim_boundary": _TWIN_CONFIDENCE_CLAIM_BOUNDARY,
        "non_actuating": True,
        "execution_disabled": True,
        "actuation_permitted": False,
        "live_merge_permitted": False,
        "hot_patch_permitted": False,
        "score_count": len(scores),
        "status_counts": status_counts,
        "worst_status_level": _STATUS_LEVELS[cast("str", summary["worst_status"])],
        "latest": latest,
        "worst": worst,
        "summary": summary,
        "series": scores,
        "backends": backends,
        "operator_summary": (
            "twin-confidence review: "
            f"{summary['tick_count']} scored ticks, worst status "
            f"{summary['worst_status']}, latest confidence "
            f"{cast('float', summary['latest_confidence']):.6g}"
        ),
        "operator_action": (
            "render as digital-twin observability evidence only; investigate "
            "warning or critical confidence drift before considering any "
            "separate reviewed control action"
        ),
    }

run_binding_spec_replay 

run_binding_spec_replay(
    spec_path: Path, *, steps: int, knobs: StudioKnobState
) -> StudioReplayResult

Run a local binding-spec replay and return Studio-ready payloads.

Parameters

spec_path : Path Filesystem path to the binding-spec file. steps : int Number of replay steps. knobs : StudioKnobState The Studio knob state.

Returns

StudioReplayResult A local binding-spec replay and return Studio-ready payloads.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/replay.py 
 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
def run_binding_spec_replay(
    spec_path: Path,
    *,
    steps: int,
    knobs: StudioKnobState,
) -> StudioReplayResult:
    """Run a local binding-spec replay and return Studio-ready payloads.

    Parameters
    ----------
    spec_path : Path
        Filesystem path to the binding-spec file.
    steps : int
        Number of replay steps.
    knobs : StudioKnobState
        The Studio knob state.

    Returns
    -------
    StudioReplayResult
        A local binding-spec replay and return Studio-ready payloads.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if isinstance(steps, bool) or not isinstance(steps, int) or steps < 1:
        raise ValueError("steps must be a positive integer")
    from scpn_phase_orchestrator.runtime.server import SimulationState

    spec = load_binding_spec(spec_path)
    sim: _ReplaySimulationState = SimulationState(spec)
    _apply_replay_knobs(sim, knobs)

    r_history: list[float] = []
    regime_history: list[str] = []
    final_state: Mapping[str, object] = sim.snapshot()
    for _ in range(steps):
        final_state = sim.step()
        r_history.append(_finite_number(final_state["R_global"], "R_global"))
        regime_history.append(_require_non_empty_text(final_state["regime"], "regime"))

    runtime = build_runtime_snapshot(
        final_state=final_state,
        knobs=knobs,
        replay_status="completed",
    )
    project_state = binding_spec_project_state(
        project_name=spec.name,
        spec_path=spec_path,
        knobs=knobs,
        runtime=runtime,
    )
    return StudioReplayResult(
        project_state=project_state,
        r_history=tuple(r_history),
        regime_history=tuple(regime_history),
        layer_table=build_layer_table(spec),
        oscillator_table=build_oscillator_table(spec),
        canvas_graph=build_canvas_graph(spec),
        connector_plan=build_live_connector_plan(spec),
        export_manifests=project_state.exports,
    )

build_layer_table 

build_layer_table(
    spec: BindingSpec,
) -> tuple[dict[str, object], ...]

Return editable layer rows for the Studio oscillator canvas.

Parameters

spec : BindingSpec The binding specification.

Returns

tuple[dict[str, object], ...] Editable layer rows for the Studio oscillator canvas.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/tables.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def build_layer_table(spec: BindingSpec) -> tuple[dict[str, object], ...]:
    """Return editable layer rows for the Studio oscillator canvas.

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

    Returns
    -------
    tuple[dict[str, object], ...]
        Editable layer rows for the Studio oscillator canvas.
    """
    return tuple(
        {
            "index": int(layer.index),
            "name": layer.name,
            "oscillator_count": len(layer.oscillator_ids),
            "family": layer.family or "",
            "omega_count": len(layer.omegas or ()),
        }
        for layer in sorted(spec.layers, key=lambda item: item.index)
    )

build_oscillator_table 

build_oscillator_table(
    spec: BindingSpec,
) -> tuple[dict[str, object], ...]

Return oscillator rows suitable for Streamlit data editing.

Parameters

spec : BindingSpec The binding specification.

Returns

tuple[dict[str, object], ...] Oscillator rows suitable for Streamlit data editing.

Source code in src/scpn_phase_orchestrator/studio/ui_helpers/tables.py 
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
def build_oscillator_table(spec: BindingSpec) -> tuple[dict[str, object], ...]:
    """Return oscillator rows suitable for Streamlit data editing.

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

    Returns
    -------
    tuple[dict[str, object], ...]
        Oscillator rows suitable for Streamlit data editing.
    """
    family_channels = {
        family_name: family.channel
        for family_name, family in spec.oscillator_families.items()
    }
    rows: list[dict[str, object]] = []
    for layer in sorted(spec.layers, key=lambda item: item.index):
        channel = family_channels.get(layer.family or "", "")
        for oscillator_id in layer.oscillator_ids:
            rows.append(
                {
                    "layer": layer.name,
                    "layer_index": int(layer.index),
                    "oscillator_id": oscillator_id,
                    "family": layer.family or "",
                    "channel": channel,
                }
            )
    return tuple(rows)

STUDIO federation manifest

federation_manifest builds the schema-A CapabilityManifest the SCPN STUDIO Hub ingests (scpn-studio-platform): the verbs SPO exposes and their honesty attributes (safety tier, side effect, timing, fidelity, proof, produces, backends). It is distinct from tools/capability_manifest.py, which generates the repository's static public inventory. Honesty is built in by construction — no verb is live-hardware, none claims a discharged formal proof, and the safety tier is research. Requires the studio optional extra. The manifest's ui_module names the federated remote built in studio-web/ (STUDIO_FEDERATION_NAME, STUDIO_EXPOSED_MODULE, STUDIO_REMOTE_ENTRY).

federation_manifest

Build the SCPN STUDIO federation CapabilityManifest for SPO.

This emits the schema-A capability manifest the SCPN STUDIO Hub ingests (scpn-studio-platform), describing the verbs SPO exposes and their honesty attributes (safety tier, side effect, timing, fidelity, proof, produces, backends). It is distinct from tools/capability_manifest.py, which generates the repository's static public inventory.

Honesty is built in by construction: SPO is review-only, so no verb is live-hardware; SPO exports PRISM/TLA+ models but does not discharge them to a proven depth, so no verb claims a formal proof; and the safety tier is research rather than certified/production because SPO is a research-grade toolkit, not a certified product. The platform SDK is an optional dependency (the studio extra); importing this module without it raises a clear error.

The manifest declares the federated UI remote through ui_module: the panel built in studio-web/ and pull-deployed to the studio's own space. Its federation name, exposed module, and remote-entry URL (:data:STUDIO_FEDERATION_NAME, :data:STUDIO_EXPOSED_MODULE, :data:STUDIO_REMOTE_ENTRY) are the contract with studio-web/module-federation.config.ts and must match it exactly.

Functions:

build_capability_manifest 

build_capability_manifest(
    *, studio_version: str
) -> _CapabilityManifest

Construct SPO's STUDIO federation CapabilityManifest.

Parameters

studio_version : str The SPO package version stamped into the manifest.

Returns

CapabilityManifest The schema-A manifest describing SPO's verbs and honesty attributes.

Raises

RuntimeError If the scpn-studio-platform SDK is not installed.

Source code in src/scpn_phase_orchestrator/studio/federation_manifest.py 
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
def build_capability_manifest(*, studio_version: str) -> _CapabilityManifest:
    """Construct SPO's STUDIO federation `CapabilityManifest`.

    Parameters
    ----------
    studio_version : str
        The SPO package version stamped into the manifest.

    Returns
    -------
    CapabilityManifest
        The schema-A manifest describing SPO's verbs and honesty attributes.

    Raises
    ------
    RuntimeError
        If the ``scpn-studio-platform`` SDK is not installed.
    """
    _require_sdk()
    verbs = _verbs()
    evidence_types = LIVE_FEED_EVIDENCE_SCHEMAS
    digest = _manifest_content_digest(verbs, evidence_types)
    return CapabilityManifest(
        studio="scpn-phase-orchestrator",
        studio_version=studio_version,
        platform_sdk=_PLATFORM_SDK,
        content_digest=digest,
        protocol_version=_PROTOCOL_VERSION,
        transport_profile=TransportProfile.LOCAL_FIRST,
        verbs=verbs,
        evidence_types=evidence_types,
        external_reference_datasets=(),
        ui_module=UiModule(
            remote_entry=STUDIO_REMOTE_ENTRY,
            exposes=(STUDIO_EXPOSED_MODULE,),
            federation=STUDIO_FEDERATION_NAME,
        ),
        contract_era="v1",
        enumeration="language-agnostic",
    )

manifest_dict 

manifest_dict(*, studio_version: str) -> dict[str, object]

Return the federation manifest as a JSON-serialisable dict.

Parameters

studio_version : str The SPO package version stamped into the manifest.

Returns

dict[str, object] CapabilityManifest.to_dict() output.

Source code in src/scpn_phase_orchestrator/studio/federation_manifest.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
def manifest_dict(*, studio_version: str) -> dict[str, object]:
    """Return the federation manifest as a JSON-serialisable dict.

    Parameters
    ----------
    studio_version : str
        The SPO package version stamped into the manifest.

    Returns
    -------
    dict[str, object]
        ``CapabilityManifest.to_dict()`` output.
    """
    result: dict[str, object] = build_capability_manifest(
        studio_version=studio_version
    ).to_dict()
    return result

STUDIO panel data

panel_data assembles the honest data the federated studio panel (studio-web/) renders: the evidence-coverage map. For each of SPO's six assurance evidence categories it lists the EU AI Act, ISO/IEC 42001 and ANSI/UL 4600 clauses the category contributes to, with the addressed / partially_addressed status taken verbatim from the assurance-case clause map (assurance.case.DEFAULT_EVIDENCE_CLAUSE_MAP), so the panel restates the repository's own self-description rather than a hand-authored copy. The rendered JSON snapshot is committed at studio-web/src/panel/evidence_coverage.json and regenerated by tools/build_studio_panel_data.py; a drift-guard test keeps the two in lock-step. Unlike federation_manifest, this module needs no optional extra — it reads only the in-tree assurance surface.

panel_data

Assemble the honest data the SPO STUDIO federation panel renders.

The federated studio panel (studio-web/) is a pure renderer: it never computes and never upgrades a claim. Its one committed data surface is the evidence-coverage map — for each of SPO's six assurance evidence categories (:data:~scpn_phase_orchestrator.assurance.evidence.EVIDENCE_CATEGORIES), the regulatory clauses that category contributes to and whether the contribution is addressed or partially_addressed. The map is the same one the assurance-case bundle uses (:data:~scpn_phase_orchestrator.assurance.case.DEFAULT_EVIDENCE_CLAUSE_MAP), so the panel restates the repository's own honest self-description rather than a hand-authored copy that could drift.

The panel bundle is inlined into the JavaScript remote at build time as studio-web/src/panel/evidence_coverage.json; :func:render_panel_data_json produces that file and a drift-guard test asserts the committed snapshot equals the live producer output, keeping the two in lock-step.

Attributes

STUDIO_ID module-attribute 

STUDIO_ID = 'scpn-phase-orchestrator'

The stable federation id, matching the STUDIO capability manifest.

PANEL_DATA_SCHEMA module-attribute 

PANEL_DATA_SCHEMA = 'spo.studio.evidence-coverage.v1'

Schema identifier for the panel evidence-coverage payload.

Functions:

build_evidence_coverage_panel 

build_evidence_coverage_panel() -> dict[str, object]

Return the JSON-serialisable evidence-coverage panel payload.

For every assurance evidence category, the payload lists the regulatory clauses the category contributes to — resolved to their catalogued standard, identifier and official title — with the addressed / partially_addressed status and rationale taken verbatim from :data:~scpn_phase_orchestrator.assurance.case.DEFAULT_EVIDENCE_CLAUSE_MAP. A summary block totals the categories, clause mappings, and coverage status counts, and names the standards spanned.

Returns

dict[str, object] The panel payload: schema, studio, disclaimer, ordered categories (each with its clauses and per-category counts), and a summary.

Raises

ValueError If :data:_CATEGORY_ORDER has drifted from the clause-map key set, so a renamed or added category can never silently drop from the panel.

Source code in src/scpn_phase_orchestrator/studio/panel_data.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
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
def build_evidence_coverage_panel() -> dict[str, object]:
    """Return the JSON-serialisable evidence-coverage panel payload.

    For every assurance evidence category, the payload lists the regulatory
    clauses the category contributes to — resolved to their catalogued standard,
    identifier and official title — with the ``addressed`` /
    ``partially_addressed`` status and rationale taken verbatim from
    :data:`~scpn_phase_orchestrator.assurance.case.DEFAULT_EVIDENCE_CLAUSE_MAP`.
    A summary block totals the categories, clause mappings, and coverage status
    counts, and names the standards spanned.

    Returns
    -------
    dict[str, object]
        The panel payload: ``schema``, ``studio``, ``disclaimer``, ordered
        ``categories`` (each with its ``clauses`` and per-category counts), and a
        ``summary``.

    Raises
    ------
    ValueError
        If :data:`_CATEGORY_ORDER` has drifted from the clause-map key set, so a
        renamed or added category can never silently drop from the panel.
    """
    if set(_CATEGORY_ORDER) != set(DEFAULT_EVIDENCE_CLAUSE_MAP):
        raise ValueError(
            "the studio panel category order has drifted from the assurance "
            "clause map; align _CATEGORY_ORDER with DEFAULT_EVIDENCE_CLAUSE_MAP"
        )

    categories: list[dict[str, object]] = []
    standards_covered: set[str] = set()
    total_addressed = 0
    total_partial = 0
    total_mappings = 0

    for category in _CATEGORY_ORDER:
        clause_records: list[dict[str, str]] = []
        addressed = 0
        partial = 0
        for key, status, rationale in DEFAULT_EVIDENCE_CLAUSE_MAP[category]:
            clause = clause_for_key(key)
            clause_records.append(
                {
                    "standard": clause.standard,
                    "clause_id": clause.clause_id,
                    "title": clause.title,
                    "status": status,
                    "rationale": rationale,
                }
            )
            standards_covered.add(clause.standard)
            if status == ADDRESSED:
                addressed += 1
            elif status == PARTIALLY_ADDRESSED:
                partial += 1
        total_addressed += addressed
        total_partial += partial
        total_mappings += len(clause_records)
        categories.append(
            {
                "category": category,
                "clause_count": len(clause_records),
                "addressed_count": addressed,
                "partially_addressed_count": partial,
                "clauses": clause_records,
            }
        )

    summary: dict[str, object] = {
        "category_count": len(categories),
        "clause_mapping_count": total_mappings,
        "addressed_count": total_addressed,
        "partially_addressed_count": total_partial,
        "standards_covered": sorted(standards_covered),
    }
    return {
        "schema": PANEL_DATA_SCHEMA,
        "studio": STUDIO_ID,
        "disclaimer": REGULATORY_DISCLAIMER,
        "categories": categories,
        "summary": summary,
    }

render_panel_data_json 

render_panel_data_json() -> str

Return the panel payload as a canonical, human-diffable JSON document.

The document is indented two spaces, keeps non-ASCII characters verbatim, and ends with a trailing newline, matching the committed studio-web/src/panel/evidence_coverage.json snapshot.

Returns

str The rendered JSON document with a trailing newline.

Source code in src/scpn_phase_orchestrator/studio/panel_data.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def render_panel_data_json() -> str:
    """Return the panel payload as a canonical, human-diffable JSON document.

    The document is indented two spaces, keeps non-ASCII characters verbatim, and
    ends with a trailing newline, matching the committed
    ``studio-web/src/panel/evidence_coverage.json`` snapshot.

    Returns
    -------
    str
        The rendered JSON document with a trailing newline.
    """
    return (
        json.dumps(build_evidence_coverage_panel(), indent=2, ensure_ascii=False) + "\n"
    )