diff --git a/docs/source/architecture/execution_progress_stream.rst b/docs/source/architecture/execution_progress_stream.rst index c6f6318..51d4472 100644 --- a/docs/source/architecture/execution_progress_stream.rst +++ b/docs/source/architecture/execution_progress_stream.rst @@ -24,7 +24,7 @@ Responsibilities - start/stop listener thread safely - consume PUB/SUB messages in non-blocking loop -- validate payload shape via ``validate_progress_payload`` +- validate the generic progress fields via ``TaskProgress.from_dict`` - dispatch payloads to callback - isolate malformed-message or callback failures without terminating the stream - retain one immutable latest observation per execution on ``ExecutionClient`` diff --git a/pyproject.toml b/pyproject.toml index a5aaa5d..1ba5d25 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "zmqruntime" -version = "0.5.0" +version = "0.6.0" description = "Generic ZMQ-based distributed execution framework" authors = [{name = "Tristan Simas", email = "tristan.simas@mail.mcgill.ca"}] license = {text = "MIT"} @@ -31,7 +31,7 @@ dependencies = [ "portalocker>=2.8.2", "pyzmq>=22.0", "psutil>=5.9", - "python-introspect>=0.1.12", + "python-introspect>=0.2.5,<0.3", "typing-extensions>=4.0.0", ] diff --git a/src/zmqruntime/__init__.py b/src/zmqruntime/__init__.py index f1db076..6084745 100644 --- a/src/zmqruntime/__init__.py +++ b/src/zmqruntime/__init__.py @@ -9,277 +9,184 @@ from __future__ import annotations from importlib.metadata import version as _distribution_version -from typing import TYPE_CHECKING - -__version__ = _distribution_version("zmqruntime") - -_LAZY_EXPORTS: dict[str, str] = { - "GlobalAckListener": "zmqruntime.ack_listener", - "EndpointConnectionCancelledError": "zmqruntime.client", - "EndpointConnectionPolicy": "zmqruntime.client", - "EndpointProcessGroup": "zmqruntime.client", - "EndpointShutdownMode": "zmqruntime.client", - "EndpointShutdownResult": "zmqruntime.client", - "ZMQClient": "zmqruntime.client", - "TransportMode": "zmqruntime.config", - "ZMQConfig": "zmqruntime.config", - "CancelRequest": "zmqruntime.messages", - "ControlErrorResponse": "zmqruntime.messages", - "ControlMessageType": "zmqruntime.messages", - "ControlResponse": "zmqruntime.messages", - "EndpointApplication": "zmqruntime.messages", - "EndpointApplicationCompatibility": "zmqruntime.messages", - "EndpointApplicationCompatibilityError": "zmqruntime.messages", - "EndpointControlCapability": "zmqruntime.messages", - "ExecuteRequest": "zmqruntime.messages", - "ExecuteResponse": "zmqruntime.messages", - "ExecutionRecord": "zmqruntime.messages", - "ExecutionStatus": "zmqruntime.messages", - "ExecutionStatusSnapshot": "zmqruntime.messages", - "ImageAck": "zmqruntime.messages", - "MessageFields": "zmqruntime.messages", - "PongResponse": "zmqruntime.messages", - "ProcessExit": "zmqruntime.messages", - "ProcessIdentity": "zmqruntime.messages", - "ProcessResourceUsage": "zmqruntime.messages", - "ProgressRegistrationRequest": "zmqruntime.messages", - "ProgressUnregistrationRequest": "zmqruntime.messages", - "QueuedExecutionInfo": "zmqruntime.messages", - "ResponseType": "zmqruntime.messages", - "ROIMessage": "zmqruntime.messages", - "RunningExecutionInfo": "zmqruntime.messages", - "ServerRole": "zmqruntime.messages", - "ShapesMessage": "zmqruntime.messages", - "SocketType": "zmqruntime.messages", - "StatusRequest": "zmqruntime.messages", - "TaskPhase": "zmqruntime.messages", - "TaskProgress": "zmqruntime.messages", - "TaskStatus": "zmqruntime.messages", - "WorkerState": "zmqruntime.messages", - "validate_progress_payload": "zmqruntime.messages", - "ExecutionLifecycleEngineABC": "zmqruntime.execution", - "ExecutionProgressObservation": "zmqruntime.execution", - "ExecutionResponseDiagnostic": "zmqruntime.execution", - "ExecutionSubmissionResponse": "zmqruntime.execution", - "ExecutionWaiter": "zmqruntime.execution", - "ExecutionWaitResult": "zmqruntime.execution", - "InMemoryExecutionLifecycleEngine": "zmqruntime.execution", - "ProgressStreamSubscriber": "zmqruntime.execution", - "WaitPolicy": "zmqruntime.execution", - "EventRegistryABC": "zmqruntime.progress", - "EventRegistryMutation": "zmqruntime.progress", - "EventRegistryMutationKind": "zmqruntime.progress", - "GenericAxisProjection": "zmqruntime.progress", - "GenericExecutionProjection": "zmqruntime.progress", - "GenericPlateProjection": "zmqruntime.progress", - "LatestEventRegistry": "zmqruntime.progress", - "ProgressProjectionAdapterABC": "zmqruntime.progress", - "build_execution_projection": "zmqruntime.progress", - "GlobalQueueTrackerRegistry": "zmqruntime.queue_tracker", - "QueueTracker": "zmqruntime.queue_tracker", - "serve_forever": "zmqruntime.runner", - "ZMQServer": "zmqruntime.server", - "EndpointStartupObserver": "zmqruntime.startup", - "EndpointStartupPhase": "zmqruntime.startup", - "EndpointStartupPresentationTarget": "zmqruntime.startup", - "EndpointStartupStatus": "zmqruntime.startup", - "EndpointStartupStatusCallback": "zmqruntime.startup", - "EndpointStartupStatusMonitor": "zmqruntime.startup", - "EndpointStartupStatusRead": "zmqruntime.startup", - "EndpointStartupStatusReader": "zmqruntime.startup", - "EndpointStartupStatusWriter": "zmqruntime.startup", - "CallbackSubscription": "zmqruntime.subscription", - "SubscriptionABC": "zmqruntime.subscription", - "OperationCancellation": "zmqruntime.timeouts", - "OperationDeadline": "zmqruntime.timeouts", - "OperationTimeoutError": "zmqruntime.timeouts", - "DataControlPortPair": "zmqruntime.transport", - "DataControlPortPairAuthority": "zmqruntime.transport", - "TcpDataControlPortPair": "zmqruntime.transport", - "TcpDataControlPortPairAuthority": "zmqruntime.transport", - "TransportEndpoint": "zmqruntime.transport", - "get_control_port": "zmqruntime.transport", - "get_control_url": "zmqruntime.transport", - "get_default_transport_mode": "zmqruntime.transport", - "get_ipc_socket_path": "zmqruntime.transport", - "get_zmq_transport_url": "zmqruntime.transport", - "is_port_in_use": "zmqruntime.transport", - "ping_control_port": "zmqruntime.transport", - "remove_ipc_socket": "zmqruntime.transport", - "resolve_transport_mode": "zmqruntime.transport", - "wait_for_endpoint_ready": "zmqruntime.transport", - "wait_for_server_ready": "zmqruntime.transport", - "ViewerAckPolicy": "zmqruntime.viewer_protocol", - "ViewerAckResponsePayload": "zmqruntime.viewer_protocol", - "ViewerBatchContextWireField": "zmqruntime.viewer_protocol", - "ViewerBatchDisplayPayload": "zmqruntime.viewer_protocol", - "ViewerBatchItemPayload": "zmqruntime.viewer_protocol", - "ViewerBatchItemWireField": "zmqruntime.viewer_protocol", - "ViewerBatchItemWireMapping": "zmqruntime.viewer_protocol", - "ViewerBatchMessageExtraInput": "zmqruntime.viewer_protocol", - "ViewerBatchMessageExtraPayload": "zmqruntime.viewer_protocol", - "ViewerBatchMessageImages": "zmqruntime.viewer_protocol", - "ViewerBatchMessagePayload": "zmqruntime.viewer_protocol", - "ViewerBatchMessageType": "zmqruntime.viewer_protocol", - "ViewerBatchMessageWirePayload": "zmqruntime.viewer_protocol", - "ViewerBatchWireField": "zmqruntime.viewer_protocol", - "ViewerComponentMetadataPayload": "zmqruntime.viewer_protocol", - "ViewerControlReplyHeader": "zmqruntime.viewer_protocol", - "ViewerControlReplyPayload": "zmqruntime.viewer_protocol", - "ViewerControlResponseField": "zmqruntime.viewer_protocol", - "ViewerDisplayConfigWireField": "zmqruntime.viewer_protocol", - "ViewerProtocolStatus": "zmqruntime.viewer_protocol", - "ViewerTransportEndpoint": "zmqruntime.viewer_protocol", - "ViewerWireMapping": "zmqruntime.viewer_protocol", - "ViewerWirePayload": "zmqruntime.viewer_protocol", - "ViewerWireScalar": "zmqruntime.viewer_protocol", - "ViewerWireValue": "zmqruntime.viewer_protocol", - "viewer_wire_key": "zmqruntime.viewer_protocol", - "ViewerInstance": "zmqruntime.viewer_state", - "ViewerState": "zmqruntime.viewer_state", - "ViewerStateManager": "zmqruntime.viewer_state", - "get_or_create_viewer": "zmqruntime.viewer_state", -} - -__all__ = [ - "EndpointConnectionCancelledError", - "EndpointConnectionPolicy", - "EndpointProcessGroup", - "EndpointShutdownMode", - "EndpointShutdownResult", - "ZMQClient", - "TransportMode", - "ZMQConfig", - "CancelRequest", - "ControlErrorResponse", - "ControlResponse", - "ControlMessageType", - "EndpointControlCapability", - "EndpointApplication", - "EndpointApplicationCompatibility", - "EndpointApplicationCompatibilityError", - "ExecuteRequest", - "ExecuteResponse", - "ExecutionRecord", - "ExecutionStatusSnapshot", - "RunningExecutionInfo", - "QueuedExecutionInfo", - "ExecutionStatus", - "ImageAck", - "MessageFields", - "ProcessExit", - "ProcessIdentity", - "PongResponse", - "ProcessResourceUsage", - "ProgressRegistrationRequest", - "ProgressUnregistrationRequest", - # Generic types - "TaskProgress", - "TaskPhase", - "TaskStatus", - "WorkerState", - "ServerRole", - "ResponseType", - "ROIMessage", - "ShapesMessage", - "SocketType", - "StatusRequest", - "validate_progress_payload", - "QueueTracker", - "GlobalQueueTrackerRegistry", - "serve_forever", - "ZMQServer", - "EndpointStartupObserver", - "EndpointStartupPhase", - "EndpointStartupPresentationTarget", - "EndpointStartupStatus", - "EndpointStartupStatusCallback", - "EndpointStartupStatusRead", - "EndpointStartupStatusMonitor", - "EndpointStartupStatusReader", - "EndpointStartupStatusWriter", - "OperationCancellation", - "OperationDeadline", - "OperationTimeoutError", - "DataControlPortPair", - "DataControlPortPairAuthority", - "TcpDataControlPortPair", - "TcpDataControlPortPairAuthority", - "TransportEndpoint", - "get_control_port", - "get_control_url", - "get_default_transport_mode", - "get_ipc_socket_path", - "get_zmq_transport_url", - "is_port_in_use", - "ping_control_port", - "remove_ipc_socket", - "resolve_transport_mode", - "wait_for_server_ready", - "wait_for_endpoint_ready", - "ViewerState", - "ViewerStateManager", - "ViewerInstance", - "get_or_create_viewer", - "ViewerAckPolicy", - "ViewerAckResponsePayload", - "ViewerBatchContextWireField", - "ViewerBatchDisplayPayload", - "ViewerBatchItemPayload", - "ViewerBatchItemWireMapping", - "ViewerBatchItemWireField", - "ViewerBatchMessageType", - "ViewerBatchMessageImages", - "ViewerBatchMessageExtraInput", - "ViewerBatchMessageExtraPayload", - "ViewerBatchMessagePayload", - "ViewerBatchMessageWirePayload", - "ViewerBatchWireField", - "ViewerDisplayConfigWireField", - "ViewerControlReplyHeader", - "ViewerControlReplyPayload", - "ViewerControlResponseField", - "ViewerComponentMetadataPayload", - "ViewerProtocolStatus", - "ViewerTransportEndpoint", - "ViewerWireMapping", - "ViewerWirePayload", - "ViewerWireScalar", - "ViewerWireValue", - "viewer_wire_key", - "ExecutionLifecycleEngineABC", - "ExecutionProgressObservation", - "InMemoryExecutionLifecycleEngine", - "ProgressStreamSubscriber", - "ExecutionWaiter", - "WaitPolicy", - "ExecutionResponseDiagnostic", - "ExecutionSubmissionResponse", - "ExecutionWaitResult", - "EventRegistryABC", - "EventRegistryMutation", - "EventRegistryMutationKind", - "CallbackSubscription", - "SubscriptionABC", - "LatestEventRegistry", - "GenericAxisProjection", - "GenericPlateProjection", - "GenericExecutionProjection", - "ProgressProjectionAdapterABC", - "build_execution_projection", -] +from python_introspect.public_api import lazy_exports -def __getattr__(name: str): - module_name = _LAZY_EXPORTS.get(name) - if module_name is None: - raise AttributeError(f"module 'zmqruntime' has no attribute {name!r}") - import importlib - - value = getattr(importlib.import_module(module_name), name) - globals()[name] = value - return value - +__version__ = _distribution_version("zmqruntime") -def __dir__() -> list[str]: - return sorted(set(globals()) | set(_LAZY_EXPORTS)) +__all__ = ( + "__version__", + *lazy_exports( + globals(), + { + ".ack_listener": ( + "GlobalAckListener", + ), + ".client": ( + "EndpointConnectionCancelledError", + "EndpointConnectionPolicy", + "EndpointProcessGroup", + "EndpointShutdownMode", + "EndpointShutdownResult", + "ZMQClient", + ), + ".config": ( + "TransportMode", + "ZMQConfig", + ), + ".execution": ( + "ExecutionLifecycleEngineABC", + "ExecutionProgressObservation", + "ExecutionResponseDiagnostic", + "ExecutionSubmissionResponse", + "ExecutionWaitResult", + "ExecutionWaiter", + "InMemoryExecutionLifecycleEngine", + "ProgressStreamSubscriber", + "WaitPolicy", + ), + ".execution.config": ( + "ExecutionTransportConfig", + ), + ".messages": ( + "CancelRequest", + "CancelResponse", + "ControlErrorResponse", + "ControlMessageType", + "ControlRequestHeader", + "ControlResponse", + "EndpointApplication", + "EndpointApplicationCompatibility", + "EndpointApplicationCompatibilityError", + "EndpointControlCapability", + "EndpointShutdownRequest", + "ExecuteRequest", + "ExecuteResponse", + "ExecutionRecord", + "ExecutionStatus", + "ExecutionStatusSnapshot", + "ImageAck", + "MessageFields", + "PongResponse", + "ProcessExit", + "ProcessIdentity", + "ProcessResourceUsage", + "ProgressRegistrationRequest", + "ProgressSubscriptionResponse", + "ProgressUnregistrationRequest", + "QueuedExecutionInfo", + "ResponseType", + "RunningExecutionInfo", + "ServerRole", + "ShutdownAck", + "SocketType", + "StatusRequest", + "TaskPhase", + "TaskProgress", + "TaskStatus", + "TypedWireMessage", + "WireMessage", + "WireView", + "WorkerState", + ), + ".process_launch": ( + "BackgroundProcessLaunchPolicy", + "BackgroundProcessLaunchSpec", + "BackgroundProcessPlatform", + ), + ".progress": ( + "EventRegistryABC", + "EventRegistryMutation", + "EventRegistryMutationKind", + "GenericAxisProjection", + "GenericExecutionProjection", + "GenericSubjectProjection", + "LatestEventRegistry", + "ProgressProjectionAdapterABC", + "build_execution_projection", + ), + ".queue_tracker": ( + "GlobalQueueTrackerRegistry", + "QueueTracker", + ), + ".runner": ( + "serve_forever", + ), + ".server": ( + "ZMQServer", + ), + ".startup": ( + "EndpointStartupObserver", + "EndpointStartupPhase", + "EndpointStartupPresentationTarget", + "EndpointStartupStatus", + "EndpointStartupStatusCallback", + "EndpointStartupStatusMonitor", + "EndpointStartupStatusRead", + "EndpointStartupStatusReader", + "EndpointStartupStatusWriter", + ), + ".subscription": ( + "CallbackSubscription", + "SubscriptionABC", + ), + ".timeouts": ( + "OperationCancellation", + "OperationDeadline", + "OperationTimeoutError", + ), + ".transport": ( + "DataControlPortPair", + "DataControlPortPairAuthority", + "TcpDataControlPortPair", + "TcpDataControlPortPairAuthority", + "TransportEndpoint", + "get_control_port", + "get_control_url", + "get_default_transport_mode", + "get_ipc_socket_path", + "get_zmq_transport_url", + "is_port_in_use", + "ping_control_port", + "remove_ipc_socket", + "resolve_transport_mode", + "wait_for_endpoint_ready", + "wait_for_server_ready", + ), + ".viewer_protocol": ( + "ViewerAckPolicy", + "ViewerAckResponsePayload", + "ViewerBatchContextWireField", + "ViewerBatchDisplayPayload", + "ViewerBatchItemPayload", + "ViewerBatchItemWireField", + "ViewerBatchItemWireMapping", + "ViewerBatchMessageExtraInput", + "ViewerBatchMessageExtraPayload", + "ViewerBatchMessageImages", + "ViewerBatchMessagePayload", + "ViewerBatchMessageType", + "ViewerBatchMessageWirePayload", + "ViewerBatchWireField", + "ViewerComponentMetadataPayload", + "ViewerControlReplyHeader", + "ViewerControlReplyPayload", + "ViewerControlResponseField", + "ViewerDisplayConfigWireField", + "ViewerProtocolStatus", + "ViewerTransportEndpoint", + "ViewerWireMapping", + "ViewerWirePayload", + "ViewerWireScalar", + "ViewerWireValue", + "viewer_wire_key", + ), + ".viewer_state": ( + "ViewerInstance", + "ViewerState", + "ViewerStateManager", + "get_or_create_viewer", + ), + }, + ), +) diff --git a/src/zmqruntime/config.py b/src/zmqruntime/config.py index 0fda403..6dbc7a2 100644 --- a/src/zmqruntime/config.py +++ b/src/zmqruntime/config.py @@ -114,6 +114,9 @@ class ZMQConfig: app_name: NonBlankString = "zmqruntime" """Application namespace included in generated transport identities and paths.""" + server_poll_interval_seconds: PositiveFloat = 0.01 + """Delay between endpoint readiness probes while an owned server starts.""" + def __post_init__(self) -> None: """Reject invalid transport topology at its declaration boundary.""" diff --git a/src/zmqruntime/execution/client.py b/src/zmqruntime/execution/client.py index 9046009..5ad165b 100644 --- a/src/zmqruntime/execution/client.py +++ b/src/zmqruntime/execution/client.py @@ -20,17 +20,22 @@ from zmqruntime.execution.responses import ( ExecutionSubmissionResponse, ExecutionWaitResult, - WireRequest, WireResponse, ) from zmqruntime.execution.wait_policy import ExecutionWaiter, WaitPolicy from zmqruntime.messages import ( + CancelRequest, ControlMessageType, ControlRequestHeader, + ExecuteRequest, ExecutionStatusSnapshot, MessageFields, + ProgressRegistrationRequest, + ProgressSubscriptionResponse, + ProgressUnregistrationRequest, PongResponse, ResponseType, + StatusRequest, ) from zmqruntime.subscription import CallbackSubscription, SubscriptionABC from zmqruntime.timeouts import OperationDeadline @@ -97,10 +102,7 @@ def submit_execution( raise RuntimeError("Failed to connect to execution server") self._ensure_progress_subscription(timeout_ms=timeout_ms) request = self.serialize_task(task, config) - if MessageFields.TYPE not in request: - request[MessageFields.TYPE] = ControlMessageType.EXECUTE.value - response = self._send_control_request(request, timeout_ms=timeout_ms) - return response + return self._send_control_request(request.to_dict(), timeout_ms=timeout_ms) def poll_status( self, @@ -108,10 +110,10 @@ def poll_status( *, timeout_ms: int = 5000, ): - request = {MessageFields.TYPE: ControlMessageType.STATUS.value} - if execution_id: - request[MessageFields.EXECUTION_ID] = execution_id - return self._send_control_request(request, timeout_ms=timeout_ms) + return self._send_control_request( + StatusRequest(execution_id=execution_id or None).to_dict(), + timeout_ms=timeout_ms, + ) def wait_for_completion( self, @@ -169,12 +171,7 @@ def execute( return response def cancel_execution(self, execution_id): - return self._send_control_request( - { - MessageFields.TYPE: ControlMessageType.CANCEL.value, - MessageFields.EXECUTION_ID: execution_id, - } - ) + return self._send_control_request(CancelRequest(execution_id=execution_id).to_dict()) def ping(self): try: @@ -281,18 +278,14 @@ def _ensure_progress_subscription(self, *, timeout_ms: int = 5000) -> None: self._start_progress_listener() return response = self._send_control_request( - { - MessageFields.TYPE: ControlMessageType.REGISTER_PROGRESS.value, - MessageFields.CLIENT_ID: self._progress_client_id, - }, + ProgressRegistrationRequest(client_id=self._progress_client_id).to_dict(), timeout_ms=timeout_ms, ) - if MessageFields.STATUS not in response: - raise RuntimeError(f"Progress registration response missing status: {response}") - if response[MessageFields.STATUS] != ResponseType.OK.value: + if response.get(MessageFields.STATUS) != ResponseType.OK.value: raise RuntimeError(f"Progress registration failed: {response}") + registration = ProgressSubscriptionResponse.from_dict(response) with self._progress_lock: - for execution_id, payload in response[MessageFields.PROGRESS_OBSERVATIONS].items(): + for execution_id, payload in registration.progress_observations.items(): baseline = ExecutionProgressObservation.from_wire(payload) previous = self._progress_by_execution_id.get(execution_id) if previous is not None: @@ -305,10 +298,7 @@ def _unregister_progress(self) -> bool: if not self.is_connected(): return True self._send_control_request( - { - MessageFields.TYPE: ControlMessageType.UNREGISTER_PROGRESS.value, - MessageFields.CLIENT_ID: self._progress_client_id, - } + ProgressUnregistrationRequest(client_id=self._progress_client_id).to_dict() ) return True @@ -400,6 +390,6 @@ def serialize_task( self, task: TaskT, config: ConfigT | None, - ) -> WireRequest: - """Serialize task for transmission. Subclass provides serialization logic.""" + ) -> ExecuteRequest: + """Declare the execution request for one task.""" raise NotImplementedError diff --git a/src/zmqruntime/execution/config.py b/src/zmqruntime/execution/config.py new file mode 100644 index 0000000..f7e629f --- /dev/null +++ b/src/zmqruntime/execution/config.py @@ -0,0 +1,108 @@ +"""Execution transport configuration.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from zmqruntime.config import ( + NonBlankString, + PositiveFloat, + PositiveInteger, + TransportMode, + ZMQConfig, +) +from zmqruntime.transport import TransportEndpoint + + +@dataclass(frozen=True, slots=True) +class ExecutionTransportConfig(ZMQConfig): + """Execution-server endpoint topology, discovery and request deadlines.""" + + 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 = TransportMode.default() + """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 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, request + 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 server + preparation to continue while genuinely stalled launches still fail. + """ + + 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 outputs written by an execution. + """ + + 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 a 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 + ), + ) diff --git a/src/zmqruntime/execution/lifecycle.py b/src/zmqruntime/execution/lifecycle.py index 82f8ede..efdf10b 100644 --- a/src/zmqruntime/execution/lifecycle.py +++ b/src/zmqruntime/execution/lifecycle.py @@ -9,10 +9,10 @@ from threading import RLock from zmqruntime.messages import ( + CancelResponse, ExecutionRecord, ExecutionStatus, ExecutionStatusSnapshot, - MessageFields, QueuedExecutionInfo, ResponseType, RunningExecutionInfo, @@ -42,23 +42,23 @@ def should_interrupt_active_work(self) -> bool: return self.transition.applied and self.transition.previous.owns_active_work - def to_response(self, *, workers_killed: int) -> dict: - """Project this outcome onto the execution control response contract.""" + def to_response(self, *, workers_killed: int) -> CancelResponse: + """The control response for this outcome.""" if self.transition.applied: - return { - MessageFields.STATUS: ResponseType.OK.value, - MessageFields.MESSAGE: f"Cancelled - killed {workers_killed} workers", - MessageFields.WORKERS_KILLED: workers_killed, - } - return { - MessageFields.STATUS: ResponseType.ERROR.value, - MessageFields.ERROR: ( + return CancelResponse( + status=ResponseType.OK, + workers_killed=workers_killed, + message=f"Cancelled - killed {workers_killed} workers", + ) + return CancelResponse( + status=ResponseType.ERROR, + workers_killed=workers_killed, + error=( f"Execution {self.transition.execution_id} is already " f"{self.transition.current.value}; cancellation was not applied" ), - MessageFields.WORKERS_KILLED: workers_killed, - } + ) class ExecutionLifecycleEngineABC(ABC): @@ -323,7 +323,7 @@ def _running_executions(self) -> tuple[RunningExecutionInfo, ...]: running.append( RunningExecutionInfo( execution_id=execution_id, - plate_id=str(record.plate_id), + subject_id=record.subject_id, start_time=start_time, elapsed=(now - start_time) if start_time > 0 else 0.0, compile_only=bool(record.compile_only), @@ -340,7 +340,7 @@ def _queued_executions(self) -> tuple[QueuedExecutionInfo, ...]: queued.append( QueuedExecutionInfo( execution_id=execution_id, - plate_id=str(record.plate_id), + subject_id=record.subject_id, queue_position=idx, ) ) diff --git a/src/zmqruntime/execution/progress_stream.py b/src/zmqruntime/execution/progress_stream.py index 04068ee..f10f51b 100644 --- a/src/zmqruntime/execution/progress_stream.py +++ b/src/zmqruntime/execution/progress_stream.py @@ -20,7 +20,7 @@ ExecutionStatusSnapshot, MessageFields, ResponseType, - validate_progress_payload, + TaskProgress, ) logger = logging.getLogger(__name__) @@ -218,7 +218,7 @@ def _dispatch_message(self, message: str) -> bool: raise ValueError("Execution stream record is not terminal") execution_id = snapshot.execution.execution_id else: - validate_progress_payload(data) + TaskProgress.from_dict(data) execution_id = data.get(MessageFields.EXECUTION_ID) with self._delivery_condition: self._dispatching_execution_id = execution_id diff --git a/src/zmqruntime/execution/server.py b/src/zmqruntime/execution/server.py index b5f4e71..0b35fdd 100644 --- a/src/zmqruntime/execution/server.py +++ b/src/zmqruntime/execution/server.py @@ -31,13 +31,14 @@ MessageFields, PongResponse, ProcessIdentity, + ProgressSubscriptionResponse, ProgressRegistrationRequest, ProgressUnregistrationRequest, - QueuedExecutionInfo, ResponseType, - RunningExecutionInfo, ServerRole, + ShutdownAck, StatusRequest, + TaskProgress, ) from zmqruntime.server import ZMQServer @@ -123,25 +124,9 @@ def _create_pong_response(self) -> PongResponse: } ), active_executions=snapshot.active_executions, - running_executions=tuple( - RunningExecutionInfo( - execution_id=info.execution_id, - plate_id=info.plate_id, - start_time=info.start_time, - elapsed=info.elapsed, - compile_only=info.compile_only, - ) - for info in running - ), - queued_executions=tuple( - QueuedExecutionInfo( - execution_id=info.execution_id, - plate_id=info.plate_id, - queue_position=info.queue_position, - ) - for info in queued - ), - workers=self._get_worker_info(), + running_executions=running, + queued_executions=queued, + workers=tuple(self._get_worker_info()), uptime=time.time() - self.start_time if self.start_time else 0, progress_subscribers=len(self._progress_subscribers), ) @@ -161,7 +146,7 @@ def process_messages(self): break logger.info( f"Publishing to ZMQ: step={progress_update.get('step')}, " - f"axis={progress_update.get('axis_id')}, plate_id={progress_update.get('plate_id')}, " + f"axis={progress_update.get('axis_id')}, subject_id={progress_update.get('subject_id')}, " f"percent={progress_update.get('percent')}" ) with self._progress_publish_lock: @@ -185,12 +170,12 @@ def handle_data_message(self, message): def _validate_and_parse(self, msg, request_class): try: request = request_class.from_dict(msg) - error = request.validate() - if error: - return None, ExecuteResponse(ResponseType.ERROR, error=error).to_dict() - return request, None - except KeyError as e: - return None, ExecuteResponse(ResponseType.ERROR, error=f"Missing field: {e}").to_dict() + except (KeyError, TypeError, ValueError) as e: + return None, ExecuteResponse(ResponseType.ERROR, error=str(e)).to_dict() + error = request.validate() + if error: + return None, ExecuteResponse(ResponseType.ERROR, error=error).to_dict() + return request, None def _start_queue_worker(self): if self.queue_worker_thread is None or not self.queue_worker_thread.is_alive(): @@ -255,7 +240,7 @@ def _handle_execute(self, msg): execution_id = str(uuid.uuid4()) record = ExecutionRecord( execution_id=execution_id, - plate_id=request.plate_id, + subject_id=request.subject_id, client_address=request.client_address, status=ExecutionStatus.QUEUED.value, compile_only=request.compile_only, @@ -286,8 +271,12 @@ def run_execution(self, execution_id, request, record): complete_transition = self._lifecycle.mark_complete( execution_id, results_summary={ - MessageFields.WELL_COUNT: len(results) if isinstance(results, dict) else 0, - MessageFields.WELLS: list(results.keys()) if isinstance(results, dict) else [], + MessageFields.PARTITION_COUNT: ( + len(results) if isinstance(results, dict) else 0 + ), + MessageFields.PARTITION_VALUES: ( + list(results.keys()) if isinstance(results, dict) else [] + ), }, ) if not complete_transition.applied: @@ -337,10 +326,6 @@ def _publish_terminal_record(self, record: ExecutionRecord) -> None: ExecutionStatusSnapshot(ResponseType.OK, execution=record).to_dict() ) - def _run_execution(self, execution_id, request, record): - """Compatibility shim for callers still bound to the old private hook.""" - self.run_execution(execution_id, request, record) - def handle_status(self, msg): execution_id = StatusRequest.from_dict(msg).execution_id if execution_id: @@ -359,10 +344,6 @@ def handle_status(self, msg): ) return snapshot.to_dict() - def _handle_status(self, msg): - """Compatibility shim for callers still bound to the old private hook.""" - return self.handle_status(msg) - def _handle_cancel(self, msg): request, error = self._validate_and_parse(msg, CancelRequest) if error: @@ -386,7 +367,7 @@ def _handle_cancel(self, msg): cancellation.transition.applied, killed, ) - return cancellation.to_response(workers_killed=killed) + return cancellation.to_response(workers_killed=killed).to_dict() def _cancel_all_executions(self): cancellations = self._lifecycle.cancel_all_active(end_time=time.time()) @@ -412,11 +393,7 @@ def _shutdown_workers(self, msg: dict[str, Any], force: bool = False) -> dict[st self.request_shutdown() msg = f"Workers killed ({killed}), server {'shutting down' if force else 'alive'}" logger.info(msg) - return { - MessageFields.TYPE: ResponseType.SHUTDOWN_ACK.value, - MessageFields.STATUS: "success", - MessageFields.MESSAGE: msg, - } + return ShutdownAck(message=msg).to_dict() def _handle_shutdown(self, msg): return self._shutdown_workers(msg, force=False) @@ -444,30 +421,26 @@ def _handle_register_progress(self, msg): for execution_id, record in self.active_executions.items() if record.progress_event is not None or ExecutionStatus(record.status).is_terminal } - return { - MessageFields.STATUS: ResponseType.OK.value, - MessageFields.MESSAGE: "Progress subscriber registered", - MessageFields.PROGRESS_OBSERVATIONS: observations, - MessageFields.CLIENT_ID: request.client_id, - MessageFields.PROGRESS_SUBSCRIBERS: len(self._progress_subscribers), - } + return ProgressSubscriptionResponse( + client_id=request.client_id, + progress_subscribers=len(self._progress_subscribers), + message="Progress subscriber registered", + progress_observations=observations, + ).to_dict() def _handle_unregister_progress(self, msg): request, error = self._validate_and_parse(msg, ProgressUnregistrationRequest) if error: return error self._progress_subscribers.discard(request.client_id) - return { - MessageFields.STATUS: ResponseType.OK.value, - MessageFields.MESSAGE: "Progress subscriber unregistered", - MessageFields.CLIENT_ID: request.client_id, - MessageFields.PROGRESS_SUBSCRIBERS: len(self._progress_subscribers), - } + return ProgressSubscriptionResponse( + client_id=request.client_id, + progress_subscribers=len(self._progress_subscribers), + message="Progress subscriber unregistered", + ).to_dict() def send_progress_update(self, progress_update: dict) -> None: - from zmqruntime.messages import validate_progress_payload - - validate_progress_payload(progress_update) + TaskProgress.from_dict(progress_update) execution_id = progress_update[MessageFields.EXECUTION_ID] with self._progress_publish_lock: record = self.active_executions[execution_id] diff --git a/src/zmqruntime/messages.py b/src/zmqruntime/messages.py index 1636d27..b72ffc6 100644 --- a/src/zmqruntime/messages.py +++ b/src/zmqruntime/messages.py @@ -12,21 +12,89 @@ import pickle import signal import sys -from abc import ABC, abstractmethod -from collections.abc import Mapping -from dataclasses import asdict, dataclass, field, fields +from collections.abc import Iterable, Mapping +from dataclasses import dataclass, field, fields, replace from enum import Enum -from typing import Any, Dict, Optional, Tuple +from typing import Any, ClassVar, Dict, Optional, Tuple from uuid import UUID import psutil from python_introspect import dataclass_from_mapping +from python_introspect.jsonable import JsonValue, to_jsonable +from typing_extensions import Self from zmqruntime.timeouts import OperationDeadline logger = logging.getLogger(__name__) +# ============================================================================= +# Declared messages: the wire mapping of a message is its declared fields +# ============================================================================= + + +class WireMessage: + """A declared message whose wire mapping is exactly its declared fields. + + Encoding is ``to_jsonable`` and decoding is ``dataclass_from_mapping``: + both read the dataclass declaration, so a message never restates its + fields. Decoding is strict: an undeclared field fails loudly. + """ + + __slots__ = () + + def to_dict(self) -> dict[str, JsonValue]: + return dict(to_jsonable(self)) + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> Self: + return dataclass_from_mapping(cls, cls.declared_items(data)) + + @classmethod + def declared_items(cls, data: Mapping[str, Any]) -> Mapping[str, Any]: + """The part of a wire mapping this message decodes.""" + + return data + + +class WireView(WireMessage): + """A declared message read out of a wider payload. + + Applications extend a generic envelope (a progress event, a status + response) with their own fields; the view decodes only the fields it + declares and leaves the rest to the application. + """ + + __slots__ = () + + @classmethod + def declared_items(cls, data: Mapping[str, Any]) -> Mapping[str, Any]: + return { + declared.name: data[declared.name] + for declared in fields(cls) + if declared.init and declared.name in data + } + + +class TypedWireMessage(WireMessage): + """A declared message that carries its kind under the wire ``type`` key.""" + + __slots__ = () + + wire_type: ClassVar[str] + + def to_dict(self) -> dict[str, JsonValue]: + return {MessageFields.TYPE: self.wire_type, **super().to_dict()} + + @classmethod + def declared_items(cls, data: Mapping[str, Any]) -> Mapping[str, Any]: + if data.get(MessageFields.TYPE) != cls.wire_type: + raise ValueError(f"{cls.__name__} requires type={cls.wire_type!r}.") + return super().declared_items( + {key: value for key, value in data.items() if key != MessageFields.TYPE} + ) + + # ============================================================================= # Generic Progress Types - Application Agnostic # ============================================================================= @@ -64,7 +132,7 @@ class TaskStatus(Enum): @dataclass(frozen=True) -class WorkerState: +class WorkerState(WireMessage): """Generic worker/process information.""" pid: int @@ -73,63 +141,25 @@ class WorkerState: memory_mb: float create_time: float | None = None - def to_dict(self) -> dict[str, Any]: - result = { - "pid": self.pid, - "status": self.status, - "cpu_percent": self.cpu_percent, - "memory_mb": self.memory_mb, - } - if self.create_time is not None: - result["create_time"] = self.create_time - return result - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "WorkerState": - known_fields = {"pid", "status", "cpu_percent", "memory_mb", "create_time"} - unexpected_fields = set(data).difference(known_fields) - if unexpected_fields: - raise ValueError(f"WorkerState received unknown fields: {sorted(unexpected_fields)!r}") - - return cls( - pid=data["pid"], - status=data["status"], - cpu_percent=data["cpu_percent"], - memory_mb=data["memory_mb"], - create_time=data.get("create_time"), - ) - @dataclass(frozen=True) -class TaskProgress: - """Generic task progress - supports both generic and app-specific enums. - - Phase and status accept EITHER strings (for transport) OR Enum types (for type safety). - Applications can use TaskPhase/TaskStatus or extend with their own enums (e.g., AxisPhase/AxisStatus). +class TaskProgress(WireView): + """The generic fields of one progress event. - The .value of the enum is stored/transmitted, allowing interop between - generic and app-specific code. + Applications publish progress events that carry these fields beside their + own; the runtime reads and validates only these. Phase and status accept + strings (as transported) or application enums (encoded by value). """ - # Required fields - task_id: str - phase: str | Enum # String for transport, Enum for type safety - status: str | Enum # String for transport, Enum for type safety + execution_id: str + phase: str | Enum + status: str | Enum percent: float timestamp: float - - # Progress tracking completed: int total: int - - # Core tracking identifiers (moved from context) - plate_id: str = "" + subject_id: str = "" axis_id: str = "" - - # Application-specific context (NOT generic runtime concerns) - context: Dict[str, Any] = field(default_factory=dict) - - # Optional error info error: Optional[str] = None traceback: Optional[str] = None @@ -139,95 +169,11 @@ def __post_init__(self): raise ValueError(f"percent must be in [0, 100], got {self.percent}") if self.completed > self.total: raise ValueError(f"completed ({self.completed}) cannot exceed total ({self.total})") - # Allow both strings and enums for phase/status if not isinstance(self.phase, (str, Enum)): raise TypeError(f"phase must be a string or Enum, got {type(self.phase)}") if not isinstance(self.status, (str, Enum)): raise TypeError(f"status must be a string or Enum, got {type(self.status)}") - def to_dict(self) -> Dict[str, Any]: - """Serialize for transport.""" - # Handle both string and Enum types for phase/status - phase_value = self.phase.value if isinstance(self.phase, Enum) else self.phase - status_value = self.status.value if isinstance(self.status, Enum) else self.status - - result = { - "type": "progress", - "task_id": self.task_id, - "execution_id": self.task_id, - "phase": phase_value, - "status": status_value, - "percent": self.percent, - "timestamp": self.timestamp, - "completed": self.completed, - "total": self.total, - "plate_id": self.plate_id, - "axis_id": self.axis_id, - **self.context, - } - if self.error: - result["error"] = self.error - if self.traceback: - result["traceback"] = self.traceback - return result - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "TaskProgress": - """Parse from transport. - - Phase and status are passed through as strings from the dict. - The application layer is responsible for converting them to appropriate Enum types. - """ - - # Separate generic fields from app-specific context - generic_fields = { - "type", - "task_id", - "execution_id", - "phase", - "status", - "percent", - "timestamp", - "completed", - "total", - "plate_id", - "axis_id", - "error", - "traceback", - "step_name", - "pid", - } - - context = {k: v for k, v in data.items() if k not in generic_fields} - - task_id = data.get("execution_id") - if task_id is None: - task_id = data.get("task_id") - if task_id is None: - raise KeyError("Missing required field: execution_id") - - # Create TaskProgress with string phase/status (no enum conversion) - return cls( - task_id=task_id, - phase=data["phase"], # Pass through as string - status=data["status"], # Pass through as string - percent=data["percent"], - timestamp=data["timestamp"], - completed=data["completed"], - total=data["total"], - plate_id=data.get("plate_id", ""), - axis_id=data.get("axis_id", ""), - context=context, - error=data.get("error"), - traceback=data.get("traceback"), - ) - - -def validate_progress_payload(payload: dict) -> dict: - """Validate progress payload using generic TaskProgress.""" - TaskProgress.from_dict(payload) # Will raise if invalid - return payload - # ============================================================================= # Message Field Constants (used by control messages) @@ -238,24 +184,18 @@ class MessageFields: TYPE = "type" OBSERVATION_BUDGET_SECONDS = "observation_budget_seconds" OBSERVATION_OPERATION = "observation_operation" - PLATE_ID = "plate_id" - EXECUTION_PLATE_ID = "execution_plate_id" + SUBJECT_ID = "subject_id" + EXECUTION_SUBJECT_ID = "execution_subject_id" SELECTED_PIPELINE_PATH = "selected_pipeline_path" PIPELINE_CODE = "pipeline_code" CONFIG_PARAMS = "config_params" CONFIG_CODE = "config_code" PIPELINE_CONFIG_CODE = "pipeline_config_code" - CLIENT_ADDRESS = "client_address" COMPILE_ONLY = "compile_only" COMPILE_ARTIFACT_ID = "compile_artifact_id" - COMPILE_STATUS = "compile_status" - COMPILE_MESSAGE = "compile_message" - MEMORY_MB = "memory_mb" - CPU_PERCENT = "cpu_percent" EXECUTION_ID = "execution_id" START_TIME = "start_time" END_TIME = "end_time" - ELAPSED = "elapsed" STATUS = "status" ERROR = "error" MESSAGE = "message" @@ -265,56 +205,27 @@ class MessageFields: SERVER = "server" SERVER_TYPE = "server_type" SERVER_ROLE = "server_role" - LOG_FILE_PATH = "log_file_path" - ACTIVE_EXECUTIONS = "active_executions" RUNNING_EXECUTIONS = "running_executions" - WORKERS = "workers" - WORKERS_KILLED = "workers_killed" - UPTIME = "uptime" - EXECUTIONS = "executions" EXECUTION = "execution" - QUEUED_EXECUTIONS = "queued_executions" - QUEUE_POSITION = "queue_position" - WELL_COUNT = "well_count" - WELLS = "wells" + PARTITION_COUNT = "partition_count" + PARTITION_VALUES = "partition_values" RESULTS_SUMMARY = "results_summary" - WELL_ID = "well_id" - STEP = "step" TIMESTAMP = "timestamp" AXIS_ID = "axis_id" - STEP_NAME = "step_name" - STEP_INDEX = "step_index" - TOTAL_STEPS = "total_steps" - PHASE = "phase" - COMPLETED = "completed" - TOTAL = "total" - PERCENT = "percent" - PATTERN = "pattern" - COMPONENT = "component" TRACEBACK = "traceback" # Acknowledgment message fields IMAGE_ID = "image_id" - VIEWER_PORT = "viewer_port" - VIEWER_TYPE = "viewer_type" # ROI message fields ROIS = "rois" - LAYER_NAME = "layer_name" SHAPES = "shapes" - COORDINATES = "coordinates" METADATA = "metadata" CLIENT_ID = "client_id" PROGRESS_SUBSCRIBERS = "progress_subscribers" PROGRESS_SEQUENCE = "progress_sequence" - PROGRESS_OBSERVATIONS = "progress_observations" - PROCESS_IDENTITY = "process_identity" - CONTROL_CAPABILITIES = "control_capabilities" - APPLICATION = "application" - APPLICATION_ID = "application_id" - APPLICATION_VERSION = "application_version" @dataclass(frozen=True) -class ProcessIdentity: +class ProcessIdentity(WireMessage): """PID-reuse-safe identity for a process visible on the local host.""" pid: int @@ -336,13 +247,6 @@ def is_current(self) -> bool: return self == self.current() - def to_dict(self) -> Dict[str, Any]: - return {"pid": self.pid, "create_time": self.create_time} - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "ProcessIdentity": - return cls(pid=int(data["pid"]), create_time=float(data["create_time"])) - def is_alive(self) -> bool | None: """Return exact local liveness, or unknown when access is denied.""" @@ -525,9 +429,19 @@ class ControlMessageType(Enum): UNREGISTER_PROGRESS = "unregister_progress" +_CONTROL_HEADER_FIELDS = frozenset({"message_type", "operation_deadline"}) + + @dataclass(frozen=True, slots=True) class ControlRequestHeader: - """Canonical control-message identity and its pickle wire boundary.""" + """A control request: its kind, its observation budget and its declared body. + + The wire mapping carries the kind under ``type`` and the remaining + observation budget as a duration (never an absolute host clock). Every + other declared field is the request body, encoded and decoded by the + declaration. Requests are read out of wider control payloads, so fields + a request does not declare are left to the payload's other readers. + """ message_type: ControlMessageType operation_deadline: OperationDeadline | None = field(default=None, kw_only=True) @@ -552,25 +466,43 @@ def admit_observation(payload: Mapping[str, object]) -> OperationDeadline: ) @classmethod - def from_dict(cls, payload: Mapping[str, object]) -> ControlRequestHeader: + def body_fields(cls) -> tuple[str, ...]: + """Declared fields carried as the request body.""" + + return tuple( + declared.name + for declared in fields(cls) + if declared.name not in _CONTROL_HEADER_FIELDS + ) + + @classmethod + def from_dict(cls, payload: Mapping[str, object]) -> Self: raw_message_type = payload[MessageFields.TYPE] if not isinstance(raw_message_type, str): raise TypeError("Control request type must be a string.") - return cls( - ControlMessageType(raw_message_type), - operation_deadline=(cls.admit_observation(payload) - if MessageFields.OBSERVATION_BUDGET_SECONDS in payload else None), + request = dataclass_from_mapping( + cls, + { + "message_type": raw_message_type, + **{name: payload[name] for name in cls.body_fields() if name in payload}, + }, ) + if MessageFields.OBSERVATION_BUDGET_SECONDS not in payload: + return request + return replace(request, operation_deadline=cls.admit_observation(payload)) @classmethod - def from_wire_payload(cls, wire_payload: bytes) -> ControlRequestHeader: + def from_wire_payload(cls, wire_payload: bytes) -> Self: payload = pickle.loads(wire_payload) if not isinstance(payload, Mapping): raise TypeError("Control request payload must be a mapping.") return cls.from_dict(payload) def to_dict(self) -> dict[str, object]: - payload = {MessageFields.TYPE: self.message_type.value} + payload = { + MessageFields.TYPE: self.message_type.value, + **{name: to_jsonable(getattr(self, name)) for name in self.body_fields()}, + } return (payload if self.operation_deadline is None else self.with_observation_budget(payload, self.operation_deadline)) @@ -584,24 +516,6 @@ class EndpointShutdownRequest(ControlRequestHeader): process_identity: ProcessIdentity | None = None - @classmethod - def from_dict(cls, payload: Mapping[str, object]) -> EndpointShutdownRequest: - header = ControlRequestHeader.from_dict(payload) - identity_data = payload.get(MessageFields.PROCESS_IDENTITY) - if identity_data is not None and not isinstance(identity_data, dict): - raise TypeError("Shutdown process identity must be a mapping") - return cls( - header.message_type, - None if identity_data is None else ProcessIdentity.from_dict(identity_data), - operation_deadline=header.operation_deadline, - ) - - def to_dict(self) -> dict[str, Any]: - payload: dict[str, Any] = ControlRequestHeader.to_dict(self) - if self.process_identity is not None: - payload[MessageFields.PROCESS_IDENTITY] = self.process_identity.to_dict() - return payload - def validate(self) -> str | None: if self.process_identity is not None and self.process_identity != ProcessIdentity.current(): return "Shutdown addressed a different native process incarnation; no mutation" @@ -703,13 +617,16 @@ def get_display_name(self): return self.value -@dataclass(frozen=True) -class ExecuteRequest: - plate_id: str +@dataclass(frozen=True, slots=True) +class ExecuteRequest(ControlRequestHeader): + """Submit one subject's work to an execution server.""" + + message_type: ControlMessageType = field(default=ControlMessageType.EXECUTE, init=False) + subject_id: str pipeline_code: str - execution_plate_id: Optional[str] = None + execution_subject_id: Optional[str] = None selected_pipeline_path: Optional[str] = None - config_params: Optional[dict] = None + config_params: Optional[Dict[str, Any]] = None config_code: Optional[str] = None pipeline_config_code: Optional[str] = None client_address: Optional[str] = None @@ -717,8 +634,8 @@ class ExecuteRequest: compile_artifact_id: Optional[str] = None def validate(self): - if not self.plate_id: - return "Missing required field: plate_id" + if not self.subject_id: + return "Missing required field: subject_id" if not self.pipeline_code: return "Missing required field: pipeline_code" if self.compile_only and self.compile_artifact_id: @@ -727,59 +644,61 @@ def validate(self): return "Missing config: provide either config_params or config_code" return None - def to_dict(self): - result: Dict[str, Any] = {} - result[MessageFields.TYPE] = ControlMessageType.EXECUTE.value - result[MessageFields.PLATE_ID] = self.plate_id - if self.execution_plate_id is not None: - result[MessageFields.EXECUTION_PLATE_ID] = self.execution_plate_id - if self.selected_pipeline_path is not None: - result[MessageFields.SELECTED_PIPELINE_PATH] = self.selected_pipeline_path - result[MessageFields.PIPELINE_CODE] = self.pipeline_code - if self.config_params is not None: - result[MessageFields.CONFIG_PARAMS] = self.config_params - if self.config_code is not None: - result[MessageFields.CONFIG_CODE] = self.config_code - if self.pipeline_config_code is not None: - result[MessageFields.PIPELINE_CONFIG_CODE] = self.pipeline_config_code - if self.client_address is not None: - result[MessageFields.CLIENT_ADDRESS] = self.client_address - if self.compile_only: - result[MessageFields.COMPILE_ONLY] = True - if self.compile_artifact_id is not None: - result[MessageFields.COMPILE_ARTIFACT_ID] = self.compile_artifact_id - return result - @classmethod - def from_dict(cls, data): - return cls( - plate_id=data[MessageFields.PLATE_ID], - pipeline_code=data[MessageFields.PIPELINE_CODE], - execution_plate_id=data.get(MessageFields.EXECUTION_PLATE_ID), - selected_pipeline_path=data.get(MessageFields.SELECTED_PIPELINE_PATH), - config_params=data.get(MessageFields.CONFIG_PARAMS), - config_code=data.get(MessageFields.CONFIG_CODE), - pipeline_config_code=data.get(MessageFields.PIPELINE_CONFIG_CODE), - client_address=data.get(MessageFields.CLIENT_ADDRESS), - compile_only=bool(data.get(MessageFields.COMPILE_ONLY, False)), - compile_artifact_id=data.get(MessageFields.COMPILE_ARTIFACT_ID), - ) +@dataclass(frozen=True, slots=True) +class StatusRequest(ControlRequestHeader): + message_type: ControlMessageType = field(default=ControlMessageType.STATUS, init=False) + execution_id: Optional[str] = None + + def validate(self): + return None + + +@dataclass(frozen=True, slots=True) +class CancelRequest(ControlRequestHeader): + message_type: ControlMessageType = field(default=ControlMessageType.CANCEL, init=False) + execution_id: str + + def validate(self): + return "Missing execution_id" if not self.execution_id else None + + +@dataclass(frozen=True, slots=True) +class ProgressRegistrationRequest(ControlRequestHeader): + message_type: ControlMessageType = field( + default=ControlMessageType.REGISTER_PROGRESS, init=False + ) + client_id: str + + def validate(self): + return "Missing client_id" if not self.client_id else None -class ControlResponse(ABC): +@dataclass(frozen=True, slots=True) +class ProgressUnregistrationRequest(ControlRequestHeader): + message_type: ControlMessageType = field( + default=ControlMessageType.UNREGISTER_PROGRESS, init=False + ) + client_id: str + + def validate(self): + return "Missing client_id" if not self.client_id else None + + +class ControlResponse(WireMessage): """Nominal response that owns its control-wire mapping.""" - @abstractmethod - def to_dict(self) -> Dict[str, Any]: - """Project this response to its control-wire representation.""" + __slots__ = () @dataclass(frozen=True) -class ControlErrorResponse(ControlResponse): - """Canonical control error and its wire projection.""" +class ControlErrorResponse(TypedWireMessage, ControlResponse): + """A control request failed; ``message`` says why.""" + + wire_type: ClassVar[str] = ResponseType.ERROR.value message: str - response_type: ResponseType = ResponseType.ERROR + status: ResponseType = ResponseType.ERROR @classmethod def from_exception(cls, error: Exception) -> "ControlErrorResponse": @@ -787,14 +706,6 @@ def from_exception(cls, error: Exception) -> "ControlErrorResponse": return cls(message=str(error)) - def to_dict(self) -> Dict[str, Any]: - response_type = self.response_type.value - return { - MessageFields.STATUS: response_type, - MessageFields.TYPE: response_type, - MessageFields.MESSAGE: self.message, - } - @dataclass(frozen=True) class ExecuteResponse(ControlResponse): @@ -803,169 +714,106 @@ class ExecuteResponse(ControlResponse): message: Optional[str] = None error: Optional[str] = None - def to_dict(self): - result: Dict[str, Any] = {MessageFields.STATUS: self.status.value} - if self.execution_id is not None: - result[MessageFields.EXECUTION_ID] = self.execution_id - if self.message is not None: - result[MessageFields.MESSAGE] = self.message - if self.error is not None: - result[MessageFields.ERROR] = self.error - return result + +@dataclass(frozen=True) +class CancelResponse(ControlResponse): + """Outcome of one cancellation request.""" + + status: ResponseType + workers_killed: int + message: Optional[str] = None + error: Optional[str] = None + + +@dataclass(frozen=True) +class ShutdownAck(TypedWireMessage, ControlResponse): + """A server accepted a shutdown request.""" + + wire_type: ClassVar[str] = ResponseType.SHUTDOWN_ACK.value + + message: str + status: str = "success" + + +@dataclass(frozen=True) +class ProgressSubscriptionResponse(ControlResponse): + """A progress subscriber was registered or released. + + A registration carries the latest observation of every execution the + server still holds, keyed by execution id, so the subscriber starts from + the server's baseline. + """ + + client_id: str + progress_subscribers: int + message: str + progress_observations: Dict[str, Any] = field(default_factory=dict) + status: ResponseType = ResponseType.OK @dataclass -class ExecutionRecord: +class ExecutionRecord(WireMessage): + """One execution's lifecycle record; its declared fields are its wire record. + + ``extras`` and ``progress_event`` are server-local attachments (an + application's orchestrator, the latest progress event). They are not + fields, so they never travel. + """ + execution_id: str - plate_id: str - client_address: Optional[str] + subject_id: str status: str + client_address: Optional[str] = None start_time: Optional[float] = None end_time: Optional[float] = None error: Optional[str] = None traceback: Optional[str] = None results_summary: Optional[Dict[str, Any]] = None compile_only: bool = False - metadata: Dict[str, Any] = field(default_factory=dict) progress_sequence: int = 0 - progress_event: dict | None = field(default=None, repr=False) def __post_init__(self) -> None: if isinstance(self.progress_sequence, bool) or not isinstance(self.progress_sequence, int): raise TypeError("Execution progress sequence must be an integer") if self.progress_sequence < 0: raise ValueError("Execution progress sequence cannot be negative") + self.extras: Dict[str, Any] = {} + self.progress_event: dict | None = None def set_extra(self, key: str, value: Any) -> None: - self.metadata[key] = value + self.extras[key] = value def get_extra(self, key: str, default: Any = None) -> Any: - if key in self.metadata: - return self.metadata[key] - return default + return self.extras.get(key, default) def pop_extra(self, key: str, default: Any = None) -> Any: - if key in self.metadata: - value = self.metadata[key] - del self.metadata[key] - return value - return default - - def to_dict(self) -> Dict[str, Any]: - def _to_transport_value(value: Any) -> Any: - if value is None or isinstance(value, (str, int, float, bool)): - return value - if isinstance(value, Enum): - return value.value - if isinstance(value, dict): - return { - str(_to_transport_value(k)): _to_transport_value(v) for k, v in value.items() - } - if isinstance(value, (list, tuple, set)): - return [_to_transport_value(v) for v in value] - return str(value) - - result: Dict[str, Any] = { - MessageFields.EXECUTION_ID: self.execution_id, - MessageFields.PLATE_ID: _to_transport_value(self.plate_id), - MessageFields.CLIENT_ADDRESS: _to_transport_value(self.client_address), - MessageFields.STATUS: _to_transport_value(self.status), - MessageFields.START_TIME: _to_transport_value(self.start_time), - MessageFields.END_TIME: _to_transport_value(self.end_time), - MessageFields.ERROR: _to_transport_value(self.error), - MessageFields.COMPILE_ONLY: self.compile_only, - MessageFields.PROGRESS_SEQUENCE: self.progress_sequence, - } - if self.traceback is not None: - result[MessageFields.TRACEBACK] = _to_transport_value(self.traceback) - if self.results_summary is not None: - result[MessageFields.RESULTS_SUMMARY] = _to_transport_value(self.results_summary) - return result - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "ExecutionRecord": - known = { - MessageFields.EXECUTION_ID, - MessageFields.PLATE_ID, - MessageFields.CLIENT_ADDRESS, - MessageFields.STATUS, - MessageFields.START_TIME, - MessageFields.END_TIME, - MessageFields.ERROR, - MessageFields.TRACEBACK, - MessageFields.RESULTS_SUMMARY, - MessageFields.COMPILE_ONLY, - MessageFields.PROGRESS_SEQUENCE, - } - metadata = {k: v for k, v in data.items() if k not in known} - return cls( - execution_id=data[MessageFields.EXECUTION_ID], - plate_id=data[MessageFields.PLATE_ID], - client_address=data.get(MessageFields.CLIENT_ADDRESS), - status=data[MessageFields.STATUS], - start_time=data.get(MessageFields.START_TIME), - end_time=data.get(MessageFields.END_TIME), - error=data.get(MessageFields.ERROR), - traceback=data.get(MessageFields.TRACEBACK), - results_summary=data.get(MessageFields.RESULTS_SUMMARY), - compile_only=bool(data.get(MessageFields.COMPILE_ONLY, False)), - metadata=metadata, - progress_sequence=data.get(MessageFields.PROGRESS_SEQUENCE, 0), - ) + return self.extras.pop(key, default) @dataclass(frozen=True) -class RunningExecutionInfo: +class RunningExecutionInfo(WireMessage): execution_id: str - plate_id: str + subject_id: str start_time: float elapsed: float compile_only: bool = False - def to_dict(self) -> Dict[str, Any]: - return { - MessageFields.EXECUTION_ID: self.execution_id, - MessageFields.PLATE_ID: self.plate_id, - MessageFields.START_TIME: self.start_time, - MessageFields.ELAPSED: self.elapsed, - MessageFields.COMPILE_ONLY: self.compile_only, - } - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "RunningExecutionInfo": - return cls( - execution_id=data[MessageFields.EXECUTION_ID], - plate_id=data.get(MessageFields.PLATE_ID, "unknown"), - start_time=float(data.get(MessageFields.START_TIME) or 0.0), - elapsed=float(data.get(MessageFields.ELAPSED) or 0.0), - compile_only=bool(data.get(MessageFields.COMPILE_ONLY, False)), - ) - @dataclass(frozen=True) -class QueuedExecutionInfo: +class QueuedExecutionInfo(WireMessage): execution_id: str - plate_id: str + subject_id: str queue_position: int - def to_dict(self) -> Dict[str, Any]: - return { - MessageFields.EXECUTION_ID: self.execution_id, - MessageFields.PLATE_ID: self.plate_id, - MessageFields.QUEUE_POSITION: self.queue_position, - } - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "QueuedExecutionInfo": - return cls( - execution_id=data[MessageFields.EXECUTION_ID], - plate_id=data.get(MessageFields.PLATE_ID, "unknown"), - queue_position=int(data.get(MessageFields.QUEUE_POSITION) or 0), - ) +@dataclass(frozen=True) +class ExecutionStatusSnapshot(WireView): + """Status of one execution, or of the server's executions. + Read as a view: status responses also carry diagnostics owned by the + response envelope (``message``, ``error``). + """ -@dataclass(frozen=True) -class ExecutionStatusSnapshot: status: ResponseType execution: Optional[ExecutionRecord] = None active_executions: Optional[int] = None @@ -974,132 +822,9 @@ class ExecutionStatusSnapshot: running_executions: Optional[Tuple[RunningExecutionInfo, ...]] = None queued_executions: Optional[Tuple[QueuedExecutionInfo, ...]] = None - def to_dict(self) -> Dict[str, Any]: - result: Dict[str, Any] = {MessageFields.STATUS: self.status.value} - if self.execution is not None: - result[MessageFields.EXECUTION] = self.execution.to_dict() - if self.active_executions is not None: - result[MessageFields.ACTIVE_EXECUTIONS] = self.active_executions - if self.uptime is not None: - result[MessageFields.UPTIME] = self.uptime - if self.executions is not None: - result[MessageFields.EXECUTIONS] = list(self.executions) - if self.running_executions is not None: - result[MessageFields.RUNNING_EXECUTIONS] = [ - info.to_dict() for info in self.running_executions - ] - if self.queued_executions is not None: - result[MessageFields.QUEUED_EXECUTIONS] = [ - info.to_dict() for info in self.queued_executions - ] - return result - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "ExecutionStatusSnapshot": - execution = None - execution_data = data.get(MessageFields.EXECUTION) - if isinstance(execution_data, dict): - execution = ExecutionRecord.from_dict(execution_data) - - running_executions = None - running_data = data.get(MessageFields.RUNNING_EXECUTIONS) - if isinstance(running_data, list): - running_executions = tuple( - RunningExecutionInfo.from_dict(entry) for entry in running_data - ) - - queued_executions = None - queued_data = data.get(MessageFields.QUEUED_EXECUTIONS) - if isinstance(queued_data, list): - queued_executions = tuple(QueuedExecutionInfo.from_dict(entry) for entry in queued_data) - - executions = None - execution_ids = data.get(MessageFields.EXECUTIONS) - if isinstance(execution_ids, list): - executions = tuple(str(eid) for eid in execution_ids) - - return cls( - status=ResponseType(data[MessageFields.STATUS]), - execution=execution, - active_executions=data.get(MessageFields.ACTIVE_EXECUTIONS), - uptime=data.get(MessageFields.UPTIME), - executions=executions, - running_executions=running_executions, - queued_executions=queued_executions, - ) - - -@dataclass(frozen=True) -class StatusRequest: - execution_id: Optional[str] = None - - def to_dict(self): - result = {MessageFields.TYPE: ControlMessageType.STATUS.value} - if self.execution_id is not None: - result[MessageFields.EXECUTION_ID] = self.execution_id - return result - - @classmethod - def from_dict(cls, data): - return cls(execution_id=data.get(MessageFields.EXECUTION_ID)) - - -@dataclass(frozen=True) -class CancelRequest: - execution_id: str - - def validate(self): - return "Missing execution_id" if not self.execution_id else None - - def to_dict(self): - return { - MessageFields.TYPE: ControlMessageType.CANCEL.value, - MessageFields.EXECUTION_ID: self.execution_id, - } - - @classmethod - def from_dict(cls, data): - return cls(execution_id=data[MessageFields.EXECUTION_ID]) - - -@dataclass(frozen=True) -class ProgressRegistrationRequest: - client_id: str - - def validate(self): - return "Missing client_id" if not self.client_id else None - - def to_dict(self): - return { - MessageFields.TYPE: ControlMessageType.REGISTER_PROGRESS.value, - MessageFields.CLIENT_ID: self.client_id, - } - - @classmethod - def from_dict(cls, data): - return cls(client_id=data[MessageFields.CLIENT_ID]) - @dataclass(frozen=True) -class ProgressUnregistrationRequest: - client_id: str - - def validate(self): - return "Missing client_id" if not self.client_id else None - - def to_dict(self): - return { - MessageFields.TYPE: ControlMessageType.UNREGISTER_PROGRESS.value, - MessageFields.CLIENT_ID: self.client_id, - } - - @classmethod - def from_dict(cls, data): - return cls(client_id=data[MessageFields.CLIENT_ID]) - - -@dataclass(frozen=True) -class ProcessResourceUsage: +class ProcessResourceUsage(WireMessage): """Process resource metrics carried by a server heartbeat.""" memory_mb: float @@ -1136,7 +861,7 @@ class EndpointControlCapability(str, Enum): @dataclass(frozen=True) -class EndpointApplication: +class EndpointApplication(WireMessage): """Application identity declared by an endpoint owner.""" identifier: str @@ -1148,12 +873,6 @@ def __post_init__(self) -> None: if not self.version: raise ValueError("Endpoint application version cannot be empty") - def to_dict(self) -> Dict[str, str]: - return { - MessageFields.APPLICATION_ID: self.identifier, - MessageFields.APPLICATION_VERSION: self.version, - } - def compatibility_with( self, observed: "EndpointApplication | None", @@ -1162,14 +881,6 @@ def compatibility_with( return EndpointApplicationCompatibility(expected=self, observed=observed) - @classmethod - def from_dict(cls, data: Mapping[str, Any]) -> "EndpointApplication": - identifier = data[MessageFields.APPLICATION_ID] - version = data[MessageFields.APPLICATION_VERSION] - if not isinstance(identifier, str) or not isinstance(version, str): - raise TypeError("Endpoint application identity fields must be strings") - return cls(identifier=identifier, version=version) - @dataclass(frozen=True) class EndpointApplicationCompatibility: @@ -1214,16 +925,22 @@ def __str__(self) -> str: @dataclass(frozen=True) -class PongResponse(ControlResponse): - """Complete typed server-heartbeat response.""" +class PongResponse(TypedWireMessage, WireView, ControlResponse): + """Complete typed server-heartbeat response. + + Read as a view: a server may extend its heartbeat with fields of its own + (a subclass), and every client still reads the declared heartbeat. + """ + + wire_type: ClassVar[str] = "pong" port: int control_port: int ready: bool server: str server_role: ServerRole - control_capabilities: frozenset[EndpointControlCapability] = frozenset( - {EndpointControlCapability.PING} + control_capabilities: Tuple[EndpointControlCapability, ...] = ( + EndpointControlCapability.PING, ) server_type: Optional[str] = None log_file_path: Optional[str] = None @@ -1239,151 +956,20 @@ class PongResponse(ControlResponse): process_usage: Optional[ProcessResourceUsage] = None application: Optional[EndpointApplication] = None - def to_dict(self) -> Dict[str, Any]: - """Serialize for transport.""" - result: Dict[str, Any] = { - MessageFields.TYPE: ResponseType.PONG.value, - MessageFields.PORT: self.port, - MessageFields.CONTROL_PORT: self.control_port, - MessageFields.READY: self.ready, - MessageFields.SERVER: self.server, - } - if self.server_type is not None: - result[MessageFields.SERVER_TYPE] = self.server_type - result[MessageFields.SERVER_ROLE] = self.server_role.value - result[MessageFields.CONTROL_CAPABILITIES] = sorted( - capability.value for capability in self.control_capabilities - ) - if self.application is not None: - result[MessageFields.APPLICATION] = self.application.to_dict() - if self.log_file_path is not None: - result[MessageFields.LOG_FILE_PATH] = self.log_file_path - if self.active_executions is not None: - result[MessageFields.ACTIVE_EXECUTIONS] = self.active_executions - if self.running_executions is not None: - result[MessageFields.RUNNING_EXECUTIONS] = [ - info.to_dict() for info in self.running_executions - ] - if self.queued_executions is not None: - result[MessageFields.QUEUED_EXECUTIONS] = [ - info.to_dict() for info in self.queued_executions - ] - if self.workers is not None: - result[MessageFields.WORKERS] = [w.to_dict() for w in self.workers] - if self.uptime is not None: - result[MessageFields.UPTIME] = self.uptime - if self.progress_subscribers is not None: - result[MessageFields.PROGRESS_SUBSCRIBERS] = self.progress_subscribers - if self.process_identity is not None: - result[MessageFields.PROCESS_IDENTITY] = self.process_identity.to_dict() - if self.compile_status is not None: - result[MessageFields.COMPILE_STATUS] = self.compile_status - if self.compile_message is not None: - result[MessageFields.COMPILE_MESSAGE] = self.compile_message - if self.process_usage is not None: - result[MessageFields.MEMORY_MB] = self.process_usage.memory_mb - result[MessageFields.CPU_PERCENT] = self.process_usage.cpu_percent - return result - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "PongResponse": - """Parse from transport.""" - if data[MessageFields.TYPE] != ResponseType.PONG.value: - raise ValueError( - f"PongResponse requires {MessageFields.TYPE}={ResponseType.PONG.value!r}." - ) - - workers_data = data.get(MessageFields.WORKERS) - workers = None - if workers_data is not None: - if not isinstance(workers_data, list) or not all( - isinstance(entry, dict) for entry in workers_data - ): - raise TypeError("PongResponse.workers must be a list of dict entries") - workers = tuple(WorkerState.from_dict(w) for w in workers_data) - - running_executions_data = data.get(MessageFields.RUNNING_EXECUTIONS) - running_executions = None - if running_executions_data is not None: - if not isinstance(running_executions_data, list): - raise TypeError("PongResponse.running_executions must be a list of dict entries") - if not all(isinstance(entry, dict) for entry in running_executions_data): - raise TypeError("PongResponse.running_executions must be a list of dict entries") - running_executions = tuple( - RunningExecutionInfo.from_dict(entry) for entry in running_executions_data - ) - - queued_executions_data = data.get(MessageFields.QUEUED_EXECUTIONS) - queued_executions = None - if queued_executions_data is not None: - if not isinstance(queued_executions_data, list): - raise TypeError("PongResponse.queued_executions must be a list of dict entries") - if not all(isinstance(entry, dict) for entry in queued_executions_data): - raise TypeError("PongResponse.queued_executions must be a list of dict entries") - queued_executions = tuple( - QueuedExecutionInfo.from_dict(entry) for entry in queued_executions_data - ) - - process_identity_data = data.get(MessageFields.PROCESS_IDENTITY) - process_identity = None - if process_identity_data is not None: - if not isinstance(process_identity_data, dict): - raise TypeError("PongResponse.process_identity must be a dict") - process_identity = ProcessIdentity.from_dict(process_identity_data) + def __post_init__(self) -> None: + """Hold advertised capabilities as one ordered set.""" - control_capabilities_data = data.get( - MessageFields.CONTROL_CAPABILITIES, - [EndpointControlCapability.PING.value], - ) - if not isinstance(control_capabilities_data, list) or not all( - isinstance(capability, str) for capability in control_capabilities_data - ): - raise TypeError("PongResponse.control_capabilities must be a list of strings") - control_capabilities = frozenset( - EndpointControlCapability(capability) for capability in control_capabilities_data + object.__setattr__( + self, + "control_capabilities", + _capability_set(self.control_capabilities), ) - application_data = data.get(MessageFields.APPLICATION) - application = None - if application_data is not None: - if not isinstance(application_data, dict): - raise TypeError("PongResponse.application must be a dict") - application = EndpointApplication.from_dict(application_data) - - memory_mb = data.get(MessageFields.MEMORY_MB) - cpu_percent = data.get(MessageFields.CPU_PERCENT) - process_usage = None - if memory_mb is not None or cpu_percent is not None: - if memory_mb is None or cpu_percent is None: - raise TypeError( - "PongResponse process usage requires both memory_mb and cpu_percent" - ) - process_usage = ProcessResourceUsage( - memory_mb=float(memory_mb), - cpu_percent=float(cpu_percent), - ) - return cls( - port=data[MessageFields.PORT], - control_port=data[MessageFields.CONTROL_PORT], - ready=data[MessageFields.READY], - server=data[MessageFields.SERVER], - server_type=data.get(MessageFields.SERVER_TYPE), - server_role=ServerRole(data[MessageFields.SERVER_ROLE]), - control_capabilities=control_capabilities, - application=application, - log_file_path=data.get(MessageFields.LOG_FILE_PATH), - active_executions=data.get(MessageFields.ACTIVE_EXECUTIONS), - running_executions=running_executions, - queued_executions=queued_executions, - workers=workers, - uptime=data.get(MessageFields.UPTIME), - progress_subscribers=data.get(MessageFields.PROGRESS_SUBSCRIBERS), - process_identity=process_identity, - compile_status=data.get(MessageFields.COMPILE_STATUS), - compile_message=data.get(MessageFields.COMPILE_MESSAGE), - process_usage=process_usage, - ) +def _capability_set( + capabilities: Iterable[EndpointControlCapability], +) -> Tuple[EndpointControlCapability, ...]: + return tuple(sorted(set(capabilities), key=lambda capability: capability.value)) # ============================================================================= @@ -1392,7 +978,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "PongResponse": @dataclass(frozen=True, slots=True) -class AckReturnRoute: +class AckReturnRoute(WireMessage): """Bound producer destination and exact listener incarnation, not a viewer port.""" url: str @@ -1406,16 +992,9 @@ def __post_init__(self) -> None: if not isinstance(self.owner, ProcessIdentity): raise TypeError("ACK route requires a ProcessIdentity") - def to_dict(self) -> dict[str, Any]: - return asdict(self) - - @classmethod - def from_dict(cls, data: Mapping[str, Any]) -> "AckReturnRoute": - return dataclass_from_mapping(cls, data) - @dataclass(frozen=True, slots=True) -class ImageTransferIdentity: +class ImageTransferIdentity(WireMessage): """Per-image return contract carried unchanged through deferred viewer work.""" image_id: str @@ -1428,29 +1007,24 @@ def __post_init__(self) -> None: if not isinstance(self.return_route, AckReturnRoute): raise TypeError("Image transfer requires an AckReturnRoute") - def to_dict(self) -> dict[str, Any]: - return asdict(self) - - @classmethod - def from_dict(cls, data: Mapping[str, Any]) -> "ImageTransferIdentity": - return dataclass_from_mapping(cls, {member.name: data[member.name] for member in fields(cls)}) - @classmethod def from_item(cls, data: Mapping[str, Any]) -> "ImageTransferIdentity | None": """Untracked control items have no image identity; tracked items require a route.""" if data.get("image_id") is None: return None - return cls.from_dict(data) + return cls.from_dict({declared.name: data[declared.name] for declared in fields(cls)}) @dataclass(frozen=True) -class ImageAck: +class ImageAck(TypedWireMessage): """Acknowledgment message sent by viewers after processing an image. Sent via PUSH socket to the original image's producer return route. Used to track real-time queue depth and show progress like '3/10 images processed'. """ + wire_type: ClassVar[str] = "image_ack" + image_id: str # UUID of the processed image viewer_port: int # Port of the viewer that processed it (for routing) viewer_type: str # 'napari' or 'fiji' @@ -1459,62 +1033,3 @@ class ImageAck: status: str = "success" # 'success', 'error', etc. timestamp: Optional[float] = None # When it was processed error: Optional[str] = None # Error message if status='error' - - def to_dict(self): - return {MessageFields.TYPE: "image_ack", **asdict(self)} - - @classmethod - def from_dict(cls, data): - if data.get(MessageFields.TYPE) != "image_ack": - raise ValueError("ImageAck requires type=image_ack") - return dataclass_from_mapping(cls, { - member.name: data[member.name] for member in fields(cls) if member.name in data - }) - - -@dataclass(frozen=True) -class ROIMessage: - """Message for streaming ROIs to viewers (Napari/Fiji). - - Sent via ZMQ to viewer servers to display ROIs in real-time. - """ - - rois: list # List of ROI dictionaries with shapes and metadata - layer_name: str = "ROIs" # Name of the layer/overlay - - def to_dict(self): - return { - MessageFields.TYPE: "rois", - MessageFields.ROIS: self.rois, - MessageFields.LAYER_NAME: self.layer_name, - } - - @classmethod - def from_dict(cls, data): - return cls( - rois=data[MessageFields.ROIS], layer_name=data.get(MessageFields.LAYER_NAME, "ROIs") - ) - - -@dataclass(frozen=True) -class ShapesMessage: - """Message for Napari shapes layer. - - Napari-specific format for displaying polygon/ellipse shapes. - """ - - shapes: list # List of shape dictionaries with type, coordinates, metadata - layer_name: str = "ROIs" - - def to_dict(self): - return { - MessageFields.TYPE: "shapes", - MessageFields.SHAPES: self.shapes, - MessageFields.LAYER_NAME: self.layer_name, - } - - @classmethod - def from_dict(cls, data): - return cls( - shapes=data[MessageFields.SHAPES], layer_name=data.get(MessageFields.LAYER_NAME, "ROIs") - ) diff --git a/src/zmqruntime/process_launch.py b/src/zmqruntime/process_launch.py new file mode 100644 index 0000000..8e1744e --- /dev/null +++ b/src/zmqruntime/process_launch.py @@ -0,0 +1,132 @@ +"""Platform policy for noninteractive background subprocesses. + +Each host family is one :class:`BackgroundProcessPlatform` subclass that owns +how a background process is launched there: console suppression, detached +process groups and the console-free interpreter. The module imports only the +standard library and ``metaclass_registry``, so early startup paths can use it +before any GUI or runtime machinery loads. +""" + +from __future__ import annotations + +import subprocess +import sys +from abc import ABC, abstractmethod +from dataclasses import dataclass +from pathlib import Path +from typing import ClassVar + +from metaclass_registry import AutoRegisterMeta + + +@dataclass(frozen=True, slots=True) +class BackgroundProcessLaunchSpec: + """Resolved keyword arguments for one subprocess launch.""" + + creationflags: int = 0 + start_new_session: bool = False + + def popen_arguments(self) -> dict[str, bool | int]: + """Return only non-default ``subprocess.Popen`` arguments.""" + + arguments: dict[str, bool | int] = {} + if self.creationflags: + arguments["creationflags"] = self.creationflags + if self.start_new_session: + arguments["start_new_session"] = True + return arguments + + +class BackgroundProcessPlatform(ABC, metaclass=AutoRegisterMeta): + """How background processes launch on one host family. + + Families register under the ``sys.platform`` value they own; a host with + no registered family launches as :class:`SessionBackgroundProcesses`. + """ + + __registry_key__ = "sys_platform" + __skip_if_no_key__ = True + + sys_platform: ClassVar[str | None] = None + + @classmethod + def current(cls) -> type[BackgroundProcessPlatform]: + """Return the family of the running host.""" + + return cls.__registry__.get(sys.platform, SessionBackgroundProcesses) + + @classmethod + @abstractmethod + def launch_spec(cls, *, detached: bool) -> BackgroundProcessLaunchSpec: + """Launch arguments for a background process, detached or attached.""" + + @classmethod + def python_executable(cls, executable: str) -> str: + """The interpreter for a background Python process family.""" + + return executable + + +class SessionBackgroundProcesses(BackgroundProcessPlatform): + """Hosts without consoles per process: detaching starts a new session.""" + + @classmethod + def launch_spec(cls, *, detached: bool) -> BackgroundProcessLaunchSpec: + return BackgroundProcessLaunchSpec(start_new_session=detached) + + +class WindowsBackgroundProcesses(BackgroundProcessPlatform): + """Windows: suppress consoles, and detach into a new process group.""" + + sys_platform = "win32" + + @classmethod + def launch_spec(cls, *, detached: bool) -> BackgroundProcessLaunchSpec: + creationflags = subprocess.CREATE_NO_WINDOW + if detached: + creationflags |= subprocess.CREATE_NEW_PROCESS_GROUP + return BackgroundProcessLaunchSpec(creationflags=creationflags) + + @classmethod + def python_executable(cls, executable: str) -> str: + """Use the colocated ``pythonw.exe`` when it exists. + + Creation flags suppress the first child console, but a child launched + through ``python.exe`` can still create visible consoles when it + starts multiprocessing descendants. ``pythonw.exe`` makes the + console-free interpreter identity transitive. + """ + + windowed_executable = Path(executable).with_name("pythonw.exe") + if windowed_executable.is_file(): + return str(windowed_executable) + return executable + + +@dataclass(frozen=True, slots=True) +class BackgroundProcessLaunchPolicy: + """Launch policy for one background process on one host family.""" + + platform: type[BackgroundProcessPlatform] + detached: bool = False + + @classmethod + def current(cls, *, detached: bool = False) -> BackgroundProcessLaunchPolicy: + """Construct the policy for the current host.""" + + return cls(platform=BackgroundProcessPlatform.current(), detached=detached) + + def resolve(self) -> BackgroundProcessLaunchSpec: + """Resolve the platform policy to a transportable launch spec.""" + + return self.platform.launch_spec(detached=self.detached) + + def popen_arguments(self) -> dict[str, bool | int]: + """Resolve directly to ``subprocess.Popen`` keyword arguments.""" + + return self.resolve().popen_arguments() + + def python_executable(self, executable: str) -> str: + """Return the interpreter for a background Python process family.""" + + return self.platform.python_executable(executable) diff --git a/src/zmqruntime/progress/__init__.py b/src/zmqruntime/progress/__init__.py index 575eb03..f0bbf5d 100644 --- a/src/zmqruntime/progress/__init__.py +++ b/src/zmqruntime/progress/__init__.py @@ -3,7 +3,7 @@ from .projection import ( GenericAxisProjection, GenericExecutionProjection, - GenericPlateProjection, + GenericSubjectProjection, ProgressProjectionAdapterABC, build_execution_projection, ) @@ -20,7 +20,7 @@ "EventRegistryMutationKind", "LatestEventRegistry", "GenericAxisProjection", - "GenericPlateProjection", + "GenericSubjectProjection", "GenericExecutionProjection", "ProgressProjectionAdapterABC", "build_execution_projection", diff --git a/src/zmqruntime/progress/projection.py b/src/zmqruntime/progress/projection.py index a5fa2c1..c9b754f 100644 --- a/src/zmqruntime/progress/projection.py +++ b/src/zmqruntime/progress/projection.py @@ -23,11 +23,11 @@ class GenericAxisProjection: @dataclass(frozen=True) -class GenericPlateProjection(Generic[TState]): - """One plate projection inside one execution snapshot.""" +class GenericSubjectProjection(Generic[TState]): + """One subject projection inside one execution snapshot.""" execution_id: str - plate_id: str + subject_id: str state: TState percent: float axis_progress: Tuple[GenericAxisProjection, ...] @@ -44,24 +44,24 @@ def active_axes(self) -> Tuple[GenericAxisProjection, ...]: @dataclass class GenericExecutionProjection(Generic[TState]): - """Cross-execution projection with de-duplicated latest plate view.""" + """Cross-execution projection with de-duplicated latest view per subject.""" - plates: List[GenericPlateProjection[TState]] = field(default_factory=list) - by_key: Dict[Tuple[str, str], GenericPlateProjection[TState]] = field( + subjects: List[GenericSubjectProjection[TState]] = field(default_factory=list) + by_key: Dict[Tuple[str, str], GenericSubjectProjection[TState]] = field( default_factory=dict ) - by_plate_latest: Dict[str, GenericPlateProjection[TState]] = field( + by_subject_latest: Dict[str, GenericSubjectProjection[TState]] = field( default_factory=dict ) state_counts: Dict[TState, int] = field(default_factory=dict) overall_percent: float = 0.0 - def get_plate( - self, plate_id: str, execution_id: Optional[str] = None - ) -> Optional[GenericPlateProjection[TState]]: + def get_subject( + self, subject_id: str, execution_id: Optional[str] = None + ) -> Optional[GenericSubjectProjection[TState]]: if execution_id is not None: - return self.by_key.get((execution_id, plate_id)) - return self.by_plate_latest.get(plate_id) + return self.by_key.get((execution_id, subject_id)) + return self.by_subject_latest.get(subject_id) def count_state(self, state: TState) -> int: return self.state_counts.get(state, 0) @@ -79,8 +79,8 @@ def pipeline_channel(self) -> str: return "pipeline" @abstractmethod - def plate_id(self, event: TEvent) -> str: - """Return plate id.""" + def subject_id(self, event: TEvent) -> str: + """Return the id of the subject this event belongs to.""" @abstractmethod def axis_id(self, event: TEvent) -> str: @@ -104,7 +104,7 @@ def channel(self, event: TEvent) -> str: @abstractmethod def known_axes(self, events: Sequence[TEvent]) -> Sequence[str]: - """Return declared axis universe for plate events when available.""" + """Return declared axis universe for one subject's events when available.""" @abstractmethod def is_failure_event(self, event: TEvent) -> bool: @@ -162,12 +162,12 @@ def _latest_by_axis( return latest -def _build_plate_projection( +def _build_subject_projection( execution_id: str, - plate_id: str, + subject_id: str, events: List[TEvent], adapter: ProgressProjectionAdapterABC[TEvent, TState], -) -> GenericPlateProjection[TState]: +) -> GenericSubjectProjection[TState]: latest_timestamp = max((adapter.timestamp(event) for event in events), default=0.0) known_axes = list(adapter.known_axes(events)) pipeline_by_axis = _latest_by_axis( @@ -232,9 +232,9 @@ def _build_plate_projection( else: state = adapter.state_executing() - return GenericPlateProjection( + return GenericSubjectProjection( execution_id=execution_id, - plate_id=plate_id, + subject_id=subject_id, state=state, percent=percent, axis_progress=tuple(axis_progress), @@ -296,18 +296,18 @@ def _build_plate_projection( ) for axis_id in axis_universe ) - return GenericPlateProjection( + return GenericSubjectProjection( execution_id=execution_id, - plate_id=plate_id, + subject_id=subject_id, state=state, percent=percent, axis_progress=axis_progress, latest_timestamp=latest_timestamp, ) - return GenericPlateProjection( + return GenericSubjectProjection( execution_id=execution_id, - plate_id=plate_id, + subject_id=subject_id, state=adapter.state_idle_from_events(events), percent=0.0, axis_progress=tuple(), @@ -323,32 +323,32 @@ def build_execution_projection( grouped: Dict[Tuple[str, str], List[TEvent]] = {} for execution_id, events in events_by_execution.items(): for event in events: - grouped.setdefault((execution_id, adapter.plate_id(event)), []).append(event) + grouped.setdefault((execution_id, adapter.subject_id(event)), []).append(event) projection: GenericExecutionProjection[TState] = GenericExecutionProjection() - for (execution_id, plate_id), plate_events in grouped.items(): - plate_projection = _build_plate_projection( + for (execution_id, subject_id), subject_events in grouped.items(): + subject_projection = _build_subject_projection( execution_id, - plate_id, - plate_events, + subject_id, + subject_events, adapter, ) - projection.plates.append(plate_projection) - projection.by_key[(execution_id, plate_id)] = plate_projection + projection.subjects.append(subject_projection) + projection.by_key[(execution_id, subject_id)] = subject_projection - for plate in projection.plates: - existing = projection.by_plate_latest.get(plate.plate_id) - if existing is None or plate.latest_timestamp > existing.latest_timestamp: - projection.by_plate_latest[plate.plate_id] = plate + for subject in projection.subjects: + existing = projection.by_subject_latest.get(subject.subject_id) + if existing is None or subject.latest_timestamp > existing.latest_timestamp: + projection.by_subject_latest[subject.subject_id] = subject - for plate in projection.by_plate_latest.values(): - projection.state_counts[plate.state] = projection.state_counts.get(plate.state, 0) + 1 + for subject in projection.by_subject_latest.values(): + projection.state_counts[subject.state] = projection.state_counts.get(subject.state, 0) + 1 - if projection.by_plate_latest: + if projection.by_subject_latest: projection.overall_percent = ( - sum(plate.percent for plate in projection.by_plate_latest.values()) - / len(projection.by_plate_latest) + sum(subject.percent for subject in projection.by_subject_latest.values()) + / len(projection.by_subject_latest) ) return projection diff --git a/tests/test_ack_return_route.py b/tests/test_ack_return_route.py index 81468f7..cbbb63f 100644 --- a/tests/test_ack_return_route.py +++ b/tests/test_ack_return_route.py @@ -146,7 +146,7 @@ def test_stale_incarnation_unknown_late_and_explicit_worker_accounting(tmp_path) def test_required_return_route_has_no_legacy_reader(): - with pytest.raises(KeyError, match="return_route"): + with pytest.raises(ValueError, match="return_route"): ImageTransferIdentity.from_dict({"image_id": "missing-route"}) with pytest.raises(ValueError, match="return_route"): ImageAck.from_dict(dict(type="image_ack", image_id="missing", viewer_port=1, diff --git a/tests/test_execution.py b/tests/test_execution.py index 39cf263..0b2661e 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -1,6 +1,5 @@ import json import multiprocessing -import pickle import platform import subprocess import sys @@ -32,6 +31,7 @@ ) from zmqruntime.execution.wait_policy import ExecutionWaiter, WaitPolicy from zmqruntime.messages import ( + CancelRequest, ControlMessageType, EndpointApplication, EndpointControlCapability, @@ -40,8 +40,11 @@ MessageFields, PongResponse, ProcessIdentity, + ProgressRegistrationRequest, + ProgressSubscriptionResponse, ResponseType, ServerRole, + StatusRequest, TaskProgress, ) from zmqruntime.transport import ( @@ -367,7 +370,7 @@ def test_shutdown_rejects_endpoint_without_advertised_capability(monkeypatch): def test_execution_server_handle_execute_and_run(): server = DummyExecutionServer(port=5555) request = ExecuteRequest( - plate_id="plate-1", + subject_id="subject-1", pipeline_code="print('hi')", config_params={"x": 1}, ) @@ -377,10 +380,10 @@ def test_execution_server_handle_execute_and_run(): record = server.active_executions[execution_id] assert record.status == ExecutionStatus.QUEUED.value - server._run_execution(execution_id, request, record) + server.run_execution(execution_id, request, record) assert record.status == ExecutionStatus.COMPLETE.value - status_response = server._handle_status( + status_response = server.handle_status( { MessageFields.TYPE: ControlMessageType.STATUS.value, MessageFields.EXECUTION_ID: execution_id, @@ -390,42 +393,10 @@ def test_execution_server_handle_execute_and_run(): assert status_response[MessageFields.EXECUTION][MessageFields.EXECUTION_ID] == execution_id -def test_execution_status_response_is_picklable_with_non_transport_values(): - server = DummyExecutionServer(port=5555) - request = ExecuteRequest( - plate_id="plate-1", - pipeline_code="print('hi')", - config_params={"x": 1}, - ) - response = server._handle_execute(request.to_dict()) - execution_id = response[MessageFields.EXECUTION_ID] - record = server.active_executions[execution_id] - - class Unserializable: - pass - - record.client_address = Unserializable - record.results_summary = {"raw": Unserializable, "nested": [Unserializable]} - - status_response = server._handle_status( - { - MessageFields.TYPE: ControlMessageType.STATUS.value, - MessageFields.EXECUTION_ID: execution_id, - } - ) - encoded = pickle.dumps(status_response) - decoded = pickle.loads(encoded) - - execution = decoded[MessageFields.EXECUTION] - assert isinstance(execution[MessageFields.CLIENT_ADDRESS], str) - assert isinstance(execution[MessageFields.RESULTS_SUMMARY]["raw"], str) - assert isinstance(execution[MessageFields.RESULTS_SUMMARY]["nested"][0], str) - - def test_failed_execution_exposes_traceback_field(): server = FailingExecutionServer(port=5555) request = ExecuteRequest( - plate_id="plate-1", + subject_id="subject-1", pipeline_code="print('hi')", config_params={"x": 1}, ) @@ -433,9 +404,9 @@ def test_failed_execution_exposes_traceback_field(): execution_id = response[MessageFields.EXECUTION_ID] record = server.active_executions[execution_id] - server._run_execution(execution_id, request, record) + server.run_execution(execution_id, request, record) - status_response = server._handle_status( + status_response = server.handle_status( { MessageFields.TYPE: ControlMessageType.STATUS.value, MessageFields.EXECUTION_ID: execution_id, @@ -449,7 +420,7 @@ def test_failed_execution_exposes_traceback_field(): def test_cancelled_execution_is_terminal_when_running_task_returns(): server = BlockingExecutionServer(port=5555) request = ExecuteRequest( - plate_id="plate-1", + subject_id="subject-1", pipeline_code="print('hi')", config_params={"x": 1}, ) @@ -482,7 +453,7 @@ def test_cancelled_execution_is_terminal_when_running_task_returns(): def test_cancel_targets_queued_execution_without_interrupting_running_execution(): server = BlockingExecutionServer(port=5555) request = ExecuteRequest( - plate_id="plate-1", + subject_id="subject-1", pipeline_code="print('hi')", config_params={"x": 1}, ) @@ -506,7 +477,7 @@ def test_cancel_targets_queued_execution_without_interrupting_running_execution( def test_cancel_rejects_terminal_execution_without_rewriting_status(): server = DummyExecutionServer(port=5555) request = ExecuteRequest( - plate_id="plate-1", + subject_id="subject-1", pipeline_code="print('hi')", config_params={"x": 1}, ) @@ -526,9 +497,9 @@ def test_cancel_rejects_terminal_execution_without_rewriting_status(): assert record.status == ExecutionStatus.COMPLETE.value -def test_task_progress_roundtrip_supports_execution_id_and_task_id(): +def test_task_progress_reads_its_fields_out_of_an_application_event(): progress = TaskProgress( - task_id="exec-123", + execution_id="exec-123", phase="running", status="running", percent=10.0, @@ -538,13 +509,10 @@ def test_task_progress_roundtrip_supports_execution_id_and_task_id(): ) payload = progress.to_dict() - parsed = TaskProgress.from_dict(payload) - assert parsed.task_id == "exec-123" - - legacy_payload = dict(payload) - del legacy_payload["execution_id"] - legacy_parsed = TaskProgress.from_dict(legacy_payload) - assert legacy_parsed.task_id == "exec-123" + assert TaskProgress.from_dict(payload) == progress + assert TaskProgress.from_dict({**payload, "application_field": [1, 2]}) == progress + with pytest.raises(ValueError): + TaskProgress.from_dict({key: value for key, value in payload.items() if key != "phase"}) class DummyExecutionClient(ExecutionClient): @@ -559,7 +527,7 @@ def send_data(self, data): return None def serialize_task(self, task, config): - return {"task": task} + return ExecuteRequest(subject_id="subject-1", pipeline_code=str(task), config_code="") def connect(self, timeout: float = 10.0): self._connection = attached_connection(self) @@ -567,17 +535,18 @@ def connect(self, timeout: float = 10.0): def _send_control_request(self, request, timeout_ms=5000): if request[MessageFields.TYPE] == ControlMessageType.REGISTER_PROGRESS.value: - return { - MessageFields.STATUS: ResponseType.OK.value, - MessageFields.PROGRESS_OBSERVATIONS: {}, - } + return ProgressSubscriptionResponse( + client_id=request[MessageFields.CLIENT_ID], + progress_subscribers=1, + message="registered", + ).to_dict() return request -def test_execution_client_submit_adds_type(): +def test_execution_client_submits_the_declared_execute_request(): client = DummyExecutionClient() response = client.submit_execution({"hello": "world"}) - assert response[MessageFields.TYPE] == ControlMessageType.EXECUTE.value + assert ExecuteRequest.from_dict(response) == client.serialize_task({"hello": "world"}, None) def test_execution_server_progress_registration_roundtrip(): @@ -621,14 +590,9 @@ def _start_progress_listener(self): def _send_control_request(self, request, timeout_ms=5000): self.sent_requests.append((request, timeout_ms)) - if request.get(MessageFields.TYPE) == ControlMessageType.REGISTER_PROGRESS.value: - return { - MessageFields.STATUS: ResponseType.OK.value, - MessageFields.PROGRESS_OBSERVATIONS: {}, - } if request.get(MessageFields.TYPE) == ControlMessageType.UNREGISTER_PROGRESS.value: return {MessageFields.STATUS: ResponseType.OK.value} - return request + return super()._send_control_request(request, timeout_ms) def test_execution_client_registers_progress_before_execute(): @@ -637,17 +601,11 @@ def test_execution_client_registers_progress_before_execute(): assert response[MessageFields.TYPE] == ControlMessageType.EXECUTE.value assert client.listener_started is True assert client.sent_requests[0] == ( - { - MessageFields.TYPE: ControlMessageType.REGISTER_PROGRESS.value, - MessageFields.CLIENT_ID: client._progress_client_id, - }, + ProgressRegistrationRequest(client_id=client._progress_client_id).to_dict(), 15000, ) assert client.sent_requests[1] == ( - { - "task": {"hello": "world"}, - MessageFields.TYPE: ControlMessageType.EXECUTE.value, - }, + client.serialize_task({"hello": "world"}, None).to_dict(), 15000, ) @@ -661,7 +619,7 @@ def test_execution_client_registers_progress_before_execute(): def test_execution_client_retains_immutable_progress_observations_per_execution(): client = ProgressAwareExecutionClient() first = TaskProgress( - task_id="execution-1", + execution_id="execution-1", phase="compile", status="running", percent=10.0, @@ -670,7 +628,7 @@ def test_execution_client_retains_immutable_progress_observations_per_execution( total=10, ).to_dict() second = TaskProgress( - task_id="execution-1", + execution_id="execution-1", phase="execute", status="running", percent=20.0, @@ -680,7 +638,7 @@ def test_execution_client_retains_immutable_progress_observations_per_execution( ).to_dict() second["nested"] = {"values": [1, 2]} other = TaskProgress( - task_id="execution-2", + execution_id="execution-2", phase="compile", status="running", percent=50.0, @@ -808,7 +766,7 @@ def send_data(self, data): return None def serialize_task(self, task, config): - return {"task": task} + return ExecuteRequest(subject_id="subject-1", pipeline_code=str(task), config_code="") def test_ipc_connect_preserves_unresponsive_live_server_endpoint(monkeypatch): @@ -1267,7 +1225,7 @@ def send_data(self, data): return None def serialize_task(self, task, config): - return {"task": task} + return ExecuteRequest(subject_id="subject-1", pipeline_code=str(task), config_code="") @pytest.mark.parametrize("transport_mode", (TransportMode.TCP, TransportMode.IPC)) @@ -1389,7 +1347,7 @@ def poll_status(_execution_id): MessageFields.STATUS: ResponseType.OK.value, MessageFields.EXECUTION: { MessageFields.EXECUTION_ID: "compile-1", - MessageFields.PLATE_ID: "plate-1", + MessageFields.SUBJECT_ID: "subject-1", MessageFields.STATUS: ExecutionStatus.COMPLETE.value, }, } @@ -1423,7 +1381,7 @@ def poll_status(_execution_id): MessageFields.STATUS: ResponseType.OK.value, MessageFields.EXECUTION: { MessageFields.EXECUTION_ID: "compile-1", - MessageFields.PLATE_ID: "plate-1", + MessageFields.SUBJECT_ID: "subject-1", MessageFields.STATUS: ExecutionStatus.COMPLETE.value, }, } @@ -1485,7 +1443,7 @@ def poll_status(_execution_id, *, timeout_ms): MessageFields.STATUS: ResponseType.OK.value, MessageFields.EXECUTION: { MessageFields.EXECUTION_ID: "compile-1", - MessageFields.PLATE_ID: "plate-1", + MessageFields.SUBJECT_ID: "subject-1", MessageFields.STATUS: ExecutionStatus.COMPLETE.value, }, } @@ -1521,10 +1479,10 @@ def test_submission_response_requires_explicit_tracking_and_diagnostics(): { MessageFields.STATUS: ResponseType.ERROR.value, MessageFields.MESSAGE: "bad request", - MessageFields.ERROR: "missing plate", + MessageFields.ERROR: "missing subject", } ) - assert failed_with_both.require_failure_text("submission") == ("bad request (missing plate)") + assert failed_with_both.require_failure_text("submission") == ("bad request (missing subject)") def test_waiter_retains_the_terminal_record_without_another_status_poll(): @@ -1532,11 +1490,11 @@ def test_waiter_retains_the_terminal_record_without_another_status_poll(): record = { "execution_id": "retained-1", - "plate_id": "/plate", + "subject_id": "/subject", "status": "complete", "start_time": 2.0, "end_time": 5.0, - "results_summary": {"well_count": 1}, + "results_summary": {"partition_count": 1}, } requests = [] @@ -1567,7 +1525,7 @@ def __setattr__(self, name, value): ) record = ObservedRecord( - "terminal-1", "/plate", None, ExecutionStatus.RUNNING.value, + "terminal-1", "/subject", ExecutionStatus.RUNNING.value, start_time=2.0, ) ExecutionStatus.COMPLETE.apply_to_record(record, timestamp=5.0) @@ -1587,12 +1545,12 @@ def test_terminal_status_waits_for_actual_progress_callback_tail(terminal_status from zmqruntime.subscription import CallbackSubscription server = DummyExecutionServer(port=5555) - record = ExecutionRecord("tail-1", "plate-1", None, ExecutionStatus.RUNNING.value) + record = ExecutionRecord("tail-1", "subject-1", ExecutionStatus.RUNNING.value) server.active_executions[record.execution_id] = record for number in (1, 2): server.send_progress_update( TaskProgress( - task_id=record.execution_id, + execution_id=record.execution_id, phase="axis_completed" if number == 2 else "step_completed", status="success", percent=number * 50.0, @@ -1666,10 +1624,10 @@ def test_progress_registration_baseline_and_loss_have_explicit_delivery_laws(): from zmqruntime.messages import ExecutionRecord server = DummyExecutionServer(port=5555) - record = ExecutionRecord("late-1", "plate-1", None, ExecutionStatus.RUNNING.value) + record = ExecutionRecord("late-1", "subject-1", ExecutionStatus.RUNNING.value) server.active_executions[record.execution_id] = record payload = TaskProgress( - task_id=record.execution_id, + execution_id=record.execution_id, phase="running", status="running", percent=50.0, @@ -1733,7 +1691,7 @@ def fail(_event): client.progress_callback = fail event = TaskProgress( - task_id="callback-1", + execution_id="callback-1", phase="axis_completed", status="success", percent=100.0, @@ -1744,7 +1702,7 @@ def fail(_event): event[MessageFields.PROGRESS_SEQUENCE] = 1 assert not client._progress_stream._dispatch_message(json.dumps(event)) record = ExecutionRecord( - "callback-1", "plate-1", None, ExecutionStatus.COMPLETE.value, progress_sequence=1 + "callback-1", "subject-1", ExecutionStatus.COMPLETE.value, progress_sequence=1 ) client.poll_status = lambda _execution_id, timeout_ms: { MessageFields.STATUS: ResponseType.OK.value, @@ -1760,12 +1718,12 @@ def test_terminal_notification_contains_finalized_summary_and_wakes_without_stat class EnrichedServer(DummyExecutionServer): def finalize_execution_record(self, record): - assert record.results_summary == {"well_count": 1, "wells": ["result"]} + assert record.results_summary == {"partition_count": 1, "partition_values": ["result"]} assert record.start_time is not None and record.end_time is not None record.results_summary["output_root"] = "/actual-output" server = EnrichedServer(port=5555) - request = ExecuteRequest(plate_id="plate-1", pipeline_code="pass", config_params={"x": 1}) + request = ExecuteRequest(subject_id="subject-1", pipeline_code="pass", config_params={"x": 1}) response = server._handle_execute(request.to_dict()) execution_id = response[MessageFields.EXECUTION_ID] server.run_execution(execution_id, request, server.active_executions[execution_id]) @@ -1794,7 +1752,7 @@ def finalize_execution_record(self, record): record.results_summary["output_root"] = "/final" server = EnrichedServer(port=5555) - request = ExecuteRequest(plate_id="plate-1", pipeline_code="pass", config_params={"x": 1}) + request = ExecuteRequest(subject_id="subject-1", pipeline_code="pass", config_params={"x": 1}) execution_id = server._handle_execute(request.to_dict())["execution_id"] runner = threading.Thread(target=server.run_execution, args=(execution_id, request, server.active_executions[execution_id])) runner.start() @@ -1802,14 +1760,14 @@ def finalize_execution_record(self, record): responses = [] def read_status(): - responses.append(server.handle_status({"execution_id": execution_id})) + responses.append(server.handle_status(StatusRequest(execution_id=execution_id).to_dict())) observed.set() reader = threading.Thread(target=read_status) registrations = [] def register(): - registrations.append(server._handle_register_progress({"client_id": "late-client"})) + registrations.append(server._handle_register_progress(ProgressRegistrationRequest(client_id="late-client").to_dict())) registered.set() subscriber = threading.Thread(target=register) @@ -1832,7 +1790,7 @@ def test_lost_terminal_notification_recovers_status_and_late_subscription_replay from zmqruntime.subscription import CallbackSubscription server = FailingExecutionServer(port=5555) - request = ExecuteRequest(plate_id="plate-1", pipeline_code="pass", config_params={"x": 1}) + request = ExecuteRequest(subject_id="subject-1", pipeline_code="pass", config_params={"x": 1}) execution_id = server._handle_execute(request.to_dict())["execution_id"] server.run_execution(execution_id, request, server.active_executions[execution_id]) terminal = server.progress_queue.get_nowait() # Deliberately lose its PUB delivery. @@ -1844,7 +1802,7 @@ def test_lost_terminal_notification_recovers_status_and_late_subscription_replay def poll(current_execution_id, *, timeout_ms): requests.append(current_execution_id) - return server.handle_status({"execution_id": current_execution_id}) + return server.handle_status(StatusRequest(execution_id=current_execution_id).to_dict()) client.poll_status = poll result = client.wait_for_completion(execution_id, poll_interval=0) @@ -1863,14 +1821,14 @@ def test_cancel_and_shutdown_seal_progress_before_terminal_notification(shutdown from zmqruntime.messages import ExecutionRecord server = DummyExecutionServer(port=5555) - record = ExecutionRecord("cancel-tail", "plate-1", None, ExecutionStatus.RUNNING.value) + record = ExecutionRecord("cancel-tail", "subject-1", ExecutionStatus.RUNNING.value) server.active_executions[record.execution_id] = record - progress = TaskProgress(task_id=record.execution_id, phase="running", status="running", percent=1.0, timestamp=1.0, completed=0, total=1).to_dict() + progress = TaskProgress(execution_id=record.execution_id, phase="running", status="running", percent=1.0, timestamp=1.0, completed=0, total=1).to_dict() server.send_progress_update(progress) if shutdown: server._cancel_all_executions() else: - server._handle_cancel({"execution_id": record.execution_id}) + server._handle_cancel(CancelRequest(execution_id=record.execution_id).to_dict()) server.send_progress_update(progress) admitted = server.progress_queue.get_nowait() terminal = server.progress_queue.get_nowait() diff --git a/tests/test_execution_logs.py b/tests/test_execution_logs.py index 844991b..dd8e8d9 100644 --- a/tests/test_execution_logs.py +++ b/tests/test_execution_logs.py @@ -49,7 +49,7 @@ def test_worker_log_observation_correlates_endpoint_heartbeat(tmp_path) -> None: running_executions=( RunningExecutionInfo( execution_id="execution-1", - plate_id="plate-1", + subject_id="subject-1", start_time=1.0, elapsed=2.0, ), diff --git a/tests/test_execution_waiter.py b/tests/test_execution_waiter.py index 9717a68..c1ac130 100644 --- a/tests/test_execution_waiter.py +++ b/tests/test_execution_waiter.py @@ -29,7 +29,7 @@ def execution_response(status: ExecutionStatus) -> WireResponse: status=ResponseType.OK, execution=ExecutionRecord( execution_id=EXECUTION_ID, - plate_id="plate", + subject_id="subject", client_address=None, status=status.value, error="worker failed" if status is ExecutionStatus.FAILED else None, @@ -91,7 +91,7 @@ def test_transport_and_decode_errors_share_the_consecutive_budget(): assert len(calls) == 3 assert result[MessageFields.STATUS] == ExecutionStatus.CANCELLED.value - assert "KeyError" in result[MessageFields.MESSAGE] + assert "ValueError" in result[MessageFields.MESSAGE] def test_valid_running_observation_resets_the_error_budget(): diff --git a/tests/test_messages.py b/tests/test_messages.py index 6de5889..447a7ff 100644 --- a/tests/test_messages.py +++ b/tests/test_messages.py @@ -75,7 +75,7 @@ def endpoint_compatibility(self): def test_execute_request_roundtrip(): request = ExecuteRequest( - plate_id="plate-1", + subject_id="subject-1", pipeline_code="print('hi')", config_params={"a": 1}, client_address="127.0.0.1", @@ -83,7 +83,7 @@ def test_execute_request_roundtrip(): data = request.to_dict() assert data[MessageFields.TYPE] == "execute" roundtrip = ExecuteRequest.from_dict(data) - assert roundtrip.plate_id == "plate-1" + assert roundtrip.subject_id == "subject-1" assert roundtrip.config_params == {"a": 1} @@ -151,7 +151,7 @@ def test_pong_response_requires_protocol_discriminator(): MessageFields.SERVER_ROLE: ServerRole.EXECUTION.value, } - with pytest.raises(KeyError): + with pytest.raises(ValueError): PongResponse.from_dict(payload) @@ -178,7 +178,7 @@ def test_pong_response_requires_explicit_server_role(): MessageFields.SERVER: "ExecutionServer", } - with pytest.raises(KeyError): + with pytest.raises(ValueError): PongResponse.from_dict(payload) @@ -192,5 +192,94 @@ def test_worker_state_projects_declared_process_identity_without_metadata_bag(): ) assert WorkerState.from_dict(state.to_dict()) == state - with pytest.raises(ValueError, match="unknown fields"): + with pytest.raises(ValueError, match="undeclared field"): WorkerState.from_dict({**state.to_dict(), "legacy_extra": "value"}) + + +def _declared_message_examples(): + import uuid + + from zmqruntime import messages + + identity = ProcessIdentity.current() + route = messages.AckReturnRoute("tcp://127.0.0.1:1", str(uuid.uuid4()), identity) + record = messages.ExecutionRecord("e-1", "subject-1", "running", start_time=1.0) + running = messages.RunningExecutionInfo("e-1", "subject-1", 1.0, 2.0) + queued = messages.QueuedExecutionInfo("e-2", "subject-1", 1) + worker = WorkerState(1, "running", 0.5, 10.0, create_time=3.0) + return ( + worker, + identity, + route, + record, + running, + queued, + ProcessResourceUsage(1.0, 2.0), + EndpointApplication("app", "1.0"), + messages.ImageTransferIdentity("image-1", route, identity), + messages.ImageAck("image-1", 1, "viewer", return_route=route, producer=identity), + messages.TaskProgress("e-1", "running", "running", 10.0, 1.0, 1, 10, subject_id="s"), + messages.ControlErrorResponse("boom"), + messages.ExecuteResponse(ResponseType.ACCEPTED, execution_id="e-1"), + messages.CancelResponse(ResponseType.OK, 2, message="cancelled"), + messages.ShutdownAck("stopping"), + messages.ProgressSubscriptionResponse("client", 1, "registered", {"e-1": {"x": 1}}), + messages.ExecutionStatusSnapshot( + ResponseType.OK, + execution=record, + running_executions=(running,), + queued_executions=(queued,), + executions=("e-1",), + ), + PongResponse( + port=1, + control_port=2, + ready=True, + server="server", + server_role=ServerRole.EXECUTION, + workers=(worker,), + running_executions=(running,), + process_identity=identity, + process_usage=ProcessResourceUsage(1.0, 2.0), + application=EndpointApplication("app", "1.0"), + ), + ControlRequestHeader(ControlMessageType.PING), + messages.EndpointShutdownRequest(ControlMessageType.SHUTDOWN, identity), + ExecuteRequest(subject_id="subject-1", pipeline_code="pass", config_params={"a": [1]}), + messages.StatusRequest(execution_id="e-1"), + CancelRequest(execution_id="e-1"), + messages.ProgressRegistrationRequest(client_id="client"), + messages.ProgressUnregistrationRequest(client_id="client"), + ) + + +def _concrete_declared_messages(): + from dataclasses import is_dataclass + + from zmqruntime import messages + + return { + value + for value in vars(messages).values() + if isinstance(value, type) + and value.__module__ == messages.__name__ + and is_dataclass(value) + and issubclass(value, (messages.WireMessage, ControlRequestHeader)) + } + + +def test_every_declared_message_round_trips_through_its_declaration() -> None: + examples = _declared_message_examples() + + assert {type(example) for example in examples} == _concrete_declared_messages() + for example in examples: + assert type(example).from_dict(example.to_dict()) == example + + +def test_typed_messages_reject_another_kind() -> None: + from zmqruntime import messages + + with pytest.raises(ValueError, match="type='image_ack'"): + messages.ImageAck.from_dict(messages.ShutdownAck("stopping").to_dict()) + with pytest.raises(ValueError, match="message_type"): + messages.StatusRequest.from_dict(CancelRequest(execution_id="e-1").to_dict()) diff --git a/tests/test_process_launch.py b/tests/test_process_launch.py new file mode 100644 index 0000000..12f0026 --- /dev/null +++ b/tests/test_process_launch.py @@ -0,0 +1,91 @@ +"""Contracts for noninteractive background subprocess creation.""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + +import pytest + +from zmqruntime.process_launch import ( + BackgroundProcessLaunchPolicy, + BackgroundProcessLaunchSpec, + BackgroundProcessPlatform, + SessionBackgroundProcesses, + WindowsBackgroundProcesses, +) + +NO_WINDOW = 0x08000000 +NEW_PROCESS_GROUP = 0x00000200 + + +@pytest.fixture +def windows_flags(monkeypatch): + monkeypatch.setattr(subprocess, "CREATE_NO_WINDOW", NO_WINDOW, raising=False) + monkeypatch.setattr( + subprocess, "CREATE_NEW_PROCESS_GROUP", NEW_PROCESS_GROUP, raising=False + ) + + +@pytest.mark.parametrize( + ("platform", "attached", "detached"), + [ + ( + WindowsBackgroundProcesses, + BackgroundProcessLaunchSpec(creationflags=NO_WINDOW), + BackgroundProcessLaunchSpec(creationflags=NO_WINDOW | NEW_PROCESS_GROUP), + ), + ( + SessionBackgroundProcesses, + BackgroundProcessLaunchSpec(), + BackgroundProcessLaunchSpec(start_new_session=True), + ), + ], +) +def test_each_platform_owns_its_launch_spec( + windows_flags, platform, attached, detached +) -> None: + assert BackgroundProcessLaunchPolicy(platform=platform).resolve() == attached + assert BackgroundProcessLaunchPolicy(platform=platform, detached=True).resolve() == detached + assert BackgroundProcessLaunchPolicy(platform=platform, detached=True).popen_arguments() == ( + detached.popen_arguments() + ) + + +def test_spec_projects_only_non_default_popen_arguments() -> None: + assert BackgroundProcessLaunchSpec().popen_arguments() == {} + assert BackgroundProcessLaunchSpec(start_new_session=True).popen_arguments() == { + "start_new_session": True + } + + +@pytest.mark.parametrize( + ("platform", "windowed_present", "expected_name"), + [ + (WindowsBackgroundProcesses, True, "pythonw.exe"), + (WindowsBackgroundProcesses, False, "python.exe"), + (SessionBackgroundProcesses, True, "python.exe"), + ], +) +def test_background_python_interpreter( + tmp_path: Path, platform, windowed_present, expected_name +) -> None: + python_executable = tmp_path / "python.exe" + python_executable.touch() + if windowed_present: + (tmp_path / "pythonw.exe").touch() + + policy = BackgroundProcessLaunchPolicy(platform=platform) + + assert policy.python_executable(str(python_executable)) == str(tmp_path / expected_name) + + +def test_current_host_selects_registered_family(monkeypatch) -> None: + monkeypatch.setattr(sys, "platform", "win32") + assert BackgroundProcessPlatform.current() is WindowsBackgroundProcesses + monkeypatch.setattr(sys, "platform", "linux") + assert BackgroundProcessPlatform.current() is SessionBackgroundProcesses + assert BackgroundProcessLaunchPolicy.current(detached=True) == BackgroundProcessLaunchPolicy( + platform=SessionBackgroundProcesses, detached=True + ) diff --git a/tests/test_progress_annotation.py b/tests/test_progress_annotation.py index b6a8f9b..6385d87 100644 --- a/tests/test_progress_annotation.py +++ b/tests/test_progress_annotation.py @@ -76,7 +76,7 @@ def test_real_openhcs_job_status_decodes_nested_progress(status): # Optional downstream integration, not a dependency or copied DTO in ZMQRuntime. execution = pytest.importorskip("openhcs.agent.dto.execution") from openhcs.agent.dto.common import SCHEMA_VERSION - from openhcs.serialization.json import to_jsonable + from python_introspect import to_jsonable original = execution.ExecutionJobStatus( schema_version=SCHEMA_VERSION, diff --git a/tests/test_progress_projection.py b/tests/test_progress_projection.py index 8cfe7db..d6411e9 100644 --- a/tests/test_progress_projection.py +++ b/tests/test_progress_projection.py @@ -9,7 +9,7 @@ @dataclass(frozen=True) class _Event: execution_id: str - plate_id: str + subject_id: str axis_id: str channel: str percent: float @@ -20,8 +20,8 @@ class _Event: class _Adapter(ProgressProjectionAdapterABC[_Event, str]): - def plate_id(self, event: _Event) -> str: - return event.plate_id + def subject_id(self, event: _Event) -> str: + return event.subject_id def axis_id(self, event: _Event) -> str: return event.axis_id @@ -75,7 +75,7 @@ def test_build_execution_projection_marks_executing_from_pipeline_channel(): "exec-1": [ _Event( execution_id="exec-1", - plate_id="plate-1", + subject_id="subject-1", axis_id="", channel="compile", percent=0.0, @@ -86,7 +86,7 @@ def test_build_execution_projection_marks_executing_from_pipeline_channel(): ), _Event( execution_id="exec-1", - plate_id="plate-1", + subject_id="subject-1", axis_id="A01", channel="pipeline", percent=50.0, @@ -98,11 +98,11 @@ def test_build_execution_projection_marks_executing_from_pipeline_channel(): } projection = build_execution_projection(events_by_execution, adapter=adapter) - plate = projection.get_plate("plate-1", "exec-1") + subject = projection.get_subject("subject-1", "exec-1") - assert plate is not None - assert plate.state == "executing" - assert round(plate.percent, 1) == 25.0 + assert subject is not None + assert subject.state == "executing" + assert round(subject.percent, 1) == 25.0 assert projection.count_state("executing") == 1 @@ -112,7 +112,7 @@ def test_build_execution_projection_marks_failed_when_axis_failed(): "exec-1": [ _Event( execution_id="exec-1", - plate_id="plate-1", + subject_id="subject-1", axis_id="", channel="compile", percent=0.0, @@ -123,7 +123,7 @@ def test_build_execution_projection_marks_failed_when_axis_failed(): ), _Event( execution_id="exec-1", - plate_id="plate-1", + subject_id="subject-1", axis_id="A01", channel="pipeline", percent=50.0, @@ -133,7 +133,7 @@ def test_build_execution_projection_marks_failed_when_axis_failed(): ), _Event( execution_id="exec-1", - plate_id="plate-1", + subject_id="subject-1", axis_id="B01", channel="pipeline", percent=100.0, @@ -145,9 +145,9 @@ def test_build_execution_projection_marks_failed_when_axis_failed(): } projection = build_execution_projection(events_by_execution, adapter=adapter) - plate = projection.get_plate("plate-1", "exec-1") + subject = projection.get_subject("subject-1", "exec-1") - assert plate is not None - assert plate.state == "failed" - assert round(plate.percent, 1) == 75.0 + assert subject is not None + assert subject.state == "failed" + assert round(subject.percent, 1) == 75.0 assert projection.count_state("failed") == 1 diff --git a/tests/test_progress_registry.py b/tests/test_progress_registry.py index 51fcd33..3c2f240 100644 --- a/tests/test_progress_registry.py +++ b/tests/test_progress_registry.py @@ -8,7 +8,7 @@ @dataclass(frozen=True) class _Event: - plate_id: str + subject_id: str axis_id: str channel: str timestamp: float @@ -20,13 +20,13 @@ def test_latest_event_registry_keeps_channels_separate(): _Event, tuple[str, str, str], ]( - key_builder=lambda event: (event.plate_id, event.axis_id, event.channel), + key_builder=lambda event: (event.subject_id, event.axis_id, event.channel), is_terminal=lambda event: event.terminal, timestamp_of=lambda event: event.timestamp, ) - registry.register_event("exec-1", _Event("plate-1", "A01", "pipeline", timestamp=1.0)) - registry.register_event("exec-1", _Event("plate-1", "A01", "step", timestamp=2.0)) + registry.register_event("exec-1", _Event("subject-1", "A01", "pipeline", timestamp=1.0)) + registry.register_event("exec-1", _Event("subject-1", "A01", "step", timestamp=2.0)) events = registry.get_events("exec-1") assert len(events) == 2 @@ -38,18 +38,18 @@ def test_latest_event_registry_cleanup_removes_old_terminal_executions(): _Event, tuple[str, str, str], ]( - key_builder=lambda event: (event.plate_id, event.axis_id, event.channel), + key_builder=lambda event: (event.subject_id, event.axis_id, event.channel), is_terminal=lambda event: event.terminal, timestamp_of=lambda event: event.timestamp, ) registry.register_event( "exec-1", - _Event("plate-1", "A01", "pipeline", timestamp=1.0, terminal=True), + _Event("subject-1", "A01", "pipeline", timestamp=1.0, terminal=True), ) registry.register_event( "exec-2", - _Event("plate-1", "A02", "pipeline", timestamp=100.0, terminal=False), + _Event("subject-1", "A02", "pipeline", timestamp=100.0, terminal=False), ) removed = registry.cleanup_old_executions(retention_seconds=10.0) @@ -63,15 +63,15 @@ def test_latest_event_registry_rejects_stale_or_duplicate_channel_events(): _Event, tuple[str, str, str], ]( - key_builder=lambda event: (event.plate_id, event.axis_id, event.channel), + key_builder=lambda event: (event.subject_id, event.axis_id, event.channel), is_terminal=lambda event: event.terminal, timestamp_of=lambda event: event.timestamp, ) listener_events = [] registry.add_listener(lambda execution_id, event: listener_events.append((execution_id, event))) - newest = _Event("plate-1", "A01", "pipeline", timestamp=2.0) - stale = _Event("plate-1", "A01", "pipeline", timestamp=1.0) + newest = _Event("subject-1", "A01", "pipeline", timestamp=2.0) + stale = _Event("subject-1", "A01", "pipeline", timestamp=1.0) assert registry.register_event("exec-1", newest) is True assert registry.register_event("exec-1", stale) is False @@ -85,20 +85,20 @@ def test_registry_mutation_listener_observes_register_and_all_clear_routes(): _Event, tuple[str, str, str], ]( - key_builder=lambda event: (event.plate_id, event.axis_id, event.channel), + key_builder=lambda event: (event.subject_id, event.axis_id, event.channel), is_terminal=lambda event: event.terminal, timestamp_of=lambda event: event.timestamp, ) mutations = [] registry.add_mutation_listener(mutations.append) - event_one = _Event("plate-1", "A01", "pipeline", timestamp=1.0) - event_two = _Event("plate-2", "A01", "pipeline", timestamp=1.0) + event_one = _Event("subject-1", "A01", "pipeline", timestamp=1.0) + event_two = _Event("subject-2", "A01", "pipeline", timestamp=1.0) registry.register_event("exec-1", event_one) registry.register_event("exec-2", event_two) registry.clear_execution("exec-1") registry.clear_all() - event_three = _Event("plate-3", "A01", "pipeline", timestamp=2.0) + event_three = _Event("subject-3", "A01", "pipeline", timestamp=2.0) registry.register_event("exec-3", event_three) assert [mutation.kind for mutation in mutations] == [ @@ -122,7 +122,7 @@ def test_registry_mutation_subscription_owns_idempotent_release(): _Event, tuple[str, str, str], ]( - key_builder=lambda event: (event.plate_id, event.axis_id, event.channel), + key_builder=lambda event: (event.subject_id, event.axis_id, event.channel), is_terminal=lambda event: event.terminal, timestamp_of=lambda event: event.timestamp, ) @@ -131,13 +131,13 @@ def test_registry_mutation_subscription_owns_idempotent_release(): registry.register_event( "exec-1", - _Event("plate-1", "A01", "pipeline", timestamp=1.0), + _Event("subject-1", "A01", "pipeline", timestamp=1.0), ) assert subscription.release() is True assert subscription.release() is False registry.register_event( "exec-2", - _Event("plate-2", "A01", "pipeline", timestamp=1.0), + _Event("subject-2", "A01", "pipeline", timestamp=1.0), ) assert len(mutations) == 1 @@ -149,7 +149,7 @@ def test_clearing_mutation_listeners_retires_owned_subscriptions(): _Event, tuple[str, str, str], ]( - key_builder=lambda event: (event.plate_id, event.axis_id, event.channel), + key_builder=lambda event: (event.subject_id, event.axis_id, event.channel), is_terminal=lambda event: event.terminal, timestamp_of=lambda event: event.timestamp, ) @@ -165,7 +165,7 @@ def test_cleanup_notifies_mutation_listener_for_each_removed_execution(): _Event, tuple[str, str, str], ]( - key_builder=lambda event: (event.plate_id, event.axis_id, event.channel), + key_builder=lambda event: (event.subject_id, event.axis_id, event.channel), is_terminal=lambda event: event.terminal, timestamp_of=lambda event: event.timestamp, ) @@ -173,7 +173,7 @@ def test_cleanup_notifies_mutation_listener_for_each_removed_execution(): registry.add_mutation_listener(mutations.append) registry.register_event( "exec-1", - _Event("plate-1", "A01", "pipeline", timestamp=1.0, terminal=True), + _Event("subject-1", "A01", "pipeline", timestamp=1.0, terminal=True), ) mutations.clear() diff --git a/tests/test_progress_stream.py b/tests/test_progress_stream.py index 78abbda..4d40425 100644 --- a/tests/test_progress_stream.py +++ b/tests/test_progress_stream.py @@ -18,7 +18,7 @@ def _message() -> str: "timestamp": 1.0, "completed": 1, "total": 2, - "plate_id": "plate-1", + "subject_id": "subject-1", "axis_id": "A01", } )