From 9d51080d2ee953b0ee5d6bde34d9b78df3eb3775 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Fri, 17 Jul 2026 19:50:01 +0300 Subject: [PATCH] feat(plugins): add runtime handshake boundary --- README.md | 19 +- docs/01_IMPLEMENTATION_PLAN.md | 1 + docs/07_MISSION_CORE_MONOREPO.md | 39 +++- ...3-device-plugin-ui-and-runtime-boundary.md | 26 ++- ...lpha2-and-experimental-device-lifecycle.md | 5 + ...n-observation-runtime-and-k1-extraction.md | 5 + ...-plugin-owned-frontend-device-workflows.md | 7 +- ...in-runtime-handshake-and-transport-seam.md | 103 +++++++++ packages/plugin-sdk/README.md | 28 ++- .../python/missioncore_plugin_sdk/py.typed | 1 + .../v0alpha2/__init__.py | 15 +- .../v0alpha2/runtime.py | 86 ++++++- .../missioncore_plugin_sdk/v0alpha2/schema.py | 13 +- plugins/xgrids-k1/README.md | 11 +- src/k1link/__init__.py | 2 +- src/k1link/device_plugins/xgrids_k1/facade.py | 22 +- .../device_plugins/xgrids_k1/legacy_api.py | 37 +-- src/k1link/web/app.py | 20 +- src/k1link/web/device_plugin_composition.py | 83 +++++-- src/k1link/web/plugin_runtime.py | 207 ++++++++++++++--- tests/test_plugin_runtime.py | 214 ++++++++++++++++-- tests/test_plugin_sdk_v0alpha2_contracts.py | 50 ++++ 22 files changed, 856 insertions(+), 138 deletions(-) create mode 100644 docs/adr/0011-laboratory-plugin-runtime-handshake-and-transport-seam.md create mode 100644 packages/plugin-sdk/python/missioncore_plugin_sdk/py.typed diff --git a/README.md b/README.md index 0f4f90a..9569907 100644 --- a/README.md +++ b/README.md @@ -25,13 +25,15 @@ unchanged. Plugin SDK v0alpha2 now provides executable, vendor-neutral identity, session, operation, runtime-action, stream, evidence and compatibility contracts. Every -backend action crosses immutable SDK `RuntimeActionInvocation` and -`RuntimeActionResult` validation in the actual dispatcher hot path. The meanings -remain a local experimental vocabulary rather than a mutation of NODE.DC -Platform Ontology. The device process is still transitional and in process: it -uses an explicitly injected K1 normalizer to produce transport-neutral local -consumer views; portable SDK stream envelopes, process isolation, durable -operations, multi-device routing and the remote Edge split remain later gates. +backend runtime must now pass a versioned descriptor/handshake against its +manifest before actions are admitted. Every action then crosses immutable SDK +`RuntimeActionInvocation` and `RuntimeActionResult` validation through a +replaceable runtime transport seam. The meanings remain a local experimental +vocabulary rather than a mutation of NODE.DC Platform Ontology. The current +transport is deliberately in process: it uses an explicitly injected K1 +normalizer to produce transport-neutral local consumer views; portable SDK +stream envelopes, process isolation, durable operations, multi-device routing +and the remote Edge split remain later gates. The current runtime cannot read K1 firmware automatically. It keeps the exact profile inactive until the operator explicitly attests firmware `3.0.2` and @@ -245,6 +247,9 @@ present. - [Vendor-neutral live observation sources](docs/adr/0006-vendor-neutral-observation-sources-and-live-only-timeline.md) - [K1 camera preview copy-remux gateway](docs/adr/0007-k1-camera-preview-copy-remux-gateway.md) - [Durable observation sessions and workspace layout](docs/adr/0008-durable-observation-sessions-and-workspace-layout.md) +- [Device-plugin observation runtime and K1 extraction](docs/adr/0009-device-plugin-observation-runtime-and-k1-extraction.md) +- [Plugin-owned frontend device workflows](docs/adr/0010-plugin-owned-frontend-device-workflows.md) +- [Laboratory plugin runtime handshake and transport seam](docs/adr/0011-laboratory-plugin-runtime-handshake-and-transport-seam.md) - [Redacted live lab report](docs/lab/001_K1_LIVE_MQTT_20260715.redacted.md) - [Session manifest schema](schemas/session-manifest.schema.json) - [Reference input provenance](docs/reference/README.md) diff --git a/docs/01_IMPLEMENTATION_PLAN.md b/docs/01_IMPLEMENTATION_PLAN.md index 9ebd075..f0fb433 100644 --- a/docs/01_IMPLEMENTATION_PLAN.md +++ b/docs/01_IMPLEMENTATION_PLAN.md @@ -19,6 +19,7 @@ Each gate produces evidence and an explicit GO, PAUSE or BLOCKED result. | Stage 6 live viewer | GO — React Control Station, embedded self-hosted Rerun cloud/trajectory and Mac pipeline metrics | | Stage 7 observation archive | GO (point/pose) — durable catalog, recovery, background RRD preparation, saved-session timeline and atomic playback verified | | Stage 7 recorded cameras | GO (contract), acceptance pending — acquisition-owned fMP4 archive and player are implemented/tested; no real archived K1 camera session exists yet | +| Plugin isolation | GO (laboratory control plane) — vendor backend/frontend are plugin-owned; manifest/runtime descriptor parity, versioned handshake, lifecycle health and transport correlation fail closed while execution remains in-process | | Stage 8 product storage | PAUSE — retention, replication, encryption, capacity monitoring and long-run browser/WASM stress remain deployment gates | USB project copying remains optional ground truth rather than a blocker for the diff --git a/docs/07_MISSION_CORE_MONOREPO.md b/docs/07_MISSION_CORE_MONOREPO.md index 8142d82..c34676a 100644 --- a/docs/07_MISSION_CORE_MONOREPO.md +++ b/docs/07_MISSION_CORE_MONOREPO.md @@ -2,8 +2,10 @@ This document maps the current repository to the target Mission Core architecture without claiming that deferred package and process boundaries -already exist. The accepted device-lifecycle decision is recorded in -[`ADR 0004`](adr/0004-plugin-sdk-v0alpha2-and-experimental-device-lifecycle.md). +already exist. The device-lifecycle decision is recorded in +[`ADR 0004`](adr/0004-plugin-sdk-v0alpha2-and-experimental-device-lifecycle.md), +and the bounded laboratory runtime seam in +[`ADR 0011`](adr/0011-laboratory-plugin-runtime-handshake-and-transport-seam.md). ## Current layout @@ -14,7 +16,7 @@ already exist. The accepted device-lifecycle decision is recorded in | `packages/plugin-sdk/` | Installable v0alpha2 Pydantic contracts and JSON Schema export | Portable host/plugin identity, lifecycle, stream and evidence boundary | | `plugins/xgrids-k1/` | v1alpha2 manifest, exact profile/loader and plugin-owned React connection/acquisition UI | Independently versioned XGRIDS device plugin | | `plugins/xgrids-k1/frontend/` | XGRIDS provisioning, acquisition/replay, diagnostics, metrics, runtime mapping and scoped CSS | Plugin-owned reviewed frontend contribution | -| `src/k1link/web/` | Local FastAPI host, plugin composition, in-memory operation/acquisition lifecycle and generic session API | Generic host APIs plus isolated plugin supervisor | +| `src/k1link/web/` | Local FastAPI host, fail-closed runtime handshake/transport seam, in-memory operation/acquisition lifecycle and generic session API | Generic host APIs plus isolated plugin supervisor | | `src/k1link/device_plugins/xgrids_k1/` | Physically isolated K1 BLE/MQTT/protobuf/LZ4/camera/replay/CLI compatibility implementation | Independently built XGRIDS device plugin process | | `src/k1link/data_plane/` | Transport-neutral decoded in-process consumer views | Local projections hydrated from portable SDK envelopes | | `src/k1link/viewer/` | Vendor-neutral Rerun consumer, metrics and recorded blueprint | Replaceable canonical scene sink and presentation adapters | @@ -57,10 +59,14 @@ The Plugin SDK v0alpha2 separately defines model/device/session identity, operation events, acquisition-related session state, canonical stream envelopes, payload handles, evidence lineage and compatibility assessments. It is importable through the root editable path dependency and independently -buildable from `packages/plugin-sdk`. Backend actions now instantiate immutable -SDK `RuntimeActionInvocation` and `RuntimeActionResult` envelopes in the real -dispatcher path. The live spatial path does not yet instantiate portable SDK -stream envelopes or use the SDK `EvidenceStore` protocol. +buildable from `packages/plugin-sdk`. A runtime-owned descriptor must match the +manifest ID, version, host API and exact action set, then complete the +`plugin-runtime/v0alpha1` handshake before its lifecycle health becomes ready. +Backend actions instantiate immutable SDK `RuntimeActionInvocation` and +`RuntimeActionResult` envelopes through the replaceable transport seam, and the +host rejects uncorrelated results. The current transport is still in-process. +The live spatial path does not yet instantiate portable SDK stream envelopes or +use the SDK `EvidenceStore` protocol. Frontend device workflows follow the same ownership rule. The generic host selects a manifest model and mounts its `device.connection` component. Concrete @@ -149,11 +155,14 @@ for the measured gate and remaining limits. ## Remaining extraction order Native replay parity, exact-profile physical point/pose regression, immutable -SDK runtime-action envelopes and physical extraction of K1 transports/codecs are -complete. Portable SDK stream/evidence envelopes remain separate from the -allocation-conscious in-process preview views. +SDK runtime-action envelopes, fail-closed runtime handshake and physical +extraction of K1 transports/codecs are complete. Portable SDK stream/evidence +envelopes remain separate from the allocation-conscious in-process preview +views. -1. Add process isolation and a durable operation/evidence supervisor. +1. When a real deployment or second execution target requires it, implement a + subprocess transport behind the current seam, then add heartbeat timeout, + restart policy and resource limits as separately accepted behavior. 2. Replace compatibility routes and singleton state with multi-device session routing. 3. Split Edge execution from the Control Station behind authenticated transport. @@ -163,6 +172,13 @@ allocation-conscious in-process preview views. plane; keep the modeling-command publisher disabled until its separate safety gate closes. +Complex equipment will likely be assembled from separately useful component +plugins into configured hardware packs, while standalone equipment remains +available for engineering work. That direction is intentionally not a current +domain schema: pack ownership, roles, compatibility and lifecycle must be +derived from real additional component families rather than inferred from the +single K1 vertical. + ## Invariants - Raw bytes are appended and OS-flushed before preview work; aligned raw and @@ -208,6 +224,7 @@ From the repository root: uv sync --frozen --group dev uv run pytest \ tests/test_plugin_sdk_v0alpha2_contracts.py \ + tests/test_plugin_runtime.py \ tests/test_plugin_catalog.py \ tests/test_xgrids_compatibility_profile.py \ tests/test_device_lifecycle.py \ diff --git a/docs/adr/0003-device-plugin-ui-and-runtime-boundary.md b/docs/adr/0003-device-plugin-ui-and-runtime-boundary.md index afbf0d1..857cfce 100644 --- a/docs/adr/0003-device-plugin-ui-and-runtime-boundary.md +++ b/docs/adr/0003-device-plugin-ui-and-runtime-boundary.md @@ -26,11 +26,12 @@ navigation, workspaces, and generic runtime code do not change. Runtime discover of arbitrary JavaScript is deliberately not supported. Backend factories are loaded only from validated, repository-local manifests. -Startup fails unless every catalog manifest produces exactly one adapter with -the same plugin ID and action set. The catalog is then frozen for the life of -the process, so a backend-catalog model cannot appear without an executable -runtime. Frontend availability remains the separate reviewed static-composition -step described below. +Startup fails unless every catalog manifest produces exactly one runtime with +the same plugin ID, version, supported host API and action set. ADR 0011 adds a +versioned descriptor/handshake before the runtime becomes ready. The catalog is +then frozen for the life of the process, so a backend-catalog model cannot +appear without an executable runtime. Frontend availability remains the +separate reviewed static-composition step described below. The local-device UX is: @@ -76,20 +77,23 @@ firmware-specific phases remain inside the XGRIDS plugin adapter. `GET /api/v1/device-models`. - Host-owned action dispatcher: `POST /api/v1/device-plugins/{pluginId}/actions/{actionId}`. +- Runtime lifecycle health: + `GET /api/v1/device-plugin-runtimes`. - Plugin-scoped event stream: `WS /api/v1/device-plugins/{pluginId}/events`, with plugin ID and sequence. - Manifest factory composition: `src/k1link/web/device_plugin_composition.py`. -- Existing K1 action endpoints are compatibility shims over that dispatcher, so - the proven physical implementation is unchanged. +- Existing K1 action endpoints are compatibility shims over the same admitted + runtime transport, so the proven physical implementation is unchanged. The XGRIDS frontend code is physically plugin-owned and statically linked into the Control Station during this v1alpha milestone. Its provisioning, acquisition/replay and diagnostics blocks are separate components, and it may import the host only through `@mission-core/plugin-sdk`. It is not yet a -separately built or signed npm package. Backend composition is manifest-driven; -frontend/backend installed-set parity is a reviewed build-time responsibility -until the host gains signed plugin bundles and a startup compatibility handshake. +separately built or signed npm package. Backend composition is manifest-driven +and performs its control-plane handshake; frontend/backend installed-set parity +is still a reviewed build-time responsibility until the host gains signed +plugin bundles and frontend loading policy. ## Required dependency rules @@ -102,7 +106,7 @@ until the host gains signed plugin bundles and a startup compatibility handshake - Passwords and other secrets cannot enter manifests, runtime snapshots, event payloads, or browser persistence. - Invalid or duplicate plugin/model IDs fail registry construction. -- Manifest/runtime ID or action drift fails backend startup. +- Manifest/runtime ID, version, host API or action drift fails backend startup. - Every v1alpha1 plugin declares exactly one model and a non-mutating, secret-free `state.read` action. - A plugin must stop or cancel active work before model selection can change. diff --git a/docs/adr/0004-plugin-sdk-v0alpha2-and-experimental-device-lifecycle.md b/docs/adr/0004-plugin-sdk-v0alpha2-and-experimental-device-lifecycle.md index 6b45acd..eaf6558 100644 --- a/docs/adr/0004-plugin-sdk-v0alpha2-and-experimental-device-lifecycle.md +++ b/docs/adr/0004-plugin-sdk-v0alpha2-and-experimental-device-lifecycle.md @@ -217,6 +217,11 @@ SDK v0alpha2 runtime-action contracts are active in the backend hot path. This does not imply that frontend state objects, live spatial frames or stored raw evidence already use every SDK contract. +ADR 0011 later adds runtime-owned descriptor, handshake and lifecycle-health +contracts plus a transport seam around this action path. Its current +implementation remains in-process and does not revise the device/acquisition +semantics accepted here. + ## Deliberately deferred The following are not complete and must not be inferred from this ADR: diff --git a/docs/adr/0009-device-plugin-observation-runtime-and-k1-extraction.md b/docs/adr/0009-device-plugin-observation-runtime-and-k1-extraction.md index 95229ea..fb593f3 100644 --- a/docs/adr/0009-device-plugin-observation-runtime-and-k1-extraction.md +++ b/docs/adr/0009-device-plugin-observation-runtime-and-k1-extraction.md @@ -87,6 +87,11 @@ operator workflows remain compatible. The `k1link` distribution name and CLI command remain transitional compatibility names; their implementation is now plugin-owned. +ADR 0011 subsequently places the action control plane behind a versioned +descriptor/handshake/health transport seam. Observation discovery and export +remain host-called plugin contributions and are not yet portable process +transport contracts. + This ADR does not claim process isolation, crash containment, a durable operation journal, signed plugin bundles, multi-device concurrency, portable SDK stream envelopes or SDK `EvidenceStore` persistence. Those require a supervisor and a diff --git a/docs/adr/0010-plugin-owned-frontend-device-workflows.md b/docs/adr/0010-plugin-owned-frontend-device-workflows.md index 20ec815..1c2e344 100644 --- a/docs/adr/0010-plugin-owned-frontend-device-workflows.md +++ b/docs/adr/0010-plugin-owned-frontend-device-workflows.md @@ -81,6 +81,7 @@ contribution and one reviewed composition import, not edits to `App`, The remaining isolation gap is deployment-level: frontend contributions are still compiled into one signed application build, and backend plugins still run -inside the control-plane process. Signed bundles, compatibility handshake, -dynamic loading policy and process/restart containment belong to the supervisor -milestone. +inside the control-plane process. ADR 0011 adds a fail-closed backend runtime +descriptor/handshake and replaceable control-plane transport seam, but not a +subprocess. Signed bundles, frontend loading/realm policy and process/restart +containment remain later deployment milestones. diff --git a/docs/adr/0011-laboratory-plugin-runtime-handshake-and-transport-seam.md b/docs/adr/0011-laboratory-plugin-runtime-handshake-and-transport-seam.md new file mode 100644 index 0000000..2deb9db --- /dev/null +++ b/docs/adr/0011-laboratory-plugin-runtime-handshake-and-transport-seam.md @@ -0,0 +1,103 @@ +# ADR 0011: laboratory plugin runtime handshake and transport seam + +- Status: accepted +- Date: 2026-07-17 +- Extends: ADR 0003, ADR 0004, ADR 0009 and ADR 0010 + +## Context + +Backend and frontend vendor code are physically plugin-owned, but the backend +dispatcher still received a raw in-process action adapter. Manifest parity was +checked through Python attributes rather than a runtime-owned wire contract. +Replacing that adapter with a later subprocess transport would therefore have +changed the dispatcher and compatibility routes as well as composition. + +Mission Core is still a development-source laboratory sandbox. The final Edge, +Control Station and equipment-composition architecture is not known. Complex +hardware packs will probably combine separately useful components, but there is +not yet enough evidence to define pack roles, lifecycle or compatibility +matrices. The next step must preserve the proven plugin boundary without +pretending that production supervision is already designed. + +## Decision + +The executable control plane gains one minimal, versioned runtime protocol: + +```text +missioncore.nodedc/plugin-runtime/v0alpha1 +``` + +Plugin SDK v0alpha2 defines closed immutable contracts for: + +- `RuntimePluginDescriptor` — plugin ID/version, supported host APIs and the + exact action set owned by the runtime; +- `RuntimeHandshakeRequest` and `RuntimeHandshakeResult` — an activation + challenge correlated by handshake ID and runtime instance ID; +- `RuntimeHealthSnapshot` — lifecycle health only: `starting`, `ready` or + `stopped`; +- the existing `RuntimeActionInvocation` and `RuntimeActionResult` envelopes. + +The generic host depends on `DevicePluginRuntimeTransport`, whose control-plane +surface is `handshake`, `health`, `invoke` and `close`. A plugin factory returns +this transport inside `DevicePluginRuntimeContribution`; it no longer exposes a +raw action adapter to composition or the dispatcher. + +Before admitting a plugin, composition validates all of the following: + +1. manifest ID equals runtime descriptor ID; +2. manifest version equals runtime descriptor version; +3. the manifest API is explicitly supported by the runtime; +4. manifest and runtime action sets are exactly equal; +5. handshake response correlates to the request and descriptor; +6. runtime lifecycle health is `ready`. + +Any failure closes the runtime and aborts composition. The dispatcher addresses +only admitted transports and rejects an action result whose invocation, plugin +or action correlation differs from the request. + +`InProcessDevicePluginRuntime` implements the protocol for the current lab. It +wraps the XGRIDS action adapter, performs no serialization and provides no crash +boundary. The deprecated XGRIDS action routes now use the same admitted +transport instead of invoking the adapter directly. The plugin-local camera +gateway remains a same-process data-plane route; portable stream/media IPC is a +separate future gate. + +`GET /api/v1/device-plugin-runtimes` exposes the small lifecycle snapshots. The +main health response reports ready/total runtime counts. These values do not +claim device connectivity, sensor health, process containment or durable +supervision. + +## Acceptance + +The automated gate proves that: + +- two unrelated synthetic plugin runtimes complete independent handshakes and + dispatch through the unchanged generic host; +- actions are rejected before successful activation; +- an unsupported host API fails closed without changing runtime health to + ready; +- uncorrelated action results are rejected; +- close is idempotent and lifecycle health becomes stopped; +- SDK handshake contracts remain closed, immutable and exported as JSON Schema. + +## Consequences + +The generic action path now has a transport-replaceable seam. A later +subprocess implementation can satisfy the same host protocol while moving +serialization, heartbeat, restart and resource policy behind the transport. +This is useful architectural order even while the product topology is still +being discovered. + +This ADR does **not** implement or promise: + +- a subprocess supervisor, crash containment, heartbeat timeout or restart; +- CPU, memory, device-permission or filesystem limits; +- dynamic plugin discovery, quarantine or hot reload; +- bundle signing, frontend realm isolation, installers or updates; +- durable operation recovery or portable stream/EvidenceStore transport; +- multi-device routing or hardware-pack composition semantics. + +The plugin remains the smallest canonical vendor-isolation unit. Whether a +future hardware pack is a host-owned composition of component plugins, a +deployable profile or another domain object is intentionally deferred until +real additional equipment makes that choice testable. diff --git a/packages/plugin-sdk/README.md b/packages/plugin-sdk/README.md index ea194b5..34e4225 100644 --- a/packages/plugin-sdk/README.md +++ b/packages/plugin-sdk/README.md @@ -36,8 +36,9 @@ The separate SDK v0alpha2 package establishes executable contracts for: - independently revisioned enrollment, connectivity, and acquisition states; - operation policy, request, acknowledgement, progress, completion, failure, timeout, cancellation, secret reference, and idempotency boundaries; -- immutable pre-session runtime-action invocation and result envelopes used by - the active backend dispatcher; +- immutable runtime descriptor, activation handshake, lifecycle health, + action-invocation and action-result envelopes used by the active backend + dispatcher; - canonical point cloud, pose, image, encoded video, and device-status streams; - immutable evidence handles, raw transport records, lineage, and store protocol; @@ -57,18 +58,21 @@ Current host implementation references: that frontend surface; - `apps/control-station/src/core/runtime/` — normalized runtime envelope; - `src/k1link/web/plugin_catalog.py` — strict backend manifest validation; -- `src/k1link/web/plugin_runtime.py` — host-owned allowlisted SDK-envelope action dispatcher; +- `src/k1link/web/plugin_runtime.py` — transport protocol, laboratory in-process + implementation and host-owned SDK-envelope action dispatcher; - `src/k1link/web/device_plugin_composition.py` — manifest factory loader and - startup parity checks; -- `docs/adr/0003-device-plugin-ui-and-runtime-boundary.md` — accepted boundary - and extraction sequence. + fail-closed descriptor/handshake checks; +- `docs/adr/0011-laboratory-plugin-runtime-handshake-and-transport-seam.md` — + accepted control-plane seam and its explicit laboratory limits. -The existing lifecycle is fail-closed: inactive provider shells perform no -I/O, events are scoped by plugin ID, and selection cannot change until the -active plugin confirms teardown. Process isolation remains a later host -milestone. v0alpha2 models secret references and operation policy, but host -authorization and the actual secret vault remain separate implementation -responsibilities. +The existing lifecycle is fail-closed: a backend runtime stays `starting` until +its ID, version, host API and exact action set pass the handshake; inactive +provider shells perform no I/O; events are scoped by plugin ID; and selection +cannot change until the active plugin confirms teardown. `ready` health means +only that this control-plane contract is admitted. Process isolation remains a +later host milestone. v0alpha2 models secret references and operation policy, +but host authorization and the actual secret vault remain separate +implementation responsibilities. The React surface is not yet an independently published npm package. Its alias is intentionally narrow so plugin source cannot import Control Station diff --git a/packages/plugin-sdk/python/missioncore_plugin_sdk/py.typed b/packages/plugin-sdk/python/missioncore_plugin_sdk/py.typed new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/packages/plugin-sdk/python/missioncore_plugin_sdk/py.typed @@ -0,0 +1 @@ + diff --git a/packages/plugin-sdk/python/missioncore_plugin_sdk/v0alpha2/__init__.py b/packages/plugin-sdk/python/missioncore_plugin_sdk/v0alpha2/__init__.py index b71a755..eb1b965 100644 --- a/packages/plugin-sdk/python/missioncore_plugin_sdk/v0alpha2/__init__.py +++ b/packages/plugin-sdk/python/missioncore_plugin_sdk/v0alpha2/__init__.py @@ -47,7 +47,15 @@ from .operations import ( validate_operation_request, ) from .payloads import InlinePayload, PayloadHandle, ReferencedPayload -from .runtime import RuntimeActionInvocation, RuntimeActionResult +from .runtime import ( + RUNTIME_PROTOCOL_VERSION, + RuntimeActionInvocation, + RuntimeActionResult, + RuntimeHandshakeRequest, + RuntimeHandshakeResult, + RuntimeHealthSnapshot, + RuntimePluginDescriptor, +) from .schema import contract_json_schemas from .session import ( AcquisitionState, @@ -121,8 +129,13 @@ __all__ = [ "Quaternion", "RawTransportRecord", "RedactionState", + "RUNTIME_PROTOCOL_VERSION", "RuntimeActionInvocation", "RuntimeActionResult", + "RuntimeHandshakeRequest", + "RuntimeHandshakeResult", + "RuntimeHealthSnapshot", + "RuntimePluginDescriptor", "ReferencedPayload", "RuleOutcome", "SecretReference", diff --git a/packages/plugin-sdk/python/missioncore_plugin_sdk/v0alpha2/runtime.py b/packages/plugin-sdk/python/missioncore_plugin_sdk/v0alpha2/runtime.py index fef9a23..c5332b8 100644 --- a/packages/plugin-sdk/python/missioncore_plugin_sdk/v0alpha2/runtime.py +++ b/packages/plugin-sdk/python/missioncore_plugin_sdk/v0alpha2/runtime.py @@ -1,12 +1,92 @@ -"""Minimal pre-session action envelopes for the executable plugin runtime.""" +"""Portable control-plane envelopes for an executable plugin runtime.""" from __future__ import annotations from typing import Literal -from pydantic import AwareDatetime, Field +from pydantic import AwareDatetime, Field, model_validator -from .common import API_VERSION, ApiVersion, ContractModel, Identifier, JsonObject +from .common import ( + API_VERSION, + ApiVersion, + ContractModel, + Identifier, + JsonObject, + ShortText, +) + +RuntimeProtocolVersion = Literal["missioncore.nodedc/plugin-runtime/v0alpha1"] +RUNTIME_PROTOCOL_VERSION: RuntimeProtocolVersion = ( + "missioncore.nodedc/plugin-runtime/v0alpha1" +) + + +class RuntimePluginDescriptor(ContractModel): + """Plugin-owned declaration used before the host admits any action.""" + + api_version: ApiVersion = API_VERSION + kind: Literal["RuntimePluginDescriptor"] = "RuntimePluginDescriptor" + runtime_protocol_version: RuntimeProtocolVersion = RUNTIME_PROTOCOL_VERSION + plugin_id: Identifier + plugin_version: ShortText + supported_host_api_versions: tuple[Identifier, ...] = Field(min_length=1) + action_ids: tuple[Identifier, ...] = Field(min_length=1) + + @model_validator(mode="after") + def declarations_are_unique(self) -> RuntimePluginDescriptor: + if len(self.supported_host_api_versions) != len( + set(self.supported_host_api_versions) + ): + raise ValueError("supported host API versions must be unique") + if len(self.action_ids) != len(set(self.action_ids)): + raise ValueError("runtime action ids must be unique") + return self + + +class RuntimeHandshakeRequest(ContractModel): + """Host compatibility challenge sent before a runtime becomes ready.""" + + api_version: ApiVersion = API_VERSION + kind: Literal["RuntimeHandshakeRequest"] = "RuntimeHandshakeRequest" + runtime_protocol_version: RuntimeProtocolVersion = RUNTIME_PROTOCOL_VERSION + handshake_id: Identifier + plugin_id: Identifier + plugin_version: ShortText + host_api_version: Identifier + requested_at: AwareDatetime + required_action_ids: tuple[Identifier, ...] = Field(min_length=1) + + @model_validator(mode="after") + def required_actions_are_unique(self) -> RuntimeHandshakeRequest: + if len(self.required_action_ids) != len(set(self.required_action_ids)): + raise ValueError("required runtime action ids must be unique") + return self + + +class RuntimeHandshakeResult(ContractModel): + """Runtime response proving identity, protocol and admitted host API.""" + + api_version: ApiVersion = API_VERSION + kind: Literal["RuntimeHandshakeResult"] = "RuntimeHandshakeResult" + handshake_id: Identifier + runtime_instance_id: Identifier + accepted_host_api_version: Identifier + descriptor: RuntimePluginDescriptor + ready_at: AwareDatetime + + +class RuntimeHealthSnapshot(ContractModel): + """Small transport-neutral liveness snapshot for laboratory supervision.""" + + api_version: ApiVersion = API_VERSION + kind: Literal["RuntimeHealthSnapshot"] = "RuntimeHealthSnapshot" + runtime_instance_id: Identifier + plugin_id: Identifier + plugin_version: ShortText + runtime_protocol_version: RuntimeProtocolVersion = RUNTIME_PROTOCOL_VERSION + status: Literal["starting", "ready", "stopped"] + observed_at: AwareDatetime + detail_code: Identifier | None = None class RuntimeActionInvocation(ContractModel): diff --git a/packages/plugin-sdk/python/missioncore_plugin_sdk/v0alpha2/schema.py b/packages/plugin-sdk/python/missioncore_plugin_sdk/v0alpha2/schema.py index 4b91e5f..e55eb06 100644 --- a/packages/plugin-sdk/python/missioncore_plugin_sdk/v0alpha2/schema.py +++ b/packages/plugin-sdk/python/missioncore_plugin_sdk/v0alpha2/schema.py @@ -8,7 +8,14 @@ from .compatibility import CompatibilityAssessment from .evidence import EvidenceRecord, RawTransportRecord from .identity import DeviceInstanceRef from .operations import OperationEvent, OperationPolicy, OperationRequest -from .runtime import RuntimeActionInvocation, RuntimeActionResult +from .runtime import ( + RuntimeActionInvocation, + RuntimeActionResult, + RuntimeHandshakeRequest, + RuntimeHandshakeResult, + RuntimeHealthSnapshot, + RuntimePluginDescriptor, +) from .session import DeviceSessionContext, DeviceSessionSnapshot from .streams import CanonicalStreamEnvelope @@ -25,6 +32,10 @@ def contract_json_schemas() -> dict[str, dict[str, object]]: "OperationEvent": OperationEvent, "RuntimeActionInvocation": RuntimeActionInvocation, "RuntimeActionResult": RuntimeActionResult, + "RuntimePluginDescriptor": RuntimePluginDescriptor, + "RuntimeHandshakeRequest": RuntimeHandshakeRequest, + "RuntimeHandshakeResult": RuntimeHandshakeResult, + "RuntimeHealthSnapshot": RuntimeHealthSnapshot, "CanonicalStreamEnvelope": CanonicalStreamEnvelope, "EvidenceRecord": EvidenceRecord, "RawTransportRecord": RawTransportRecord, diff --git a/plugins/xgrids-k1/README.md b/plugins/xgrids-k1/README.md index fde93bd..b61e8d3 100644 --- a/plugins/xgrids-k1/README.md +++ b/plugins/xgrids-k1/README.md @@ -50,10 +50,12 @@ other vendor state. The frontend contribution imports the host only through `@mission-core/plugin-sdk`; its CSS is scoped below `.xgrids-k1-plugin`. Backend startup loads the reviewed factory declared by `backendEntrypoint`, then -cross-checks its plugin ID and complete action set against this manifest. +cross-checks the runtime-owned plugin ID, version, supported host API and exact +action set against this manifest. The runtime must complete the versioned +control-plane handshake and report `ready` before any action is admitted. Actions enter through the host dispatcher and the transitional facade `src/k1link/device_plugins/xgrids_k1/facade.py`. The old flat routes live in the plugin's -legacy router; both paths delegate to the same proven +legacy router; both paths now cross the same runtime transport and delegate to the proven `XgridsK1CompatibilityService` methods. Synchronous capture/runtime operations run outside the FastAPI event loop. @@ -62,8 +64,9 @@ semantic `acquisition.stop` in `capture-only` mode; replay and pre-v1alpha2 sessions retain the legacy `stream.stop` shim. Neither path claims that the physical K1 stopped without separate operator evidence. BLE, MQTT, codec, archive discovery and RRD export modules now live under the plugin package -after replay parity and physical K1 regression. Process isolation and signed -deployment remain later supervisor gates. +after replay parity and physical K1 regression. The current transport remains +in-process and its health is lifecycle-only; process isolation, crash/restart +containment and signed deployment remain later supervisor gates. The compatibility profile is descriptive and cannot itself authorize a vendor write. The current runtime cannot inspect K1 firmware: it keeps the profile diff --git a/src/k1link/__init__.py b/src/k1link/__init__.py index 1a9005c..2768709 100644 --- a/src/k1link/__init__.py +++ b/src/k1link/__init__.py @@ -1,3 +1,3 @@ -"""Mission Core runtime with the transitional XGRIDS K1 compatibility adapter.""" +"""Mission Core control, observation, and device-plugin runtime.""" __version__ = "0.1.0" diff --git a/src/k1link/device_plugins/xgrids_k1/facade.py b/src/k1link/device_plugins/xgrids_k1/facade.py index 1753b66..61ddfc9 100644 --- a/src/k1link/device_plugins/xgrids_k1/facade.py +++ b/src/k1link/device_plugins/xgrids_k1/facade.py @@ -15,7 +15,10 @@ from pathlib import Path from typing import Any, Literal, Protocol, cast from bleak.exc import BleakError -from missioncore_plugin_sdk.v0alpha2 import RuntimeActionInvocation +from missioncore_plugin_sdk.v0alpha2 import ( + RuntimeActionInvocation, + RuntimePluginDescriptor, +) from pydantic import BaseModel, ConfigDict, Field, SecretStr, ValidationError from k1link.artifacts import write_json_atomic @@ -50,11 +53,13 @@ from k1link.web.device_lifecycle import ( ) from k1link.web.plugin_runtime import ( DevicePluginRuntimeContribution, + InProcessDevicePluginRuntime, PluginActionNotFoundError, PluginExecutionError, ) XGRIDS_K1_PLUGIN_ID = "nodedc.device.xgrids-lixelkity-k1" +XGRIDS_K1_PLUGIN_VERSION = "0.3.0" XGRIDS_K1_MODEL_ID = "xgrids.lixelkity-k1" XGRIDS_K1_COMPATIBILITY_PROFILE_ID = "xgrids.lixelkity-k1.fw-3.0.2.direct-lan.v1" @@ -1821,15 +1826,24 @@ def build_xgrids_k1_plugin(repository_root: Path) -> DevicePluginRuntimeContribu _validate_installed_compatibility_profile(repository_root) service = XgridsK1CompatibilityService(repository_root) adapter = XgridsK1PluginFacade(service) + runtime = InProcessDevicePluginRuntime( + adapter, + RuntimePluginDescriptor( + plugin_id=XGRIDS_K1_PLUGIN_ID, + plugin_version=XGRIDS_K1_PLUGIN_VERSION, + supported_host_api_versions=("missioncore.nodedc/v1alpha2",), + action_ids=tuple(sorted(adapter.action_ids)), + ), + close=service.close, + ) return DevicePluginRuntimeContribution( - adapter=adapter, + runtime=runtime, legacy_routers=( - build_xgrids_k1_legacy_router(adapter), + build_xgrids_k1_legacy_router(runtime), build_xgrids_k1_camera_router( service.camera_preview, XGRIDS_K1_PLUGIN_ID, ), ), observation=build_xgrids_k1_observation(repository_root), - close=service.close, ) diff --git a/src/k1link/device_plugins/xgrids_k1/legacy_api.py b/src/k1link/device_plugins/xgrids_k1/legacy_api.py index 518a122..7366cd6 100644 --- a/src/k1link/device_plugins/xgrids_k1/legacy_api.py +++ b/src/k1link/device_plugins/xgrids_k1/legacy_api.py @@ -18,25 +18,28 @@ from k1link.device_plugins.xgrids_k1.facade import ( LiveRequest, ReplayRequest, ViewerSettingsRequest, - XgridsK1PluginFacade, ) -from k1link.web.plugin_runtime import PluginExecutionError, invoke_device_plugin_adapter +from k1link.web.plugin_runtime import ( + DevicePluginRuntimeTransport, + PluginExecutionError, + invoke_device_plugin_runtime, +) -def build_xgrids_k1_legacy_router(adapter: XgridsK1PluginFacade) -> APIRouter: +def build_xgrids_k1_legacy_router(runtime: DevicePluginRuntimeTransport) -> APIRouter: """Temporary flat API kept for scripts created before the plugin boundary.""" router = APIRouter(include_in_schema=True) @router.get("/api/state", deprecated=True) async def get_state() -> dict[str, Any]: - return await invoke_device_plugin_adapter(adapter, ACTION_STATE_READ, {}) + return await invoke_device_plugin_runtime(runtime, ACTION_STATE_READ, {}) @router.post("/api/ble/scan", deprecated=True) async def scan_ble(request: BleScanRequest) -> dict[str, Any]: try: - return await invoke_device_plugin_adapter( - adapter, + return await invoke_device_plugin_runtime( + runtime, ACTION_DISCOVERY_SCAN, request.model_dump(), ) @@ -46,8 +49,8 @@ def build_xgrids_k1_legacy_router(adapter: XgridsK1PluginFacade) -> APIRouter: @router.post("/api/connect", deprecated=True) async def connect(request: ConnectRequest) -> dict[str, Any]: try: - return await invoke_device_plugin_adapter( - adapter, + return await invoke_device_plugin_runtime( + runtime, ACTION_NETWORK_PROVISION, request.model_dump(), ) @@ -60,8 +63,8 @@ def build_xgrids_k1_legacy_router(adapter: XgridsK1PluginFacade) -> APIRouter: @router.post("/api/session/live", deprecated=True) async def start_live(request: LiveRequest) -> dict[str, Any]: try: - return await invoke_device_plugin_adapter( - adapter, + return await invoke_device_plugin_runtime( + runtime, ACTION_STREAM_START_LIVE, request.model_dump(), ) @@ -71,8 +74,8 @@ def build_xgrids_k1_legacy_router(adapter: XgridsK1PluginFacade) -> APIRouter: @router.post("/api/session/replay", deprecated=True) async def start_replay(request: ReplayRequest) -> dict[str, Any]: try: - return await invoke_device_plugin_adapter( - adapter, + return await invoke_device_plugin_runtime( + runtime, ACTION_STREAM_START_REPLAY, request.model_dump(), ) @@ -81,12 +84,12 @@ def build_xgrids_k1_legacy_router(adapter: XgridsK1PluginFacade) -> APIRouter: @router.post("/api/session/stop", deprecated=True) async def stop_session() -> dict[str, Any]: - return await invoke_device_plugin_adapter(adapter, ACTION_STREAM_STOP, {}) + return await invoke_device_plugin_runtime(runtime, ACTION_STREAM_STOP, {}) @router.post("/api/viewer/settings", deprecated=True) async def update_viewer_settings(request: ViewerSettingsRequest) -> dict[str, Any]: - return await invoke_device_plugin_adapter( - adapter, + return await invoke_device_plugin_runtime( + runtime, ACTION_VIEWER_SETTINGS_UPDATE, request.model_dump(), ) @@ -98,8 +101,8 @@ def build_xgrids_k1_legacy_router(adapter: XgridsK1PluginFacade) -> APIRouter: while True: await websocket.send_json( { - "state": await invoke_device_plugin_adapter( - adapter, + "state": await invoke_device_plugin_runtime( + runtime, ACTION_STATE_READ, {}, ) diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index ec01232..a9c9b08 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -32,6 +32,7 @@ from k1link.web.plugin_runtime import ( PluginActionNotFoundError, PluginExecutionError, PluginNotFoundError, + PluginRuntimeUnavailableError, ) from k1link.web.session_api import build_session_router @@ -175,11 +176,17 @@ async def request_validation_error_handler( @app.get("/api/health") def health() -> dict[str, Any]: + runtime_health = plugin_environment.runtime_health + runtimes_ready = all(item["status"] == "ready" for item in runtime_health) return { - "ok": True, - "status": "ok", + "ok": runtimes_ready, + "status": "ok" if runtimes_ready else "degraded", "service": "mission-core-control-plane", "version": __version__, + "plugin_runtimes": { + "ready": sum(item["status"] == "ready" for item in runtime_health), + "total": len(runtime_health), + }, } @@ -199,6 +206,11 @@ def get_device_models() -> dict[str, Any]: raise HTTPException(status_code=500, detail=str(exc)) from exc +@app.get("/api/v1/device-plugin-runtimes") +def get_device_plugin_runtimes() -> dict[str, Any]: + return {"items": list(plugin_environment.runtime_health)} + + @app.post("/api/v1/device-plugins/{plugin_id}/actions/{action_id}") async def invoke_device_plugin_action( plugin_id: str, @@ -216,6 +228,8 @@ async def invoke_device_plugin_action( raise HTTPException(status_code=400, detail=str(exc)) from exc except PluginExecutionError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc + except PluginRuntimeUnavailableError as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc @app.websocket("/api/v1/device-plugins/{plugin_id}/events") @@ -232,7 +246,7 @@ async def device_plugin_events(websocket: WebSocket, plugin_id: str) -> None: return except (PluginNotFoundError, PluginActionNotFoundError): await websocket.close(code=1008, reason="Device plugin is not available") - except PluginExecutionError: + except (PluginExecutionError, PluginRuntimeUnavailableError): await websocket.close(code=1011, reason="Device plugin state stream failed") diff --git a/src/k1link/web/device_plugin_composition.py b/src/k1link/web/device_plugin_composition.py index 8b18adc..2644af0 100644 --- a/src/k1link/web/device_plugin_composition.py +++ b/src/k1link/web/device_plugin_composition.py @@ -1,11 +1,14 @@ from __future__ import annotations from dataclasses import dataclass +from datetime import UTC, datetime from importlib import import_module from pathlib import Path from typing import Any +from uuid import uuid4 from fastapi import APIRouter +from missioncore_plugin_sdk.v0alpha2 import RuntimeHandshakeRequest from k1link.sessions.plugin_contract import ObservationArchiveSource, RecordingExporter from k1link.web.plugin_catalog import DevicePluginCatalog, DevicePluginManifest @@ -38,11 +41,20 @@ class InstalledDevicePluginEnvironment: @property def recording_exporters(self) -> dict[str, RecordingExporter]: return { - contribution.adapter.plugin_id: contribution.observation.recording_exporter + contribution.runtime.descriptor.plugin_id: ( + contribution.observation.recording_exporter + ) for contribution in self._contributions if contribution.observation is not None } + @property + def runtime_health(self) -> tuple[dict[str, Any], ...]: + return tuple( + contribution.runtime.health().model_dump(mode="json") + for contribution in self._contributions + ) + def close(self) -> None: cleanup_errors = _close_contributions(self._contributions) if cleanup_errors: @@ -60,7 +72,7 @@ def _close_contributions( errors: list[Exception] = [] for contribution in reversed(contributions): try: - contribution.close() + contribution.runtime.close() except Exception as exc: errors.append(exc) return errors @@ -110,42 +122,83 @@ def _load_contribution( ) try: - adapter = contribution.adapter - if adapter.plugin_id != manifest.metadata.id: + runtime = contribution.runtime + descriptor = runtime.descriptor + if descriptor.plugin_id != manifest.metadata.id: raise DevicePluginCompositionError( "Device-plugin manifest/runtime id mismatch: " - f"{manifest.metadata.id} != {adapter.plugin_id}" + f"{manifest.metadata.id} != {descriptor.plugin_id}" + ) + if descriptor.plugin_version != manifest.metadata.version: + raise DevicePluginCompositionError( + "Device-plugin manifest/runtime version mismatch: " + f"{manifest.metadata.version} != {descriptor.plugin_version}" + ) + if manifest.apiVersion not in descriptor.supported_host_api_versions: + raise DevicePluginCompositionError( + f"Device-plugin runtime does not support {manifest.apiVersion}" ) manifest_actions = frozenset(action.id for action in manifest.spec.actions) - if adapter.action_ids != manifest_actions: + if frozenset(descriptor.action_ids) != manifest_actions: raise DevicePluginCompositionError( - f"Device-plugin manifest/runtime actions mismatch for {adapter.plugin_id}" + "Device-plugin manifest/runtime actions mismatch for " + f"{descriptor.plugin_id}" ) observation = contribution.observation if observation is not None: if not observation.archives: raise DevicePluginCompositionError( - f"Device-plugin observation contribution is empty for {adapter.plugin_id}" + "Device-plugin observation contribution is empty for " + f"{descriptor.plugin_id}" ) archive_ids: set[str] = set() archive_roots: set[Path] = set() for archive in observation.archives: - if archive.plugin_id != adapter.plugin_id: + if archive.plugin_id != descriptor.plugin_id: raise DevicePluginCompositionError( "Device-plugin observation/runtime id mismatch: " - f"{archive.plugin_id} != {adapter.plugin_id}" + f"{archive.plugin_id} != {descriptor.plugin_id}" ) if archive.archive_id in archive_ids: raise DevicePluginCompositionError( - f"Duplicate observation archive id for {adapter.plugin_id}" + f"Duplicate observation archive id for {descriptor.plugin_id}" ) resolved_root = archive.root.expanduser().resolve() if resolved_root in archive_roots: raise DevicePluginCompositionError( - f"Duplicate observation archive root for {adapter.plugin_id}" + f"Duplicate observation archive root for {descriptor.plugin_id}" ) archive_ids.add(archive.archive_id) archive_roots.add(resolved_root) + handshake_request = RuntimeHandshakeRequest( + handshake_id=uuid4().hex, + plugin_id=manifest.metadata.id, + plugin_version=manifest.metadata.version, + host_api_version=manifest.apiVersion, + requested_at=datetime.now(UTC), + required_action_ids=tuple(sorted(manifest_actions)), + ) + try: + handshake = runtime.handshake(handshake_request) + runtime_health = runtime.health() + except Exception as exc: + raise DevicePluginCompositionError( + f"Device-plugin runtime handshake failed for {descriptor.plugin_id}" + ) from exc + if ( + handshake.handshake_id != handshake_request.handshake_id + or handshake.descriptor != descriptor + or handshake.accepted_host_api_version != manifest.apiVersion + or runtime_health.runtime_instance_id != handshake.runtime_instance_id + or runtime_health.plugin_id != descriptor.plugin_id + or runtime_health.plugin_version != descriptor.plugin_version + or runtime_health.runtime_protocol_version + != descriptor.runtime_protocol_version + or runtime_health.status != "ready" + ): + raise DevicePluginCompositionError( + f"Device-plugin runtime handshake is inconsistent for {descriptor.plugin_id}" + ) except Exception as exc: _add_cleanup_notes(exc, _close_contributions((contribution,))) raise @@ -155,7 +208,7 @@ def _load_contribution( def load_installed_device_plugins( repository_root: Path, ) -> InstalledDevicePluginEnvironment: - """Load only local, validated manifest entrypoints and cross-check every adapter.""" + """Load local manifests and admit only compatible plugin runtimes.""" catalog = DevicePluginCatalog(repository_root) manifests = catalog.manifests() @@ -167,7 +220,9 @@ def load_installed_device_plugins( _add_cleanup_notes(exc, _close_contributions(loaded)) raise contributions = tuple(loaded) - dispatcher = DevicePluginDispatcher([contribution.adapter for contribution in contributions]) + dispatcher = DevicePluginDispatcher( + [contribution.runtime for contribution in contributions] + ) catalog_ids = {manifest.metadata.id for manifest in manifests} if set(dispatcher.action_declarations) != catalog_ids: raise DevicePluginCompositionError( diff --git a/src/k1link/web/plugin_runtime.py b/src/k1link/web/plugin_runtime.py index b8113ad..7ca017e 100644 --- a/src/k1link/web/plugin_runtime.py +++ b/src/k1link/web/plugin_runtime.py @@ -3,11 +3,19 @@ from __future__ import annotations from collections.abc import Callable, Mapping from dataclasses import dataclass from datetime import UTC, datetime -from typing import Any, Protocol +from threading import RLock +from typing import Any, Literal, Protocol from uuid import uuid4 from fastapi import APIRouter -from missioncore_plugin_sdk.v0alpha2 import RuntimeActionInvocation, RuntimeActionResult +from missioncore_plugin_sdk.v0alpha2 import ( + RuntimeActionInvocation, + RuntimeActionResult, + RuntimeHandshakeRequest, + RuntimeHandshakeResult, + RuntimeHealthSnapshot, + RuntimePluginDescriptor, +) from pydantic import BaseModel, ConfigDict, Field from k1link.sessions.plugin_contract import ObservationRuntimeContribution @@ -32,14 +40,18 @@ def _noop() -> None: return None -@dataclass(frozen=True) -class DevicePluginRuntimeContribution: - """One reviewed backend plugin contribution loaded from its manifest factory.""" +class DevicePluginRuntimeTransport(Protocol): + """Replaceable control-plane seam between Core and one plugin runtime.""" - adapter: DevicePluginActionAdapter - legacy_routers: tuple[APIRouter, ...] = () - observation: ObservationRuntimeContribution | None = None - close: Callable[[], None] = _noop + descriptor: RuntimePluginDescriptor + + def handshake(self, request: RuntimeHandshakeRequest) -> RuntimeHandshakeResult: ... + + def health(self) -> RuntimeHealthSnapshot: ... + + async def invoke(self, invocation: RuntimeActionInvocation) -> RuntimeActionResult: ... + + def close(self) -> None: ... class PluginNotFoundError(LookupError): @@ -54,19 +66,150 @@ class PluginExecutionError(RuntimeError): """A validated plugin action failed while talking to its device/runtime.""" +class PluginRuntimeCompatibilityError(RuntimeError): + """A plugin runtime cannot satisfy the reviewed manifest/host contract.""" + + +class PluginRuntimeUnavailableError(RuntimeError): + """A plugin runtime has not completed activation or has already stopped.""" + + +class InProcessDevicePluginRuntime: + """Laboratory transport preserving the future process boundary in-process.""" + + def __init__( + self, + adapter: DevicePluginActionAdapter, + descriptor: RuntimePluginDescriptor, + *, + close: Callable[[], None] = _noop, + ) -> None: + if adapter.plugin_id != descriptor.plugin_id: + raise PluginRuntimeCompatibilityError( + "In-process adapter/runtime descriptor plugin id mismatch" + ) + if adapter.action_ids != frozenset(descriptor.action_ids): + raise PluginRuntimeCompatibilityError( + "In-process adapter/runtime descriptor actions mismatch" + ) + self._adapter = adapter + self.descriptor = descriptor + self._close = close + self._runtime_instance_id = f"{descriptor.plugin_id}:{uuid4().hex}" + self._status: Literal["starting", "ready", "stopped"] = "starting" + self._lock = RLock() + + def handshake(self, request: RuntimeHandshakeRequest) -> RuntimeHandshakeResult: + with self._lock: + if self._status == "stopped": + raise PluginRuntimeUnavailableError( + f"Device plugin runtime is stopped: {self.descriptor.plugin_id}" + ) + if request.plugin_id != self.descriptor.plugin_id: + raise PluginRuntimeCompatibilityError( + "Runtime handshake plugin id does not match the descriptor" + ) + if request.plugin_version != self.descriptor.plugin_version: + raise PluginRuntimeCompatibilityError( + "Runtime handshake plugin version does not match the descriptor" + ) + if request.runtime_protocol_version != self.descriptor.runtime_protocol_version: + raise PluginRuntimeCompatibilityError( + "Runtime handshake protocol version is not supported" + ) + if request.host_api_version not in self.descriptor.supported_host_api_versions: + raise PluginRuntimeCompatibilityError( + "Runtime does not support the manifest host API version" + ) + if frozenset(request.required_action_ids) != frozenset( + self.descriptor.action_ids + ): + raise PluginRuntimeCompatibilityError( + "Runtime handshake actions do not match the manifest" + ) + self._status = "ready" + return RuntimeHandshakeResult( + handshake_id=request.handshake_id, + runtime_instance_id=self._runtime_instance_id, + accepted_host_api_version=request.host_api_version, + descriptor=self.descriptor, + ready_at=datetime.now(UTC), + ) + + def health(self) -> RuntimeHealthSnapshot: + with self._lock: + status = self._status + return RuntimeHealthSnapshot( + runtime_instance_id=self._runtime_instance_id, + plugin_id=self.descriptor.plugin_id, + plugin_version=self.descriptor.plugin_version, + runtime_protocol_version=self.descriptor.runtime_protocol_version, + status=status, + observed_at=datetime.now(UTC), + ) + + async def invoke(self, invocation: RuntimeActionInvocation) -> RuntimeActionResult: + with self._lock: + if self._status != "ready": + raise PluginRuntimeUnavailableError( + f"Device plugin runtime is not ready: {self.descriptor.plugin_id}" + ) + if invocation.plugin_id != self.descriptor.plugin_id: + raise PluginActionNotFoundError("Runtime invocation targets another plugin") + if invocation.action_id not in self.descriptor.action_ids: + raise PluginActionNotFoundError( + f"Device plugin {self.descriptor.plugin_id} does not declare action " + f"{invocation.action_id}" + ) + output = await self._adapter.invoke(invocation) + return RuntimeActionResult( + invocation_id=invocation.invocation_id, + plugin_id=invocation.plugin_id, + action_id=invocation.action_id, + completed_at=datetime.now(UTC), + output=dict(output), + ) + + def close(self) -> None: + with self._lock: + if self._status == "stopped": + return + self._status = "stopped" + self._close() + + +@dataclass(frozen=True) +class DevicePluginRuntimeContribution: + """One reviewed backend plugin contribution loaded from its manifest factory.""" + + runtime: DevicePluginRuntimeTransport + legacy_routers: tuple[APIRouter, ...] = () + observation: ObservationRuntimeContribution | None = None + + class DevicePluginDispatcher: """Host-owned dispatcher for allowlisted, namespaced plugin actions.""" - def __init__(self, adapters: list[DevicePluginActionAdapter]) -> None: - self._adapters: dict[str, DevicePluginActionAdapter] = {} - for adapter in adapters: - if adapter.plugin_id in self._adapters: - raise ValueError(f"Duplicate runtime device-plugin id: {adapter.plugin_id}") - self._adapters[adapter.plugin_id] = adapter + def __init__(self, runtimes: list[DevicePluginRuntimeTransport]) -> None: + self._runtimes: dict[str, DevicePluginRuntimeTransport] = {} + for runtime in runtimes: + plugin_id = runtime.descriptor.plugin_id + if plugin_id in self._runtimes: + raise ValueError(f"Duplicate runtime device-plugin id: {plugin_id}") + self._runtimes[plugin_id] = runtime @property def action_declarations(self) -> dict[str, frozenset[str]]: - return {plugin_id: adapter.action_ids for plugin_id, adapter in self._adapters.items()} + return { + plugin_id: frozenset(runtime.descriptor.action_ids) + for plugin_id, runtime in self._runtimes.items() + } + + @property + def health_snapshots(self) -> dict[str, RuntimeHealthSnapshot]: + return { + plugin_id: runtime.health() for plugin_id, runtime in self._runtimes.items() + } async def invoke( self, @@ -74,36 +217,36 @@ class DevicePluginDispatcher: action_id: str, payload: Mapping[str, Any], ) -> dict[str, Any]: - adapter = self._adapters.get(plugin_id) - if adapter is None: + runtime = self._runtimes.get(plugin_id) + if runtime is None: raise PluginNotFoundError(f"Device plugin is not installed: {plugin_id}") - return await invoke_device_plugin_adapter(adapter, action_id, payload) + return await invoke_device_plugin_runtime(runtime, action_id, payload) -async def invoke_device_plugin_adapter( - adapter: DevicePluginActionAdapter, +async def invoke_device_plugin_runtime( + runtime: DevicePluginRuntimeTransport, action_id: str, payload: Mapping[str, Any], ) -> dict[str, Any]: """Validate SDK request/result envelopes around one plugin action call.""" - if action_id not in adapter.action_ids: + descriptor = runtime.descriptor + if action_id not in descriptor.action_ids: raise PluginActionNotFoundError( - f"Device plugin {adapter.plugin_id} does not declare action {action_id}" + f"Device plugin {descriptor.plugin_id} does not declare action {action_id}" ) invocation = RuntimeActionInvocation( invocation_id=uuid4().hex, - plugin_id=adapter.plugin_id, + plugin_id=descriptor.plugin_id, action_id=action_id, requested_at=datetime.now(UTC), parameters=dict(payload), ) - output = await adapter.invoke(invocation) - result = RuntimeActionResult( - invocation_id=invocation.invocation_id, - plugin_id=invocation.plugin_id, - action_id=invocation.action_id, - completed_at=datetime.now(UTC), - output=dict(output), - ) + result = await runtime.invoke(invocation) + if ( + result.invocation_id != invocation.invocation_id + or result.plugin_id != invocation.plugin_id + or result.action_id != invocation.action_id + ): + raise PluginExecutionError("Plugin runtime returned an uncorrelated action result") return dict(result.output) diff --git a/tests/test_plugin_runtime.py b/tests/test_plugin_runtime.py index 45a27ee..58c2d70 100644 --- a/tests/test_plugin_runtime.py +++ b/tests/test_plugin_runtime.py @@ -3,11 +3,19 @@ from __future__ import annotations import asyncio import json import threading +from datetime import UTC, datetime from pathlib import Path from typing import Any +from uuid import uuid4 import pytest -from missioncore_plugin_sdk.v0alpha2 import RuntimeActionInvocation +from missioncore_plugin_sdk.v0alpha2 import ( + RuntimeActionInvocation, + RuntimeActionResult, + RuntimeHandshakeRequest, + RuntimeHandshakeResult, + RuntimePluginDescriptor, +) from pydantic import ValidationError import k1link.web.device_plugin_composition as plugin_composition @@ -19,6 +27,7 @@ from k1link.device_plugins.xgrids_k1.facade import ( ACTION_STREAM_STOP, ACTION_VIEWER_SETTINGS_UPDATE, XGRIDS_K1_PLUGIN_ID, + XGRIDS_K1_PLUGIN_VERSION, CompatibilityAttestationRequest, ConnectRequest, ViewerSettingsRequest, @@ -33,8 +42,12 @@ from k1link.web.plugin_catalog import DevicePluginCatalog from k1link.web.plugin_runtime import ( DevicePluginDispatcher, DevicePluginRuntimeContribution, + InProcessDevicePluginRuntime, PluginActionNotFoundError, + PluginExecutionError, PluginNotFoundError, + PluginRuntimeCompatibilityError, + PluginRuntimeUnavailableError, ) @@ -76,6 +89,38 @@ class FakeXgridsService: return {"phase": "idle", "viewer_settings": request.model_dump()} +def _in_process_runtime( + adapter: Any, + *, + plugin_version: str = XGRIDS_K1_PLUGIN_VERSION, + host_api_version: str = "missioncore.nodedc/v1alpha2", + close: Any | None = None, + activate: bool = True, +) -> InProcessDevicePluginRuntime: + runtime = InProcessDevicePluginRuntime( + adapter, + RuntimePluginDescriptor( + plugin_id=adapter.plugin_id, + plugin_version=plugin_version, + supported_host_api_versions=(host_api_version,), + action_ids=tuple(sorted(adapter.action_ids)), + ), + **({"close": close} if close is not None else {}), + ) + if activate: + runtime.handshake( + RuntimeHandshakeRequest( + handshake_id=uuid4().hex, + plugin_id=adapter.plugin_id, + plugin_version=plugin_version, + host_api_version=host_api_version, + requested_at=datetime.now(UTC), + required_action_ids=tuple(sorted(adapter.action_ids)), + ) + ) + return runtime + + def test_manifest_and_runtime_facade_declare_identical_actions() -> None: repository_root = Path(__file__).resolve().parents[1] manifest = next( @@ -112,8 +157,11 @@ def test_composition_closes_a_runtime_that_does_not_match_its_manifest( ) closed: list[bool] = [] contribution = DevicePluginRuntimeContribution( - adapter=XgridsK1PluginFacade(FakeXgridsService()), - close=lambda: closed.append(True), + runtime=_in_process_runtime( + XgridsK1PluginFacade(FakeXgridsService()), + close=lambda: closed.append(True), + activate=False, + ), ) monkeypatch.setattr( plugin_composition, @@ -127,6 +175,47 @@ def test_composition_closes_a_runtime_that_does_not_match_its_manifest( assert closed == [True] +def test_composition_rejects_an_uncorrelated_runtime_handshake( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class UncorrelatedHandshakeRuntime(InProcessDevicePluginRuntime): + def handshake( + self, + request: RuntimeHandshakeRequest, + ) -> RuntimeHandshakeResult: + result = super().handshake(request) + return result.model_copy(update={"handshake_id": "another-handshake"}) + + repository_root = Path(__file__).resolve().parents[1] + manifest = next( + candidate + for candidate in DevicePluginCatalog(repository_root).manifests() + if candidate.metadata.id == XGRIDS_K1_PLUGIN_ID + ) + closed: list[bool] = [] + adapter = XgridsK1PluginFacade(FakeXgridsService()) + runtime = UncorrelatedHandshakeRuntime( + adapter, + RuntimePluginDescriptor( + plugin_id=adapter.plugin_id, + plugin_version=XGRIDS_K1_PLUGIN_VERSION, + supported_host_api_versions=(manifest.apiVersion,), + action_ids=tuple(sorted(adapter.action_ids)), + ), + close=lambda: closed.append(True), + ) + monkeypatch.setattr( + plugin_composition, + "_load_factory", + lambda _: lambda __: DevicePluginRuntimeContribution(runtime=runtime), + ) + + with pytest.raises(DevicePluginCompositionError, match="handshake is inconsistent"): + plugin_composition._load_contribution(repository_root, manifest) + + assert closed == [True] + + def test_composition_loads_and_dispatches_two_synthetic_plugins( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, @@ -149,8 +238,13 @@ def test_composition_loads_and_dispatches_two_synthetic_plugins( closed: list[str] = [] contributions = { f"synthetic:{suffix}": DevicePluginRuntimeContribution( - adapter=SyntheticAdapter(f"example.{suffix}"), - close=lambda suffix=suffix: closed.append(suffix), + runtime=_in_process_runtime( + SyntheticAdapter(f"example.{suffix}"), + plugin_version="0.1.0", + host_api_version="missioncore.nodedc/v1alpha1", + close=lambda suffix=suffix: closed.append(suffix), + activate=False, + ), ) for suffix in ("first", "second") } @@ -205,6 +299,7 @@ def test_composition_loads_and_dispatches_two_synthetic_plugins( environment.dispatcher.invoke(f"example.{suffix}", "state.read", {}) ) assert state == {"plugin_id": f"example.{suffix}"} + assert {item["status"] for item in environment.runtime_health} == {"ready"} finally: environment.close() @@ -226,12 +321,16 @@ def test_environment_shutdown_attempts_every_plugin_after_close_failure( legacy_routers=(), _contributions=( DevicePluginRuntimeContribution( - adapter=XgridsK1PluginFacade(FakeXgridsService()), - close=lambda: closed.append("healthy"), + runtime=_in_process_runtime( + XgridsK1PluginFacade(FakeXgridsService()), + close=lambda: closed.append("healthy"), + ), ), DevicePluginRuntimeContribution( - adapter=XgridsK1PluginFacade(FakeXgridsService()), - close=failing_close, + runtime=_in_process_runtime( + XgridsK1PluginFacade(FakeXgridsService()), + close=failing_close, + ), ), ), ) @@ -245,7 +344,9 @@ def test_environment_shutdown_attempts_every_plugin_after_close_failure( def test_dispatcher_routes_allowlisted_action_to_xgrids_facade() -> None: service = FakeXgridsService() - dispatcher = DevicePluginDispatcher([XgridsK1PluginFacade(service)]) + dispatcher = DevicePluginDispatcher( + [_in_process_runtime(XgridsK1PluginFacade(service))] + ) state = asyncio.run( dispatcher.invoke( @@ -260,7 +361,9 @@ def test_dispatcher_routes_allowlisted_action_to_xgrids_facade() -> None: def test_dispatcher_rejects_unknown_plugin_and_action() -> None: - dispatcher = DevicePluginDispatcher([XgridsK1PluginFacade(FakeXgridsService())]) + dispatcher = DevicePluginDispatcher( + [_in_process_runtime(XgridsK1PluginFacade(FakeXgridsService()))] + ) with pytest.raises(PluginNotFoundError): asyncio.run(dispatcher.invoke("missing.plugin", ACTION_STREAM_STOP, {})) @@ -270,7 +373,9 @@ def test_dispatcher_rejects_unknown_plugin_and_action() -> None: def test_facade_validates_payload_before_calling_service() -> None: service = FakeXgridsService() - dispatcher = DevicePluginDispatcher([XgridsK1PluginFacade(service)]) + dispatcher = DevicePluginDispatcher( + [_in_process_runtime(XgridsK1PluginFacade(service))] + ) with pytest.raises(ValidationError): asyncio.run( @@ -332,7 +437,9 @@ def test_facade_preserves_existing_runtime_operations( expected_call: str, ) -> None: service = FakeXgridsService() - dispatcher = DevicePluginDispatcher([XgridsK1PluginFacade(service)]) + dispatcher = DevicePluginDispatcher( + [_in_process_runtime(XgridsK1PluginFacade(service))] + ) asyncio.run(dispatcher.invoke(XGRIDS_K1_PLUGIN_ID, action_id, payload)) @@ -348,10 +455,89 @@ def test_sync_runtime_actions_run_outside_the_api_event_loop() -> None: return super().stop() service = ThreadAwareService() - dispatcher = DevicePluginDispatcher([XgridsK1PluginFacade(service)]) + dispatcher = DevicePluginDispatcher( + [_in_process_runtime(XgridsK1PluginFacade(service))] + ) event_loop_thread = threading.get_ident() asyncio.run(dispatcher.invoke(XGRIDS_K1_PLUGIN_ID, ACTION_STREAM_STOP, {})) assert service.thread_id is not None assert service.thread_id != event_loop_thread + + +def test_runtime_requires_successful_handshake_before_dispatch() -> None: + adapter = XgridsK1PluginFacade(FakeXgridsService()) + runtime = _in_process_runtime(adapter, activate=False) + dispatcher = DevicePluginDispatcher([runtime]) + + assert runtime.health().status == "starting" + with pytest.raises(PluginRuntimeUnavailableError, match="not ready"): + asyncio.run(dispatcher.invoke(XGRIDS_K1_PLUGIN_ID, ACTION_STREAM_STOP, {})) + + with pytest.raises(PluginRuntimeCompatibilityError, match="host API"): + runtime.handshake( + RuntimeHandshakeRequest( + handshake_id="handshake-incompatible", + plugin_id=XGRIDS_K1_PLUGIN_ID, + plugin_version=XGRIDS_K1_PLUGIN_VERSION, + host_api_version="missioncore.nodedc/v1alpha1", + requested_at=datetime.now(UTC), + required_action_ids=tuple(sorted(adapter.action_ids)), + ) + ) + + assert runtime.health().status == "starting" + + +def test_runtime_health_transitions_to_stopped_and_close_is_idempotent() -> None: + closed: list[bool] = [] + runtime = _in_process_runtime( + XgridsK1PluginFacade(FakeXgridsService()), + close=lambda: closed.append(True), + ) + + assert runtime.health().status == "ready" + runtime.close() + runtime.close() + + assert runtime.health().status == "stopped" + assert closed == [True] + + +def test_dispatcher_rejects_uncorrelated_transport_result() -> None: + class UncorrelatedRuntime(InProcessDevicePluginRuntime): + async def invoke( + self, + invocation: RuntimeActionInvocation, + ) -> RuntimeActionResult: + result = await super().invoke(invocation) + return result.model_copy(update={"invocation_id": "another-invocation"}) + + adapter = XgridsK1PluginFacade(FakeXgridsService()) + descriptor = RuntimePluginDescriptor( + plugin_id=adapter.plugin_id, + plugin_version=XGRIDS_K1_PLUGIN_VERSION, + supported_host_api_versions=("missioncore.nodedc/v1alpha2",), + action_ids=tuple(sorted(adapter.action_ids)), + ) + runtime = UncorrelatedRuntime(adapter, descriptor) + runtime.handshake( + RuntimeHandshakeRequest( + handshake_id="handshake-correlation", + plugin_id=adapter.plugin_id, + plugin_version=XGRIDS_K1_PLUGIN_VERSION, + host_api_version="missioncore.nodedc/v1alpha2", + requested_at=datetime.now(UTC), + required_action_ids=descriptor.action_ids, + ) + ) + + with pytest.raises(PluginExecutionError, match="uncorrelated"): + asyncio.run( + DevicePluginDispatcher([runtime]).invoke( + XGRIDS_K1_PLUGIN_ID, + ACTION_STREAM_STOP, + {}, + ) + ) diff --git a/tests/test_plugin_sdk_v0alpha2_contracts.py b/tests/test_plugin_sdk_v0alpha2_contracts.py index 01bae1a..b4eb5a6 100644 --- a/tests/test_plugin_sdk_v0alpha2_contracts.py +++ b/tests/test_plugin_sdk_v0alpha2_contracts.py @@ -339,6 +339,52 @@ def test_plugin_sdk_v0alpha2_compatibility_blocks_unproven_control() -> None: ) +def test_plugin_sdk_v0alpha2_runtime_handshake_is_versioned_and_closed() -> None: + descriptor = sdk.RuntimePluginDescriptor( + plugin_id="nodedc.device.synthetic", + plugin_version="0.1.0", + supported_host_api_versions=("missioncore.nodedc/v1alpha2",), + action_ids=("state.read", "acquisition.start"), + ) + request = sdk.RuntimeHandshakeRequest( + handshake_id="handshake-001", + plugin_id=descriptor.plugin_id, + plugin_version=descriptor.plugin_version, + host_api_version="missioncore.nodedc/v1alpha2", + requested_at=NOW, + required_action_ids=descriptor.action_ids, + ) + result = sdk.RuntimeHandshakeResult( + handshake_id=request.handshake_id, + runtime_instance_id="runtime.synthetic-001", + accepted_host_api_version=request.host_api_version, + descriptor=descriptor, + ready_at=NOW, + ) + health = sdk.RuntimeHealthSnapshot( + runtime_instance_id=result.runtime_instance_id, + plugin_id=descriptor.plugin_id, + plugin_version=descriptor.plugin_version, + status="ready", + observed_at=NOW, + ) + + assert descriptor.runtime_protocol_version == sdk.RUNTIME_PROTOCOL_VERSION + assert result.descriptor == descriptor + assert health.status == "ready" + with pytest.raises(ValidationError, match="runtime action ids must be unique"): + sdk.RuntimePluginDescriptor.model_validate( + { + **descriptor.model_dump(), + "action_ids": ["state.read", "state.read"], + } + ) + with pytest.raises(ValidationError, match="extra"): + sdk.RuntimeHandshakeRequest.model_validate( + {**request.model_dump(), "process_id": 42} + ) + + def test_plugin_sdk_v0alpha2_exports_closed_json_schemas_and_vocabulary() -> None: schemas = sdk.contract_json_schemas() @@ -348,6 +394,10 @@ def test_plugin_sdk_v0alpha2_exports_closed_json_schemas_and_vocabulary() -> Non "OperationEvent", "RuntimeActionInvocation", "RuntimeActionResult", + "RuntimePluginDescriptor", + "RuntimeHandshakeRequest", + "RuntimeHandshakeResult", + "RuntimeHealthSnapshot", "CanonicalStreamEnvelope", "EvidenceRecord", "CompatibilityAssessment",