diff --git a/docs/refactor/consolidation-20261009/L3-zmqruntime-vocabulary.md b/docs/refactor/consolidation-20261009/L3-zmqruntime-vocabulary.md new file mode 100644 index 0000000000..5aef972911 --- /dev/null +++ b/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. diff --git a/external/PolyStore b/external/PolyStore index 473c3dafed..7ac9d01110 160000 --- a/external/PolyStore +++ b/external/PolyStore @@ -1 +1 @@ -Subproject commit 473c3dafedff695afd7ba1c368cdcd57ba2d0f66 +Subproject commit 7ac9d011100b18f4ad1a54635ee94f52a47efb30 diff --git a/external/pyqt-reactive b/external/pyqt-reactive index 291b1be4ae..269e896ba8 160000 --- a/external/pyqt-reactive +++ b/external/pyqt-reactive @@ -1 +1 @@ -Subproject commit 291b1be4aee9e1e9ea4d6096321a7ad54dbaa7b9 +Subproject commit 269e896ba8ec3050ae305e4ac9c46a1c01c2cb9b diff --git a/external/zmqruntime b/external/zmqruntime index 079a1c41d2..a1db4a5439 160000 --- a/external/zmqruntime +++ b/external/zmqruntime @@ -1 +1 @@ -Subproject commit 079a1c41d28c02adce84a3852e4baab817ea4f27 +Subproject commit a1db4a5439caceaddbc5d89fa43300dec09aad35 diff --git a/openhcs/core/progress/projection.py b/openhcs/core/progress/projection.py index ddc0acc960..ad584c022d 100644 --- a/openhcs/core/progress/projection.py +++ b/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: diff --git a/openhcs/core/progress/types.py b/openhcs/core/progress/types.py index 305931fbe5..52f6f7d012 100644 --- a/openhcs/core/progress/types.py +++ b/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 diff --git a/openhcs/desktop/restart.py b/openhcs/desktop/restart.py index 4211f2a62c..7a2a870c21 100644 --- a/openhcs/desktop/restart.py +++ b/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 ( diff --git a/openhcs/desktop/update.py b/openhcs/desktop/update.py index acb46b922d..cbcc0afee1 100644 --- a/openhcs/desktop/update.py +++ b/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 ( diff --git a/openhcs/gui_startup.py b/openhcs/gui_startup.py index 62e46f86b5..604668c452 100644 --- a/openhcs/gui_startup.py +++ b/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" diff --git a/openhcs/omero/plugin/omero_openhcs/views.py b/openhcs/omero/plugin/omero_openhcs/views.py index 9e5b07afcb..98211eac7b 100644 --- a/openhcs/omero/plugin/omero_openhcs/views.py +++ b/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,7 +60,9 @@ 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 @@ -59,29 +70,24 @@ def ping(self): 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(): diff --git a/openhcs/processing/backends/lib_registry/registry_service.py b/openhcs/processing/backends/lib_registry/registry_service.py index 3e446429de..b706eff7f5 100644 --- a/openhcs/processing/backends/lib_registry/registry_service.py +++ b/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 diff --git a/openhcs/runtime/viewer_protocol.py b/openhcs/runtime/viewer_protocol.py index 923f9143b8..28d3f7daa0 100644 --- a/openhcs/runtime/viewer_protocol.py +++ b/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 ( diff --git a/openhcs/runtime/viewer_snapshot.py b/openhcs/runtime/viewer_snapshot.py index f7f56342e8..b5a4c48fd4 100644 --- a/openhcs/runtime/viewer_snapshot.py +++ b/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): diff --git a/openhcs/runtime/zmq_config.py b/openhcs/runtime/zmq_config.py index 3285aeefb0..4c93c6a696 100644 --- a/openhcs/runtime/zmq_config.py +++ b/openhcs/runtime/zmq_config.py @@ -4,19 +4,13 @@ from dataclasses import dataclass -from zmqruntime.config import ( - ZMQConfig, - NonBlankString, - PositiveFloat, - PositiveInteger, - TransportMode, -) -from zmqruntime.transport import TransportEndpoint, get_default_transport_mode +from zmqruntime.config import NonBlankString +from zmqruntime.execution.config import ExecutionTransportConfig @dataclass(frozen=True, slots=True) -class OpenHCSZMQConfig(ZMQConfig): - """Process-level execution transport topology and lifecycle defaults.""" +class OpenHCSZMQConfig(ExecutionTransportConfig): + """zmqruntime's execution transport under OpenHCS's application names.""" ipc_socket_prefix: NonBlankString = "openhcs-zmq" """OpenHCS namespace prefix for generated IPC data and control sockets.""" @@ -24,96 +18,5 @@ class OpenHCSZMQConfig(ZMQConfig): app_name: NonBlankString = "openhcs" """OpenHCS application namespace used in transport identities and runtime paths.""" - client_host: NonBlankString = "localhost" - """Execution-server hostname used by clients in TCP mode. - - IPC mode is local and ignores this network hostname. - """ - - server_host: NonBlankString = "*" - """Network interface bound by execution servers in TCP mode. - - ``*`` accepts connections on every interface; use a loopback address to - restrict the server to this machine. IPC mode ignores this field. - """ - - transport_mode: TransportMode = get_default_transport_mode() - """Transport used for execution data and control sockets. - - IPC is local-only and owns filesystem socket cleanup. TCP uses - ``client_host`` and ``server_host`` and can cross machine boundaries. - """ - - persistent: bool = True - """Reuse an execution client's open sockets across successive operations. - - Disable this when each request should create and close its own connection; - it does not control execution-server lifetime. - """ - - control_timeout_ms: PositiveInteger = 5000 - """Default deadline for short execution control requests such as status and stop. - - Individual agent or client requests may supply a narrower or wider timeout. - """ - - execution_submission_timeout_ms: PositiveInteger = 30000 - """Total deadline for preparing and sending one execution submission. - - This spans endpoint connection, progress registration, pipeline and configuration - serialisation, and the execute request. It does not limit the submitted execution. - """ - - server_info_timeout_ms: PositiveInteger = 500 - """Deadline for retrieving the typed descriptor of one discovered server endpoint.""" - - server_scan_timeout_ms: PositiveInteger = 200 - """Per-endpoint heartbeat deadline used while scanning for execution servers. - - Larger values tolerate slow hosts but multiply the duration of a full port - scan. - """ - - client_connect_timeout_seconds: PositiveFloat = 15.0 - """Inactivity deadline while a client waits for an execution endpoint to become ready. - - Reported startup activity resets this deadline, allowing slow registry - preparation to continue while genuinely stalled launches still fail. - """ - - server_poll_interval_seconds: PositiveFloat = 0.01 - """Delay between endpoint readiness probes during execution-server startup.""" - - ports_per_server_type: PositiveInteger = 10 - """Consecutive candidate data ports reserved and scanned for each server role. - - Increasing this permits more concurrent role instances but makes discovery - inspect more endpoints. - """ - - compiled_artifact_ttl_seconds: PositiveFloat = 30.0 * 60.0 - """Retention time for server-side compiled-artifact inspection records. - - Expired records are discarded from the execution server's inspection cache; - this does not delete pipeline outputs or compiled bundles saved to disk. - """ - - def client_endpoint( - self, - port: int | None = None, - *, - host: str | None = None, - transport_mode: TransportMode | None = None, - ) -> TransportEndpoint: - """Return the exact execution endpoint declared for an OpenHCS client.""" - - return TransportEndpoint( - host=self.client_host if host is None else host, - port=self.default_port if port is None else port, - transport_mode=TransportMode.resolve( - self.transport_mode if transport_mode is None else transport_mode - ), - ) - OPENHCS_ZMQ_CONFIG = OpenHCSZMQConfig() diff --git a/openhcs/runtime/zmq_execution_client.py b/openhcs/runtime/zmq_execution_client.py index 1d1795f6ec..cea410626f 100644 --- a/openhcs/runtime/zmq_execution_client.py +++ b/openhcs/runtime/zmq_execution_client.py @@ -17,7 +17,6 @@ from typing import TYPE_CHECKING, Any, TypeAlias from arraybridge import MemoryType -from pyqt_reactive.process_launch import BackgroundProcessLaunchPolicy from typing_extensions import override from zmqruntime import ( EndpointApplicationCompatibility, @@ -40,11 +39,13 @@ ControlMessageType, ControlRequestHeader, EndpointApplicationCompatibilityError, + ExecuteRequest, ExecutionRecord, - MessageFields, PongResponse, ServerRole, + StatusRequest, ) +from zmqruntime.process_launch import BackgroundProcessLaunchPolicy from zmqruntime.startup import ( EndpointStartupObserver, EndpointStartupPhase, @@ -556,43 +557,20 @@ def request_payload(self) -> ZMQExecutionRequestPayload: compile_control=self.compile_control, ) - def request(self) -> "ZMQRequest": - return ZMQRequest.from_items( - ( - (MessageFields.TYPE, ControlMessageType.EXECUTE.value), - (MessageFields.PIPELINE_CODE, self.pipeline_code), - *self.identity.request_items(), - *self.compile_control.request_items(), - *self.config_projection.request_items(), - ) + def request(self) -> ExecuteRequest: + source_fields = self.config_projection.source_fields + return ExecuteRequest( + subject_id=self.identity.plate_id, + execution_subject_id=self.identity.execution_plate_id, + selected_pipeline_path=self.identity.selected_pipeline_path, + pipeline_code=self.pipeline_code, + config_params=self.config_params, + config_code=None if source_fields is None else source_fields.config_code, + compile_only=self.compile_control.compile_only, + compile_artifact_id=self.compile_control.compile_artifact_id, ) -@dataclass(frozen=True, slots=True) -class ZMQRequest: - """Nominal OpenHCS execution request before lowering to zmqruntime.""" - - values: Mapping[str, ZMQValue] - - @classmethod - def from_items( - cls, - items: Sequence[tuple[str, ZMQValue]], - ) -> "ZMQRequest": - return cls(values=dict(items)) - - def with_items( - self, - items: Sequence[tuple[str, ZMQValue]], - ) -> "ZMQRequest": - values = dict(self.values) - values.update(items) - return ZMQRequest(values=values) - - def as_wire_payload(self) -> dict[str, ZMQValue]: - return dict(self.values) - - @dataclass(frozen=True, slots=True) class ZMQConfigParamsBoundary: """Boundary for zmqruntime's auxiliary params field, not config authority.""" @@ -619,20 +597,11 @@ def with_updates( merged_params.update(updates) return ZMQConfigParamsBoundary(params=merged_params) - def request_items(self) -> tuple[tuple[str, ZMQValue], ...]: - return ((MessageFields.CONFIG_PARAMS, self.params),) - @dataclass(frozen=True, slots=True) class ZMQConfigSourceFields: config_code: str | None = None - def request_items(self) -> tuple[tuple[str, ZMQValue], ...]: - items: list[tuple[str, ZMQValue]] = [] - if self.config_code is not None: - items.append((MessageFields.CONFIG_CODE, self.config_code)) - return tuple(items) - @dataclass(frozen=True, slots=True) class ZMQConfigProjection: @@ -656,14 +625,6 @@ def from_task( config_sha=hashlib.sha256(config_source.encode("utf-8")).hexdigest()[:12], ) - def request_items(self) -> tuple[tuple[str, ZMQValue], ...]: - items: list[tuple[str, ZMQValue]] = [] - if self.params_boundary is not None: - items.extend(self.params_boundary.request_items()) - if self.source_fields is not None: - items.extend(self.source_fields.request_items()) - return tuple(items) - def signature_transport(self) -> ZMQExecutionConfigTransport: source_fields = self.source_fields return ZMQExecutionConfigTransport( @@ -1050,12 +1011,12 @@ def serialize_task( self, task: OpenHCSExecutionSubmission, config=None, - ) -> dict[str, ZMQValue]: + ) -> ExecuteRequest: return self.serialize_prepared_request(ZMQExecutionRequestBuilder.from_task(task)) def serialize_prepared_request( self, request_builder: ZMQExecutionRequestBuilder - ) -> dict[str, ZMQValue]: + ) -> ExecuteRequest: request = request_builder.request() request_payload = request_builder.request_payload logger.info( @@ -1066,7 +1027,7 @@ def serialize_prepared_request( request_payload.pipeline_sha, request_builder.config_projection.config_sha, ) - return request.as_wire_payload() + return request def submit_pipeline( self, @@ -1135,11 +1096,8 @@ def get_status( *, timeout_ms: int | None = None, ): - request = {MessageFields.TYPE: ControlMessageType.STATUS.value} - if execution_id: - request[MessageFields.EXECUTION_ID] = execution_id return self._send_control_request( - request, + StatusRequest(execution_id=execution_id or None).to_dict(), timeout_ms=self._control_timeout_ms(timeout_ms), ) @@ -1171,7 +1129,7 @@ def _submission_timeout_ms(self, timeout_ms: int | None) -> int: def _submit_submission( self, - build_request: Callable[[], dict[str, ZMQValue]], + build_request: Callable[[], ExecuteRequest], *, timeout_ms: int, ): @@ -1192,9 +1150,7 @@ def _submit_submission( self._ensure_progress_subscription( timeout_ms=deadline.remaining_milliseconds() ) - request = build_request() - if MessageFields.TYPE not in request: - request[MessageFields.TYPE] = ControlMessageType.EXECUTE.value + request = build_request().to_dict() request_timeout_ms = deadline.remaining_milliseconds() except TimeoutError as exc: raise ExecutionSubmissionPreparationTimeoutError( diff --git a/openhcs/runtime/zmq_execution_server.py b/openhcs/runtime/zmq_execution_server.py index c9335ddd80..08d167b716 100644 --- a/openhcs/runtime/zmq_execution_server.py +++ b/openhcs/runtime/zmq_execution_server.py @@ -8,7 +8,7 @@ import sys import time from collections.abc import Sequence -from dataclasses import dataclass +from dataclasses import dataclass, replace from types import ModuleType from polystore import cleanup_backend_connections @@ -31,7 +31,7 @@ from openhcs.core.orchestrator.cancellation import ExecutionCancelledError from openhcs.core.orchestrator.worker_execution import PreparedForkWorkerLaneRunner from openhcs.core.pipeline_document import PipelineDocumentCodec -from openhcs.core.progress import ProgressEvent +from openhcs.core.progress import ProgressEvent, ProgressExecutionContext from openhcs.core.steps.function_step import FunctionStep from openhcs.runtime.environment_provenance import RuntimeEnvironmentSnapshot from openhcs.runtime.zmq_application import OPENHCS_ENDPOINT_APPLICATION @@ -54,7 +54,6 @@ ) from openhcs.runtime.zmq_progress import ZMQCompilerProgressQueue, ZMQProgressEmitter from openhcs.runtime.zmq_server_hooks import ( - ZMQPongResponseEnricher, ZMQResultsSummaryEnricher, ZMQWorkerCleanup, ) @@ -124,8 +123,9 @@ def validate_compile_request(self) -> None: def progress_context(self) -> dict: return { - MessageFields.EXECUTION_ID: self.execution_id, - MessageFields.PLATE_ID: self.plate_id, + **ProgressExecutionContext( + execution_id=self.execution_id, plate_id=self.plate_id + ).to_transport_fields(), MessageFields.AXIS_ID: "", } @@ -303,10 +303,12 @@ def _cleanup_compiled_artifacts(self) -> None: def _create_pong_response(self): self._cleanup_compiled_artifacts() - return ZMQPongResponseEnricher( - active_executions=self.active_executions, - compile_status=self._get_compile_status, - ).enrich(super()._create_pong_response()) + compile_status, compile_message = self._get_compile_status() + return replace( + super()._create_pong_response(), + compile_status=compile_status, + compile_message=compile_message, + ) def _enqueue_progress(self, progress_update: dict) -> None: event = ProgressEvent.from_dict(progress_update) diff --git a/openhcs/runtime/zmq_execution_signature.py b/openhcs/runtime/zmq_execution_signature.py index db76c61f2f..183e77997f 100644 --- a/openhcs/runtime/zmq_execution_signature.py +++ b/openhcs/runtime/zmq_execution_signature.py @@ -29,8 +29,6 @@ | tuple["TransportValue", ...] ) TransportRequestItems = tuple[tuple[str, TransportValue], ...] -EXECUTION_PLATE_ID_FIELD = "execution_plate_id" -SELECTED_PIPELINE_PATH_FIELD = "selected_pipeline_path" class ZMQAuxiliaryParamField(Enum): @@ -218,21 +216,11 @@ class ZMQExecutionIdentity: execution_plate_id: str | None = None selected_pipeline_path: str | None = None - def request_items(self) -> TransportRequestItems: - items: list[tuple[str, TransportValue]] = [ - (MessageFields.PLATE_ID, self.plate_id), - ] - if self.execution_plate_id is not None: - items.append((EXECUTION_PLATE_ID_FIELD, self.execution_plate_id)) - if self.selected_pipeline_path is not None: - items.append((SELECTED_PIPELINE_PATH_FIELD, self.selected_pipeline_path)) - return tuple(items) - def signature_items(self) -> TransportRequestItems: return ( - (MessageFields.PLATE_ID, self.plate_id), - (EXECUTION_PLATE_ID_FIELD, self.execution_plate_id), - (SELECTED_PIPELINE_PATH_FIELD, self.selected_pipeline_path), + (MessageFields.SUBJECT_ID, self.plate_id), + (MessageFields.EXECUTION_SUBJECT_ID, self.execution_plate_id), + (MessageFields.SELECTED_PIPELINE_PATH, self.selected_pipeline_path), ) @@ -270,14 +258,6 @@ def validate(self) -> None: if self.compile_only and self.compile_artifact_id: raise ValueError("compile_only and compile_artifact_id cannot both be set") - def request_items(self) -> TransportRequestItems: - items: list[tuple[str, TransportValue]] = [] - if self.compile_only: - items.append((MessageFields.COMPILE_ONLY, True)) - if self.compile_artifact_id is not None: - items.append((MessageFields.COMPILE_ARTIFACT_ID, self.compile_artifact_id)) - return tuple(items) - @dataclass(frozen=True, slots=True) class ZMQExecutionConfigTransport: @@ -320,8 +300,8 @@ def from_execute_request( ) -> "ZMQExecutionRequestPayload": return cls( identity=ZMQExecutionIdentity( - plate_id=request.plate_id, - execution_plate_id=request.execution_plate_id, + plate_id=request.subject_id, + execution_plate_id=request.execution_subject_id, selected_pipeline_path=request.selected_pipeline_path, ), pipeline_code=request.pipeline_code, diff --git a/openhcs/runtime/zmq_progress.py b/openhcs/runtime/zmq_progress.py index c7b22ceb7d..507143daa8 100644 --- a/openhcs/runtime/zmq_progress.py +++ b/openhcs/runtime/zmq_progress.py @@ -34,7 +34,7 @@ class ZMQCompilerProgressQueue(ZMQProgressTarget, ProgressQueue): def put(self, progress_update: dict) -> None: canonical_update = dict(progress_update) - canonical_update[MessageFields.PLATE_ID] = self.plate_id + canonical_update[MessageFields.SUBJECT_ID] = self.plate_id self.enqueue(canonical_update) diff --git a/openhcs/runtime/zmq_server_hooks.py b/openhcs/runtime/zmq_server_hooks.py index fffc87aff0..b9903c29e6 100644 --- a/openhcs/runtime/zmq_server_hooks.py +++ b/openhcs/runtime/zmq_server_hooks.py @@ -3,14 +3,12 @@ from __future__ import annotations import logging -from dataclasses import dataclass, replace -from typing import Any, Callable +from dataclasses import dataclass +from typing import Any from zmqruntime.messages import ( ExecutionStatus, MessageFields, - PongResponse, - QueuedExecutionInfo, ResponseType, ) @@ -24,37 +22,6 @@ logger = logging.getLogger(__name__) -@dataclass(frozen=True, slots=True) -class ZMQPongResponseEnricher: - """Adds OpenHCS queue and compile-status data to base pong responses.""" - - active_executions: dict[str, Any] - compile_status: Callable[[], tuple[str | None, str | None]] - - def enrich(self, response: PongResponse) -> PongResponse: - compile_status, compile_message = self.compile_status() - - queued = [ - (execution_id, record) - for execution_id, record in self.active_executions.items() - if record.status == ExecutionStatus.QUEUED.value - ] - queued_executions = tuple( - QueuedExecutionInfo( - execution_id=execution_id, - plate_id=str(record.plate_id), - queue_position=index + 1, - ) - for index, (execution_id, record) in enumerate(queued) - ) - return replace( - response, - compile_status=compile_status, - compile_message=compile_message, - queued_executions=queued_executions, - ) - - @dataclass(frozen=True, slots=True) class ZMQResultsSummaryEnricher: """Projects OpenHCS output-plate metadata into execution summaries.""" diff --git a/pyproject.toml b/pyproject.toml index 3c75da728e..d5fb1c3106 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -99,14 +99,14 @@ dependencies = [ "psutil>=5.9.0", # External dependencies (PyPI versions for production builds) - "zmqruntime>=0.5.0,<0.6", + "zmqruntime>=0.6,<0.7", "pycodify>=0.1.3,<0.2", "objectstate>=1.3.1,<2", "python-introspect>=0.2.5,<0.3", "metaclass-registry>=0.3.0,<0.4", "arraybridge>=0.3.8,<0.4", - "polystore>=0.5.0,<0.6", - "pyqt-reactive>=0.3.30,<0.4", + "polystore>=0.6,<0.7", + "pyqt-reactive>=0.4,<0.5", ] [project.optional-dependencies] diff --git a/tests/integration/test_compiled_artifact_inspection.py b/tests/integration/test_compiled_artifact_inspection.py index 2b5d15eb7a..1645175621 100644 --- a/tests/integration/test_compiled_artifact_inspection.py +++ b/tests/integration/test_compiled_artifact_inspection.py @@ -341,7 +341,7 @@ def apply_runtime_notification(record) -> None: # The server publishes progress only for executions it admitted. server.active_executions["run-1"] = ExecutionRecord( execution_id="run-1", - plate_id="/plates/one", + subject_id="/plates/one", client_address=None, status="running", ) diff --git a/tests/integration/test_source_store_code_mode_zmq.py b/tests/integration/test_source_store_code_mode_zmq.py index 87ab15bc8f..d1f0d69f71 100644 --- a/tests/integration/test_source_store_code_mode_zmq.py +++ b/tests/integration/test_source_store_code_mode_zmq.py @@ -15,7 +15,6 @@ ExecutionSubmissionResponse, ExecutionWaitResult, ) -from zmqruntime.messages import MessageFields from openhcs.constants.constants import Backend from openhcs.constants.input_source import InputSource @@ -251,12 +250,12 @@ def test_code_mode_and_zmq_wire_preserve_mixed_store_sources( observation_path=tmp_path / "non_live_observation.pkl", ) wire = ZMQExecutionClient().serialize_task(submission.compile_request()) - wire_pipeline = _exec_pipeline_source(wire[MessageFields.PIPELINE_CODE]) + wire_pipeline = _exec_pipeline_source(wire.pipeline_code) wire_pipeline_document = PipelineDocumentCodec.from_namespace(wire_pipeline) - wire_global_config = _exec_config_source(wire[MessageFields.CONFIG_CODE]) + wire_global_config = _exec_config_source(wire.config_code) - assert wire[MessageFields.COMPILE_ONLY] is True - assert MessageFields.PIPELINE_CONFIG_CODE not in wire + assert wire.compile_only is True + assert wire.pipeline_config_code is None assert ( FunctionStepTransportAuthority.source_from_pipeline( wire_pipeline["pipeline_steps"] diff --git a/tests/unit/agent/test_agent_services.py b/tests/unit/agent/test_agent_services.py index 26b1009a04..8d774f2de9 100644 --- a/tests/unit/agent/test_agent_services.py +++ b/tests/unit/agent/test_agent_services.py @@ -432,7 +432,7 @@ def server_info(self, connection, *, timeout_ms: int): running_executions=( RunningExecutionInfo( execution_id=_ExecutionTestId.EXECUTE, - plate_id="plate-1", + subject_id="plate-1", start_time=0.0, elapsed=0.0, ), diff --git a/tests/unit/agent/test_cold_inspection_feedback.py b/tests/unit/agent/test_cold_inspection_feedback.py index 15c9bd66f9..275c79e3c3 100644 --- a/tests/unit/agent/test_cold_inspection_feedback.py +++ b/tests/unit/agent/test_cold_inspection_feedback.py @@ -49,7 +49,7 @@ def test_original_progress_codec_is_the_only_event_store_and_rejects_bad_enum(): invalid["phase"] = "not-a-declared-phase" with pytest.raises(ValueError, match="Invalid phase"): queue.put(invalid) - with pytest.raises(KeyError): + with pytest.raises(ValueError, match="missing required field"): queue.put({"phase": "compile", "status": "running"}) assert queue.events == [event] assert len(statuses) == 1 diff --git a/tests/unit/agent/test_mcp_dev_client_viewer_renderers.py b/tests/unit/agent/test_mcp_dev_client_viewer_renderers.py index b52e1c58ce..7b8469ba12 100644 --- a/tests/unit/agent/test_mcp_dev_client_viewer_renderers.py +++ b/tests/unit/agent/test_mcp_dev_client_viewer_renderers.py @@ -445,7 +445,7 @@ def test_runtime_scan_info_and_status_render_server_facts(): execution_id="run-1", status="ok", response={"status": "ok", "active_executions": 1, "uptime": 12.34, "executions": ["run-1", "run-2"], "queued_executions": [], - "running_executions": [{"execution_id": "run-1", "plate_id": "p", + "running_executions": [{"execution_id": "run-1", "subject_id": "p", "start_time": 1.0, "elapsed": 2.0}]}, ) rendered = RuntimeExecutionStatusRenderer.render( diff --git a/tests/unit/desktop/test_desktop_update.py b/tests/unit/desktop/test_desktop_update.py index 01cedd463a..c9549fe798 100644 --- a/tests/unit/desktop/test_desktop_update.py +++ b/tests/unit/desktop/test_desktop_update.py @@ -10,7 +10,10 @@ from packaging.version import Version from PyQt6.QtNetwork import QNetworkReply, QNetworkRequest from PyQt6.QtWidgets import QMessageBox, QWidget -from pyqt_reactive.process_launch import BackgroundProcessPlatform +from zmqruntime.process_launch import ( + BackgroundProcessPlatform, + WindowsBackgroundProcesses, +) from pyqt_reactive.services.scope_token_service import ScopeTokenService from pyqt_reactive.theming import ColorScheme @@ -671,7 +674,7 @@ def test_windows_update_worker_uses_windowed_interpreter_and_no_console( monkeypatch.setattr( BackgroundProcessPlatform, "current", - classmethod(lambda cls: cls.WINDOWS), + classmethod(lambda cls: WindowsBackgroundProcesses), ) monkeypatch.setattr( DesktopRuntimeEnvironment, diff --git a/tests/unit/progress/test_projection.py b/tests/unit/progress/test_projection.py index 2e88550c00..23749ad7d2 100644 --- a/tests/unit/progress/test_projection.py +++ b/tests/unit/progress/test_projection.py @@ -229,14 +229,14 @@ def test_server_lifecycle_entries_reconcile_by_execution_and_plate_identity(): queued_executions=( QueuedExecutionInfo( execution_id="exec-next", - plate_id="/tmp/plate", + subject_id="/tmp/plate", queue_position=4, ), ), running_executions=( RunningExecutionInfo( execution_id="exec-running", - plate_id="/tmp/other", + subject_id="/tmp/other", start_time=2.0, elapsed=0.5, compile_only=True, @@ -260,14 +260,14 @@ def test_running_entry_only_supersedes_matching_queued_identity(): queued_executions=( QueuedExecutionInfo( execution_id="shared-exec", - plate_id="/tmp/queued-plate", + subject_id="/tmp/queued-plate", queue_position=2, ), ), running_executions=( RunningExecutionInfo( execution_id="shared-exec", - plate_id="/tmp/running-plate", + subject_id="/tmp/running-plate", start_time=2.0, elapsed=0.5, ), @@ -285,7 +285,7 @@ def test_running_entry_only_supersedes_matching_queued_identity(): def test_server_snapshot_refreshes_queue_position_for_same_identity(): queued_entry = QueuedExecutionInfo( execution_id="exec-queued", - plate_id="/tmp/plate", + subject_id="/tmp/plate", queue_position=4, ) projection = build_execution_runtime_projection( @@ -317,7 +317,7 @@ def test_running_snapshot_advances_compiled_identity_into_execution(): running_executions=( RunningExecutionInfo( execution_id="exec-shared", - plate_id="/tmp/plate", + subject_id="/tmp/plate", start_time=2.0, elapsed=0.5, ), @@ -342,7 +342,7 @@ def test_lagging_server_snapshot_does_not_downgrade_terminal_event_state(): running_executions=( RunningExecutionInfo( execution_id="exec-terminal", - plate_id="/tmp/plate", + subject_id="/tmp/plate", start_time=0.5, elapsed=0.5, ), diff --git a/tests/unit/pyqt_gui/test_logging_config.py b/tests/unit/pyqt_gui/test_logging_config.py index 6b9c4e137b..b7425ae620 100644 --- a/tests/unit/pyqt_gui/test_logging_config.py +++ b/tests/unit/pyqt_gui/test_logging_config.py @@ -190,7 +190,7 @@ def scan_ports(scan_service, ports): running_executions=( RunningExecutionInfo( execution_id=execution_id, - plate_id="plate-1", + subject_id="plate-1", start_time=1.0, elapsed=2.0, ), diff --git a/tests/unit/pyqt_gui/test_progress_tree_aggregation.py b/tests/unit/pyqt_gui/test_progress_tree_aggregation.py index 286556e2f3..0612dabfaa 100644 --- a/tests/unit/pyqt_gui/test_progress_tree_aggregation.py +++ b/tests/unit/pyqt_gui/test_progress_tree_aggregation.py @@ -260,12 +260,12 @@ def test_queued_plates_are_projected_without_progress_events(): queued=[ { "execution_id": "exec-123", - "plate_id": "/tmp/plate_a", + "subject_id": "/tmp/plate_a", "queue_position": 1, }, { "execution_id": "exec-456", - "plate_id": "/tmp/plate_b", + "subject_id": "/tmp/plate_b", "queue_position": 2, }, ] @@ -293,7 +293,7 @@ def test_new_queued_execution_replaces_old_compiled_identity_for_same_plate(): queued=[ { "execution_id": "exec-run", - "plate_id": "/tmp/plate_a", + "subject_id": "/tmp/plate_a", "queue_position": 2, } ] @@ -313,7 +313,9 @@ def test_running_plate_is_projected_without_progress_events(): running=[ { "execution_id": "exec-run", - "plate_id": "/tmp/plate_a", + "subject_id": "/tmp/plate_a", + "start_time": 0.0, + "elapsed": 0.0, "compile_only": True, } ] @@ -328,7 +330,14 @@ def test_running_snapshot_advances_init_only_projection_to_executing(): nodes = _nodes( {"exec-run": [_init_event(execution_id="exec-run")]}, _execution_server_info( - running=[{"execution_id": "exec-run", "plate_id": "/tmp/plate"}] + running=[ + { + "execution_id": "exec-run", + "subject_id": "/tmp/plate", + "start_time": 0.0, + "elapsed": 0.0, + } + ] ), ) @@ -410,7 +419,14 @@ def get_events(self, execution_id): tree = QTreeWidget() server_info = _execution_server_info( - running=[{"execution_id": "exec-1", "plate_id": "/tmp/plate"}] + running=[ + { + "execution_id": "exec-1", + "subject_id": "/tmp/plate", + "start_time": 0.0, + "elapsed": 0.0, + } + ] ) server_item = QTreeWidgetItem(["Execution Server", "", ""]) server_item.setData(0, Qt.ItemDataRole.UserRole, server_info) diff --git a/tests/unit/pyqt_gui/test_session_batch_workflows.py b/tests/unit/pyqt_gui/test_session_batch_workflows.py index 7d236de362..4c1924639f 100644 --- a/tests/unit/pyqt_gui/test_session_batch_workflows.py +++ b/tests/unit/pyqt_gui/test_session_batch_workflows.py @@ -640,7 +640,7 @@ def test_server_queue_snapshot_projects_queued_executions(monkeypatch): queued_execution_entries=( QueuedExecutionInfo( execution_id="exec-queued", - plate_id="plate-queued", + subject_id="plate-queued", queue_position=3, ), ), diff --git a/tests/unit/pyqt_gui/test_zmq_server_info.py b/tests/unit/pyqt_gui/test_zmq_server_info.py index 6c3a3221c0..8afe0a6115 100644 --- a/tests/unit/pyqt_gui/test_zmq_server_info.py +++ b/tests/unit/pyqt_gui/test_zmq_server_info.py @@ -28,12 +28,14 @@ def test_execution_server_payload_parses_to_typed_info(): "running_executions": [ { "execution_id": "exec-1", - "plate_id": "/tmp/p1", + "subject_id": "/tmp/p1", + "start_time": 1.0, + "elapsed": 2.0, "compile_only": True, }, ], "queued_executions": [ - {"execution_id": "exec-2", "plate_id": "/tmp/p2", "queue_position": 1}, + {"execution_id": "exec-2", "subject_id": "/tmp/p2", "queue_position": 1}, ], "compile_status": "compiled success", "compile_message": "ok", @@ -47,7 +49,7 @@ def test_execution_server_payload_parses_to_typed_info(): assert len(info.workers) == 1 assert info.running_executions == ("exec-1",) assert info.queued_executions == ("exec-2",) - assert info.running_execution_entries[0].plate_id == "/tmp/p1" + assert info.running_execution_entries[0].subject_id == "/tmp/p1" assert info.running_execution_entries[0].compile_only is True assert info.queued_execution_entries[0].queue_position == 1 assert info.response.compile_status == "compiled success" @@ -64,8 +66,7 @@ def test_viewer_server_payload_parses_to_typed_info(): "server_type": "napari", "server_role": "viewer", "log_file_path": "/tmp/napari.log", - "memory_mb": 1024.5, - "cpu_percent": 8.25, + "process_usage": {"memory_mb": 1024.5, "cpu_percent": 8.25}, } info = BaseServerInfo.from_response(PongResponse.from_dict(payload)) diff --git a/tests/unit/test_benchmark_control.py b/tests/unit/test_benchmark_control.py index 90ffad2845..72b1f62db7 100644 --- a/tests/unit/test_benchmark_control.py +++ b/tests/unit/test_benchmark_control.py @@ -549,7 +549,7 @@ def test_measured_finalization_refuses_missing_server_timing_without_receipt( request=ZMQExecutionRequestBuilder.from_task(submission), record=ExecutionRecord( execution_id="execution-1", - plate_id=str(tmp_path), + subject_id=str(tmp_path), client_address=None, status="complete", results_summary={}, diff --git a/tests/unit/test_benchmark_timing.py b/tests/unit/test_benchmark_timing.py index 2456962945..cfbc12a1fb 100644 --- a/tests/unit/test_benchmark_timing.py +++ b/tests/unit/test_benchmark_timing.py @@ -97,7 +97,7 @@ def test_phase_timing_rejects_invalid_duration(seconds: float) -> None: def test_completed_server_duration_uses_exact_ordinary_job() -> None: record = ExecutionRecord( execution_id="execution-1", - plate_id="plate", + subject_id="plate", client_address=None, status="complete", start_time=10.0, @@ -120,7 +120,7 @@ def test_completed_server_duration_rejects_invalid_bounds( ) -> None: record = ExecutionRecord( execution_id="execution-1", - plate_id="plate", + subject_id="plate", client_address=None, status="complete", start_time=start_time, diff --git a/tests/unit/test_function_step_transport.py b/tests/unit/test_function_step_transport.py index 60c0432661..f5e0095e10 100644 --- a/tests/unit/test_function_step_transport.py +++ b/tests/unit/test_function_step_transport.py @@ -230,14 +230,12 @@ def test_request_builder_stores_explicit_pipeline_code_directly() -> None: assert compile_submission.compile_only is True assert builder.pipeline_code == pipeline_source assert builder.request_payload.pipeline_code == pipeline_source - assert builder.request().values[MessageFields.PIPELINE_CODE] == pipeline_source + assert builder.request().pipeline_code == pipeline_source assert not hasattr(builder, "pipeline_transport") - assert builder.request().values[MessageFields.COMPILE_ONLY] is True + assert builder.request().compile_only is True assert artifact_builder.pipeline_code == pipeline_source - assert artifact_builder.request().values[MessageFields.COMPILE_ARTIFACT_ID] == ( - "compile-1" - ) - assert MessageFields.COMPILE_ONLY not in artifact_builder.request().values + assert artifact_builder.request().compile_artifact_id == "compile-1" + assert artifact_builder.request().compile_only is False with pytest.raises(ValueError, match="compile_artifact_id cannot be empty"): submission.with_compile_artifact_id("") @@ -277,13 +275,11 @@ def test_zmq_execution_submission_serializes_default_plate_config_on_client() -> global_config=GlobalPipelineConfig(), ) - payload = ZMQExecutionClient().serialize_task(submission) - document = PipelineDocumentCodec.from_source( - payload[MessageFields.PIPELINE_CODE] - ) + request = ZMQExecutionClient().serialize_task(submission) + document = PipelineDocumentCodec.from_source(request.pipeline_code) - assert MessageFields.CONFIG_CODE in payload - assert MessageFields.PIPELINE_CONFIG_CODE not in payload + assert request.config_code is not None + assert request.pipeline_config_code is None assert isinstance(document.pipeline_config, PipelineConfig) diff --git a/tests/unit/test_l3_zmqruntime_vocabulary_guards.py b/tests/unit/test_l3_zmqruntime_vocabulary_guards.py new file mode 100644 index 0000000000..cdec3ed7c1 --- /dev/null +++ b/tests/unit/test_l3_zmqruntime_vocabulary_guards.py @@ -0,0 +1,65 @@ +"""L3 guards: zmqruntime is domain-blind and owns its codec, launch policy and config.""" + +from __future__ import annotations + +import ast +import re +from dataclasses import fields +from pathlib import Path + +import zmqruntime.messages as messages +from zmqruntime.execution.config import ExecutionTransportConfig + +from openhcs.runtime.zmq_config import OpenHCSZMQConfig + +REPO = Path(__file__).resolve().parents[2] +ZMQRUNTIME_SRC = REPO / "external" / "zmqruntime" / "src" / "zmqruntime" +DOMAIN_WORD = re.compile(r"plate|\bwells?\b|well_", re.IGNORECASE) + + +def test_zmqruntime_names_no_microscopy_word() -> None: + offenders = [ + f"{path.relative_to(REPO)}:{number}" + for path in ZMQRUNTIME_SRC.rglob("*.py") + for number, line in enumerate(path.read_text().splitlines(), start=1) + if DOMAIN_WORD.search(line.replace("Template", "")) + ] + assert offenders == [] + + +def test_messages_inherit_their_codec_from_the_declared_bases() -> None: + bases = {"WireMessage", "WireView", "TypedWireMessage", "ControlRequestHeader"} + tree = ast.parse(Path(messages.__file__).read_text()) + offenders = [ + f"{node.name}.{member.name}" + for node in tree.body + if isinstance(node, ast.ClassDef) and node.name not in bases + for member in node.body + if isinstance(member, ast.FunctionDef) and member.name in {"to_dict", "from_dict"} + ] + assert offenders == [] + + +def test_launch_policy_lives_in_zmqruntime() -> None: + assert not ( + REPO / "external" / "pyqt-reactive" / "src" / "pyqt_reactive" / "process_launch.py" + ).exists() + importers = [ + str(path.relative_to(REPO)) + for root in ("openhcs", "tests") + for path in (REPO / root).rglob("*.py") + if "pyqt_reactive.process_launch" in path.read_text() + and path.name != Path(__file__).name + ] + assert importers == [] + + +def test_openhcs_transport_config_declares_only_its_application_names() -> None: + inherited = {declared.name for declared in fields(ExecutionTransportConfig)} + own = {declared.name for declared in fields(OpenHCSZMQConfig)} - inherited + assert own == set() + assert set(OpenHCSZMQConfig.__dataclass_fields__) == inherited + overridden = { + name for name in OpenHCSZMQConfig.__annotations__ if name in inherited + } + assert overridden == {"app_name", "ipc_socket_prefix"} diff --git a/tests/unit/test_matched_cellprofiler_batch.py b/tests/unit/test_matched_cellprofiler_batch.py index 2ffe2e73bf..c9c7a8bef2 100644 --- a/tests/unit/test_matched_cellprofiler_batch.py +++ b/tests/unit/test_matched_cellprofiler_batch.py @@ -597,7 +597,7 @@ def _axis_event(axis_id: str, phase: str, pid: int, timestamp: float) -> Progres return ProgressEvent.from_dict( { "execution_id": "job-1", - "plate_id": "plate-1", + "subject_id": "plate-1", "axis_id": axis_id, "step_name": "pipeline", "phase": phase, diff --git a/tests/unit/test_nonpersistent_viewer_state_evidence.py b/tests/unit/test_nonpersistent_viewer_state_evidence.py index fdc45c95e6..7a233ce8c8 100644 --- a/tests/unit/test_nonpersistent_viewer_state_evidence.py +++ b/tests/unit/test_nonpersistent_viewer_state_evidence.py @@ -79,7 +79,7 @@ def execute_compiled_plate(self, **kwargs): record = ExecutionRecord( execution_id="exec", - plate_id="plate", + subject_id="plate", client_address=None, status=ExecutionStatus.RUNNING.value, ) diff --git a/tests/unit/test_openhcs_adapter.py b/tests/unit/test_openhcs_adapter.py index 9f9ee31225..c95894cb84 100644 --- a/tests/unit/test_openhcs_adapter.py +++ b/tests/unit/test_openhcs_adapter.py @@ -337,7 +337,7 @@ def poll_status(self, execution_id): "status": "ok", "execution": { "execution_id": execution_id, - "plate_id": "/tmp/execution_plate", + "subject_id": "/tmp/execution_plate", "client_address": None, "status": "complete", "results_summary": {"output_plate_root": str(tmp_path)}, diff --git a/tests/unit/test_openhcs_measured_run.py b/tests/unit/test_openhcs_measured_run.py index df501b08e8..688d9e297c 100644 --- a/tests/unit/test_openhcs_measured_run.py +++ b/tests/unit/test_openhcs_measured_run.py @@ -105,7 +105,7 @@ def poll_status(self, execution_id): "status": "ok", "execution": { "execution_id": execution_id, - "plate_id": str(tmp_path), + "subject_id": str(tmp_path), "client_address": None, "status": "complete", "start_time": 10.0, @@ -262,7 +262,7 @@ def poll_status(self, execution_id): "status": "ok", "execution": { "execution_id": execution_id, - "plate_id": str(tmp_path), + "subject_id": str(tmp_path), "client_address": None, "status": "complete", "start_time": 10.0, diff --git a/tests/unit/test_well_throughput_scaling.py b/tests/unit/test_well_throughput_scaling.py index 9198f2f7b7..a060ce845e 100644 --- a/tests/unit/test_well_throughput_scaling.py +++ b/tests/unit/test_well_throughput_scaling.py @@ -1374,7 +1374,7 @@ def fake_execute( timing_observer( { "execution_id": "job-1", - "plate_id": "plate-1", + "subject_id": "plate-1", "axis_id": f"W{index:03d}", "step_name": "pipeline", "phase": phase, @@ -1464,7 +1464,7 @@ def test_well_throughput_rejects_two_worker_label_from_one_pid() -> None: events = [ { "execution_id": "job-1", - "plate_id": "plate-1", + "subject_id": "plate-1", "axis_id": axis_id, "step_name": "pipeline", "phase": phase, diff --git a/tests/unit/test_zmq_compiled_pipeline_run.py b/tests/unit/test_zmq_compiled_pipeline_run.py index d2403ec0cb..dcbca81bf7 100644 --- a/tests/unit/test_zmq_compiled_pipeline_run.py +++ b/tests/unit/test_zmq_compiled_pipeline_run.py @@ -56,7 +56,7 @@ def wait_for_completion(self, execution_id, poll_interval=0.5): "execution_id": execution_id, "execution": { "execution_id": execution_id, - "plate_id": "/tmp/plate", + "subject_id": "/tmp/plate", "status": self.compile_status if execution_id == "compile-1" else self.execute_status, "start_time": 10.0, "end_time": 12.0, diff --git a/tests/unit/test_zmq_execution_server.py b/tests/unit/test_zmq_execution_server.py index b2d85f9603..07e2c9fc87 100644 --- a/tests/unit/test_zmq_execution_server.py +++ b/tests/unit/test_zmq_execution_server.py @@ -104,7 +104,7 @@ def test_axis_outcomes_qualify_server_terminal_after_export(monkeypatch, failed) server = ZMQExecutionServer(port=5555) record = ExecutionRecord( execution_id="axis-outcome-check", - plate_id="/tmp/plate", + subject_id="/tmp/plate", client_address=None, status=ExecutionStatus.QUEUED.value, ) @@ -151,7 +151,7 @@ def test_axis_outcomes_qualify_server_terminal_after_export(monkeypatch, failed) monkeypatch.setattr(server, "_kill_worker_processes", lambda: 0) server.run_execution( record.execution_id, - ExecuteRequest(plate_id=record.plate_id, pipeline_code="pass"), + ExecuteRequest(subject_id=record.subject_id, pipeline_code="pass"), record, ) terminal = server.progress_queue.get_nowait()[MessageFields.EXECUTION] @@ -163,7 +163,7 @@ def test_axis_outcomes_qualify_server_terminal_after_export(monkeypatch, failed) if failed: assert "A02: error: producer lineage missing" in terminal[MessageFields.ERROR] else: - assert terminal[MessageFields.RESULTS_SUMMARY][MessageFields.WELLS] == [ + assert terminal[MessageFields.RESULTS_SUMMARY][MessageFields.PARTITION_VALUES] == [ "A01", "A02", ] @@ -180,7 +180,7 @@ def test_terminal_notification_contains_openhcs_summary_before_cleanup( monkeypatch.setattr(server, "_kill_worker_processes", lambda: 0) record = ExecutionRecord( execution_id="exec-finalized", - plate_id="/tmp/plate", + subject_id="/tmp/plate", client_address=None, status=ExecutionStatus.QUEUED.value, ) @@ -199,7 +199,7 @@ def test_terminal_notification_contains_openhcs_summary_before_cleanup( server.run_execution( record.execution_id, - ExecuteRequest(plate_id=record.plate_id, pipeline_code="pass"), + ExecuteRequest(subject_id=record.subject_id, pipeline_code="pass"), record, ) @@ -208,8 +208,8 @@ def test_terminal_notification_contains_openhcs_summary_before_cleanup( assert terminal[MessageFields.START_TIME] is not None assert terminal[MessageFields.END_TIME] is not None summary = terminal[MessageFields.RESULTS_SUMMARY] - assert summary[MessageFields.WELL_COUNT] == 1 - assert summary[MessageFields.WELLS] == ["W001"] + assert summary[MessageFields.PARTITION_COUNT] == 1 + assert summary[MessageFields.PARTITION_VALUES] == ["W001"] if invalid_output_metadata: assert "Failed to attach output_plate_root" in caplog.text assert "output_plate_root" not in summary @@ -295,7 +295,7 @@ def test_server_exports_outcomes_without_projecting_compiled_values( server._server_environment = None record = ExecutionRecord( execution_id="execution-1", - plate_id="plate-1", + subject_id="plate-1", client_address=None, status=ExecutionStatus.RUNNING.value, ) @@ -524,7 +524,7 @@ def test_zmq_server_admits_all_progress_through_generic_terminal_watermark() -> from openhcs.runtime.zmq_execution_client import ZMQExecutionClient server = ZMQExecutionServer() - record = ExecutionRecord("execution-1", "plate-1", None, ExecutionStatus.RUNNING.value) + record = ExecutionRecord("execution-1", "plate-1", ExecutionStatus.RUNNING.value) server.active_executions[record.execution_id] = record server._worker_assignments_by_execution = { "execution-1": {"worker_0": ["A01", "B01"]} @@ -532,7 +532,7 @@ def test_zmq_server_admits_all_progress_through_generic_terminal_watermark() -> worker_queue = SimpleQueue() progress_context = ProgressExecutionContext( execution_id=record.execution_id, - plate_id=record.plate_id, + plate_id=record.subject_id, ) parent = create_event( ProgressEventPayload( @@ -587,7 +587,7 @@ def test_zmq_server_records_the_compilation_output_plate_value_without_rebuildin server._worker_assignments_by_execution = {} record = ExecutionRecord( execution_id="execution-1", - plate_id="plate-1", + subject_id="plate-1", client_address=None, status=ExecutionStatus.QUEUED.value, ) @@ -603,7 +603,7 @@ def test_zmq_server_records_the_compilation_output_plate_value_without_rebuildin server._record_compilation_outputs(record.execution_id, compilation) - assert record.metadata == { + assert record.extras == { ExecutionOutputPlateSummary.EXECUTION_RECORD_KEY: output_plate } assert ( @@ -799,7 +799,7 @@ def test_compiled_request_keeps_config_scope_without_resolving_configuration(mon ) server.active_executions[context.execution_id] = ExecutionRecord( execution_id=context.execution_id, - plate_id=context.plate_id, + subject_id=context.plate_id, client_address=None, status=ExecutionStatus.RUNNING.value, ) diff --git a/tests/unit/test_zmq_progress.py b/tests/unit/test_zmq_progress.py index 16ee0d55bf..d9f2aa820c 100644 --- a/tests/unit/test_zmq_progress.py +++ b/tests/unit/test_zmq_progress.py @@ -11,7 +11,7 @@ def test_compiler_progress_queue_uses_request_plate_identity(): compiler_update = { "execution_id": "exec-1", - "plate_id": "/tmp/plate", + "subject_id": "/tmp/plate", "axis_id": "A01", "step_name": "compilation", "phase": ProgressPhase.COMPILE.value, @@ -23,8 +23,8 @@ def test_compiler_progress_queue_uses_request_plate_identity(): queue.put(compiler_update) - assert emitted[0]["plate_id"] == "/tmp/plate#openhcs-cppipe=Analysis.cppipe" - assert compiler_update["plate_id"] == "/tmp/plate" + assert emitted[0]["subject_id"] == "/tmp/plate#openhcs-cppipe=Analysis.cppipe" + assert compiler_update["subject_id"] == "/tmp/plate" def test_compile_failed_emits_failed_compile_events_for_axes(): @@ -71,7 +71,7 @@ def test_compile_heartbeat_emits_running_pipeline_compile_event(): assert len(emitted) == 1 assert emitted[0]["execution_id"] == "exec-1" - assert emitted[0]["plate_id"] == "/tmp/plate" + assert emitted[0]["subject_id"] == "/tmp/plate" assert emitted[0]["axis_id"] == "" assert emitted[0]["step_name"] == "pipeline" assert emitted[0]["phase"] == ProgressPhase.COMPILE.value diff --git a/tests/unit/test_zmq_submission_timeout.py b/tests/unit/test_zmq_submission_timeout.py index 53dc24eaa3..5f2cdaa16f 100644 --- a/tests/unit/test_zmq_submission_timeout.py +++ b/tests/unit/test_zmq_submission_timeout.py @@ -7,6 +7,7 @@ from zmqruntime.execution import ExecutionClient from zmqruntime.messages import ( ControlMessageType, + ExecuteRequest, MessageFields, PongResponse, ServerRole, @@ -50,7 +51,7 @@ def ensure_progress_subscription(self, *, timeout_ms: int): observed.append(("progress", timeout_ms)) def serialize_task(self, _task, _config=None): - return {} + return ExecuteRequest(subject_id="subject", pipeline_code="pass", config_code="") def send_control_request(self, request, *, timeout_ms: int): observed.append((request[MessageFields.TYPE], timeout_ms))