Skip to content

SPDX-License-Identifier: AGPL-3.0-or-later

Commercial license available

© Concepts 1996–2026 Miroslav Šotek. All rights reserved.

© Code 2020–2026 Miroslav Šotek. All rights reserved.

ORCID: 0009-0009-3560-0851

Contact: www.anulum.li | protoscience@anulum.li

SCPN Phase Orchestrator — Plugin API reference

Plugins

The plugin interface defines metadata and compatibility checks for extension packages that provide domainpacks, phase extractors, monitors, actuators, or bridges.

Plugins expose a manifest through the scpn_phase_orchestrator.plugins Python entry-point group. The manifest is validated before a marketplace, CI job, or deployment imports domain-specific implementation code.

from scpn_phase_orchestrator.plugins import (
    PluginCapability,
    PluginManifest,
    validate_plugin_manifest,
)

manifest = PluginManifest(
    name="grid_pack",
    version="0.1.0",
    package="grid_pack",
    capabilities=(
        PluginCapability(
            kind="extractor",
            name="pmu",
            target="grid_pack.extractors:PMUExtractor",
            channels=("P",),
        ),
    ),
)

validate_plugin_manifest(manifest)

Supported capability kinds are domainpack, extractor, monitor, actuator, and bridge. Monitor capabilities must declare at least one channel so registry, marketplace, and Rust-dispatch metadata can expose the observed signal surface explicitly.

Marketplace and CI tooling can package discovered manifests into a deterministic metadata catalogue without importing plugin implementation targets:

from scpn_phase_orchestrator.plugins import build_plugin_marketplace_catalog

catalogue = build_plugin_marketplace_catalog((manifest,))

The catalogue includes the package manifest, compatibility result, reason list, SPO version, schema version, and capability counts for every supported capability kind. By default incompatible manifests are counted but omitted from the published plugins list; pass include_incompatible=True when a review job needs the full rejection report.

Rust-side dispatchers can consume a flattened metadata registry without importing plugin implementation targets:

from scpn_phase_orchestrator.plugins import build_rust_plugin_registry

registry = build_rust_plugin_registry((manifest,))

The Rust registry uses schema scpn_rust_plugin_registry_v1 and flattens every compatible capability into records containing plugin name, package, kind, target, channels, knobs, and compatibility status. This is the metadata bridge for Rust loaders that need stable JSON before any Python callback is invoked.

Rust runtime handoff review jobs can consume a stricter no-load handoff:

from scpn_phase_orchestrator.plugins import build_rust_plugin_runtime_handoff

handoff = build_rust_plugin_runtime_handoff((manifest,))

The handoff uses schema scpn_rust_plugin_runtime_handoff_v1, groups compatible capabilities by kind for dispatch review, records deterministic target hashes, keeps incompatible capabilities in a blocked review list when requested, and sets loading_permitted to false. It is a runtime preflight contract for native dispatchers, not a dynamic plugin loader.

Python-owned runtime loading is a separate explicit gate. It is disabled by default, validates the manifest before import, only resolves a declared capability by kind and name, keeps implementation targets inside the manifest package by default, and returns deterministic audit metadata beside the loaded callable:

from scpn_phase_orchestrator.plugins import (
    PluginRuntimeExecutionPolicy,
    PluginRuntimeLoadPolicy,
    execute_plugin_capability,
    load_plugin_capability,
)

loaded = load_plugin_capability(
    manifest,
    "extractor",
    "pmu",
    policy=PluginRuntimeLoadPolicy(loading_permitted=True),
)

extractor_cls = loaded.target_object
audit_record = loaded.audit_record

Runtime loading is intended for approved deployment boundaries after manifest review. Do not use it as a generic import helper: incompatible manifests, undeclared capabilities, non-callable targets, domainpack metadata records, and targets outside the plugin package fail closed.

Approved runtime invocation is a second gate. It is also disabled by default and records invocation shape without serialising argument values:

executed = execute_plugin_capability(
    manifest,
    "monitor",
    "frequency_drift",
    args=(phase_window,),
    kwargs={"sample_rate_hz": 100.0},
    policy=PluginRuntimeExecutionPolicy(
        loading_permitted=True,
        execution_permitted=True,
    ),
)

result = executed.result
execution_audit = executed.audit_record

The execution record uses schema scpn_plugin_runtime_execute_v1, links back to the load hash and target hash, records argument count, keyword names, result type, and a deterministic execution hash. Argument payloads are intentionally not copied into the audit record.

Deterministic runtime execution plans (review only)

build_plugin_execution_plan() is a planning-only stage and does not load modules, resolve callables, or execute any capability implementation.

The plan uses schema scpn_plugin_runtime_execution_plan_v1 and records only immutable review metadata:

  • manifest compatibility state and manifest identity
  • target hash and load hash values
  • immutable plan_hash
  • invocation shape only: positional argument count and keyword names
  • load/execution policy flags and policy hash-approval intent
  • blocked reasons and fallback status

The plan builder is explicit fail-closed:

  • execution disabled by policy
  • invalid capability kind
  • missing compatibility declaration
  • out-of-package target
  • unsupported kind for the active policy
  • missing target hash approval when required

No module import occurs before these checks complete. Tests in this checkout call build_plugin_execution_plan() with a monkeypatched import guard to confirm no target import happens during planning.

Argument payloads are not copied into the plan record. The audit shape contains argument_count and keyword_names only.

Approval artifacts do not execute anything by themselves.

build_plugin_execution_request() consumes two metadata artifacts:

  • reviewed execution plan
  • operator approval artifact

The request builder verifies that the bound values match (plan_hash, target_hash, plugin, kind, name, and operator) and returns a request envelope containing deployment policy, not runtime invocation data. The request phase still does not import modules or invoke targets.

Runtime permissions in the emitted envelope follow these semantics:

  • loading_permitted and execution_permitted are set true only when require_target_hash_approval=true
  • approved_target_hashes contains the reviewed target_hash

Operators can materialize this request through:

spo plugins request-execution PLAN_JSON APPROVAL_JSON

execute_plugin_execution_request() is the first owned runtime consumption point for that envelope. It rebuilds the non-importing plan for the actual call shape, checks the request plan_hash and target_hash, and only then delegates to the explicit runtime execution path. A stale request, changed keyword set, or different manifest fails before target import.

Stored request envelopes should pass validate_plugin_execution_request() immediately before use. The validator recomputes the request hash, rejects tampered audit records, and accepts a deployment-owned revoked-request hash set so rotated approvals fail closed before plugin import.

Deployment stores should persist request envelopes beside a build_plugin_execution_request_storage_manifest() record. The storage manifest binds the request hash, approval hash, target hash, storage URI, storage backend, retention policy, creator, and current revoked-request hash set into a deterministic manifest hash. validate_plugin_execution_request_storage_manifest() recomputes that manifest before use so request storage changes and approval rotation are visible in audit records.

For local deployment stores, build_plugin_execution_request_storage_bundle() combines the request envelope and storage manifest into schema scpn_plugin_execution_request_storage_bundle_v1. validate_plugin_execution_request_storage_bundle() recomputes the bundle, request, storage-manifest, and revocation hashes. The write_plugin_execution_request_storage_bundle() adapter writes that bundle to a local file atomically and refuses to overwrite an existing file unless the caller explicitly opts in.

Non-local deployment stores use build_plugin_execution_request_storage_adapter_manifest() as a deterministic handoff contract. The adapter manifest validates the backend/URI pair for s3_object, gcs_object, azure_blob, oci_object, and https_api, rejects credential-bearing URIs, binds the storage-manifest hash and bundle hash, and marks the write mode as deployment-owned. It does not perform object-store or HTTP writes from the SPO CLI.

Operators can persist the same bundle from reviewed request JSON:

spo plugins persist-execution-request REQUEST_JSON bundle.json \
  --storage-uri file:///var/lib/spo/plugin-requests/bundle.json \
  --created-by deployment_gate

Operators can also emit an external storage handoff without writing:

spo plugins storage-adapter-manifest REQUEST_JSON \
  --storage-uri s3://spo-prod/plugin-requests/request.json \
  --storage-backend s3_object \
  --created-by deployment_gate

Request lifecycle rotation is explicit. Use build_plugin_execution_request_revocation() or the CLI command below to emit a hash-linked revocation artefact for the request hash. Deployment stores then add that request hash to their revoked-request set before accepting any replacement bundle.

spo plugins revoke-execution-request REQUEST_JSON \
  --revoked-by deployment_gate \
  --revocation-reference REV-2026-05-20-01 \
  --revocation-reason operator_rotation

Multiple revocation artefacts can be consolidated with build_plugin_execution_request_revocation_list(). The resulting deterministic list exposes as_revoked_request_hashes(), which is the input shape expected by request validation and request-bound runtime execution.

spo plugins revocation-list REVOCATION_JSON --created-by deployment_gate

The same list can be bound into local persistence:

spo plugins persist-execution-request REQUEST_JSON bundle.json \
  --storage-uri file:///var/lib/spo/plugin-requests/bundle.json \
  --created-by deployment_gate \
  --revocation-list REVOCATION_LIST_JSON

Operator lifecycle status is represented by build_plugin_execution_request_lifecycle_record(). It consumes a request plus optional storage bundle and revocation list evidence, then emits one deterministic approved, stored, or revoked status record without loading or executing plugin targets.

spo plugins lifecycle-status REQUEST_JSON \
  --storage-bundle bundle.json \
  --revocation-list REVOCATION_LIST_JSON \
  --created-by deployment_gate

Batch lifecycle review is represented by build_plugin_execution_request_lifecycle_summary(). It validates each lifecycle record, rejects duplicate request hashes, and emits deterministic status counts plus storage-missing and renewal-required queues for operator planning.

spo plugins lifecycle-summary LIFECYCLE_JSON ... \
  --created-by deployment_gate

Operator dashboard policy is represented by build_plugin_execution_request_lifecycle_policy_report(). It validates the lifecycle summary, binds optional storage-adapter manifests, and emits action counts for request persistence, adapter registration, approval renewal, and external-write confirmation.

spo plugins lifecycle-policy-report SUMMARY_JSON \
  --storage-adapter ADAPTER_JSON \
  --created-by deployment_gate

Operator approval binding

Execution approval must be held in an operator-owned artefact that binds:

  • operator identity/reference
  • reviewed plan_hash
  • reviewed target_hash

Use the reviewed plan_hash/target_hash pair as the invariant approval key and enable execution only when all three conditions are met:

  • loading_permitted=True
  • execution_permitted=True
  • require_target_hash_approval=True with the approved target_hash

If the operator hash pairing is absent or mismatched, deployment should keep runtime loading disabled until approval is re-issued.

Deployments can also require a pre-approved target hash before import:

policy = PluginRuntimeExecutionPolicy(
    loading_permitted=True,
    execution_permitted=True,
    approved_target_hashes=("64_hex_characters_from_reviewed_load_audit",),
    require_target_hash_approval=True,
)

When require_target_hash_approval is true, the loader computes the declared target hash from manifest metadata and policy shape before import. Targets that do not match the approved set fail closed without loading the implementation module.

A runnable metadata-only example is available at examples/plugin_marketplace_catalog.py. It builds a validated extractor/actuator manifest and prints the resulting catalogue JSON without loading the implementation targets declared in the manifest.

The same catalogue is available from the command line:

spo plugins catalog
spo plugins catalog --include-incompatible
spo plugins catalog --rust-registry
spo plugins catalog --rust-runtime-handoff

The default output omits incompatible entries from plugins while still counting them. Use --include-incompatible in CI or review jobs that need the full rejection reason list. Use --rust-registry when a Rust-side dispatcher needs the flattened capability registry instead of the marketplace catalogue. Use --rust-runtime-handoff when a runtime review job needs grouped dispatch records, target hashes, and explicit no-load policy. --rust-registry and --rust-runtime-handoff are mutually exclusive output modes.

The spo plugins plan-execution <plugin> <kind> <capability> command emits the same deterministic plan JSON for operator sign-off, keeps argument payloads out of the audit record, and enforces --require-target-hash-approval before any future request artifact can be produced.

registry

Plugin manifest registry: validation, execution governance, and runtime loading.

This package splits the plugin registry into responsibility modules (manifest, policy, request, storage, revocation, lifecycle, runtime, rust hand-off) while preserving the original public import surface. importlib, metadata, and __version__ are re-exported so tests and callers can resolve them on this package namespace; runtime loading remains Python-owned and disabled by default.

Classes

PluginExecutionRequestLifecyclePolicyReport dataclass 

PluginExecutionRequestLifecyclePolicyReport(
    schema: str,
    version: str,
    summary_hash: str,
    request_count: int,
    policy_action_counts: dict[str, int],
    storage_missing_request_hashes: tuple[str, ...],
    renewal_required_request_hashes: tuple[str, ...],
    missing_adapter_request_hashes: tuple[str, ...],
    local_storage_request_hashes: tuple[str, ...],
    non_local_storage_request_hashes: tuple[str, ...],
    external_write_followup_request_hashes: tuple[str, ...],
    storage_adapter_hashes: tuple[str, ...],
    created_by: str,
    policy_hash: str,
    audit_record: dict[str, object],
)

Deterministic operator policy report for plugin lifecycle batches.

PluginExecutionRequestLifecycleRecord dataclass 

PluginExecutionRequestLifecycleRecord(
    schema: str,
    version: str,
    request_hash: str,
    status: Literal["approved", "stored", "revoked"],
    plugin: str,
    kind: PluginKind,
    name: str,
    operator_identity: str,
    approval_reference: str,
    storage_manifest_hash: str | None,
    storage_backend: str | None,
    storage_uri: str | None,
    revoked: bool,
    revocation_list_hash: str | None,
    revocation_hash: str | None,
    revoked_by: str | None,
    revocation_reference: str | None,
    created_by: str,
    lifecycle_hash: str,
    audit_record: dict[str, object],
)

Operator-facing lifecycle status for one approved execution request.

PluginExecutionRequestLifecycleSummary dataclass 

PluginExecutionRequestLifecycleSummary(
    schema: str,
    version: str,
    request_count: int,
    status_counts: dict[str, int],
    lifecycle_hashes: tuple[str, ...],
    approved_request_hashes: tuple[str, ...],
    stored_request_hashes: tuple[str, ...],
    revoked_request_hashes: tuple[str, ...],
    storage_missing_request_hashes: tuple[str, ...],
    renewal_required_request_hashes: tuple[str, ...],
    created_by: str,
    summary_hash: str,
    audit_record: dict[str, object],
)

Deterministic operator summary for lifecycle-review batches.

PluginCapability dataclass 

PluginCapability(
    kind: PluginKind,
    name: str,
    target: str,
    version: str = "0.1.0",
    channels: tuple[str, ...] = (),
    knobs: tuple[str, ...] = (),
)

One declared extension capability.

PluginCompatibilityReport dataclass 

PluginCompatibilityReport(
    manifest: PluginManifest,
    compatible: bool,
    reasons: tuple[str, ...],
)

Compatibility result for one plugin manifest.

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

Return a serialisable compatibility record.

Returns

dict[str, object] A serialisable compatibility record.

Source code in src/scpn_phase_orchestrator/plugins/registry/manifest.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
def to_audit_record(self) -> dict[str, object]:
    """Return a serialisable compatibility record.

    Returns
    -------
    dict[str, object]
        A serialisable compatibility record.
    """
    return {
        "manifest": self.manifest.to_audit_record(),
        "compatible": self.compatible,
        "reasons": list(self.reasons),
    }

PluginManifest dataclass 

PluginManifest(
    name: str,
    version: str,
    package: str,
    capabilities: tuple[PluginCapability, ...],
    min_spo_version: str | None = None,
)

Versioned plugin manifest for marketplace and CI validation.

Methods:
from_mapping classmethod 
from_mapping(payload: dict[str, Any]) -> PluginManifest

Construct a manifest from a JSON/YAML-style mapping.

Parameters

payload : dict[str, Any] The payload mapping or bytes.

Returns

PluginManifest A manifest from a JSON/YAML-style mapping.

Source code in src/scpn_phase_orchestrator/plugins/registry/manifest.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@classmethod
def from_mapping(cls, payload: dict[str, Any]) -> PluginManifest:
    """Construct a manifest from a JSON/YAML-style mapping.

    Parameters
    ----------
    payload : dict[str, Any]
        The payload mapping or bytes.

    Returns
    -------
    PluginManifest
        A manifest from a JSON/YAML-style mapping.
    """
    capabilities = tuple(
        PluginCapability(
            kind=item["kind"],
            name=item["name"],
            target=item["target"],
            version=item.get("version", "0.1.0"),
            channels=tuple(item.get("channels", ())),
            knobs=tuple(item.get("knobs", ())),
        )
        for item in payload.get("capabilities", ())
    )
    return cls(
        name=payload["name"],
        version=payload["version"],
        package=payload["package"],
        capabilities=capabilities,
        min_spo_version=payload.get("min_spo_version"),
    )
to_audit_record 
to_audit_record() -> dict[str, object]

Return a serialisable manifest record.

Returns

dict[str, object] A serialisable manifest record.

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

    Returns
    -------
    dict[str, object]
        A serialisable manifest record.
    """
    return {
        "name": self.name,
        "version": self.version,
        "package": self.package,
        "min_spo_version": self.min_spo_version,
        "capabilities": [
            {
                "kind": capability.kind,
                "name": capability.name,
                "target": capability.target,
                "version": capability.version,
                "channels": list(capability.channels),
                "knobs": list(capability.knobs),
            }
            for capability in self.capabilities
        ],
    }

PluginRuntimeExecutionPolicy dataclass 

PluginRuntimeExecutionPolicy(
    loading_permitted: bool = False,
    execution_permitted: bool = False,
    allowed_kinds: tuple[
        PluginKind, ...
    ] = _DEFAULT_RUNTIME_LOAD_KINDS,
    require_package_target: bool = True,
    approved_target_hashes: tuple[str, ...] = (),
    require_target_hash_approval: bool = False,
)

Explicit policy gate for invoking Python-owned plugin runtime targets.

Methods:
to_load_policy 
to_load_policy() -> PluginRuntimeLoadPolicy

Return the corresponding load policy for target resolution.

Returns

PluginRuntimeLoadPolicy The corresponding load policy for target resolution.

Source code in src/scpn_phase_orchestrator/plugins/registry/policy.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
def to_load_policy(self) -> PluginRuntimeLoadPolicy:
    """Return the corresponding load policy for target resolution.

    Returns
    -------
    PluginRuntimeLoadPolicy
        The corresponding load policy for target resolution.
    """
    return PluginRuntimeLoadPolicy(
        loading_permitted=self.loading_permitted,
        allowed_kinds=self.allowed_kinds,
        require_package_target=self.require_package_target,
    )

PluginRuntimeLoadPolicy dataclass 

PluginRuntimeLoadPolicy(
    loading_permitted: bool = False,
    allowed_kinds: tuple[
        PluginKind, ...
    ] = _DEFAULT_RUNTIME_LOAD_KINDS,
    require_package_target: bool = True,
)

Explicit policy gate for Python-owned plugin runtime loading.

PluginExecutionApproval dataclass 

PluginExecutionApproval(
    schema: str,
    version: str,
    plan_hash: str,
    target_hash: str,
    plugin: str,
    kind: PluginKind,
    name: str,
    operator_identity: str,
    approval_reference: str,
    approval_reason: str,
    approved: bool,
    execution_permitted: bool,
    approval_hash: str,
    audit_record: dict[str, object],
)

Operator approval artefact for a plugin execution plan.

PluginExecutionRequest dataclass 

PluginExecutionRequest(
    schema: str,
    version: str,
    plan_hash: str,
    approval_hash: str,
    target_hash: str,
    plugin: str,
    kind: PluginKind,
    name: str,
    operator_identity: str,
    approval_reference: str,
    loading_permitted: bool,
    execution_permitted: bool,
    require_target_hash_approval: bool,
    approved_target_hashes: tuple[str, ...],
    allowed_kinds: tuple[PluginKind, ...],
    require_package_target: bool,
    audit_record: dict[str, object],
)

Operator-approved request artefact consumed by runtime execution.

Methods:
to_execution_policy 
to_execution_policy() -> PluginRuntimeExecutionPolicy

Construct an explicit runtime execution policy from request fields.

Returns

PluginRuntimeExecutionPolicy An explicit runtime execution policy from request fields.

Source code in src/scpn_phase_orchestrator/plugins/registry/request.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def to_execution_policy(self) -> PluginRuntimeExecutionPolicy:
    """Construct an explicit runtime execution policy from request fields.

    Returns
    -------
    PluginRuntimeExecutionPolicy
        An explicit runtime execution policy from request fields.
    """
    return PluginRuntimeExecutionPolicy(
        loading_permitted=self.loading_permitted,
        execution_permitted=self.execution_permitted,
        allowed_kinds=self.allowed_kinds,
        require_package_target=self.require_package_target,
        require_target_hash_approval=self.require_target_hash_approval,
        approved_target_hashes=self.approved_target_hashes,
    )

PluginExecutionRequestRevocation dataclass 

PluginExecutionRequestRevocation(
    schema: str,
    version: str,
    request_hash: str,
    plan_hash: str,
    approval_hash: str,
    target_hash: str,
    plugin: str,
    kind: PluginKind,
    name: str,
    operator_identity: str,
    approval_reference: str,
    revoked_by: str,
    revocation_reference: str,
    revocation_reason: str,
    revoked: bool,
    revocation_hash: str,
    audit_record: dict[str, object],
)

Operator lifecycle artefact that revokes an execution request hash.

PluginExecutionRequestRevocationList dataclass 

PluginExecutionRequestRevocationList(
    schema: str,
    version: str,
    request_hashes: tuple[str, ...],
    revocation_hashes: tuple[str, ...],
    revocation_count: int,
    created_by: str,
    revocation_list_hash: str,
    audit_record: dict[str, object],
)

Deployment-owned aggregate of request revocation artefacts.

Methods:
as_revoked_request_hashes 
as_revoked_request_hashes() -> tuple[str, ...]

Return the revoked request hash set for validation calls.

Returns

tuple[str, ...] The revoked request hash set for validation calls.

Source code in src/scpn_phase_orchestrator/plugins/registry/revocation.py 
111
112
113
114
115
116
117
118
119
def as_revoked_request_hashes(self) -> tuple[str, ...]:
    """Return the revoked request hash set for validation calls.

    Returns
    -------
    tuple[str, ...]
        The revoked request hash set for validation calls.
    """
    return self.request_hashes

ExecutedPluginCapability dataclass 

ExecutedPluginCapability(
    loaded: LoadedPluginCapability,
    result: object,
    audit_record: dict[str, object],
)

Result of an explicitly approved plugin runtime invocation.

LoadedPluginCapability dataclass 

LoadedPluginCapability(
    manifest: PluginManifest,
    capability: PluginCapability,
    target_object: object,
    audit_record: dict[str, object],
)

Resolved plugin runtime target with deterministic audit evidence.

PluginExecutionPlan dataclass 

PluginExecutionPlan(
    manifest: PluginManifest,
    capability: PluginCapability,
    argument_count: int,
    keyword_names: tuple[str, ...],
    target_hash: str,
    plan_hash: str,
    audit_record: dict[str, object],
)

Prepared plugin-capability invocation plan without executing it.

PluginExecutionRequestStorageAdapterManifest dataclass 

PluginExecutionRequestStorageAdapterManifest(
    schema: str,
    version: str,
    request_hash: str,
    storage_manifest_hash: str,
    storage_backend: str,
    storage_uri: str,
    storage_scheme: str,
    adapter_mode: str,
    bundle_hash: str,
    write_performed: bool,
    created_by: str,
    adapter_hash: str,
    audit_record: dict[str, object],
)

Deterministic handoff record for deployment-owned request storage.

PluginExecutionRequestStorageManifest dataclass 

PluginExecutionRequestStorageManifest(
    schema: str,
    version: str,
    request_hash: str,
    plan_hash: str,
    approval_hash: str,
    target_hash: str,
    plugin: str,
    kind: PluginKind,
    name: str,
    operator_identity: str,
    approval_reference: str,
    storage_uri: str,
    storage_backend: str,
    retention_policy: str,
    created_by: str,
    revoked_request_hashes: tuple[str, ...],
    revocation_hash: str,
    manifest_hash: str,
    audit_record: dict[str, object],
)

Deployment-owned storage manifest for an approved execution request.

Functions:

build_plugin_execution_request_lifecycle_policy_report 

build_plugin_execution_request_lifecycle_policy_report(
    summary: PluginExecutionRequestLifecycleSummary,
    *,
    created_by: str,
    storage_adapters: tuple[
        PluginExecutionRequestStorageAdapterManifest, ...
    ] = (),
) -> PluginExecutionRequestLifecyclePolicyReport

Build a deterministic operator policy report for lifecycle batches.

Parameters

summary : PluginExecutionRequestLifecycleSummary The lifecycle summary. created_by : str Identifier of the creating actor. storage_adapters : tuple[PluginExecutionRequestStorageAdapterManifest, ...] Storage-adapter manifests.

Returns

PluginExecutionRequestLifecyclePolicyReport A deterministic operator policy report for lifecycle batches.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/plugins/registry/lifecycle.py 
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
def build_plugin_execution_request_lifecycle_policy_report(
    summary: PluginExecutionRequestLifecycleSummary,
    *,
    created_by: str,
    storage_adapters: tuple[PluginExecutionRequestStorageAdapterManifest, ...] = (),
) -> PluginExecutionRequestLifecyclePolicyReport:
    """Build a deterministic operator policy report for lifecycle batches.

    Parameters
    ----------
    summary : PluginExecutionRequestLifecycleSummary
        The lifecycle summary.
    created_by : str
        Identifier of the creating actor.
    storage_adapters : tuple[PluginExecutionRequestStorageAdapterManifest, ...]
        Storage-adapter manifests.

    Returns
    -------
    PluginExecutionRequestLifecyclePolicyReport
        A deterministic operator policy report for lifecycle batches.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    validate_plugin_execution_request_lifecycle_summary(summary)
    _require_identifier(created_by, "lifecycle policy creator")
    request_hashes = set(summary.approved_request_hashes)
    request_hashes.update(summary.stored_request_hashes)
    request_hashes.update(summary.revoked_request_hashes)
    adapter_by_request: dict[str, PluginExecutionRequestStorageAdapterManifest] = {}
    for adapter in storage_adapters:
        validate_plugin_execution_request_storage_adapter_manifest(adapter)
        if adapter.request_hash not in request_hashes:
            raise ValueError("storage adapter request hash is not in lifecycle summary")
        if adapter.request_hash in adapter_by_request:
            raise ValueError("duplicate storage adapter request hash")
        adapter_by_request[adapter.request_hash] = adapter
    storage_relevant = tuple(
        sorted(
            (*summary.storage_missing_request_hashes, *summary.stored_request_hashes)
        )
    )
    missing_adapters = tuple(
        request_hash
        for request_hash in storage_relevant
        if request_hash not in adapter_by_request
    )
    local_storage = tuple(
        sorted(
            request_hash
            for request_hash, adapter in adapter_by_request.items()
            if adapter.storage_backend == "local_file"
        )
    )
    non_local_storage = tuple(
        sorted(
            request_hash
            for request_hash, adapter in adapter_by_request.items()
            if adapter.storage_backend != "local_file"
        )
    )
    external_followup = tuple(
        sorted(
            request_hash
            for request_hash, adapter in adapter_by_request.items()
            if adapter.storage_backend != "local_file" and not adapter.write_performed
        )
    )
    action_counts = {
        "confirm_external_write": len(external_followup),
        "persist_request": len(summary.storage_missing_request_hashes),
        "register_storage_adapter": len(missing_adapters),
        "renew_approval": len(summary.renewal_required_request_hashes),
    }
    audit_record: dict[str, object] = {
        "schema": "scpn_plugin_execution_request_lifecycle_policy_v1",
        "version": "1.0.0",
        "summary_hash": summary.summary_hash,
        "request_count": summary.request_count,
        "status_counts": dict(summary.status_counts),
        "policy_action_counts": action_counts,
        "storage_missing_request_hashes": list(summary.storage_missing_request_hashes),
        "renewal_required_request_hashes": list(
            summary.renewal_required_request_hashes
        ),
        "missing_adapter_request_hashes": list(missing_adapters),
        "local_storage_request_hashes": list(local_storage),
        "non_local_storage_request_hashes": list(non_local_storage),
        "external_write_followup_request_hashes": list(external_followup),
        "storage_adapter_hashes": sorted(
            adapter.adapter_hash for adapter in storage_adapters
        ),
        "created_by": created_by,
    }
    audit_record["policy_hash"] = _record_hash(audit_record)
    return PluginExecutionRequestLifecyclePolicyReport(
        schema="scpn_plugin_execution_request_lifecycle_policy_v1",
        version="1.0.0",
        summary_hash=summary.summary_hash,
        request_count=summary.request_count,
        policy_action_counts=action_counts,
        storage_missing_request_hashes=summary.storage_missing_request_hashes,
        renewal_required_request_hashes=summary.renewal_required_request_hashes,
        missing_adapter_request_hashes=missing_adapters,
        local_storage_request_hashes=local_storage,
        non_local_storage_request_hashes=non_local_storage,
        external_write_followup_request_hashes=external_followup,
        storage_adapter_hashes=tuple(
            sorted(adapter.adapter_hash for adapter in storage_adapters)
        ),
        created_by=created_by,
        policy_hash=str(audit_record["policy_hash"]),
        audit_record=audit_record,
    )

build_plugin_execution_request_lifecycle_record 

build_plugin_execution_request_lifecycle_record(
    request: PluginExecutionRequest,
    *,
    created_by: str,
    storage_manifest: PluginExecutionRequestStorageManifest
    | None = None,
    revocation_list: PluginExecutionRequestRevocationList
    | None = None,
) -> PluginExecutionRequestLifecycleRecord

Build a deterministic operator lifecycle status record.

Parameters

request : PluginExecutionRequest The plugin execution request. created_by : str Identifier of the creating actor. storage_manifest : PluginExecutionRequestStorageManifest | None The storage manifest. revocation_list : PluginExecutionRequestRevocationList | None The aggregate revocation list.

Returns

PluginExecutionRequestLifecycleRecord A deterministic operator lifecycle status record.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/plugins/registry/lifecycle.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
def build_plugin_execution_request_lifecycle_record(
    request: PluginExecutionRequest,
    *,
    created_by: str,
    storage_manifest: PluginExecutionRequestStorageManifest | None = None,
    revocation_list: PluginExecutionRequestRevocationList | None = None,
) -> PluginExecutionRequestLifecycleRecord:
    """Build a deterministic operator lifecycle status record.

    Parameters
    ----------
    request : PluginExecutionRequest
        The plugin execution request.
    created_by : str
        Identifier of the creating actor.
    storage_manifest : PluginExecutionRequestStorageManifest | None
        The storage manifest.
    revocation_list : PluginExecutionRequestRevocationList | None
        The aggregate revocation list.

    Returns
    -------
    PluginExecutionRequestLifecycleRecord
        A deterministic operator lifecycle status record.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    validate_plugin_execution_request(request)
    _require_identifier(created_by, "lifecycle creator")
    request_hash = str(request.audit_record["request_hash"])
    storage_manifest_hash: str | None = None
    storage_backend: str | None = None
    storage_uri: str | None = None
    if storage_manifest is not None:
        validate_plugin_execution_request_storage_manifest(request, storage_manifest)
        storage_manifest_hash = storage_manifest.manifest_hash
        storage_backend = storage_manifest.storage_backend
        storage_uri = storage_manifest.storage_uri

    revocation_list_hash: str | None = None
    revocation_hash: str | None = None
    revoked_by: str | None = None
    revocation_reference: str | None = None
    if revocation_list is not None:
        validate_plugin_execution_request_revocation_list(revocation_list)
        revocation_list_hash = revocation_list.revocation_list_hash
        revocations = revocation_list.audit_record.get("revocations")
        # Defensive: validate_..._revocation_list above already guarantees
        # ``revocations`` is a list of dict records.
        if not isinstance(revocations, (list, tuple)):  # pragma: no cover
            raise ValueError("revocation list revocations must be a sequence")
        for revocation in revocations:
            if not isinstance(revocation, dict):  # pragma: no cover
                raise ValueError("revocation list revocations must be object records")
            if revocation.get("request_hash") == request_hash:
                revocation_hash = str(revocation["revocation_hash"])
                revoked_by = str(revocation["revoked_by"])
                revocation_reference = str(revocation["revocation_reference"])
                break

    revoked = revocation_hash is not None
    status: Literal["approved", "stored", "revoked"]
    if revoked:
        status = "revoked"
    elif storage_manifest is not None:
        status = "stored"
    else:
        status = "approved"
    audit_record: dict[str, object] = {
        "schema": "scpn_plugin_execution_request_lifecycle_v1",
        "version": "1.0.0",
        "request_hash": request_hash,
        "plan_hash": request.plan_hash,
        "approval_hash": request.approval_hash,
        "target_hash": request.target_hash,
        "plugin": request.plugin,
        "kind": request.kind,
        "name": request.name,
        "operator_identity": request.operator_identity,
        "approval_reference": request.approval_reference,
        "status": status,
        "loading_permitted": request.loading_permitted,
        "execution_permitted": request.execution_permitted,
        "storage_manifest_hash": storage_manifest_hash,
        "storage_backend": storage_backend,
        "storage_uri": storage_uri,
        "revoked": revoked,
        "revocation_list_hash": revocation_list_hash,
        "revocation_hash": revocation_hash,
        "revoked_by": revoked_by,
        "revocation_reference": revocation_reference,
        "created_by": created_by,
    }
    audit_record["lifecycle_hash"] = _record_hash(audit_record)
    return PluginExecutionRequestLifecycleRecord(
        schema="scpn_plugin_execution_request_lifecycle_v1",
        version="1.0.0",
        request_hash=request_hash,
        status=status,
        plugin=request.plugin,
        kind=request.kind,
        name=request.name,
        operator_identity=request.operator_identity,
        approval_reference=request.approval_reference,
        storage_manifest_hash=storage_manifest_hash,
        storage_backend=storage_backend,
        storage_uri=storage_uri,
        revoked=revoked,
        revocation_list_hash=revocation_list_hash,
        revocation_hash=revocation_hash,
        revoked_by=revoked_by,
        revocation_reference=revocation_reference,
        created_by=created_by,
        lifecycle_hash=str(audit_record["lifecycle_hash"]),
        audit_record=audit_record,
    )

build_plugin_execution_request_lifecycle_summary 

build_plugin_execution_request_lifecycle_summary(
    lifecycle_records: tuple[
        PluginExecutionRequestLifecycleRecord, ...
    ],
    *,
    created_by: str,
) -> PluginExecutionRequestLifecycleSummary

Build a deterministic batch summary for operator lifecycle review.

Parameters

lifecycle_records : tuple[PluginExecutionRequestLifecycleRecord, ...] Lifecycle status records. created_by : str Identifier of the creating actor.

Returns

PluginExecutionRequestLifecycleSummary A deterministic batch summary for operator lifecycle review.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/plugins/registry/lifecycle.py 
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
def build_plugin_execution_request_lifecycle_summary(
    lifecycle_records: tuple[PluginExecutionRequestLifecycleRecord, ...],
    *,
    created_by: str,
) -> PluginExecutionRequestLifecycleSummary:
    """Build a deterministic batch summary for operator lifecycle review.

    Parameters
    ----------
    lifecycle_records : tuple[PluginExecutionRequestLifecycleRecord, ...]
        Lifecycle status records.
    created_by : str
        Identifier of the creating actor.

    Returns
    -------
    PluginExecutionRequestLifecycleSummary
        A deterministic batch summary for operator lifecycle review.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    _require_identifier(created_by, "lifecycle summary creator")
    if not lifecycle_records:
        raise ValueError("lifecycle summary requires at least one record")
    validated = tuple(
        validate_plugin_execution_request_lifecycle_record(record)
        for record in lifecycle_records
    )
    request_hashes = tuple(record.request_hash for record in validated)
    if len(set(request_hashes)) != len(request_hashes):
        raise ValueError("lifecycle summary contains duplicate request hashes")
    records = tuple(sorted(validated, key=lambda item: item.request_hash))
    approved = tuple(
        record.request_hash for record in records if record.status == "approved"
    )
    stored = tuple(
        record.request_hash for record in records if record.status == "stored"
    )
    revoked = tuple(
        record.request_hash for record in records if record.status == "revoked"
    )
    status_counts = {
        "approved": len(approved),
        "stored": len(stored),
        "revoked": len(revoked),
    }
    audit_record: dict[str, object] = {
        "schema": "scpn_plugin_execution_request_lifecycle_summary_v1",
        "version": "1.0.0",
        "request_count": len(records),
        "status_counts": status_counts,
        "lifecycle_hashes": [record.lifecycle_hash for record in records],
        "request_hashes": [record.request_hash for record in records],
        "approved_request_hashes": list(approved),
        "stored_request_hashes": list(stored),
        "revoked_request_hashes": list(revoked),
        "storage_missing_request_hashes": list(approved),
        "renewal_required_request_hashes": list(revoked),
        "created_by": created_by,
    }
    audit_record["summary_hash"] = _record_hash(audit_record)
    return PluginExecutionRequestLifecycleSummary(
        schema="scpn_plugin_execution_request_lifecycle_summary_v1",
        version="1.0.0",
        request_count=len(records),
        status_counts=status_counts,
        lifecycle_hashes=tuple(record.lifecycle_hash for record in records),
        approved_request_hashes=approved,
        stored_request_hashes=stored,
        revoked_request_hashes=revoked,
        storage_missing_request_hashes=approved,
        renewal_required_request_hashes=revoked,
        created_by=created_by,
        summary_hash=str(audit_record["summary_hash"]),
        audit_record=audit_record,
    )

validate_plugin_execution_request_lifecycle_record 

validate_plugin_execution_request_lifecycle_record(
    lifecycle: PluginExecutionRequestLifecycleRecord,
) -> PluginExecutionRequestLifecycleRecord

Validate a stored plugin execution request lifecycle record.

Parameters

lifecycle : PluginExecutionRequestLifecycleRecord The lifecycle status record.

Returns

PluginExecutionRequestLifecycleRecord A stored plugin execution request lifecycle record.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/plugins/registry/lifecycle.py 
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
def validate_plugin_execution_request_lifecycle_record(
    lifecycle: PluginExecutionRequestLifecycleRecord,
) -> PluginExecutionRequestLifecycleRecord:
    """Validate a stored plugin execution request lifecycle record.

    Parameters
    ----------
    lifecycle : PluginExecutionRequestLifecycleRecord
        The lifecycle status record.

    Returns
    -------
    PluginExecutionRequestLifecycleRecord
        A stored plugin execution request lifecycle record.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    record = lifecycle.audit_record
    if record.get("schema") != "scpn_plugin_execution_request_lifecycle_v1":
        raise ValueError("lifecycle schema mismatch")
    if record.get("version") != "1.0.0":
        raise ValueError("lifecycle version must be 1.0.0")
    expected_hash = record.get("lifecycle_hash")
    if not isinstance(expected_hash, str):
        raise ValueError("lifecycle record is missing lifecycle_hash")
    _validate_sha256(expected_hash, "lifecycle hash")
    payload = dict(record)
    payload.pop("lifecycle_hash", None)
    if _record_hash(payload) != expected_hash:
        raise ValueError("lifecycle hash mismatch")
    field_checks = {
        "request_hash": lifecycle.request_hash,
        "status": lifecycle.status,
        "plugin": lifecycle.plugin,
        "kind": lifecycle.kind,
        "name": lifecycle.name,
        "operator_identity": lifecycle.operator_identity,
        "approval_reference": lifecycle.approval_reference,
        "storage_manifest_hash": lifecycle.storage_manifest_hash,
        "storage_backend": lifecycle.storage_backend,
        "storage_uri": lifecycle.storage_uri,
        "revoked": lifecycle.revoked,
        "revocation_list_hash": lifecycle.revocation_list_hash,
        "revocation_hash": lifecycle.revocation_hash,
        "revoked_by": lifecycle.revoked_by,
        "revocation_reference": lifecycle.revocation_reference,
        "created_by": lifecycle.created_by,
    }
    for field_name, expected in field_checks.items():
        if record.get(field_name) != expected:
            raise ValueError(f"lifecycle {field_name} field mismatch")
    return lifecycle

validate_plugin_execution_request_lifecycle_summary 

validate_plugin_execution_request_lifecycle_summary(
    summary: PluginExecutionRequestLifecycleSummary,
) -> PluginExecutionRequestLifecycleSummary

Validate a stored plugin execution request lifecycle summary.

Parameters

summary : PluginExecutionRequestLifecycleSummary The lifecycle batch summary.

Returns

PluginExecutionRequestLifecycleSummary A stored plugin execution request lifecycle summary.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/plugins/registry/lifecycle.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def validate_plugin_execution_request_lifecycle_summary(
    summary: PluginExecutionRequestLifecycleSummary,
) -> PluginExecutionRequestLifecycleSummary:
    """Validate a stored plugin execution request lifecycle summary.

    Parameters
    ----------
    summary : PluginExecutionRequestLifecycleSummary
        The lifecycle batch summary.

    Returns
    -------
    PluginExecutionRequestLifecycleSummary
        A stored plugin execution request lifecycle summary.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    record = summary.audit_record
    if record.get("schema") != "scpn_plugin_execution_request_lifecycle_summary_v1":
        raise ValueError("lifecycle summary schema mismatch")
    if record.get("version") != "1.0.0":
        raise ValueError("lifecycle summary version must be 1.0.0")
    expected_hash = record.get("summary_hash")
    if not isinstance(expected_hash, str):
        raise ValueError("lifecycle summary is missing summary_hash")
    _validate_sha256(expected_hash, "lifecycle summary hash")
    payload = dict(record)
    payload.pop("summary_hash", None)
    if _record_hash(payload) != expected_hash:
        raise ValueError("lifecycle summary hash mismatch")
    for field_name, expected in (
        ("request_count", summary.request_count),
        ("status_counts", summary.status_counts),
        ("lifecycle_hashes", list(summary.lifecycle_hashes)),
        ("approved_request_hashes", list(summary.approved_request_hashes)),
        ("stored_request_hashes", list(summary.stored_request_hashes)),
        ("revoked_request_hashes", list(summary.revoked_request_hashes)),
        (
            "storage_missing_request_hashes",
            list(summary.storage_missing_request_hashes),
        ),
        (
            "renewal_required_request_hashes",
            list(summary.renewal_required_request_hashes),
        ),
        ("created_by", summary.created_by),
    ):
        if record.get(field_name) != expected:
            raise ValueError(f"lifecycle summary {field_name} field mismatch")
    return summary

build_plugin_marketplace_catalog 

build_plugin_marketplace_catalog(
    manifests: tuple[PluginManifest, ...],
    *,
    include_incompatible: bool = False,
) -> dict[str, object]

Build a deterministic catalogue payload for marketplace tooling.

The catalogue is metadata-only: it uses manifest declarations and compatibility reports, and it never imports plugin implementation targets.

Parameters

manifests : tuple[PluginManifest, ...] The manifest records. include_incompatible : bool Whether to include incompatible items.

Returns

dict[str, object] A deterministic catalogue payload for marketplace tooling.

Source code in src/scpn_phase_orchestrator/plugins/registry/manifest.py 
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
def build_plugin_marketplace_catalog(
    manifests: tuple[PluginManifest, ...],
    *,
    include_incompatible: bool = False,
) -> dict[str, object]:
    """Build a deterministic catalogue payload for marketplace tooling.

    The catalogue is metadata-only: it uses manifest declarations and
    compatibility reports, and it never imports plugin implementation targets.

    Parameters
    ----------
    manifests : tuple[PluginManifest, ...]
        The manifest records.
    include_incompatible : bool
        Whether to include incompatible items.

    Returns
    -------
    dict[str, object]
        A deterministic catalogue payload for marketplace tooling.
    """
    reports = tuple(compatibility_report(manifest) for manifest in manifests)
    selected = (
        reports
        if include_incompatible
        else tuple(report for report in reports if report.compatible)
    )
    sorted_reports = tuple(
        sorted(
            selected,
            key=lambda report: (report.manifest.name, report.manifest.version),
        )
    )
    return {
        "schema_version": "1.0.0",
        "spo_version": __version__,
        "plugins": [report.to_audit_record() for report in sorted_reports],
        "plugin_count": len(sorted_reports),
        "compatible_count": sum(1 for report in reports if report.compatible),
        "incompatible_count": sum(1 for report in reports if not report.compatible),
        "capability_counts": _capability_counts(sorted_reports),
    }

compatibility_report 

compatibility_report(
    manifest: PluginManifest,
) -> PluginCompatibilityReport

Return a non-throwing compatibility report for a manifest.

Parameters

manifest : PluginManifest The manifest object.

Returns

PluginCompatibilityReport A non-throwing compatibility report for a manifest.

Source code in src/scpn_phase_orchestrator/plugins/registry/manifest.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def compatibility_report(manifest: PluginManifest) -> PluginCompatibilityReport:
    """Return a non-throwing compatibility report for a manifest.

    Parameters
    ----------
    manifest : PluginManifest
        The manifest object.

    Returns
    -------
    PluginCompatibilityReport
        A non-throwing compatibility report for a manifest.
    """
    reasons: list[str] = []
    if manifest.min_spo_version is not None and _version_tuple(
        __version__
    ) < _version_tuple(manifest.min_spo_version):
        reasons.append(
            f"requires SPO >= {manifest.min_spo_version}, current {__version__}"
        )
    seen: set[tuple[str, str]] = set()
    for capability in manifest.capabilities:
        key = (capability.kind, capability.name)
        if key in seen:
            reasons.append(f"duplicate capability {capability.kind}:{capability.name}")
        seen.add(key)
        if capability.kind == "extractor" and not capability.channels:
            reasons.append(f"extractor {capability.name} must declare channels")
        if capability.kind == "monitor" and not capability.channels:
            reasons.append(f"monitor {capability.name} must declare channels")
        if capability.kind == "actuator" and not capability.knobs:
            reasons.append(f"actuator {capability.name} must declare knobs")
    return PluginCompatibilityReport(
        manifest=manifest,
        compatible=not reasons,
        reasons=tuple(reasons),
    )

discover_plugin_manifests 

discover_plugin_manifests(
    entry_point_group: str = _ENTRY_POINT_GROUP,
) -> tuple[PluginManifest, ...]

Discover plugin manifests from Python entry points.

Parameters

entry_point_group : str Python entry-point group name.

Returns

tuple[PluginManifest, ...] Plugin manifests from Python entry points.

Source code in src/scpn_phase_orchestrator/plugins/registry/manifest.py 
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 discover_plugin_manifests(
    entry_point_group: str = _ENTRY_POINT_GROUP,
) -> tuple[PluginManifest, ...]:
    """Discover plugin manifests from Python entry points.

    Parameters
    ----------
    entry_point_group : str
        Python entry-point group name.

    Returns
    -------
    tuple[PluginManifest, ...]
        Plugin manifests from Python entry points.
    """
    entry_points = metadata.entry_points()
    selected = entry_points.select(group=entry_point_group)

    manifests: list[PluginManifest] = []
    for entry_point in selected:
        loaded = entry_point.load()
        payload = loaded() if callable(loaded) else loaded
        manifest = (
            payload
            if isinstance(payload, PluginManifest)
            else PluginManifest.from_mapping(payload)
        )
        validate_plugin_manifest(manifest)
        manifests.append(manifest)
    return tuple(manifests)

validate_plugin_manifest 

validate_plugin_manifest(
    manifest: PluginManifest,
) -> PluginManifest

Validate and return a plugin manifest.

Dataclass construction performs structural validation; this function exists as a stable public compatibility gate for tooling.

Parameters

manifest : PluginManifest The manifest object.

Returns

PluginManifest And return a plugin manifest.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/plugins/registry/manifest.py 
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
def validate_plugin_manifest(manifest: PluginManifest) -> PluginManifest:
    """Validate and return a plugin manifest.

    Dataclass construction performs structural validation; this function
    exists as a stable public compatibility gate for tooling.

    Parameters
    ----------
    manifest : PluginManifest
        The manifest object.

    Returns
    -------
    PluginManifest
        And return a plugin manifest.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    compatibility = compatibility_report(manifest)
    if not compatibility.compatible:
        raise ValueError("; ".join(compatibility.reasons))
    return manifest

build_plugin_execution_approval 

build_plugin_execution_approval(
    plan: PluginExecutionPlan,
    *,
    operator_identity: str,
    approval_reference: str,
    approval_reason: str,
) -> PluginExecutionApproval

Build a deterministic operator approval artefact for an execution plan.

Parameters

plan : PluginExecutionPlan The execution plan. operator_identity : str Identifier of the operator. approval_reference : str External approval reference. approval_reason : str Reason recorded with the approval.

Returns

PluginExecutionApproval A deterministic operator approval artefact for an execution plan.

Raises

PermissionError If the operation is not permitted by policy.

Source code in src/scpn_phase_orchestrator/plugins/registry/request.py 
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
def build_plugin_execution_approval(
    plan: PluginExecutionPlan,
    *,
    operator_identity: str,
    approval_reference: str,
    approval_reason: str,
) -> PluginExecutionApproval:
    """Build a deterministic operator approval artefact for an execution plan.

    Parameters
    ----------
    plan : PluginExecutionPlan
        The execution plan.
    operator_identity : str
        Identifier of the operator.
    approval_reference : str
        External approval reference.
    approval_reason : str
        Reason recorded with the approval.

    Returns
    -------
    PluginExecutionApproval
        A deterministic operator approval artefact for an execution plan.

    Raises
    ------
    PermissionError
        If the operation is not permitted by policy.
    """
    _validate_sha256(plan.plan_hash, "plan hash")
    _validate_sha256(plan.target_hash, "target hash")
    _require_identifier(operator_identity, "operator identity")
    _require_identifier(approval_reference, "approval reference")
    _require_non_empty(approval_reason, "approval reason")

    execution_permitted = bool(plan.audit_record.get("execution_permitted"))
    if not execution_permitted:
        raise PermissionError("plugin runtime execution must be permitted for approval")

    require_target_hash_approval = bool(
        plan.audit_record.get("require_target_hash_approval")
    )
    if require_target_hash_approval and not bool(
        plan.audit_record.get("target_hash_approved")
    ):
        raise PermissionError(
            f"plugin runtime target hash {plan.target_hash} is not approved"
        )

    audit_record = {
        "schema": "scpn_plugin_execution_approval_v1",
        "version": "1.0.0",
        "plan_hash": plan.plan_hash,
        "target_hash": plan.target_hash,
        "plugin": plan.manifest.name,
        "kind": plan.capability.kind,
        "name": plan.capability.name,
        "operator_identity": operator_identity,
        "approval_reference": approval_reference,
        "approval_reason": approval_reason,
        "approved": True,
        "execution_permitted": execution_permitted,
    }
    audit_record["approval_hash"] = _record_hash(audit_record)
    return PluginExecutionApproval(
        schema="scpn_plugin_execution_approval_v1",
        version="1.0.0",
        plan_hash=plan.plan_hash,
        target_hash=plan.target_hash,
        plugin=plan.manifest.name,
        kind=plan.capability.kind,
        name=plan.capability.name,
        operator_identity=operator_identity,
        approval_reference=approval_reference,
        approval_reason=approval_reason,
        approved=True,
        execution_permitted=execution_permitted,
        approval_hash=str(audit_record["approval_hash"]),
        audit_record=audit_record,
    )

build_plugin_execution_request 

build_plugin_execution_request(
    plan: PluginExecutionPlan,
    approval: PluginExecutionApproval,
) -> PluginExecutionRequest

Build a deterministic, non-importing request artefact for execution.

Parameters

plan : PluginExecutionPlan The execution plan. approval : PluginExecutionApproval The execution-plan approval artefact.

Returns

PluginExecutionRequest A deterministic, non-importing request artefact for execution.

Raises

ValueError If the inputs are invalid or inconsistent. PermissionError If the operation is not permitted by policy.

Source code in src/scpn_phase_orchestrator/plugins/registry/request.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
def build_plugin_execution_request(
    plan: PluginExecutionPlan,
    approval: PluginExecutionApproval,
) -> PluginExecutionRequest:
    """Build a deterministic, non-importing request artefact for execution.

    Parameters
    ----------
    plan : PluginExecutionPlan
        The execution plan.
    approval : PluginExecutionApproval
        The execution-plan approval artefact.

    Returns
    -------
    PluginExecutionRequest
        A deterministic, non-importing request artefact for execution.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    PermissionError
        If the operation is not permitted by policy.
    """
    if approval.schema != "scpn_plugin_execution_approval_v1":
        raise ValueError("approval schema must be scpn_plugin_execution_approval_v1")
    if not approval.approved:
        raise PermissionError("approval must be granted before request construction")
    if not approval.execution_permitted:
        raise PermissionError(
            "approval execution_permitted must be true before request construction"
        )
    _validate_sha256(plan.plan_hash, "plan hash")
    _validate_sha256(plan.target_hash, "target hash")
    _validate_sha256(approval.plan_hash, "approval plan hash")
    _validate_sha256(approval.target_hash, "approval target hash")
    if plan.plan_hash != approval.plan_hash:
        raise ValueError("plan hash mismatch")
    if plan.target_hash != approval.target_hash:
        raise ValueError("target hash mismatch")
    if plan.manifest.name != approval.plugin:
        raise ValueError("plugin name mismatch")
    if plan.capability.kind != approval.kind:
        raise ValueError("capability kind mismatch")
    if plan.capability.name != approval.name:
        raise ValueError("capability name mismatch")

    loading_permitted = bool(plan.audit_record.get("loading_permitted"))
    execution_permitted = bool(plan.audit_record.get("execution_permitted"))
    if not loading_permitted:
        raise PermissionError("plugin runtime loading must be permitted before request")
    if not execution_permitted:
        raise PermissionError(
            "plugin runtime execution must be permitted before request"
        )

    require_target_hash_approval = bool(
        plan.audit_record.get("require_target_hash_approval")
    )
    if require_target_hash_approval and not bool(
        plan.audit_record.get("target_hash_approved")
    ):
        raise PermissionError(
            f"plugin runtime target hash {plan.target_hash} is not approved"
        )

    raw_allowed_kinds = plan.audit_record.get("allowed_kinds")
    if not isinstance(raw_allowed_kinds, (list, tuple)):
        raise ValueError("plan audit record allowed_kinds must be a sequence")
    validated_allowed_kinds: list[PluginKind] = []
    if not raw_allowed_kinds:
        raise ValueError("plan audit record is missing allowed_kinds")
    for raw_kind in raw_allowed_kinds:
        if raw_kind == "domainpack":
            validated_allowed_kinds.append("domainpack")
        elif raw_kind == "extractor":
            validated_allowed_kinds.append("extractor")
        elif raw_kind == "monitor":
            validated_allowed_kinds.append("monitor")
        elif raw_kind == "actuator":
            validated_allowed_kinds.append("actuator")
        elif raw_kind == "bridge":
            validated_allowed_kinds.append("bridge")
        else:
            raise ValueError(f"unsupported runtime load kind: {raw_kind}")
    allowed_kinds: tuple[PluginKind, ...] = tuple(validated_allowed_kinds)

    require_package_target = bool(plan.audit_record.get("require_package_target", True))
    approved_target_hashes = (plan.target_hash,)
    audit_record = _runtime_execution_request_audit_record(
        plan_hash=plan.plan_hash,
        approval_hash=approval.approval_hash,
        target_hash=plan.target_hash,
        plugin=plan.manifest.name,
        kind=plan.capability.kind,
        name=plan.capability.name,
        operator_identity=approval.operator_identity,
        approval_reference=approval.approval_reference,
        loading_permitted=loading_permitted,
        execution_permitted=execution_permitted,
        require_target_hash_approval=require_target_hash_approval,
        approved_target_hashes=approved_target_hashes,
        allowed_kinds=allowed_kinds,
        require_package_target=require_package_target,
    )
    return PluginExecutionRequest(
        schema="scpn_plugin_runtime_execution_request_v1",
        version="1.0.0",
        plan_hash=plan.plan_hash,
        approval_hash=approval.approval_hash,
        target_hash=plan.target_hash,
        plugin=plan.manifest.name,
        kind=plan.capability.kind,
        name=plan.capability.name,
        operator_identity=approval.operator_identity,
        approval_reference=approval.approval_reference,
        loading_permitted=loading_permitted,
        execution_permitted=execution_permitted,
        require_target_hash_approval=require_target_hash_approval,
        approved_target_hashes=approved_target_hashes,
        allowed_kinds=tuple(allowed_kinds),
        require_package_target=require_package_target,
        audit_record=audit_record,
    )

validate_plugin_execution_request 

validate_plugin_execution_request(
    request: PluginExecutionRequest,
    *,
    revoked_request_hashes: tuple[str, ...] = (),
) -> PluginExecutionRequest

Validate a stored plugin execution request before runtime consumption.

Parameters

request : PluginExecutionRequest The plugin execution request. revoked_request_hashes : tuple[str, ...] Hashes of revoked execution requests.

Returns

PluginExecutionRequest A stored plugin execution request before runtime consumption.

Raises

ValueError If the inputs are invalid or inconsistent. PermissionError If the operation is not permitted by policy.

Source code in src/scpn_phase_orchestrator/plugins/registry/request.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def validate_plugin_execution_request(
    request: PluginExecutionRequest,
    *,
    revoked_request_hashes: tuple[str, ...] = (),
) -> PluginExecutionRequest:
    """Validate a stored plugin execution request before runtime consumption.

    Parameters
    ----------
    request : PluginExecutionRequest
        The plugin execution request.
    revoked_request_hashes : tuple[str, ...]
        Hashes of revoked execution requests.

    Returns
    -------
    PluginExecutionRequest
        A stored plugin execution request before runtime consumption.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    PermissionError
        If the operation is not permitted by policy.
    """
    request_hash = request.audit_record.get("request_hash")
    if not isinstance(request_hash, str):
        raise ValueError("request audit record is missing request_hash")
    _validate_sha256(request_hash, "request hash")
    normalised_revocations: set[str] = set()
    for revoked_hash in revoked_request_hashes:
        _validate_sha256(revoked_hash, "revoked request hash")
        normalised_revocations.add(revoked_hash.lower())
    if request_hash.lower() in normalised_revocations:
        raise PermissionError("plugin execution request has been revoked")
    if not request.require_target_hash_approval:
        raise PermissionError(
            "plugin execution request must require target hash approval"
        )
    if request.approved_target_hashes != (request.target_hash,):
        raise PermissionError(
            "plugin execution request must approve exactly the request target hash"
        )

    expected_record = _runtime_execution_request_audit_record(
        plan_hash=request.plan_hash,
        approval_hash=request.approval_hash,
        target_hash=request.target_hash,
        plugin=request.plugin,
        kind=request.kind,
        name=request.name,
        operator_identity=request.operator_identity,
        approval_reference=request.approval_reference,
        loading_permitted=request.loading_permitted,
        execution_permitted=request.execution_permitted,
        require_target_hash_approval=request.require_target_hash_approval,
        approved_target_hashes=request.approved_target_hashes,
        allowed_kinds=request.allowed_kinds,
        require_package_target=request.require_package_target,
    )
    if request.audit_record != expected_record:
        raise ValueError("request audit record mismatch")
    return request

build_plugin_execution_request_revocation 

build_plugin_execution_request_revocation(
    request: PluginExecutionRequest,
    *,
    revoked_by: str,
    revocation_reference: str,
    revocation_reason: str,
) -> PluginExecutionRequestRevocation

Build a deterministic revocation artefact for an execution request.

Parameters

request : PluginExecutionRequest The plugin execution request. revoked_by : str Identifier of the revoking actor. revocation_reference : str External revocation reference. revocation_reason : str Reason recorded with the revocation.

Returns

PluginExecutionRequestRevocation A deterministic revocation artefact for an execution request.

Source code in src/scpn_phase_orchestrator/plugins/registry/revocation.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
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
def build_plugin_execution_request_revocation(
    request: PluginExecutionRequest,
    *,
    revoked_by: str,
    revocation_reference: str,
    revocation_reason: str,
) -> PluginExecutionRequestRevocation:
    """Build a deterministic revocation artefact for an execution request.

    Parameters
    ----------
    request : PluginExecutionRequest
        The plugin execution request.
    revoked_by : str
        Identifier of the revoking actor.
    revocation_reference : str
        External revocation reference.
    revocation_reason : str
        Reason recorded with the revocation.

    Returns
    -------
    PluginExecutionRequestRevocation
        A deterministic revocation artefact for an execution request.
    """
    validate_plugin_execution_request(request)
    _require_identifier(revoked_by, "revocation actor")
    _require_identifier(revocation_reference, "revocation reference")
    _require_non_empty(revocation_reason, "revocation reason")
    request_hash = str(request.audit_record["request_hash"])
    audit_record = {
        "schema": "scpn_plugin_execution_request_revocation_v1",
        "version": "1.0.0",
        "request_hash": request_hash,
        "plan_hash": request.plan_hash,
        "approval_hash": request.approval_hash,
        "target_hash": request.target_hash,
        "plugin": request.plugin,
        "kind": request.kind,
        "name": request.name,
        "operator_identity": request.operator_identity,
        "approval_reference": request.approval_reference,
        "revoked_by": revoked_by,
        "revocation_reference": revocation_reference,
        "revocation_reason": revocation_reason,
        "revoked": True,
    }
    audit_record["revocation_hash"] = _record_hash(audit_record)
    return PluginExecutionRequestRevocation(
        schema="scpn_plugin_execution_request_revocation_v1",
        version="1.0.0",
        request_hash=request_hash,
        plan_hash=request.plan_hash,
        approval_hash=request.approval_hash,
        target_hash=request.target_hash,
        plugin=request.plugin,
        kind=request.kind,
        name=request.name,
        operator_identity=request.operator_identity,
        approval_reference=request.approval_reference,
        revoked_by=revoked_by,
        revocation_reference=revocation_reference,
        revocation_reason=revocation_reason,
        revoked=True,
        revocation_hash=str(audit_record["revocation_hash"]),
        audit_record=audit_record,
    )

build_plugin_execution_request_revocation_list 

build_plugin_execution_request_revocation_list(
    revocations: tuple[
        PluginExecutionRequestRevocation, ...
    ],
    *,
    created_by: str,
) -> PluginExecutionRequestRevocationList

Build a deterministic deployment revocation-list artefact.

Parameters

revocations : tuple[PluginExecutionRequestRevocation, ...] The revocation artefacts. created_by : str Identifier of the creating actor.

Returns

PluginExecutionRequestRevocationList A deterministic deployment revocation-list artefact.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/plugins/registry/revocation.py 
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
def build_plugin_execution_request_revocation_list(
    revocations: tuple[PluginExecutionRequestRevocation, ...],
    *,
    created_by: str,
) -> PluginExecutionRequestRevocationList:
    """Build a deterministic deployment revocation-list artefact.

    Parameters
    ----------
    revocations : tuple[PluginExecutionRequestRevocation, ...]
        The revocation artefacts.
    created_by : str
        Identifier of the creating actor.

    Returns
    -------
    PluginExecutionRequestRevocationList
        A deterministic deployment revocation-list artefact.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    _require_identifier(created_by, "revocation list creator")
    if not revocations:
        raise ValueError("revocation list requires at least one revocation")
    records: list[dict[str, object]] = []
    for revocation in revocations:
        expected_record = dict(revocation.audit_record)
        expected_record.pop("revocation_hash", None)
        if _record_hash(expected_record) != revocation.revocation_hash:
            raise ValueError("revocation audit record mismatch")
        records.append(dict(revocation.audit_record))
    records.sort(
        key=lambda record: (
            str(record["request_hash"]),
            str(record["revocation_hash"]),
        )
    )
    request_hashes = tuple(str(record["request_hash"]) for record in records)
    if len(set(request_hashes)) != len(request_hashes):
        raise ValueError("revocation list contains duplicate request hashes")
    revocation_hashes = tuple(str(record["revocation_hash"]) for record in records)
    audit_record = {
        "schema": "scpn_plugin_execution_request_revocation_list_v1",
        "version": "1.0.0",
        "created_by": created_by,
        "revocation_count": len(records),
        "request_hashes": list(request_hashes),
        "revocation_hashes": list(revocation_hashes),
        "revocations": records,
    }
    audit_record["revocation_list_hash"] = _record_hash(audit_record)
    return PluginExecutionRequestRevocationList(
        schema="scpn_plugin_execution_request_revocation_list_v1",
        version="1.0.0",
        request_hashes=request_hashes,
        revocation_hashes=revocation_hashes,
        revocation_count=len(records),
        created_by=created_by,
        revocation_list_hash=str(audit_record["revocation_list_hash"]),
        audit_record=audit_record,
    )

validate_plugin_execution_request_revocation_list 

validate_plugin_execution_request_revocation_list(
    revocation_list: PluginExecutionRequestRevocationList,
) -> PluginExecutionRequestRevocationList

Validate a stored aggregate request-revocation list.

Parameters

revocation_list : PluginExecutionRequestRevocationList The aggregate revocation list.

Returns

PluginExecutionRequestRevocationList A stored aggregate request-revocation list.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/plugins/registry/revocation.py 
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def validate_plugin_execution_request_revocation_list(
    revocation_list: PluginExecutionRequestRevocationList,
) -> PluginExecutionRequestRevocationList:
    """Validate a stored aggregate request-revocation list.

    Parameters
    ----------
    revocation_list : PluginExecutionRequestRevocationList
        The aggregate revocation list.

    Returns
    -------
    PluginExecutionRequestRevocationList
        A stored aggregate request-revocation list.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    record = revocation_list.audit_record
    if record.get("schema") != "scpn_plugin_execution_request_revocation_list_v1":
        raise ValueError("revocation list schema mismatch")
    if record.get("version") != "1.0.0":
        raise ValueError("revocation list version must be 1.0.0")
    expected_hash = record.get("revocation_list_hash")
    if not isinstance(expected_hash, str):
        raise ValueError("revocation list is missing revocation_list_hash")
    _validate_sha256(expected_hash, "revocation list hash")
    payload = dict(record)
    payload.pop("revocation_list_hash", None)
    if _record_hash(payload) != expected_hash:
        raise ValueError("revocation list hash mismatch")
    request_hashes = record.get("request_hashes")
    revocation_hashes = record.get("revocation_hashes")
    revocations = record.get("revocations")
    if not isinstance(request_hashes, list) or not all(
        isinstance(item, str) for item in request_hashes
    ):
        raise ValueError("revocation list request_hashes must be a string list")
    if not isinstance(revocation_hashes, list) or not all(
        isinstance(item, str) for item in revocation_hashes
    ):
        raise ValueError("revocation list revocation_hashes must be a string list")
    if not isinstance(revocations, list) or not all(
        isinstance(item, dict) for item in revocations
    ):
        raise ValueError("revocation list revocations must be object records")
    if len(set(request_hashes)) != len(request_hashes):
        raise ValueError("revocation list contains duplicate request hashes")
    if revocation_list.request_hashes != tuple(request_hashes):
        raise ValueError("revocation list request hash field mismatch")
    if revocation_list.revocation_hashes != tuple(revocation_hashes):
        raise ValueError("revocation list revocation hash field mismatch")
    if revocation_list.revocation_count != len(request_hashes):
        raise ValueError("revocation list count mismatch")
    for revocation in revocations:
        revocation_hash = revocation.get("revocation_hash")
        if not isinstance(revocation_hash, str):
            raise ValueError("revocation record is missing revocation_hash")
        _validate_sha256(revocation_hash, "revocation hash")
        revocation_payload = dict(revocation)
        revocation_payload.pop("revocation_hash", None)
        if _record_hash(revocation_payload) != revocation_hash:
            raise ValueError("revocation audit record mismatch")
    return revocation_list

build_plugin_execution_plan 

build_plugin_execution_plan(
    manifest: PluginManifest,
    kind: PluginKind,
    name: str,
    *,
    args: tuple[object, ...] = (),
    kwargs: dict[str, object] | None = None,
    policy: PluginRuntimeExecutionPolicy | None = None,
) -> PluginExecutionPlan

Build a deterministic runtime invocation plan without executing it.

Parameters

manifest : PluginManifest The manifest object. kind : PluginKind The plugin capability kind. name : str The capability or resource name. args : tuple[object, ...] Positional arguments for the capability. kwargs : dict[str, object] | None Keyword arguments for the capability, or None. policy : PluginRuntimeExecutionPolicy | None The runtime policy, or None.

Returns

PluginExecutionPlan A deterministic runtime invocation plan without executing it.

Raises

PermissionError If the operation is not permitted by policy. TypeError If an argument has the wrong type. ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/plugins/registry/runtime.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def build_plugin_execution_plan(
    manifest: PluginManifest,
    kind: PluginKind,
    name: str,
    *,
    args: tuple[object, ...] = (),
    kwargs: dict[str, object] | None = None,
    policy: PluginRuntimeExecutionPolicy | None = None,
) -> PluginExecutionPlan:
    """Build a deterministic runtime invocation plan without executing it.

    Parameters
    ----------
    manifest : PluginManifest
        The manifest object.
    kind : PluginKind
        The plugin capability kind.
    name : str
        The capability or resource name.
    args : tuple[object, ...]
        Positional arguments for the capability.
    kwargs : dict[str, object] | None
        Keyword arguments for the capability, or ``None``.
    policy : PluginRuntimeExecutionPolicy | None
        The runtime policy, or ``None``.

    Returns
    -------
    PluginExecutionPlan
        A deterministic runtime invocation plan without executing it.

    Raises
    ------
    PermissionError
        If the operation is not permitted by policy.
    TypeError
        If an argument has the wrong type.
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if policy is None:
        policy = PluginRuntimeExecutionPolicy()
    if not policy.execution_permitted:
        raise PermissionError("plugin runtime execution is disabled by policy")
    if not isinstance(args, tuple):
        raise TypeError("args must be a tuple")
    if kwargs is None:
        kwargs = {}
    if not isinstance(kwargs, dict):
        raise TypeError("kwargs must be a dictionary")
    for key in kwargs:
        _require_identifier(key, "plugin execution keyword")

    validate_plugin_manifest(manifest)
    _require_identifier(name, "capability name")
    if kind not in _VALID_KINDS:
        raise ValueError(f"unsupported plugin capability kind: {kind}")

    capability = _select_capability(manifest, kind, name)
    load_policy = policy.to_load_policy()
    if kind not in load_policy.allowed_kinds:
        raise ValueError(f"{kind} capability is not permitted by runtime load policy")

    module_name, _attribute_path = _parse_target(capability.target)
    if load_policy.require_package_target and not _target_within_package(
        module_name,
        manifest.package,
    ):
        raise ValueError(
            f"capability target {capability.target!r} is outside plugin package "
            f"{manifest.package!r}"
        )

    target_hash = _preimport_target_hash(
        manifest=manifest,
        capability=capability,
        policy=policy,
    )
    _assert_execution_target_hash_approved(
        manifest=manifest,
        capability=capability,
        policy=policy,
    )

    keyword_names = tuple(sorted(kwargs))
    audit_record = _runtime_execution_plan_audit_record(
        manifest=manifest,
        capability=capability,
        policy=policy,
        target_hash=target_hash,
        argument_count=len(args),
        keyword_names=keyword_names,
    )
    return PluginExecutionPlan(
        manifest=manifest,
        capability=capability,
        argument_count=len(args),
        keyword_names=keyword_names,
        target_hash=target_hash,
        plan_hash=str(audit_record["plan_hash"]),
        audit_record=audit_record,
    )

execute_plugin_capability 

execute_plugin_capability(
    manifest: PluginManifest,
    kind: PluginKind,
    name: str,
    *,
    args: tuple[object, ...] = (),
    kwargs: dict[str, object] | None = None,
    policy: PluginRuntimeExecutionPolicy | None = None,
) -> ExecutedPluginCapability

Invoke a declared plugin capability under an explicit execution policy.

Execution is denied before any target import unless both loading and execution are explicitly permitted. Audit metadata records the invocation shape without serialising argument values, so secrets and large payloads are not copied into the audit record.

Parameters

manifest : PluginManifest The manifest object. kind : PluginKind The plugin capability kind. name : str The capability or resource name. args : tuple[object, ...] Positional arguments for the capability. kwargs : dict[str, object] | None Keyword arguments for the capability, or None. policy : PluginRuntimeExecutionPolicy | None The runtime policy, or None.

Returns

ExecutedPluginCapability Invoke a declared plugin capability under an explicit execution policy.

Raises

TypeError If an argument has the wrong type.

Source code in src/scpn_phase_orchestrator/plugins/registry/runtime.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def execute_plugin_capability(
    manifest: PluginManifest,
    kind: PluginKind,
    name: str,
    *,
    args: tuple[object, ...] = (),
    kwargs: dict[str, object] | None = None,
    policy: PluginRuntimeExecutionPolicy | None = None,
) -> ExecutedPluginCapability:
    """Invoke a declared plugin capability under an explicit execution policy.

    Execution is denied before any target import unless both loading and
    execution are explicitly permitted. Audit metadata records the invocation
    shape without serialising argument values, so secrets and large payloads are
    not copied into the audit record.

    Parameters
    ----------
    manifest : PluginManifest
        The manifest object.
    kind : PluginKind
        The plugin capability kind.
    name : str
        The capability or resource name.
    args : tuple[object, ...]
        Positional arguments for the capability.
    kwargs : dict[str, object] | None
        Keyword arguments for the capability, or ``None``.
    policy : PluginRuntimeExecutionPolicy | None
        The runtime policy, or ``None``.

    Returns
    -------
    ExecutedPluginCapability
        Invoke a declared plugin capability under an explicit execution policy.

    Raises
    ------
    TypeError
        If an argument has the wrong type.
    """
    if kwargs is None:
        kwargs = {}
    if policy is None:
        policy = PluginRuntimeExecutionPolicy()
    plan = build_plugin_execution_plan(
        manifest,
        kind,
        name,
        args=args,
        kwargs=kwargs,
        policy=policy,
    )
    loaded = load_plugin_capability(
        manifest,
        kind,
        name,
        policy=policy.to_load_policy(),
    )
    target = loaded.target_object
    if not callable(target):
        raise TypeError("loaded plugin target must be callable")
    result = target(*args, **kwargs)
    audit_record = _runtime_execute_audit_record(
        loaded=loaded,
        policy=policy,
        args=args,
        kwargs=kwargs,
        result=result,
        plan_hash=plan.plan_hash,
    )
    return ExecutedPluginCapability(
        loaded=loaded,
        result=result,
        audit_record=audit_record,
    )

execute_plugin_execution_request 

execute_plugin_execution_request(
    manifest: PluginManifest,
    request: PluginExecutionRequest,
    *,
    args: tuple[object, ...] = (),
    kwargs: dict[str, object] | None = None,
    revoked_request_hashes: tuple[str, ...] = (),
) -> ExecutedPluginCapability

Invoke a plugin only when the approved request matches this call shape.

The request-bound path validates manifest identity, invocation shape, plan hash, and target hash before importing the plugin module. Argument values remain outside audit records; only positional count and keyword names participate in the plan hash.

Parameters

manifest : PluginManifest The manifest object. request : PluginExecutionRequest The plugin execution request. args : tuple[object, ...] Positional arguments for the capability. kwargs : dict[str, object] | None Keyword arguments for the capability, or None. revoked_request_hashes : tuple[str, ...] Hashes of revoked execution requests.

Returns

ExecutedPluginCapability Invoke a plugin only when the approved request matches this call shape.

Raises

TypeError If an argument has the wrong type. ValueError If the inputs are invalid or inconsistent. PermissionError If the operation is not permitted by policy.

Source code in src/scpn_phase_orchestrator/plugins/registry/runtime.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
def execute_plugin_execution_request(
    manifest: PluginManifest,
    request: PluginExecutionRequest,
    *,
    args: tuple[object, ...] = (),
    kwargs: dict[str, object] | None = None,
    revoked_request_hashes: tuple[str, ...] = (),
) -> ExecutedPluginCapability:
    """Invoke a plugin only when the approved request matches this call shape.

    The request-bound path validates manifest identity, invocation shape, plan
    hash, and target hash before importing the plugin module. Argument values
    remain outside audit records; only positional count and keyword names
    participate in the plan hash.

    Parameters
    ----------
    manifest : PluginManifest
        The manifest object.
    request : PluginExecutionRequest
        The plugin execution request.
    args : tuple[object, ...]
        Positional arguments for the capability.
    kwargs : dict[str, object] | None
        Keyword arguments for the capability, or ``None``.
    revoked_request_hashes : tuple[str, ...]
        Hashes of revoked execution requests.

    Returns
    -------
    ExecutedPluginCapability
        Invoke a plugin only when the approved request matches this call shape.

    Raises
    ------
    TypeError
        If an argument has the wrong type.
    ValueError
        If the inputs are invalid or inconsistent.
    PermissionError
        If the operation is not permitted by policy.
    """
    if kwargs is None:
        kwargs = {}
    if not isinstance(args, tuple):
        raise TypeError("args must be a tuple")
    if not isinstance(kwargs, dict):
        raise TypeError("kwargs must be a dictionary")
    if request.schema != "scpn_plugin_runtime_execution_request_v1":
        raise ValueError(
            "request schema must be scpn_plugin_runtime_execution_request_v1"
        )
    validate_plugin_execution_request(
        request,
        revoked_request_hashes=revoked_request_hashes,
    )
    if manifest.name != request.plugin:
        raise ValueError("request plugin does not match manifest")

    policy = request.to_execution_policy()
    plan = build_plugin_execution_plan(
        manifest,
        request.kind,
        request.name,
        args=args,
        kwargs=kwargs,
        policy=policy,
    )
    if plan.plan_hash != request.plan_hash:
        raise PermissionError("execution request plan hash mismatch")
    if plan.target_hash != request.target_hash:
        raise PermissionError("execution request target hash mismatch")

    executed = execute_plugin_capability(
        manifest,
        request.kind,
        request.name,
        args=args,
        kwargs=kwargs,
        policy=policy,
    )
    audit_record = {
        **executed.audit_record,
        "request_hash": request.audit_record["request_hash"],
        "approval_hash": request.approval_hash,
        "operator_identity": request.operator_identity,
        "approval_reference": request.approval_reference,
    }
    audit_record["execution_hash"] = _record_hash(audit_record)
    return ExecutedPluginCapability(
        loaded=executed.loaded,
        result=executed.result,
        audit_record=audit_record,
    )

load_plugin_capability 

load_plugin_capability(
    manifest: PluginManifest,
    kind: PluginKind,
    name: str,
    *,
    policy: PluginRuntimeLoadPolicy | None = None,
) -> LoadedPluginCapability

Resolve a declared plugin capability under an explicit runtime policy.

Runtime loading is Python-owned and disabled by default. Callers must pass a policy with loading_permitted=True after their deployment boundary has approved the manifest. The target must be declared by the manifest, match the requested capability kind/name, and resolve inside the plugin package unless the policy explicitly relaxes that check.

Parameters

manifest : PluginManifest The manifest object. kind : PluginKind The plugin capability kind. name : str The capability or resource name. policy : PluginRuntimeLoadPolicy | None The runtime policy, or None.

Returns

LoadedPluginCapability A declared plugin capability under an explicit runtime policy.

Raises

ValueError If the inputs are invalid or inconsistent. PermissionError If the operation is not permitted by policy. TypeError If an argument has the wrong type.

Source code in src/scpn_phase_orchestrator/plugins/registry/runtime.py 
 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
def load_plugin_capability(
    manifest: PluginManifest,
    kind: PluginKind,
    name: str,
    *,
    policy: PluginRuntimeLoadPolicy | None = None,
) -> LoadedPluginCapability:
    """Resolve a declared plugin capability under an explicit runtime policy.

    Runtime loading is Python-owned and disabled by default. Callers must pass a
    policy with ``loading_permitted=True`` after their deployment boundary has
    approved the manifest. The target must be declared by the manifest, match the
    requested capability kind/name, and resolve inside the plugin package unless
    the policy explicitly relaxes that check.

    Parameters
    ----------
    manifest : PluginManifest
        The manifest object.
    kind : PluginKind
        The plugin capability kind.
    name : str
        The capability or resource name.
    policy : PluginRuntimeLoadPolicy | None
        The runtime policy, or ``None``.

    Returns
    -------
    LoadedPluginCapability
        A declared plugin capability under an explicit runtime policy.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    PermissionError
        If the operation is not permitted by policy.
    TypeError
        If an argument has the wrong type.
    """
    validate_plugin_manifest(manifest)
    if policy is None:
        policy = _DEFAULT_RUNTIME_LOAD_POLICY
    _require_identifier(name, "capability name")
    if kind not in _VALID_KINDS:
        raise ValueError(f"unsupported plugin capability kind: {kind}")
    if not policy.loading_permitted:
        raise PermissionError("plugin runtime loading is disabled by policy")
    if kind not in policy.allowed_kinds:
        raise ValueError(f"{kind} capability is not permitted by runtime load policy")

    capability = _select_capability(manifest, kind, name)
    module_name, attribute_path = _parse_target(capability.target)
    if policy.require_package_target and not _target_within_package(
        module_name,
        manifest.package,
    ):
        raise ValueError(
            f"capability target {capability.target!r} is outside plugin package "
            f"{manifest.package!r}"
        )

    module = importlib.import_module(module_name)
    target_object = _resolve_attribute_path(module, attribute_path, capability.target)
    if not callable(target_object):
        raise TypeError(f"capability target {capability.target!r} must be callable")

    audit_record = _runtime_load_audit_record(
        manifest=manifest,
        capability=capability,
        policy=policy,
        module_name=module_name,
        callable_target=True,
    )
    return LoadedPluginCapability(
        manifest=manifest,
        capability=capability,
        target_object=target_object,
        audit_record=audit_record,
    )

build_rust_plugin_registry 

build_rust_plugin_registry(
    manifests: tuple[PluginManifest, ...],
    *,
    include_incompatible: bool = False,
) -> dict[str, object]

Build a flattened metadata registry for Rust-side dispatchers.

The payload avoids Python object graphs and implementation imports. Rust consumers can parse capabilities, targets, channel/knob declarations, and compatibility flags from stable JSON before deciding whether to hand a target back to Python.

Parameters

manifests : tuple[PluginManifest, ...] The manifest records. include_incompatible : bool Whether to include incompatible items.

Returns

dict[str, object] A flattened metadata registry for Rust-side dispatchers.

Raises

TypeError If an argument has the wrong type.

Source code in src/scpn_phase_orchestrator/plugins/registry/rust_handoff.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def build_rust_plugin_registry(
    manifests: tuple[PluginManifest, ...],
    *,
    include_incompatible: bool = False,
) -> dict[str, object]:
    """Build a flattened metadata registry for Rust-side dispatchers.

    The payload avoids Python object graphs and implementation imports. Rust
    consumers can parse capabilities, targets, channel/knob declarations, and
    compatibility flags from stable JSON before deciding whether to hand a
    target back to Python.

    Parameters
    ----------
    manifests : tuple[PluginManifest, ...]
        The manifest records.
    include_incompatible : bool
        Whether to include incompatible items.

    Returns
    -------
    dict[str, object]
        A flattened metadata registry for Rust-side dispatchers.

    Raises
    ------
    TypeError
        If an argument has the wrong type.
    """
    catalog = build_plugin_marketplace_catalog(
        manifests,
        include_incompatible=include_incompatible,
    )
    plugins = catalog["plugins"]
    if not isinstance(plugins, list):
        raise TypeError("plugin catalogue payload is malformed")

    capabilities: list[dict[str, object]] = []
    for plugin in plugins:
        manifest = plugin["manifest"]
        compatible = bool(plugin["compatible"])
        if not isinstance(manifest, dict):
            raise TypeError("plugin manifest payload is malformed")
        manifest_capabilities = manifest["capabilities"]
        if not isinstance(manifest_capabilities, list):
            raise TypeError("plugin capabilities payload is malformed")
        for capability in manifest_capabilities:
            if not isinstance(capability, dict):
                raise TypeError("plugin capability payload is malformed")
            capabilities.append(
                {
                    "plugin": manifest["name"],
                    "plugin_version": manifest["version"],
                    "package": manifest["package"],
                    "kind": capability["kind"],
                    "name": capability["name"],
                    "target": capability["target"],
                    "version": capability["version"],
                    "channels": capability["channels"],
                    "knobs": capability["knobs"],
                    "compatible": compatible,
                }
            )

    capabilities.sort(
        key=lambda item: (
            str(item["plugin"]),
            str(item["kind"]),
            str(item["name"]),
            str(item["version"]),
        )
    )
    return {
        "schema": "scpn_rust_plugin_registry_v1",
        "spo_version": catalog["spo_version"],
        "include_incompatible": include_incompatible,
        "capability_count": len(capabilities),
        "capabilities": capabilities,
        "capability_counts": catalog["capability_counts"],
    }

build_rust_plugin_runtime_handoff 

build_rust_plugin_runtime_handoff(
    manifests: tuple[PluginManifest, ...],
    *,
    include_incompatible: bool = False,
) -> dict[str, object]

Build a guarded metadata handoff for a future Rust runtime loader.

The handoff is intentionally non-executing: it groups capabilities by kind, hashes every target record, carries compatibility state, and records that native/plugin loading remains disabled. Rust can consume this as a stable preflight contract before any future loader is allowed to resolve or call implementation targets.

Parameters

manifests : tuple[PluginManifest, ...] The manifest records. include_incompatible : bool Whether to include incompatible items.

Returns

dict[str, object] A guarded metadata handoff for a future Rust runtime loader.

Raises

TypeError If an argument has the wrong type.

Source code in src/scpn_phase_orchestrator/plugins/registry/rust_handoff.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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def build_rust_plugin_runtime_handoff(
    manifests: tuple[PluginManifest, ...],
    *,
    include_incompatible: bool = False,
) -> dict[str, object]:
    """Build a guarded metadata handoff for a future Rust runtime loader.

    The handoff is intentionally non-executing: it groups capabilities by kind,
    hashes every target record, carries compatibility state, and records that
    native/plugin loading remains disabled. Rust can consume this as a stable
    preflight contract before any future loader is allowed to resolve or call
    implementation targets.

    Parameters
    ----------
    manifests : tuple[PluginManifest, ...]
        The manifest records.
    include_incompatible : bool
        Whether to include incompatible items.

    Returns
    -------
    dict[str, object]
        A guarded metadata handoff for a future Rust runtime loader.

    Raises
    ------
    TypeError
        If an argument has the wrong type.
    """
    registry = build_rust_plugin_registry(
        manifests,
        include_incompatible=include_incompatible,
    )
    capabilities = registry["capabilities"]
    if not isinstance(capabilities, list):
        raise TypeError("rust plugin registry payload is malformed")

    dispatch_groups: dict[str, list[dict[str, object]]] = {
        kind: [] for kind in sorted(_VALID_KINDS)
    }
    blocked: list[dict[str, object]] = []
    target_hashes: dict[str, str] = {}
    for capability in capabilities:
        if not isinstance(capability, dict):
            raise TypeError("rust plugin capability payload is malformed")
        kind = str(capability["kind"])
        if kind not in _VALID_KINDS:
            raise TypeError("rust plugin capability kind is malformed")
        record = {
            "plugin": capability["plugin"],
            "plugin_version": capability["plugin_version"],
            "package": capability["package"],
            "kind": capability["kind"],
            "name": capability["name"],
            "target": capability["target"],
            "version": capability["version"],
            "channels": capability["channels"],
            "knobs": capability["knobs"],
            "compatible": capability["compatible"],
            "loading_permitted": False,
            "load_policy": "metadata_only_review",
        }
        target_hash = _record_hash(record)
        record["target_hash"] = target_hash
        target_hashes[
            (
                f"{record['plugin']}:{record['kind']}:"
                f"{record['name']}:{record['version']}"
            )
        ] = target_hash
        if record["compatible"] is True:
            dispatch_groups[kind].append(record)
        else:
            blocked.append(
                {
                    **record,
                    "blocked_reason": "incompatible_manifest",
                }
            )

    for records in dispatch_groups.values():
        records.sort(
            key=lambda item: (
                str(item["plugin"]),
                str(item["name"]),
                str(item["version"]),
            )
        )
    blocked.sort(
        key=lambda item: (
            str(item["plugin"]),
            str(item["kind"]),
            str(item["name"]),
            str(item["version"]),
        )
    )
    handoff = {
        "schema": "scpn_rust_plugin_runtime_handoff_v1",
        "registry_schema": registry["schema"],
        "spo_version": registry["spo_version"],
        "include_incompatible": include_incompatible,
        "loading_permitted": False,
        "load_policy": "metadata_only_review",
        "dispatch_groups": dispatch_groups,
        "target_hashes": dict(sorted(target_hashes.items())),
        "compatible_capability_count": sum(
            len(records) for records in dispatch_groups.values()
        ),
        "blocked_capability_count": len(blocked),
        "blocked_capabilities": blocked,
    }
    handoff["handoff_hash"] = _record_hash(handoff)
    return handoff

build_plugin_execution_request_storage_adapter_manifest 

build_plugin_execution_request_storage_adapter_manifest(
    request: PluginExecutionRequest,
    manifest: PluginExecutionRequestStorageManifest,
    *,
    write_performed: bool = False,
) -> PluginExecutionRequestStorageAdapterManifest

Build a deterministic handoff manifest for local or external stores.

Parameters

request : PluginExecutionRequest The plugin execution request. manifest : PluginExecutionRequestStorageManifest The manifest object. write_performed : bool Whether a write was performed.

Returns

PluginExecutionRequestStorageAdapterManifest A deterministic handoff manifest for local or external stores.

Raises

PermissionError If the operation is not permitted by policy.

Source code in src/scpn_phase_orchestrator/plugins/registry/storage.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
def build_plugin_execution_request_storage_adapter_manifest(
    request: PluginExecutionRequest,
    manifest: PluginExecutionRequestStorageManifest,
    *,
    write_performed: bool = False,
) -> PluginExecutionRequestStorageAdapterManifest:
    """Build a deterministic handoff manifest for local or external stores.

    Parameters
    ----------
    request : PluginExecutionRequest
        The plugin execution request.
    manifest : PluginExecutionRequestStorageManifest
        The manifest object.
    write_performed : bool
        Whether a write was performed.

    Returns
    -------
    PluginExecutionRequestStorageAdapterManifest
        A deterministic handoff manifest for local or external stores.

    Raises
    ------
    PermissionError
        If the operation is not permitted by policy.
    """
    validate_plugin_execution_request_storage_manifest(request, manifest)
    bundle = build_plugin_execution_request_storage_bundle(request, manifest)
    storage_scheme = _validate_storage_backend_uri(
        manifest.storage_backend,
        manifest.storage_uri,
    )
    adapter_mode = (
        "local_file_atomic_write"
        if manifest.storage_backend == "local_file"
        else "deployment_owned_external_write"
    )
    if manifest.storage_backend in _NON_LOCAL_STORAGE_BACKENDS and write_performed:
        raise PermissionError("non-local storage adapters must not write implicitly")
    audit_record: dict[str, object] = {
        "schema": "scpn_plugin_execution_request_storage_adapter_v1",
        "version": "1.0.0",
        "request_hash": manifest.request_hash,
        "storage_manifest_hash": manifest.manifest_hash,
        "storage_backend": manifest.storage_backend,
        "storage_uri": manifest.storage_uri,
        "storage_scheme": storage_scheme,
        "adapter_mode": adapter_mode,
        "bundle_hash": str(bundle["bundle_hash"]),
        "write_performed": write_performed,
        "created_by": manifest.created_by,
    }
    audit_record["adapter_hash"] = _record_hash(audit_record)
    return PluginExecutionRequestStorageAdapterManifest(
        schema="scpn_plugin_execution_request_storage_adapter_v1",
        version="1.0.0",
        request_hash=manifest.request_hash,
        storage_manifest_hash=manifest.manifest_hash,
        storage_backend=manifest.storage_backend,
        storage_uri=manifest.storage_uri,
        storage_scheme=storage_scheme,
        adapter_mode=adapter_mode,
        bundle_hash=str(bundle["bundle_hash"]),
        write_performed=write_performed,
        created_by=manifest.created_by,
        adapter_hash=str(audit_record["adapter_hash"]),
        audit_record=audit_record,
    )

build_plugin_execution_request_storage_bundle 

build_plugin_execution_request_storage_bundle(
    request: PluginExecutionRequest,
    manifest: PluginExecutionRequestStorageManifest,
) -> dict[str, object]

Build a deterministic local persistence bundle for a request envelope.

Parameters

request : PluginExecutionRequest The plugin execution request. manifest : PluginExecutionRequestStorageManifest The manifest object.

Returns

dict[str, object] A deterministic local persistence bundle for a request envelope.

Source code in src/scpn_phase_orchestrator/plugins/registry/storage.py 
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
def build_plugin_execution_request_storage_bundle(
    request: PluginExecutionRequest,
    manifest: PluginExecutionRequestStorageManifest,
) -> dict[str, object]:
    """Build a deterministic local persistence bundle for a request envelope.

    Parameters
    ----------
    request : PluginExecutionRequest
        The plugin execution request.
    manifest : PluginExecutionRequestStorageManifest
        The manifest object.

    Returns
    -------
    dict[str, object]
        A deterministic local persistence bundle for a request envelope.
    """
    validate_plugin_execution_request_storage_manifest(request, manifest)
    bundle: dict[str, object] = {
        "schema": "scpn_plugin_execution_request_storage_bundle_v1",
        "version": "1.0.0",
        "request": dict(request.audit_record),
        "storage_manifest": dict(manifest.audit_record),
    }
    bundle["bundle_hash"] = _record_hash(bundle)
    return bundle

build_plugin_execution_request_storage_manifest 

build_plugin_execution_request_storage_manifest(
    request: PluginExecutionRequest,
    *,
    storage_uri: str,
    storage_backend: str,
    retention_policy: str,
    created_by: str,
    revoked_request_hashes: tuple[str, ...] = (),
) -> PluginExecutionRequestStorageManifest

Build a deterministic storage manifest for an execution request.

Parameters

request : PluginExecutionRequest The plugin execution request. storage_uri : str Storage URI. storage_backend : str Storage backend identifier. retention_policy : str Retention policy label. created_by : str Identifier of the creating actor. revoked_request_hashes : tuple[str, ...] Hashes of revoked execution requests.

Returns

PluginExecutionRequestStorageManifest A deterministic storage manifest for an execution request.

Source code in src/scpn_phase_orchestrator/plugins/registry/storage.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def build_plugin_execution_request_storage_manifest(
    request: PluginExecutionRequest,
    *,
    storage_uri: str,
    storage_backend: str,
    retention_policy: str,
    created_by: str,
    revoked_request_hashes: tuple[str, ...] = (),
) -> PluginExecutionRequestStorageManifest:
    """Build a deterministic storage manifest for an execution request.

    Parameters
    ----------
    request : PluginExecutionRequest
        The plugin execution request.
    storage_uri : str
        Storage URI.
    storage_backend : str
        Storage backend identifier.
    retention_policy : str
        Retention policy label.
    created_by : str
        Identifier of the creating actor.
    revoked_request_hashes : tuple[str, ...]
        Hashes of revoked execution requests.

    Returns
    -------
    PluginExecutionRequestStorageManifest
        A deterministic storage manifest for an execution request.
    """
    validate_plugin_execution_request(
        request,
        revoked_request_hashes=revoked_request_hashes,
    )
    _require_non_empty(storage_uri, "storage URI")
    _require_identifier(storage_backend, "storage backend")
    storage_scheme = _validate_storage_backend_uri(storage_backend, storage_uri)
    _require_identifier(retention_policy, "retention policy")
    _require_identifier(created_by, "storage manifest creator")
    normalised_revocations = tuple(
        sorted(revoked_hash.lower() for revoked_hash in revoked_request_hashes)
    )
    for revoked_hash in normalised_revocations:
        _validate_sha256(revoked_hash, "revoked request hash")
    revocation_hash = _request_revocation_hash(normalised_revocations)
    request_hash = str(request.audit_record["request_hash"])
    audit_record: dict[str, object] = {
        "schema": "scpn_plugin_execution_request_storage_manifest_v1",
        "version": "1.0.0",
        "request_hash": request_hash,
        "plan_hash": request.plan_hash,
        "approval_hash": request.approval_hash,
        "target_hash": request.target_hash,
        "plugin": request.plugin,
        "kind": request.kind,
        "name": request.name,
        "operator_identity": request.operator_identity,
        "approval_reference": request.approval_reference,
        "storage_uri": storage_uri,
        "storage_backend": storage_backend,
        "storage_scheme": storage_scheme,
        "retention_policy": retention_policy,
        "created_by": created_by,
        "revoked_request_hashes": list(normalised_revocations),
        "revocation_hash": revocation_hash,
    }
    audit_record["manifest_hash"] = _record_hash(audit_record)
    return PluginExecutionRequestStorageManifest(
        schema="scpn_plugin_execution_request_storage_manifest_v1",
        version="1.0.0",
        request_hash=request_hash,
        plan_hash=request.plan_hash,
        approval_hash=request.approval_hash,
        target_hash=request.target_hash,
        plugin=request.plugin,
        kind=request.kind,
        name=request.name,
        operator_identity=request.operator_identity,
        approval_reference=request.approval_reference,
        storage_uri=storage_uri,
        storage_backend=storage_backend,
        retention_policy=retention_policy,
        created_by=created_by,
        revoked_request_hashes=normalised_revocations,
        revocation_hash=revocation_hash,
        manifest_hash=str(audit_record["manifest_hash"]),
        audit_record=audit_record,
    )

validate_plugin_execution_request_storage_adapter_manifest 

validate_plugin_execution_request_storage_adapter_manifest(
    adapter: PluginExecutionRequestStorageAdapterManifest,
) -> PluginExecutionRequestStorageAdapterManifest

Validate a stored plugin request storage-adapter handoff manifest.

Parameters

adapter : PluginExecutionRequestStorageAdapterManifest The storage-adapter manifest.

Returns

PluginExecutionRequestStorageAdapterManifest A stored plugin request storage-adapter handoff manifest.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/plugins/registry/storage.py 
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
def validate_plugin_execution_request_storage_adapter_manifest(
    adapter: PluginExecutionRequestStorageAdapterManifest,
) -> PluginExecutionRequestStorageAdapterManifest:
    """Validate a stored plugin request storage-adapter handoff manifest.

    Parameters
    ----------
    adapter : PluginExecutionRequestStorageAdapterManifest
        The storage-adapter manifest.

    Returns
    -------
    PluginExecutionRequestStorageAdapterManifest
        A stored plugin request storage-adapter handoff manifest.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    record = adapter.audit_record
    if record.get("schema") != "scpn_plugin_execution_request_storage_adapter_v1":
        raise ValueError("storage adapter schema mismatch")
    if record.get("version") != "1.0.0":
        raise ValueError("storage adapter version must be 1.0.0")
    expected_hash = record.get("adapter_hash")
    if not isinstance(expected_hash, str):
        raise ValueError("storage adapter is missing adapter_hash")
    _validate_sha256(expected_hash, "storage adapter hash")
    payload = dict(record)
    payload.pop("adapter_hash", None)
    if _record_hash(payload) != expected_hash:
        raise ValueError("storage adapter hash mismatch")
    for field_name, expected in (
        ("request_hash", adapter.request_hash),
        ("storage_manifest_hash", adapter.storage_manifest_hash),
        ("storage_backend", adapter.storage_backend),
        ("storage_uri", adapter.storage_uri),
        ("storage_scheme", adapter.storage_scheme),
        ("adapter_mode", adapter.adapter_mode),
        ("bundle_hash", adapter.bundle_hash),
        ("write_performed", adapter.write_performed),
        ("created_by", adapter.created_by),
    ):
        if record.get(field_name) != expected:
            raise ValueError(f"storage adapter {field_name} field mismatch")
    return adapter

validate_plugin_execution_request_storage_bundle 

validate_plugin_execution_request_storage_bundle(
    bundle: dict[str, object],
) -> dict[str, object]

Validate an on-disk plugin execution request storage bundle.

Parameters

bundle : dict[str, object] The on-disk storage bundle.

Returns

dict[str, object] An on-disk plugin execution request storage bundle.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/plugins/registry/storage.py 
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
def validate_plugin_execution_request_storage_bundle(
    bundle: dict[str, object],
) -> dict[str, object]:
    """Validate an on-disk plugin execution request storage bundle.

    Parameters
    ----------
    bundle : dict[str, object]
        The on-disk storage bundle.

    Returns
    -------
    dict[str, object]
        An on-disk plugin execution request storage bundle.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if bundle.get("schema") != "scpn_plugin_execution_request_storage_bundle_v1":
        raise ValueError(
            "storage bundle schema must be "
            "scpn_plugin_execution_request_storage_bundle_v1"
        )
    if bundle.get("version") != "1.0.0":
        raise ValueError("storage bundle version must be 1.0.0")
    bundle_hash = bundle.get("bundle_hash")
    if not isinstance(bundle_hash, str):
        raise ValueError("storage bundle is missing bundle_hash")
    _validate_sha256(bundle_hash, "storage bundle hash")
    expected_bundle = dict(bundle)
    expected_bundle.pop("bundle_hash", None)
    if _record_hash(expected_bundle) != bundle_hash:
        raise ValueError("storage bundle hash mismatch")

    request_record = bundle.get("request")
    manifest_record = bundle.get("storage_manifest")
    if not isinstance(request_record, dict):
        raise ValueError("storage bundle request must be an object")
    if not isinstance(manifest_record, dict):
        raise ValueError("storage bundle manifest must be an object")
    _validate_request_audit_record(request_record)
    _validate_storage_manifest_audit_record(manifest_record)
    for field in ("request_hash", "plan_hash", "approval_hash", "target_hash"):
        if request_record.get(field) != manifest_record.get(field):
            raise ValueError(f"storage bundle {field} mismatch")
    return bundle

validate_plugin_execution_request_storage_manifest 

validate_plugin_execution_request_storage_manifest(
    request: PluginExecutionRequest,
    manifest: PluginExecutionRequestStorageManifest,
) -> PluginExecutionRequestStorageManifest

Validate that a storage manifest still matches its request envelope.

Parameters

request : PluginExecutionRequest The plugin execution request. manifest : PluginExecutionRequestStorageManifest The manifest object.

Returns

PluginExecutionRequestStorageManifest That a storage manifest still matches its request envelope.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/plugins/registry/storage.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def validate_plugin_execution_request_storage_manifest(
    request: PluginExecutionRequest,
    manifest: PluginExecutionRequestStorageManifest,
) -> PluginExecutionRequestStorageManifest:
    """Validate that a storage manifest still matches its request envelope.

    Parameters
    ----------
    request : PluginExecutionRequest
        The plugin execution request.
    manifest : PluginExecutionRequestStorageManifest
        The manifest object.

    Returns
    -------
    PluginExecutionRequestStorageManifest
        That a storage manifest still matches its request envelope.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    request_hash = str(request.audit_record.get("request_hash", ""))
    if manifest.request_hash != request_hash:
        raise ValueError("storage manifest request hash mismatch")
    if manifest.plan_hash != request.plan_hash:
        raise ValueError("storage manifest plan hash mismatch")
    if manifest.approval_hash != request.approval_hash:
        raise ValueError("storage manifest approval hash mismatch")
    if manifest.target_hash != request.target_hash:
        raise ValueError("storage manifest target hash mismatch")
    _validate_storage_backend_uri(manifest.storage_backend, manifest.storage_uri)
    validate_plugin_execution_request(
        request,
        revoked_request_hashes=manifest.revoked_request_hashes,
    )
    expected = build_plugin_execution_request_storage_manifest(
        request,
        storage_uri=manifest.storage_uri,
        storage_backend=manifest.storage_backend,
        retention_policy=manifest.retention_policy,
        created_by=manifest.created_by,
        revoked_request_hashes=manifest.revoked_request_hashes,
    )
    if manifest.audit_record != expected.audit_record:
        raise ValueError("storage manifest audit record mismatch")
    return manifest

write_plugin_execution_request_storage_bundle 

write_plugin_execution_request_storage_bundle(
    request: PluginExecutionRequest,
    manifest: PluginExecutionRequestStorageManifest,
    path: str | Path,
    *,
    overwrite: bool = False,
) -> dict[str, object]

Atomically persist a validated local-file execution request bundle.

Parameters

request : PluginExecutionRequest The plugin execution request. manifest : PluginExecutionRequestStorageManifest The manifest object. path : str | Path Filesystem path to the target file. overwrite : bool Whether to overwrite an existing artefact.

Returns

dict[str, object] A validated local-file execution request bundle.

Raises

ValueError If the inputs are invalid or inconsistent. FileExistsError If the target already exists and overwrite is false. IsADirectoryError If the target path is a directory.

Source code in src/scpn_phase_orchestrator/plugins/registry/storage.py 
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
def write_plugin_execution_request_storage_bundle(
    request: PluginExecutionRequest,
    manifest: PluginExecutionRequestStorageManifest,
    path: str | Path,
    *,
    overwrite: bool = False,
) -> dict[str, object]:
    """Atomically persist a validated local-file execution request bundle.

    Parameters
    ----------
    request : PluginExecutionRequest
        The plugin execution request.
    manifest : PluginExecutionRequestStorageManifest
        The manifest object.
    path : str | Path
        Filesystem path to the target file.
    overwrite : bool
        Whether to overwrite an existing artefact.

    Returns
    -------
    dict[str, object]
        A validated local-file execution request bundle.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    FileExistsError
        If the target already exists and overwrite is false.
    IsADirectoryError
        If the target path is a directory.
    """
    if manifest.storage_backend != "local_file":
        raise ValueError("only local_file request storage bundles can be written")
    bundle = build_plugin_execution_request_storage_bundle(request, manifest)
    output_path = Path(path)
    if output_path.exists() and not overwrite:
        raise FileExistsError(f"request storage bundle already exists: {output_path}")
    if output_path.exists() and output_path.is_dir():
        raise IsADirectoryError(str(output_path))
    output_path.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(bundle, indent=2, sort_keys=True) + "\n"
    temporary_path = output_path.with_name(
        f".{output_path.name}.{bundle['bundle_hash']}.tmp"
    )
    temporary_path.write_text(payload, encoding="utf-8")
    temporary_path.replace(output_path)
    return bundle