L3: zmqruntime runtime vocabulary, declared messages, launch policy and transport config

docs/refactor/consolidation-20261009/L3-zmqruntime-vocabulary.md

@@ -0,0 +1,54 @@
+# L3: zmqruntime owns its runtime vocabulary, launch policy and transport config
+
+**Head audited:** `openhcs` `main` at `89b8fdbff` (#1185); `zmqruntime` `main` at `079a1c4` (0.5.0); `pyqt-reactive` `main` at `291b1be` (0.3.30); `PolyStore` `main` at `45a9bc1` (0.5.0). **Rules:** [00-RULES.md](00-RULES.md). **Step 3.**
+**Shared abstractions** ([02-SHARED-ABSTRACTIONS.md](02-SHARED-ABSTRACTIONS.md)). *Builds:* zmqruntime's declared-message base. *Uses:* the L2 codec (`python_introspect.to_jsonable`, `dataclass_from_mapping`), `AutoRegisterMeta`.
+
+## What is wrong
+
+**A generic runtime library speaks one domain's words, writes its codec by hand 45 times, and two pieces of its own machinery live in its consumers.**
+
+- **Domain words.** 107 microscopy tokens in six zmqruntime modules: `plate_id` on `TaskProgress`, `ExecuteRequest`, `ExecutionRecord`, `RunningExecutionInfo`, `QueuedExecutionInfo` (`messages.py`, 29); `GenericPlateProjection`, `plates`, `by_plate_latest`, `get_plate`, `adapter.plate_id` (`progress/projection.py`, 60); `MessageFields.WELL_COUNT/WELLS/WELL_ID` and the results summary written with them (`execution/server.py:289-290`); `execution_plate_id`. The kernel's word for the values of the parallel axis is `partition_values` (G2: `ZMQCompilationRequest.partition_values`).
+- **Hand-written codec.** `messages.py` defines 24 `to_dict` and 21 `from_dict` (45) over 21 message dataclasses. Each restates its field list two or three times; strictness differs per class (`WorkerState` rejects unknown keys, `ExecutionRecord` sweeps them into `metadata`, the rest drop them); `ExecutionRecord.to_dict` carries a private `_to_transport_value` encoder, a fourth encoder beside `to_jsonable`. `PongResponse.from_dict` is 98 lines.
+- **Launch policy in a widget library.** `pyqt_reactive/process_launch.py` (standard-library only) owns console suppression and detached sessions for every background process. Ten openhcs call sites in six production modules (`gui_startup`, `processing/backends/lib_registry/registry_service`, `desktop/restart`, `desktop/update`, `runtime/zmq_execution_client`, `runtime/viewer_protocol`) and three in pyqt-reactive (`log_highlight_client`, `system_metrics_sampler` x2) import a process policy from a Qt library. The policy is an enum (`BackgroundProcessPlatform.WINDOWS/OTHER`) with `if platform is WINDOWS` branches in `resolve` and `python_executable` (rule 1a).
+- **Transport config in the application.** `openhcs/runtime/zmq_config.py` (`OpenHCSZMQConfig`, 119 lines) declares twelve generic execution-endpoint fields (hosts, transport mode, request deadlines, discovery scan, readiness poll, port range, artifact TTL) and `client_endpoint`. zmqruntime's own base client reads `config.server_poll_interval_seconds` (`client.py:1291`), a field its `ZMQConfig` does not declare.
+- **Compatibility shims** in the execution server: `_run_execution` and `_handle_status` ("Compatibility shim for callers still bound to the old private hook").
+- **Dead:** `ROIMessage`, `ShapesMessage` (no reader outside their own module), `MessageFields.WELL_ID`.
+
+## Target
+
+- **Vocabulary.** `plate_id` → `subject_id` (what one execution request is about), `execution_plate_id` → `execution_subject_id`, `GenericPlateProjection` → `GenericSubjectProjection` (`subjects`, `by_subject_latest`, `get_subject`, `adapter.subject_id`). The results summary carries `partition_count`/`partition_values`. No microscopy word remains in zmqruntime.
+- **Declared messages.** One base, `WireMessage`: `to_dict()` is `to_jsonable(self)`, `from_dict()` is `dataclass_from_mapping(cls, data)`. Two capabilities compose on it: `TypedWireMessage` (a `wire_type` class attribute carried under `type`) and `WireView` (a declared message read out of a wider payload, for application progress envelopes and status responses). Control requests derive their body from their declared fields under `ControlRequestHeader`, which alone encodes the relative observation budget. No message class writes `to_dict`/`from_dict`. `ExecutionRecord`'s server-local attachments (`extras`, `progress_event`) stop being dataclass fields, so its declared fields are exactly its wire record.
+- **Launch policy** moves to `zmqruntime.process_launch`. `BackgroundProcessPlatform` becomes an `AutoRegisterMeta` family keyed by `sys.platform`: `WindowsBackgroundProcesses` owns creation flags and the `pythonw.exe` interpreter; the default family member owns `start_new_session`. pyqt-reactive and openhcs import it from zmqruntime; `pyqt_reactive/process_launch.py` is deleted.
+- **Transport config** moves to `zmqruntime.execution.config.ExecutionTransportConfig(ZMQConfig)`; `server_poll_interval_seconds` moves onto `ZMQConfig`, where the base client reads it. `OpenHCSZMQConfig` keeps only OpenHCS's two names (`app_name`, `ipc_socket_prefix`), so saved UI preferences that name the type load unchanged.
+- **Deleted:** the 45 methods, `_to_transport_value`, `ExecutionRecord.metadata` as a field, `ROIMessage`, `ShapesMessage`, `MessageFields` keys no reader uses, both compatibility shims, `pyqt_reactive/process_launch.py`, the generic fields of `OpenHCSZMQConfig`, and openhcs's hand-assembled `ExecuteRequest` items.
+
+## Persisted state
+
+Runtime only (ZMQ messages between processes of one install). The UI preference file stores `OpenHCSZMQConfig` field values; every field keeps its name and owner type, so it loads unchanged.
+
+## Releases
+
+zmqruntime 0.6.0 (breaking); pyqt-reactive 0.4.0 (drops `process_launch`, `zmqruntime>=0.6,<0.7`); PolyStore 0.6.0 (`zmqruntime>=0.6,<0.7`).
+
+## Guards
+
+`tests/unit/test_l3_zmqruntime_vocabulary_guards.py`:
+- No microscopy word (`plate`, `well`) in any `external/zmqruntime/src` module.
+- No class in `zmqruntime.messages` defines `to_dict` or `from_dict` except the declared bases.
+- `pyqt_reactive/process_launch.py` does not exist; nothing under `openhcs/` imports `pyqt_reactive.process_launch`.
+- `OpenHCSZMQConfig` declares no field beyond `app_name` and `ipc_socket_prefix`.
+
+## Tests
+
+- zmqruntime: one round-trip test over every declared message (registry of `WireMessage` subclasses), one test that a typed message rejects another type, one for envelope reading, the launch-policy family test (moved from pyqt-reactive), the transport config validation test.
+- openhcs: the existing ZMQ execution, progress and streaming tests; tests that built wire dicts by hand use the declared types.
+
+## New-case experiments
+
+- A new control message: today a dataclass plus a hand-written `to_dict`/`from_dict` pair restating its fields and its `type`. After: the dataclass under `ControlRequestHeader` or `TypedWireMessage`.
+- A non-microscopy application (the remote-sensing witness) submitting work: today its request is about a `plate_id` and its results report `wells`. After: `subject_id` and `partition_values`.
+- A new host family for background processes: today an enum member plus a branch in two methods. After: one subclass.
+
+## Done when
+
+The guards pass; the three library PRs are open with CI green; openhcs points at the PR branches and its runtime, ZMQ and streaming tests, the `test_main.py` disk ZMQ cases and the Official30 parity check over ZMQ (29/30) pass.

openhcs/core/progress/projection.py

@@ -12,7 +12,7 @@
 from zmqruntime.progress import (
     GenericAxisProjection,
     GenericExecutionProjection,
-    GenericPlateProjection,
+    GenericSubjectProjection,
     ProgressProjectionAdapterABC,
     build_execution_projection,
 )
@@ -287,11 +287,11 @@ class PlateRuntimeIdentity:
     @classmethod
     def from_generic_plate(
         cls,
-        generic_plate: GenericPlateProjection[PlateRuntimeState],
+        generic_plate: GenericSubjectProjection[PlateRuntimeState],
     ) -> "PlateRuntimeIdentity":
         return cls(
             execution_id=generic_plate.execution_id,
-            plate_id=generic_plate.plate_id,
+            plate_id=generic_plate.subject_id,
         )
 
 
@@ -308,7 +308,7 @@ class PlateRuntimeProjection:
     @classmethod
     def from_generic_plate(
         cls,
-        generic_plate: GenericPlateProjection[PlateRuntimeState],
+        generic_plate: GenericSubjectProjection[PlateRuntimeState],
         *,
         state_channel: ProgressChannel | None = None,
     ) -> "PlateRuntimeProjection":
@@ -388,7 +388,7 @@ def from_generic_projection(
     ) -> "ExecutionRuntimeProjection":
         projection = cls()
 
-        for generic_plate in generic_projection.plates:
+        for generic_plate in generic_projection.subjects:
             identity = PlateRuntimeIdentity.from_generic_plate(generic_plate)
             state_declaration = PlateRuntimeStateDeclarationBase.for_state(
                 generic_plate.state
@@ -404,7 +404,7 @@ def from_generic_projection(
                 )
             )
 
-        for generic_plate in generic_projection.by_plate_latest.values():
+        for generic_plate in generic_projection.by_subject_latest.values():
             projection.mark_latest(
                 PlateRuntimeIdentity.from_generic_plate(generic_plate)
             )
@@ -463,11 +463,11 @@ def reconcile_server_executions(
         """Project the authoritative live server queue over retained event history."""
 
         running_identities = {
-            PlateRuntimeIdentity(entry.execution_id, entry.plate_id)
+            PlateRuntimeIdentity(entry.execution_id, entry.subject_id)
             for entry in running_executions
         }
         for entry in queued_executions:
-            identity = PlateRuntimeIdentity(entry.execution_id, entry.plate_id)
+            identity = PlateRuntimeIdentity(entry.execution_id, entry.subject_id)
             if identity in running_identities:
                 continue
             current = self.by_identity.get(identity)
@@ -489,7 +489,7 @@ def reconcile_server_executions(
             self.mark_latest(identity)
 
         for entry in running_executions:
-            identity = PlateRuntimeIdentity(entry.execution_id, entry.plate_id)
+            identity = PlateRuntimeIdentity(entry.execution_id, entry.subject_id)
             current = self.by_identity.get(identity)
             server_state = (
                 PlateRuntimeState.COMPILING
@@ -535,7 +535,7 @@ def recalculate_summary(self) -> None:
 class _OpenHCSProjectionAdapter(
     ProgressProjectionAdapterABC[ProgressEvent, PlateRuntimeState]
 ):
-    def plate_id(self, event: ProgressEvent) -> str:
+    def subject_id(self, event: ProgressEvent) -> str:
         return event.plate_id
 
     def axis_id(self, event: ProgressEvent) -> str:

openhcs/core/progress/types.py

@@ -451,7 +451,7 @@ class ProgressIdentity:
     def from_transport_fields(cls, data: dict[str, Any]) -> "ProgressIdentity":
         return cls(
             execution_id=str(data["execution_id"]),
-            plate_id=str(data["plate_id"]),
+            plate_id=str(data["subject_id"]),
             axis_id=str(data["axis_id"]),
             step_name=str(data["step_name"]),
         )
@@ -647,7 +647,7 @@ def from_dict(cls, data: dict[str, Any]) -> "ProgressEvent":
         # Validate OpenHCS-required fields
         required_fields = {
             "execution_id",
-            "plate_id",
+            "subject_id",
             "axis_id",
             "step_name",
             "phase",
@@ -718,7 +718,7 @@ def to_dict(self) -> dict[str, Any]:
         """
         result = {
             "execution_id": self.execution_id,
-            "plate_id": self.plate_id,
+            "subject_id": self.plate_id,
             "axis_id": self.axis_id,
             "step_name": self.step_name,
             "phase": self.phase.value,  # Enum → string

openhcs/desktop/restart.py

@@ -7,7 +7,7 @@
 from pathlib import Path
 import subprocess
 
-from pyqt_reactive.process_launch import BackgroundProcessLaunchPolicy
+from zmqruntime.process_launch import BackgroundProcessLaunchPolicy
 from PyQt6.QtCore import QTimer
 
 from openhcs.desktop.update import (

openhcs/desktop/update.py

@@ -24,8 +24,8 @@
 from PyQt6.QtGui import QDesktopServices
 from PyQt6.QtNetwork import QNetworkAccessManager, QNetworkReply, QNetworkRequest
 from PyQt6.QtWidgets import QMessageBox
-from pyqt_reactive.process_launch import BackgroundProcessLaunchPolicy
 from python_introspect import dataclass_from_mapping, to_jsonable
+from zmqruntime.process_launch import BackgroundProcessLaunchPolicy
 
 from openhcs import __version__ as OPENHCS_VERSION
 from openhcs.desktop.deployment import (

openhcs/gui_startup.py

@@ -22,7 +22,7 @@
 from threading import Event, Lock, Thread
 from typing import IO
 
-from pyqt_reactive.process_launch import BackgroundProcessLaunchPolicy
+from zmqruntime.process_launch import BackgroundProcessLaunchPolicy
 
 _STARTUP_WINDOW_CHILD_ARGUMENT = "--startup-window-child"
 _STARTUP_PROGRESS_ENVIRONMENT = "OPENHCS_STARTUP_PROGRESS"

openhcs/omero/plugin/omero_openhcs/views.py

@@ -14,7 +14,16 @@
 from django.shortcuts import render
 from django.views.decorators.http import require_http_methods
 from omeroweb.webclient.decorators import login_required
-from zmqruntime import TransportEndpoint, TransportMode, ZMQConfig
+from zmqruntime import (
+    CancelRequest,
+    ControlMessageType,
+    ControlRequestHeader,
+    ExecuteRequest,
+    StatusRequest,
+    TransportEndpoint,
+    TransportMode,
+    ZMQConfig,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -51,37 +60,34 @@ def _send_request(self, request):
     def ping(self):
         """Ping server to check if alive."""
         try:
-            response = self._send_request({"type": "ping"})
+            response = self._send_request(
+                ControlRequestHeader(ControlMessageType.PING).to_dict()
+            )
         except zmq.ZMQError:
             logger.debug("ZMQ execution endpoint is unavailable", exc_info=True)
             return False
         return response.get("type") == "pong" and response.get("ready")
 
     def get_server_info(self):
         """Get detailed server info including workers."""
-        return self._send_request({"type": "ping"})
+        return self._send_request(ControlRequestHeader(ControlMessageType.PING).to_dict())
 
     def get_status(self, execution_id=None):
         """Get execution status."""
-        request = {"type": "status"}
-        if execution_id:
-            request["execution_id"] = execution_id
-        return self._send_request(request)
+        return self._send_request(StatusRequest(execution_id=execution_id or None).to_dict())
 
     def execute_pipeline(self, plate_id, pipeline_code, config_code):
         """Execute pipeline on server."""
-        request = {
-            "type": "execute",
-            "plate_id": str(plate_id),
-            "pipeline_code": pipeline_code,
-            "config_code": config_code,
-        }
-        return self._send_request(request)
+        request = ExecuteRequest(
+            subject_id=str(plate_id),
+            pipeline_code=pipeline_code,
+            config_code=config_code,
+        )
+        return self._send_request(request.to_dict())
 
     def cancel(self, execution_id):
         """Cancel execution."""
-        request = {"type": "cancel", "execution_id": execution_id}
-        return self._send_request(request)
+        return self._send_request(CancelRequest(execution_id=execution_id).to_dict())
 
 
 def _get_zmq_client():

openhcs/processing/backends/lib_registry/registry_service.py

@@ -20,7 +20,7 @@
 from arraybridge import MemoryType
 from objectstate import LazyDataclassFactory
 from python_introspect import SignatureAnalyzer
-from pyqt_reactive.process_launch import BackgroundProcessLaunchPolicy
+from zmqruntime.process_launch import BackgroundProcessLaunchPolicy
 from zmqruntime import OperationCancellation
 from zmqruntime.client import endpoint_process
 

openhcs/runtime/viewer_protocol.py

@@ -31,7 +31,7 @@
 from pydantic import StrictInt
 from polystore.backend_registry import register_cleanup_callback
 from polystore.streaming_constants import StreamingDataType
-from pyqt_reactive.process_launch import BackgroundProcessLaunchPolicy
+from zmqruntime.process_launch import BackgroundProcessLaunchPolicy
 from zmqruntime.client import EndpointProcessGroup, endpoint_process
 from zmqruntime.config import NonBlankString, TransportMode, ZMQConfig
 from zmqruntime.messages import (

openhcs/runtime/viewer_snapshot.py

@@ -38,7 +38,15 @@ def from_control_error(
         )
 
     def to_dict(self) -> dict[str, object]:
-        return {**super().to_dict(), "observation": self.observation}
+        """The control error's wire mapping, carrying the native observation as is."""
+
+        error = ControlErrorResponse(
+            **{
+                declared.name: getattr(self, declared.name)
+                for declared in fields(ControlErrorResponse)
+            }
+        )
+        return {**error.to_dict(), "observation": self.observation}
 
 
 class ViewerWindowSnapshotService(QtWindowSnapshotService):