From dd9b58d0f979002efd5e1b19d276961a37395033 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 17:07:57 -0400 Subject: [PATCH 01/15] B7: receipt for execution-server isolation Co-Authored-By: Claude Opus 5.5 --- .../B7-server-isolation.md | 105 ++++++++++++++++++ 1 file changed, 105 insertions(+) create mode 100644 docs/refactor/consolidation-20261009/B7-server-isolation.md diff --git a/docs/refactor/consolidation-20261009/B7-server-isolation.md b/docs/refactor/consolidation-20261009/B7-server-isolation.md new file mode 100644 index 0000000000..8b9f160bdb --- /dev/null +++ b/docs/refactor/consolidation-20261009/B7-server-isolation.md @@ -0,0 +1,105 @@ +# B7: an execution client reaches only a server it may use + +**Head audited:** `openhcs` `main` at `625150118` (#1207); `zmqruntime` `main` at `dc327e6` (0.6.1). **Rules:** [00-RULES.md](00-RULES.md). **Origin:** both (the attach rule is zmqruntime's; the shared namespace and the launch are OpenHCS's). +**Shared abstractions.** *Builds:* `zmqruntime.EndpointLaunch` (where a server's code is, which process started it, who may use it) and the `EndpointAudience` family. *Uses:* `AutoRegisterMeta`, `WireMessage`, `ProcessIdentity`, `OpenHCSRuntimeImportAuthority`. + +## What is wrong + +**Nothing relates a client to the server it uses: a client attaches to whatever answers at an address, every OpenHCS process of one user shares one address namespace, and nothing ever stops a server whose starter is gone.** + +How a client picks a server today: + +- **One namespace per user.** `OPENHCS_ZMQ_CONFIG` is a module singleton, the default of 30 parameters. Its endpoint is data port 7777 and control port 8777. On Linux and macOS the transport is IPC, and `IpcTransportDeclaration.socket_path` (`zmqruntime/transport_modes.py:383`) maps a port to `~/.openhcs/ipc/openhcs-zmq-.sock`. Every checkout, worktree and installed copy run by one user resolves the same file for the same port. On Windows the transport is TCP and the namespace is the machine's loopback ports. +- **Attach to whatever answers.** `ZMQClient._connect_locked` (`zmqruntime/client.py:1060`) takes the startup lock, and if the socket file exists it pings the control socket. Any reply with `ready=True` makes it an `AttachedEndpointConnection`. The reply already carries `process_identity` and `application`; neither is read. `connect_existing` does the same. +- **The only identity check is somewhere else and compares too little.** `EndpointApplication` is `(identifier, version)`. `ZMQClientService` (GUI and session) and the function-catalog client compare it after connecting. A plain `ZMQExecutionClient.connect()` (scripts, the benchmark harness, `ExecutionConnectionSpec.execution_client` for MCP) never compares it. Two worktrees have the same identifier and version, so the check passes between them anyway. +- **"A free port" is not reserved.** `DataControlPortPairAuthority.acquire` returns the first pair from 7777 up whose socket files are absent. It reserves nothing. Three harness sites and 9 test files call it and then build a client on that port. +- **Discovery scan.** `ZMQClient.scan_servers` pings a port range (`ports_per_server_type` = 10 per role) in the same namespace and returns every reply. The GUI server browser and `RuntimeServerService.scan` list them; `ZMQServerScanService` is what `test_zmq_browser_scan_service_discovers_ui_bridge_default_transport` exercises, against a `UiBridgeControlServer` bound in the user's real `~/.openhcs/ipc`. +- **UI-bridge descriptors.** The running GUI writes `ui_bridge_*.json` into one per-user runtime directory (`UiBridgeDescriptorDirectoryAuthority`), and `UiBridgeProcessAdvertisedDescriptorCatalog` also reads `/proc/*/environ` of every process for `OPENHCS_UI_BRIDGE_DESCRIPTOR_FILE`. An MCP client with no explicit descriptor takes the only live one, from any checkout. Requests carry `(identifier, version)` and an auth token. This is a client driving a GUI, not choosing an execution server; it is U4's and is recorded under Reassigned. + +The three observed failures are one sequence: + +- The parity test picks one port with `acquire` and keeps it for all 30 workflows (`tests/integration/test_cellprofiler_official30_zmq.py:201`). +- Each workflow builds `ZMQExecutionClient(port=port, persistent=False)`, which starts a server and terminates it at the end of the workflow. Between workflows the socket file is absent. +- A second process (another parity run, or any OpenHCS process on the default port, because `acquire` starts at 7777) sees the port free and starts its own server there. +- The first run's next workflow finds the socket, gets a ready reply and attaches. It now runs on code imported from the other worktree. +- When the other process ends, its server stops under the first run. `benchmark/openhcs_measured_run.py:297` reports "client entered without a connected endpoint" for the same collision seen from the other side. + +`require_owned_server` exists in the harness (`openhcs_measured_run.py:305`) and would have refused; the parity path does not pass it. + +Why a server lived 31 hours: + +- `ExecutionTransportConfig.persistent` defaults to `True`. `ExecutionRuntimeLaunchPlan.spawn` launches a persistent server with `start_new_session=True` (its own session and process group), and `OwnedEndpointConnection.close_client(persistent=True)` does nothing. +- The server holds no reference to the process that started it. `serve_forever` loops until a shutdown request or a signal. There is no idle limit and no check on the starter. +- A non-persistent server is stopped only by its client's `disconnect()`. A client that is killed (a pytest timeout, an agent stopped mid-run) never calls it. +- The code is in memory, so deleting the worktree changes nothing. No stale group is running at this head to inspect; the cause is read from the code. + +## Target + +**The relation, declared once.** A server states its launch, and `ZMQClient` checks that statement at the one place a connection is established. + +```python +class EndpointAudience(ABC, metaclass=AutoRegisterMeta): # who may use a server + def admits(owner, client) -> bool + def orphan_stops(unfinished_work) -> bool +class OwnerOnlyAudience(EndpointAudience): ... # a non-persistent server +class AnySessionAudience(EndpointAudience): ... # a persistent server + +@dataclass(frozen=True) +class EndpointLaunch(WireMessage): + code_root: str # directory the server's top-level package is imported from + owner: ProcessIdentity # the process that started it + audience: str # wire name, derived from the family registry +``` + +- `ZMQServer` holds its `EndpointLaunch` and reports it in every heartbeat (`PongResponse.launch`). `code_root` is derived from the server class, never passed by a caller. +- **A client that starts a server** hands it the launch (`owner` = the client process, audience from `persistent`) and connects only if the server that becomes ready reports exactly that launch. +- **A client that attaches** (an occupied endpoint in `connect`, or `connect_existing`) is refused with `EndpointLaunchMismatchError` unless the server reports a launch, its `code_root` equals the client's, and its audience admits the client process. The error names both code roots and the owner. +- The check applies to endpoints on this host. A remote TCP host is an endpoint the user named; its paths and process ids mean nothing here. +- **Namespace.** `ZMQConfig.socket_root()` owns where socket and lock files live. `OpenHCSZMQConfig` derives it from the import root: `~/.openhcs//`. Two checkouts no longer share addresses, so the refusal is the second line, not the normal case. `OPENHCS_SOCKET_ROOT` replaces the root for one process tree. +- **Orphans.** The server checks `launch.owner` once a second in its message loop. When the owner is gone, an owner-only server cancels its work and stops; an any-session server stops once it has no running or queued execution. +- **Tests and harness.** `tests/conftest.py` sets `OPENHCS_SOCKET_ROOT` to a short private directory before OpenHCS is imported (one per pytest process, so per xdist worker). The harness uses `ZMQExecutionClient.with_private_server`, which takes its address from `TransportDeclaration.private_endpoint`. + +Identity that must match: code location and owning process (through the audience). Version stays where it is: `EndpointApplication`, compared by `ZMQClientService`. A server edited after it started has the same root and version and is not detected; a source digest would be needed and no surface owns that. + +Deleted: the three `DataControlPortPairAuthority.acquire` + `ZMQExecutionClient(port=...)` pairs in the harness, `require_owned_server`, the parity test's `_free_zmq_port_pair`, and the hand-built `Path.home() / ".openhcs" / "ipc"` in `viewer_endpoint_discovery.py`. + +## Persisted state + +Runtime only. Socket and lock files move from `~/.openhcs/ipc` and `~/.openhcs/tcp` to `~/.openhcs//`; servers started before the upgrade are not found and are not reused. The saved `OpenHCSZMQConfig` preference keeps every field. + +## Guards + +- No `Path.home()` in `openhcs/` or `zmqruntime` builds a socket path outside `socket_root()`. +- No harness or test file pairs `DataControlPortPairAuthority.acquire(OPENHCS_ZMQ_CONFIG` with an execution client. +- `ZMQClient` assigns `_connection` only after the launch check. + +## Tests + +- Two servers from two code roots: each client reaches only its own; a client aimed at the other's address is refused with both roots in the message. +- An owner-only server refuses a second process and admits its owner. +- An orphaned server exits: owner-only at once, any-session after its work. +- zmqruntime: the same three at the library level with a minimal server. + +## New-case experiments + +A new kind of server (a viewer, a bridge) that must not be shared across checkouts: today it would need its own check after connecting. After: it subclasses `ZMQServer` and its clients subclass `ZMQClient`; nothing else. + +## Done when + +- A client cannot hold a connection to a local server whose launch it did not verify. +- A server whose owner is gone exits without a request. +- The parity check passes 30/30 with no port chosen by the test. +- The guards pass; the structural ratchet and nra-r1 show no regression. + +## Reassigned + +| Item | Evidence | Belongs to | +|---|---|---| +| UI-bridge descriptor discovery takes the only live descriptor from any checkout | `ui_bridge_service.py:1563-1585`, `/proc` environment scan at `:1599` | U4 | + +## Progress notes + +- [x] Mechanism measured, receipt written. +- [ ] zmqruntime: `EndpointLaunch`, client check, orphan stop, `socket_root`, `private_endpoint`. +- [ ] OpenHCS: launch argument, socket root, harness, conftest, guards. +- [ ] Parity, ratchet, nra, CI. From 2219b288b02d47ab0bcd4b32120423e7835615cd Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 20:20:57 -0400 Subject: [PATCH 02/15] WIP B7: point zmqruntime at fix/B7-server-isolation Co-Authored-By: Claude Opus 5.5 --- external/zmqruntime | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/external/zmqruntime b/external/zmqruntime index dc327e6ca2..0c2ad2f99c 160000 --- a/external/zmqruntime +++ b/external/zmqruntime @@ -1 +1 @@ -Subproject commit dc327e6ca2496aa4ad405479832f4e9e219750e4 +Subproject commit 0c2ad2f99c5bee202ec3297abfb2f37884920a3e From d8deeacc0eef3f7d1cd1412a720c90f72edfd802 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 20:22:25 -0400 Subject: [PATCH 03/15] WIP B7: launch record, checkout socket root, private harness servers Co-Authored-By: Claude Opus 5.5 --- benchmark/matched_cellprofiler_batch.py | 8 +--- benchmark/openhcs_measured_run.py | 41 ++++++++++++------- benchmark/well_throughput_scaling.py | 14 ++----- .../services/viewer_endpoint_discovery.py | 6 +-- openhcs/runtime/zmq_config.py | 28 +++++++++++++ openhcs/runtime/zmq_execution_client.py | 39 ++++++++++++++++-- openhcs/runtime/zmq_execution_server.py | 3 ++ .../runtime/zmq_execution_server_launcher.py | 8 ++++ .../test_cellprofiler_official30_zmq.py | 15 ------- 9 files changed, 107 insertions(+), 55 deletions(-) diff --git a/benchmark/matched_cellprofiler_batch.py b/benchmark/matched_cellprofiler_batch.py index d4cb5be31a..333ec8b219 100644 --- a/benchmark/matched_cellprofiler_batch.py +++ b/benchmark/matched_cellprofiler_batch.py @@ -30,7 +30,6 @@ ensure_global_config_context, rebuild_lazy_config_with_new_global_reference, ) -from zmqruntime import DataControlPortPairAuthority from zmqruntime.messages import ExecutionStatusSnapshot from benchmark.adapters.cellprofiler import ( @@ -114,7 +113,6 @@ from openhcs.interop.cellprofiler.plate_workspace import ( prepare_cellprofiler_input_workspace, ) -from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG from openhcs.runtime.zmq_execution_client import ( OpenHCSExecutionSubmission, ZMQExecutionClient, @@ -1206,11 +1204,7 @@ def main(argv: list[str] | None = None) -> int: if args.preflight_native_references: execution_scope = nullcontext(None) else: - port = DataControlPortPairAuthority.acquire( - OPENHCS_ZMQ_CONFIG, - transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, - ).data_port - execution_scope = ZMQExecutionClient(port=port, persistent=False) + execution_scope = ZMQExecutionClient.with_private_server() receipts = {} with execution_scope as client: for case in selected_cases: diff --git a/benchmark/openhcs_measured_run.py b/benchmark/openhcs_measured_run.py index 9523226820..67e9103f90 100644 --- a/benchmark/openhcs_measured_run.py +++ b/benchmark/openhcs_measured_run.py @@ -8,6 +8,7 @@ import sys import time from collections.abc import Callable, Mapping +from contextlib import AbstractContextManager from dataclasses import ( dataclass, field, @@ -15,7 +16,6 @@ from pathlib import Path from typing import Any -from zmqruntime import DataControlPortPairAuthority from zmqruntime.messages import PongResponse from benchmark.contracts.measured_run_receipt import ( @@ -34,7 +34,6 @@ RuntimeArtifactExecutionObservation, ) from openhcs.runtime.zmq_application import OPENHCS_ENDPOINT_APPLICATION -from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG from openhcs.runtime.zmq_execution_client import ( ZMQExecutionRequestBuilder, OpenHCSExecutionSubmission, @@ -247,6 +246,29 @@ def _require_measured_submission( raise ValueError("Measured OpenHCS observation export path must be absolute.") +def measured_execution_client( + *, + execution_port: int | None, + progress_callback=None, +) -> AbstractContextManager[ZMQExecutionClient]: + """The client of one measured run. + + With no port the run starts a server at an address of its own. A port is + an address the caller named; the client still refuses a server there that + another process started for its own use or that runs other code. + """ + + if execution_port is None: + return ZMQExecutionClient.with_private_server( + progress_callback=progress_callback + ) + return ZMQExecutionClient( + port=execution_port, + persistent=False, + progress_callback=progress_callback, + ) + + def execute_measured_openhcs_pipeline( *, submission: OpenHCSExecutionSubmission, @@ -259,18 +281,9 @@ def execute_measured_openhcs_pipeline( """Measure an ordinary pipeline submission with its requested observation.""" _require_measured_submission(submission, expected_axis_count) - client_port = execution_port - if client_port is None and require_owned_server: - client_port = DataControlPortPairAuthority.acquire( - OPENHCS_ZMQ_CONFIG, - transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, - ).data_port - client = ZMQExecutionClient( - port=client_port, - persistent=False, - progress_callback=timing_observer, - ) - with client: + with measured_execution_client( + execution_port=execution_port, progress_callback=timing_observer + ) as client: return execute_measured_openhcs_pipeline_on_client( client=client, submission=submission, diff --git a/benchmark/well_throughput_scaling.py b/benchmark/well_throughput_scaling.py index a088abe9b6..aa2571cef9 100644 --- a/benchmark/well_throughput_scaling.py +++ b/benchmark/well_throughput_scaling.py @@ -17,7 +17,6 @@ import psutil from objectstate.lazy_factory import rebuild_lazy_config_with_new_global_reference -from zmqruntime import DataControlPortPairAuthority if TYPE_CHECKING: from benchmark.reports.cppipe_figures import BenchmarkMetricRow @@ -34,6 +33,7 @@ _ZMQProgressTimingObserver, execute_measured_openhcs_pipeline, execute_measured_openhcs_pipeline_on_client, + measured_execution_client, ) from benchmark.timing import BenchmarkPhase, PhaseTimingTrace from openhcs.core.components.parser_metaprogramming import FilenameParseResult @@ -68,7 +68,6 @@ prepare_cellprofiler_input_workspace, ) from openhcs.core.dataset_sources.source_schema import SourceSchemaFilenameParser -from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG from openhcs.runtime.zmq_execution_client import ( OpenHCSExecutionSubmission, ZMQExecutionClient, @@ -2001,16 +2000,9 @@ def run_well_throughput_suite( if observation_key in completed or observation_key in skipped: continue if reuse_execution_server and shared_client is None: - client_port = execution_port - if client_port is None: - client_port = DataControlPortPairAuthority.acquire( - OPENHCS_ZMQ_CONFIG, - transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, - ).data_port shared_client = client_stack.enter_context( - ZMQExecutionClient( - port=client_port, - persistent=False, + measured_execution_client( + execution_port=execution_port, progress_callback=shared_observer, ) ) diff --git a/openhcs/agent/services/viewer_endpoint_discovery.py b/openhcs/agent/services/viewer_endpoint_discovery.py index 538d268e15..1e7dbab417 100644 --- a/openhcs/agent/services/viewer_endpoint_discovery.py +++ b/openhcs/agent/services/viewer_endpoint_discovery.py @@ -53,11 +53,7 @@ def as_dict(self) -> dict[str, Any]: def viewer_ipc_directory() -> Path: """Return the OpenHCS IPC socket directory for the configured transport.""" - return ( - Path.home() - / f".{OPENHCS_ZMQ_CONFIG.app_name}" - / OPENHCS_ZMQ_CONFIG.ipc_socket_dir - ) + return OPENHCS_ZMQ_CONFIG.socket_root() / OPENHCS_ZMQ_CONFIG.ipc_socket_dir def candidate_viewer_ports() -> tuple[int, ...]: diff --git a/openhcs/runtime/zmq_config.py b/openhcs/runtime/zmq_config.py index 4c93c6a696..8f2ad559fa 100644 --- a/openhcs/runtime/zmq_config.py +++ b/openhcs/runtime/zmq_config.py @@ -2,11 +2,35 @@ from __future__ import annotations +import hashlib +import os from dataclasses import dataclass +from pathlib import Path from zmqruntime.config import NonBlankString from zmqruntime.execution.config import ExecutionTransportConfig +from openhcs.runtime.import_authority import OpenHCSRuntimeImportAuthority + +SOCKET_ROOT_ENV = "OPENHCS_SOCKET_ROOT" +"""Directory that replaces the socket root for one process tree. + +Child processes inherit it, so a test run or a script and every server and +viewer it starts share one namespace that nothing else on the machine sees. +""" + + +def checkout_socket_root() -> Path: + """Socket root of this OpenHCS checkout or installation. + + Port ``n`` names one socket file per root. Deriving the root from the + import root keeps two checkouts run by one user from sharing addresses. + """ + + import_root = OpenHCSRuntimeImportAuthority.current().import_root + digest = hashlib.sha256(str(import_root).encode("utf-8")).hexdigest()[:8] + return Path.home() / ".openhcs" / digest + @dataclass(frozen=True, slots=True) class OpenHCSZMQConfig(ExecutionTransportConfig): @@ -18,5 +42,9 @@ class OpenHCSZMQConfig(ExecutionTransportConfig): app_name: NonBlankString = "openhcs" """OpenHCS application namespace used in transport identities and runtime paths.""" + def socket_root(self) -> Path: + declared = os.environ.get(SOCKET_ROOT_ENV) + return Path(declared) if declared else checkout_socket_root() + OPENHCS_ZMQ_CONFIG = OpenHCSZMQConfig() diff --git a/openhcs/runtime/zmq_execution_client.py b/openhcs/runtime/zmq_execution_client.py index 14efb43965..c36bd7429f 100644 --- a/openhcs/runtime/zmq_execution_client.py +++ b/openhcs/runtime/zmq_execution_client.py @@ -7,10 +7,10 @@ import subprocess import sys import time -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Iterator, Mapping, Sequence from concurrent.futures import CancelledError from copy import deepcopy -from contextlib import AbstractContextManager, nullcontext +from contextlib import AbstractContextManager, contextmanager, nullcontext from dataclasses import dataclass, replace from enum import Enum from pathlib import Path @@ -39,6 +39,7 @@ ControlMessageType, ControlRequestHeader, EndpointApplicationCompatibilityError, + EndpointLaunch, ExecuteRequest, ExecutionRecord, PongResponse, @@ -157,6 +158,7 @@ def spawn( config: OpenHCSZMQConfig, *, persistent: bool, + launch: EndpointLaunch, ) -> subprocess.Popen: """Materialize this admitted plan through the canonical process policy. @@ -191,6 +193,8 @@ def spawn( str(self.startup_status_file), "--config-source", _pycodify_config_source(server_config), + "--launch", + launch.to_argument(), ] root_logger = logging.getLogger() current_log_level = root_logger.getEffectiveLevel() @@ -981,6 +985,30 @@ def __init__( connection_status_callback=connection_status_callback, ) + @classmethod + @contextmanager + def with_private_server( + cls, + *, + progress_callback=None, + config: OpenHCSZMQConfig = OPENHCS_ZMQ_CONFIG, + ) -> Iterator[ZMQExecutionClient]: + """A connected client on a server it started at an address of its own. + + The server stops when the block ends. The address comes from the + transport, so no other process is given it: a run that needs its own + server uses this instead of picking a free port. + """ + + private_endpoint = config.transport_mode.declaration.private_endpoint + with private_endpoint(config) as private_config: + with cls( + persistent=False, + progress_callback=progress_callback, + config=private_config, + ) as client: + yield client + def connect( self, timeout: float | None = None, @@ -1280,7 +1308,12 @@ def _spawn_server_process(self) -> subprocess.Popen: plan = self.runtime_launch_plan() self._runtime_launch_plan = None self._startup_status_path = plan.startup_status_file - return plan.spawn(self.endpoint, self.config, persistent=self.persistent) + return plan.spawn( + self.endpoint, + self.config, + persistent=self.persistent, + launch=self.launch(), + ) def runtime_launch_plan(self) -> ExecutionRuntimeLaunchPlan: """Resolve owner defaults without creating directories or warming.""" diff --git a/openhcs/runtime/zmq_execution_server.py b/openhcs/runtime/zmq_execution_server.py index 49cd9c8607..60a7dcfc75 100644 --- a/openhcs/runtime/zmq_execution_server.py +++ b/openhcs/runtime/zmq_execution_server.py @@ -15,6 +15,7 @@ from zmqruntime.config import TransportMode from zmqruntime.execution import ExecutionServer from zmqruntime.messages import ( + EndpointLaunch, ExecuteRequest, ExecutionRecord, ExecutionStatus, @@ -173,6 +174,7 @@ def __init__( log_file_path: str | None = None, transport_mode: TransportMode | None = None, config: OpenHCSZMQConfig = OPENHCS_ZMQ_CONFIG, + launch: EndpointLaunch | None = None, ): super().__init__( port=config.default_port if port is None else port, @@ -183,6 +185,7 @@ def __init__( ), config=config, application=OPENHCS_ENDPOINT_APPLICATION, + launch=launch, ) self._compile_status: str | None = None self._compile_message: str | None = None diff --git a/openhcs/runtime/zmq_execution_server_launcher.py b/openhcs/runtime/zmq_execution_server_launcher.py index 477fd88c7e..b7ff3f2b14 100644 --- a/openhcs/runtime/zmq_execution_server_launcher.py +++ b/openhcs/runtime/zmq_execution_server_launcher.py @@ -12,6 +12,7 @@ from pathlib import Path from zmqruntime.config import TransportMode +from zmqruntime.messages import EndpointLaunch from zmqruntime.startup import ( EndpointStartupPhase, EndpointStartupStatusWriter, @@ -69,6 +70,12 @@ def main( action="store_true", help="Prepare endpoint-owned capability caches without binding sockets", ) + parser.add_argument( + "--launch", + type=EndpointLaunch.from_argument, + default=None, + help="Launch record of the client that starts this server", + ) parser.add_argument( "--startup-status-path", type=str, @@ -192,6 +199,7 @@ def main( server = execution_server_type( log_file_path=args.log_file_path, config=config, + launch=args.launch, ) status_reporter.emit( diff --git a/tests/integration/test_cellprofiler_official30_zmq.py b/tests/integration/test_cellprofiler_official30_zmq.py index a298d1d30d..bb4b91c859 100644 --- a/tests/integration/test_cellprofiler_official30_zmq.py +++ b/tests/integration/test_cellprofiler_official30_zmq.py @@ -198,10 +198,6 @@ def _assert_manifest_reference_parity( persistent=False, ), ) - execution_port = _free_zmq_port_pair( - set(), - transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, - ) observations = run_comparison_suite( cases, output_root=output_root, @@ -211,7 +207,6 @@ def _assert_manifest_reference_parity( openhcs_global_config=global_config, discard_openhcs_outputs=True, continue_on_error=True, - openhcs_execution_port=execution_port, ) assert len(observations) == expected_case_count @@ -605,10 +600,6 @@ def test_official30_nonpersistent_napari_isolated_per_case( native_reference_root = _native_reference_root() assert len(_OFFICIAL30_CASES) == 30 excluded_ports: set[int] = set() - execution_port = _free_zmq_port_pair( - excluded_ports, - transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, - ) viewer_transport_mode = TransportMode.default() viewer_port = _free_zmq_port_pair( excluded_ports, @@ -664,7 +655,6 @@ def test_official30_nonpersistent_napari_isolated_per_case( native_reference_root=native_reference_root, require_native_reference=True, openhcs_global_config=global_config, - openhcs_execution_port=execution_port, discard_openhcs_outputs=False, continue_on_error=True, ) @@ -755,10 +745,6 @@ def test_official30_persistent_fiji_variants_isolated_per_case( assert len(_OFFICIAL30_CASES) == 30 excluded_ports: set[int] = set() - execution_port = _free_zmq_port_pair( - excluded_ports, - transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, - ) ports_by_viewer = { viewer_type: _free_zmq_port_pair( excluded_ports, @@ -812,7 +798,6 @@ def test_official30_persistent_fiji_variants_isolated_per_case( native_reference_root=native_reference_root, require_native_reference=True, openhcs_global_config=global_config, - openhcs_execution_port=execution_port, discard_openhcs_outputs=False, continue_on_error=True, ) From f9f5c67f4a86f05c24ef97faad0184951d5e08a4 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 20:28:08 -0400 Subject: [PATCH 04/15] WIP B7: private socket root for tests, real-server isolation tests Co-Authored-By: Claude Opus 5.5 --- external/zmqruntime | 2 +- openhcs/runtime/zmq_execution_client.py | 13 ++- openhcs/runtime/zmq_execution_server.py | 5 + tests/conftest.py | 12 ++ .../test_execution_server_isolation.py | 108 ++++++++++++++++++ 5 files changed, 137 insertions(+), 3 deletions(-) create mode 100644 tests/integration/test_execution_server_isolation.py diff --git a/external/zmqruntime b/external/zmqruntime index 0c2ad2f99c..3a0a24a3d7 160000 --- a/external/zmqruntime +++ b/external/zmqruntime @@ -1 +1 @@ -Subproject commit 0c2ad2f99c5bee202ec3297abfb2f37884920a3e +Subproject commit 3a0a24a3d7a30c34a72aab07d35e6921536b90b6 diff --git a/openhcs/runtime/zmq_execution_client.py b/openhcs/runtime/zmq_execution_client.py index c36bd7429f..6300d2c702 100644 --- a/openhcs/runtime/zmq_execution_client.py +++ b/openhcs/runtime/zmq_execution_client.py @@ -985,6 +985,10 @@ def __init__( connection_status_callback=connection_status_callback, ) + @classmethod + def code_root(cls) -> str: + return str(OpenHCSRuntimeImportAuthority.current().import_root) + @classmethod @contextmanager def with_private_server( @@ -1002,12 +1006,17 @@ def with_private_server( private_endpoint = config.transport_mode.declaration.private_endpoint with private_endpoint(config) as private_config: - with cls( + client = cls( persistent=False, progress_callback=progress_callback, config=private_config, - ) as client: + ) + if not client.connect(): + raise RuntimeError("The private execution server did not start") + try: yield client + finally: + client.disconnect() def connect( self, diff --git a/openhcs/runtime/zmq_execution_server.py b/openhcs/runtime/zmq_execution_server.py index 60a7dcfc75..788214bdc8 100644 --- a/openhcs/runtime/zmq_execution_server.py +++ b/openhcs/runtime/zmq_execution_server.py @@ -34,6 +34,7 @@ from openhcs.core.progress import ProgressEvent, ProgressExecutionContext from openhcs.core.steps.function_step import FunctionStep from openhcs.runtime.environment_provenance import RuntimeEnvironmentSnapshot +from openhcs.runtime.import_authority import OpenHCSRuntimeImportAuthority from openhcs.runtime.zmq_application import OPENHCS_ENDPOINT_APPLICATION from openhcs.runtime.zmq_compilation import ( ZMQCompilationRequest, @@ -167,6 +168,10 @@ class ZMQExecutionServer(FunctionCatalogExecutionServer): _server_type = "execution" + @classmethod + def code_root(cls) -> str: + return str(OpenHCSRuntimeImportAuthority.current().import_root) + def __init__( self, port: int | None = None, diff --git a/tests/conftest.py b/tests/conftest.py index 068fef6fc7..e3de3e9047 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,8 @@ """Global pytest configuration for OpenHCS integration tests.""" +import atexit import os +import shutil import signal import tempfile import uuid @@ -11,6 +13,16 @@ ensure_source_checkout_external_paths() +from openhcs.runtime.zmq_config import SOCKET_ROOT_ENV + +# Every socket and lock file of this pytest process, and of the servers and +# viewers it starts, lives under a root no other run sees. Each xdist worker +# imports this module, so each has its own. The path is short: ``sun_path`` +# holds about 104-108 bytes. +_SOCKET_ROOT = tempfile.mkdtemp(prefix="ohcs-sock-") +os.environ[SOCKET_ROOT_ENV] = _SOCKET_ROOT +atexit.register(shutil.rmtree, _SOCKET_ROOT, ignore_errors=True) + import psutil import pytest from zmqruntime.messages import AckReturnRoute, ProcessIdentity diff --git a/tests/integration/test_execution_server_isolation.py b/tests/integration/test_execution_server_isolation.py new file mode 100644 index 0000000000..58d4bd856f --- /dev/null +++ b/tests/integration/test_execution_server_isolation.py @@ -0,0 +1,108 @@ +"""A real execution client reaches only a server it may use (B7).""" + +from __future__ import annotations + +import subprocess +import sys +from dataclasses import replace + +import psutil +import pytest +from zmqruntime.messages import ( + EndpointLaunch, + EndpointLaunchMismatchError, + OwnerOnlyAudience, + ProcessIdentity, +) + +from openhcs.runtime.import_authority import OpenHCSRuntimeImportAuthority +from openhcs.runtime.zmq_execution_client import ZMQExecutionClient + +SERVER_START_TIMEOUT_SECONDS = 600 +ANOTHER_PROCESS = ProcessIdentity(pid=1, create_time=0.0) + + +class _ClientOfAnotherCheckout(ZMQExecutionClient): + def launch(self) -> EndpointLaunch: + return replace(super().launch(), code_root="/another/checkout") + + +class _ClientOfAnotherProcess(ZMQExecutionClient): + def launch(self) -> EndpointLaunch: + return replace(super().launch(), owner=ANOTHER_PROCESS) + + +@pytest.fixture(scope="module") +def private_client(): + with ZMQExecutionClient.with_private_server() as client: + yield client + + +def test_a_started_server_reports_the_launch_its_client_handed_it(private_client): + launch = private_client.connected_endpoint.launch + + assert launch == private_client.launch() + assert launch.code_root == str( + OpenHCSRuntimeImportAuthority.current().import_root + ) + assert launch.owner == ProcessIdentity.current() + assert launch.audience_type is OwnerOnlyAudience + assert private_client.owned_server_process_is_alive() is True + + +@pytest.mark.parametrize( + ("stray_type", "named_in_refusal"), + [ + (_ClientOfAnotherCheckout, "/another/checkout"), + (_ClientOfAnotherProcess, "for its own use"), + ], +) +def test_a_client_from_elsewhere_is_refused_at_the_same_address( + private_client, stray_type, named_in_refusal +): + stray = stray_type(persistent=False, config=private_client.config) + + with pytest.raises(EndpointLaunchMismatchError) as refused: + stray.connect(timeout=10) + + assert named_in_refusal in str(refused.value) + assert not stray.is_connected() + assert private_client.owned_server_process_is_alive() is True + assert private_client.endpoint.ping(private_client.config) is not None + + +_OWNER_SCRIPT = """ +import sys, time +from openhcs._source_dependencies import ensure_source_checkout_external_paths +ensure_source_checkout_external_paths() +from openhcs.runtime.zmq_execution_client import ZMQExecutionClient +with ZMQExecutionClient.with_private_server() as client: + print(client.connected_endpoint.process_identity.pid, flush=True) + time.sleep(3600) +""" + + +@pytest.mark.timeout(SERVER_START_TIMEOUT_SECONDS + 60) +def test_an_execution_server_exits_after_its_owner_is_killed(reap_spawned_processes): + import_root = str(OpenHCSRuntimeImportAuthority.current().import_root) + owner = subprocess.Popen( + [ + sys.executable, + "-c", + f"import sys; sys.path.insert(0, {import_root!r})" + _OWNER_SCRIPT, + ], + stdout=subprocess.PIPE, + text=True, + ) + try: + server = psutil.Process(int(owner.stdout.readline())) + workers_and_server = [server, *server.children(recursive=True)] + assert server.is_running() + + owner.kill() + owner.wait(timeout=10) + + _gone, alive = psutil.wait_procs(workers_and_server, timeout=30) + assert alive == [] + finally: + owner.kill() From 8f01769c51f8e92decf34c63505c6fd1eb436e94 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 20:44:31 -0400 Subject: [PATCH 05/15] WIP B7: guards and adapted harness tests Co-Authored-By: Claude Opus 5.5 --- .../agent/test_owned_runtime_bootstrap.py | 19 ++- tests/unit/test_b7_server_isolation_guards.py | 119 ++++++++++++++++++ tests/unit/test_matched_cellprofiler_batch.py | 5 +- tests/unit/test_openhcs_adapter.py | 8 ++ tests/unit/test_openhcs_measured_run.py | 15 ++- tests/unit/test_well_throughput_scaling.py | 17 ++- 6 files changed, 163 insertions(+), 20 deletions(-) create mode 100644 tests/unit/test_b7_server_isolation_guards.py diff --git a/tests/unit/agent/test_owned_runtime_bootstrap.py b/tests/unit/agent/test_owned_runtime_bootstrap.py index 6faed35b1a..6dbd0020a3 100644 --- a/tests/unit/agent/test_owned_runtime_bootstrap.py +++ b/tests/unit/agent/test_owned_runtime_bootstrap.py @@ -21,7 +21,9 @@ from openhcs.agent.path_policy import AgentPathPolicy, AgentPathPolicyError from openhcs.agent.services.runtime_server_service import RuntimeServerService from openhcs.mcp.context import OpenHCSAgentContext -from openhcs.runtime.zmq_config import OpenHCSZMQConfig +from zmqruntime.messages import EndpointAudience, EndpointLaunch + +from openhcs.runtime.zmq_config import SOCKET_ROOT_ENV, OpenHCSZMQConfig from openhcs.runtime.zmq_execution_client import ( ExecutionRuntimeLaunchPlan, ZMQExecutionClient, @@ -54,6 +56,7 @@ def stop(self, timeout=5, kill_timeout=2): @pytest.fixture def setup(tmp_path, monkeypatch): monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.delenv(SOCKET_ROOT_ENV) monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "data")) monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path / "cache")) policy = AgentPathPolicy.with_roots( @@ -159,7 +162,15 @@ def test_admitted_plan_owns_materialization_and_child_configuration( module.MemoryType, "subprocess_environment", lambda: environment ) - assert plan.spawn(endpoint, config, persistent=persistent) is native_process + launch_record = EndpointLaunch.by_this_process( + "/code", EndpointAudience.of_persistent(persistent) + ) + assert ( + plan.spawn(endpoint, config, persistent=persistent, launch=launch_record) + is native_process + ) + command = launch.call_args.args[0] + assert command[command.index("--launch") + 1] == launch_record.to_argument() intercepted_spawn.assert_not_called() launch.assert_called_once() policy_selection.assert_called_once_with(detached=persistent) @@ -202,7 +213,8 @@ def test_client_consumes_exact_admitted_plan_on_spawn_failure(setup, monkeypatch calls = [] - def consume(selected_plan, endpoint, config, *, persistent): + def consume(selected_plan, endpoint, config, *, persistent, launch): + assert launch == client.launch() calls.append((selected_plan, endpoint, config, persistent)) raise OSError("controlled no-child spawn failure") @@ -656,6 +668,7 @@ def test_close_native_transport_write_admission_precedes_lifecycle( new_home = tmp_path / "not-authorised" new_home.mkdir() monkeypatch.setenv("HOME", str(new_home)) + monkeypatch.delenv(SOCKET_ROOT_ENV, raising=False) policy = AgentPathPolicy.with_roots( readable_roots=(tmp_path,), writable_roots=(tmp_path / "data", tmp_path / "cache", tmp_path / ".openhcs"), diff --git a/tests/unit/test_b7_server_isolation_guards.py b/tests/unit/test_b7_server_isolation_guards.py new file mode 100644 index 0000000000..7d29d8a6b7 --- /dev/null +++ b/tests/unit/test_b7_server_isolation_guards.py @@ -0,0 +1,119 @@ +"""Guards for B7: the checks that keep a client on a server it may use.""" + +from __future__ import annotations + +import ast +import inspect +from pathlib import Path + +import zmqruntime.client +import zmqruntime.transport_modes +from zmqruntime.client import ZMQClient +from zmqruntime.config import ZMQConfig + +import openhcs +from openhcs.runtime.import_authority import OpenHCSRuntimeImportAuthority +from openhcs.runtime.zmq_config import ( + OPENHCS_ZMQ_CONFIG, + SOCKET_ROOT_ENV, + checkout_socket_root, +) +from openhcs.runtime.zmq_execution_client import ZMQExecutionClient +from openhcs.runtime.zmq_execution_server import ZMQExecutionServer + +REPOSITORY_ROOT = Path(openhcs.__file__).resolve().parents[1] +LAUNCH_CHECKS = {"_require_started_endpoint", "_require_usable_endpoint"} + + +def _functions_assigning_a_connection() -> dict[str, ast.FunctionDef]: + tree = ast.parse(inspect.getsource(ZMQClient)) + assigning = {} + for function in ast.walk(tree): + if not isinstance(function, ast.FunctionDef): + continue + for node in ast.walk(function): + if not isinstance(node, ast.Assign): + continue + targets = [ast.unparse(target) for target in node.targets] + cleared = isinstance(node.value, ast.Constant) and node.value.value is None + if "self._connection" in targets and not cleared: + assigning[function.name] = function + return assigning + + +def test_a_client_holds_a_connection_only_after_the_launch_check(): + assigning = _functions_assigning_a_connection() + + assert assigning, "ZMQClient no longer assigns _connection where this guard looks" + for name, function in assigning.items(): + if name == "__init__": + continue + called = { + node.func.attr + for node in ast.walk(function) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) + } + assert called & LAUNCH_CHECKS, ( + f"ZMQClient.{name} establishes a connection without checking " + "the server's launch" + ) + + +def test_socket_paths_come_from_the_socket_root(): + for module in (zmqruntime.transport_modes, zmqruntime.client): + assert "Path.home()" not in inspect.getsource(module), module.__name__ + # ZMQ endpoints only. The MCP resident-server socket (openhcs/mcp/socket.py) + # is a different transport and already names its checkout; it is U4's. + for package in ("runtime", "agent", "core", "pyqt_gui"): + for source in (REPOSITORY_ROOT / "openhcs" / package).rglob("*.py"): + if source.name == "zmq_config.py": + continue + text = source.read_text(encoding="utf-8") + assert "ipc_socket_dir" not in text or "Path.home()" not in text, source + + +def test_the_harness_picks_no_free_port_for_an_execution_server(): + for source in (REPOSITORY_ROOT / "benchmark").glob("*.py"): + assert "DataControlPortPairAuthority" not in source.read_text( + encoding="utf-8" + ), source + + +def test_client_and_server_state_the_import_root_as_their_code_root(): + import_root = str(OpenHCSRuntimeImportAuthority.current().import_root) + + assert ZMQExecutionClient.code_root() == import_root + assert ZMQExecutionServer.code_root() == import_root + + class ScriptClient(ZMQExecutionClient): + """A subclass defined elsewhere is still this checkout's client.""" + + assert ScriptClient.code_root() == import_root + + +def test_each_checkout_has_its_own_socket_root(monkeypatch, tmp_path): + assert OPENHCS_ZMQ_CONFIG.socket_root() != ZMQConfig(app_name="openhcs").socket_root() + + monkeypatch.delenv(SOCKET_ROOT_ENV) + here = OPENHCS_ZMQ_CONFIG.socket_root() + assert here == checkout_socket_root() + + other_checkout = tmp_path / "checkout" + (other_checkout / "openhcs").mkdir(parents=True) + (other_checkout / "openhcs" / "__init__.py").write_text("", encoding="utf-8") + monkeypatch.setattr( + OpenHCSRuntimeImportAuthority, + "current", + classmethod(lambda cls: cls(other_checkout)), + ) + assert OPENHCS_ZMQ_CONFIG.socket_root() != here + + monkeypatch.setenv(SOCKET_ROOT_ENV, str(tmp_path)) + assert OPENHCS_ZMQ_CONFIG.socket_root() == tmp_path + + +def test_this_test_run_has_a_private_socket_root(): + root = OPENHCS_ZMQ_CONFIG.socket_root() + + assert root != checkout_socket_root() + assert root.name.startswith("ohcs-sock-") diff --git a/tests/unit/test_matched_cellprofiler_batch.py b/tests/unit/test_matched_cellprofiler_batch.py index c9c7a8bef2..9949dfe78e 100644 --- a/tests/unit/test_matched_cellprofiler_batch.py +++ b/tests/unit/test_matched_cellprofiler_batch.py @@ -361,9 +361,10 @@ def reject_endpoint(*args, **kwargs): monkeypatch.setattr(matched_batch, "_run_case", admitted_case) monkeypatch.setattr( - matched_batch.DataControlPortPairAuthority, "acquire", reject_endpoint + matched_batch, + "ZMQExecutionClient", + SimpleNamespace(with_private_server=reject_endpoint), ) - monkeypatch.setattr(matched_batch, "ZMQExecutionClient", reject_endpoint) output = tmp_path / "preflight" assert ( matched_batch.main( diff --git a/tests/unit/test_openhcs_adapter.py b/tests/unit/test_openhcs_adapter.py index 766cc3ac75..d505b991fa 100644 --- a/tests/unit/test_openhcs_adapter.py +++ b/tests/unit/test_openhcs_adapter.py @@ -250,6 +250,10 @@ class FakeZMQExecutionClient: submitted: list[OpenHCSExecutionSubmission] = [] waits: list[str] = [] + @classmethod + def with_private_server(cls, *, progress_callback=None): + return cls(port=None, persistent=False, progress_callback=progress_callback) + def __init__(self, *, port, persistent, progress_callback): assert port is None self.progress_callback = progress_callback @@ -419,6 +423,10 @@ def test_benchmark_rejects_incompatible_execution_endpoint( tmp_path: Path, ) -> None: class IncompatibleZMQExecutionClient: + @classmethod + def with_private_server(cls, *, progress_callback=None): + return cls(port=None, persistent=False, progress_callback=progress_callback) + def __init__(self, *, port, persistent, progress_callback): self.connected_endpoint = SimpleNamespace( application=EndpointApplication("openhcs", "0.8.4"), diff --git a/tests/unit/test_openhcs_measured_run.py b/tests/unit/test_openhcs_measured_run.py index c23e6275ad..9e3095b4ec 100644 --- a/tests/unit/test_openhcs_measured_run.py +++ b/tests/unit/test_openhcs_measured_run.py @@ -49,6 +49,10 @@ def test_measured_run_validates_an_ordinary_pipeline_document( clients = [] class FakeClient: + @classmethod + def with_private_server(cls, *, progress_callback=None): + return cls(port=None, persistent=False, progress_callback=progress_callback) + def __init__(self, *, port, persistent, progress_callback): assert port is None assert persistent is False @@ -613,6 +617,10 @@ def test_measured_run_refuses_to_reuse_an_unowned_server( selected_ports = [] class AttachedClient: + @classmethod + def with_private_server(cls, *, progress_callback=None): + return cls(port=None, persistent=False, progress_callback=progress_callback) + def __init__(self, **kwargs): selected_ports.append(kwargs["port"]) self.connected_endpoint = SimpleNamespace( @@ -632,11 +640,6 @@ def owned_server_process_is_alive(self): return None monkeypatch.setattr(measured_run, "ZMQExecutionClient", AttachedClient) - monkeypatch.setattr( - measured_run.DataControlPortPairAuthority, - "acquire", - lambda *_args, **_kwargs: SimpleNamespace(data_port=5555), - ) submission = OpenHCSExecutionSubmission( subject_id=tmp_path, pipeline_document=PipelineDocumentCodec.from_values( @@ -658,4 +661,4 @@ def owned_server_process_is_alive(self): timing_observer=measured_run._ZMQProgressTimingObserver(), require_owned_server=True, ) - assert selected_ports == [5555] + assert selected_ports == [None] diff --git a/tests/unit/test_well_throughput_scaling.py b/tests/unit/test_well_throughput_scaling.py index 8d784fece1..ea139c7abb 100644 --- a/tests/unit/test_well_throughput_scaling.py +++ b/tests/unit/test_well_throughput_scaling.py @@ -88,19 +88,12 @@ def test_reused_server_suite_keeps_one_client_and_distinct_resume_identity( monkeypatch.setattr( well_throughput_scaling, "load_comparison_cases", lambda _: cases ) - monkeypatch.setattr( - well_throughput_scaling, - "DataControlPortPairAuthority", - SimpleNamespace( - acquire=lambda *_args, **_kwargs: SimpleNamespace(data_port=7801) - ), - ) opened: list[object] = [] closed: list[object] = [] class FakeClient: def __init__(self, *, port, persistent, progress_callback): - assert (port, persistent) == (7801, False) + assert (port, persistent) == (None, False) self.progress_callback = progress_callback def __enter__(self): @@ -110,7 +103,13 @@ def __enter__(self): def __exit__(self, *_args): closed.append(self) - monkeypatch.setattr(well_throughput_scaling, "ZMQExecutionClient", FakeClient) + monkeypatch.setattr( + well_throughput_scaling, + "measured_execution_client", + lambda *, execution_port, progress_callback: FakeClient( + port=execution_port, persistent=False, progress_callback=progress_callback + ), + ) observed: list[tuple[str, object]] = [] def fake_run_case_well_throughput(**kwargs): From 45dfa272550e84a025b4477e7272c3afbce44854 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 20:51:49 -0400 Subject: [PATCH 06/15] B7: receipt matches the built design; require zmqruntime 0.6.2 Co-Authored-By: Claude Opus 5.5 --- .../B7-server-isolation.md | 45 +++++++++++-------- pyproject.toml | 2 +- 2 files changed, 27 insertions(+), 20 deletions(-) diff --git a/docs/refactor/consolidation-20261009/B7-server-isolation.md b/docs/refactor/consolidation-20261009/B7-server-isolation.md index 8b9f160bdb..6deb8e4786 100644 --- a/docs/refactor/consolidation-20261009/B7-server-isolation.md +++ b/docs/refactor/consolidation-20261009/B7-server-isolation.md @@ -24,7 +24,7 @@ The three observed failures are one sequence: - The first run's next workflow finds the socket, gets a ready reply and attaches. It now runs on code imported from the other worktree. - When the other process ends, its server stops under the first run. `benchmark/openhcs_measured_run.py:297` reports "client entered without a connected endpoint" for the same collision seen from the other side. -`require_owned_server` exists in the harness (`openhcs_measured_run.py:305`) and would have refused; the parity path does not pass it. +`require_owned_server` exists in the harness (`openhcs_measured_run.py:305`) and would have refused; the parity path does not pass it. It stays: a measured run may still be given a port, and then needs a server it started. Why a server lived 31 hours: @@ -51,17 +51,24 @@ class EndpointLaunch(WireMessage): audience: str # wire name, derived from the family registry ``` -- `ZMQServer` holds its `EndpointLaunch` and reports it in every heartbeat (`PongResponse.launch`). `code_root` is derived from the server class, never passed by a caller. +- `ZMQServer` holds its `EndpointLaunch` and reports it in every heartbeat (`PongResponse.launch`). +- `code_root()` is a classmethod on clients and servers. Its default is the directory the class's top-level package is imported from. OpenHCS overrides it on both with `OpenHCSRuntimeImportAuthority.current().import_root`, the root its child processes are already launched from, so a client subclass defined in a script or a test is still this checkout's client. - **A client that starts a server** hands it the launch (`owner` = the client process, audience from `persistent`) and connects only if the server that becomes ready reports exactly that launch. - **A client that attaches** (an occupied endpoint in `connect`, or `connect_existing`) is refused with `EndpointLaunchMismatchError` unless the server reports a launch, its `code_root` equals the client's, and its audience admits the client process. The error names both code roots and the owner. - The check applies to endpoints on this host. A remote TCP host is an endpoint the user named; its paths and process ids mean nothing here. -- **Namespace.** `ZMQConfig.socket_root()` owns where socket and lock files live. `OpenHCSZMQConfig` derives it from the import root: `~/.openhcs//`. Two checkouts no longer share addresses, so the refusal is the second line, not the normal case. `OPENHCS_SOCKET_ROOT` replaces the root for one process tree. +- **Namespace.** `ZMQConfig.socket_root()` owns where socket and lock files live. `OpenHCSZMQConfig` derives it from the import root: `~/.openhcs//`. Two checkouts no longer share addresses, so the refusal is the second line, not the normal case. `OPENHCS_SOCKET_ROOT` replaces the root for one process tree; it is read at each call, and child processes inherit it. An absolute `ipc_socket_dir` still selects an explicit directory. - **Orphans.** The server checks `launch.owner` once a second in its message loop. When the owner is gone, an owner-only server cancels its work and stops; an any-session server stops once it has no running or queued execution. -- **Tests and harness.** `tests/conftest.py` sets `OPENHCS_SOCKET_ROOT` to a short private directory before OpenHCS is imported (one per pytest process, so per xdist worker). The harness uses `ZMQExecutionClient.with_private_server`, which takes its address from `TransportDeclaration.private_endpoint`. +- **Tests and harness.** `tests/conftest.py` sets `OPENHCS_SOCKET_ROOT` to a short private directory (one per pytest process, so per xdist worker). The harness uses `ZMQExecutionClient.with_private_server`, which takes its address from `TransportDeclaration.private_endpoint` (a new socket directory for IPC, the first unbound pair for TCP) and raises if the server does not start. `benchmark.openhcs_measured_run.measured_execution_client` is the one place the harness builds a client: private with no port, the named port otherwise. Identity that must match: code location and owning process (through the audience). Version stays where it is: `EndpointApplication`, compared by `ZMQClientService`. A server edited after it started has the same root and version and is not detected; a source digest would be needed and no surface owns that. -Deleted: the three `DataControlPortPairAuthority.acquire` + `ZMQExecutionClient(port=...)` pairs in the harness, `require_owned_server`, the parity test's `_free_zmq_port_pair`, and the hand-built `Path.home() / ".openhcs" / "ipc"` in `viewer_endpoint_discovery.py`. +Deleted (all **derived**: `private_endpoint` and `socket_root` now own the fact): + +- the three `DataControlPortPairAuthority.acquire` + `ZMQExecutionClient(port=...)` pairs in the harness; +- the execution port the three official30 tests chose (`_free_zmq_port_pair` stays for viewer ports); +- the hand-built `Path.home() / ".openhcs" / "ipc"` in `viewer_endpoint_discovery.py`. + +Behaviour removed: a measured run with no port and without `require_owned_server` no longer uses the shared default port 7777; it starts a private server (**contract**: this surface's ownership rule). ## Persisted state @@ -69,16 +76,19 @@ Runtime only. Socket and lock files move from `~/.openhcs/ipc` and `~/.openhcs/t ## Guards -- No `Path.home()` in `openhcs/` or `zmqruntime` builds a socket path outside `socket_root()`. -- No harness or test file pairs `DataControlPortPairAuthority.acquire(OPENHCS_ZMQ_CONFIG` with an execution client. -- `ZMQClient` assigns `_connection` only after the launch check. +`tests/unit/test_b7_server_isolation_guards.py`: + +- Every `ZMQClient` method that assigns a connection calls a launch check (AST). +- No ZMQ endpoint code (`zmqruntime.transport_modes`, `zmqruntime.client`, `openhcs/{runtime,agent,core,pyqt_gui}`) builds a socket path from `Path.home()`. +- No module in `benchmark/` names `DataControlPortPairAuthority`. +- Client and server state the import root as their code root, including a subclass defined elsewhere. +- Each checkout has its own socket root, and the test run's root is private. ## Tests -- Two servers from two code roots: each client reaches only its own; a client aimed at the other's address is refused with both roots in the message. -- An owner-only server refuses a second process and admits its owner. -- An orphaned server exits: owner-only at once, any-session after its work. -- zmqruntime: the same three at the library level with a minimal server. +- zmqruntime `tests/test_endpoint_launch.py`, with real server processes: two servers from two code roots, each client reaches only its own, and a client aimed at the other's address is refused with both roots in the message; an owner-only server refuses another process and a persistent one admits it; a server that reports no launch is refused; a server exits after its owner is killed (both audiences); a persistent orphan finishes its work first. +- OpenHCS `tests/integration/test_execution_server_isolation.py`, with the real launcher: the started server reports the launch its client handed it; a client of another checkout and a client of another process are refused at the same address and the server is unharmed; the server and its children exit after the owner is killed. +- Both orphan tests and the two-root test were run against the code with the check removed and fail there. ## New-case experiments @@ -96,10 +106,7 @@ A new kind of server (a viewer, a bridge) that must not be shared across checkou | Item | Evidence | Belongs to | |---|---|---| | UI-bridge descriptor discovery takes the only live descriptor from any checkout | `ui_bridge_service.py:1563-1585`, `/proc` environment scan at `:1599` | U4 | - -## Progress notes - -- [x] Mechanism measured, receipt written. -- [ ] zmqruntime: `EndpointLaunch`, client check, orphan stop, `socket_root`, `private_endpoint`. -- [ ] OpenHCS: launch argument, socket root, harness, conftest, guards. -- [ ] Parity, ratchet, nra, CI. +| The MCP resident-server socket restates the checkout digest and `~/.openhcs/ipc` | `openhcs/mcp/socket.py:48,61`; it should read `OPENHCS_ZMQ_CONFIG.socket_root()` | U4 | +| `connect` kills the listener of an unresponsive TCP address before starting its own server | `zmqruntime/client.py` `_kill_processes_on_port`; the process may be another session's | zmqruntime (L3) | +| The application version is compared after connecting, and only by `ZMQClientService` and the catalog client | `authoring/session/execution_client.py:202`; a plain client attaches to a same-root server of another version | U1 | +| `viewer_endpoint_discovery.py:212` names an undefined `ViewerEndpointDiscoveryResult` | ruff F821 at this head | U4 | diff --git a/pyproject.toml b/pyproject.toml index d6f3007b95..96b060a695 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -99,7 +99,7 @@ dependencies = [ "psutil>=5.9.0", # External dependencies (PyPI versions for production builds) - "zmqruntime>=0.6.1,<0.7", + "zmqruntime>=0.6.2,<0.7", "pycodify>=0.2.0,<0.3", "objectstate>=2.1,<3", "python-introspect>=0.2.8,<0.3", From 1a7be0dcbb6bd4e56618598b8edbdd1a480ba593 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 21:19:13 -0400 Subject: [PATCH 07/15] B7: client and server inherit their code root from one mixin Co-Authored-By: Claude Opus 5.5 --- openhcs/runtime/import_authority.py | 13 +++++++++++++ openhcs/runtime/zmq_execution_client.py | 11 +++++------ openhcs/runtime/zmq_execution_server.py | 8 ++------ 3 files changed, 20 insertions(+), 12 deletions(-) diff --git a/openhcs/runtime/import_authority.py b/openhcs/runtime/import_authority.py index 17916eeac2..4cc50df0a4 100644 --- a/openhcs/runtime/import_authority.py +++ b/openhcs/runtime/import_authority.py @@ -67,3 +67,16 @@ def module_process_arguments(self, module_name: str) -> tuple[str, str]: """Project one module launch to interpreter arguments.""" return "-c", self.module_bootstrap_code(module_name) + + +class ImportRootEndpoint: + """A ZMQ client or server whose code root is the OpenHCS import root. + + Both sides of an execution endpoint inherit this, so the root a server + reports and the root a client expects are one declaration: the root every + OpenHCS child process is launched from. + """ + + @classmethod + def code_root(cls) -> str: + return str(OpenHCSRuntimeImportAuthority.current().import_root) diff --git a/openhcs/runtime/zmq_execution_client.py b/openhcs/runtime/zmq_execution_client.py index 6300d2c702..c96ff66961 100644 --- a/openhcs/runtime/zmq_execution_client.py +++ b/openhcs/runtime/zmq_execution_client.py @@ -67,7 +67,10 @@ ) from openhcs.core.steps.function_step import FunctionStep from openhcs.core.xdg_paths import get_openhcs_data_dir, get_openhcs_log_dir -from openhcs.runtime.import_authority import OpenHCSRuntimeImportAuthority +from openhcs.runtime.import_authority import ( + ImportRootEndpoint, + OpenHCSRuntimeImportAuthority, +) from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG, OpenHCSZMQConfig from openhcs.runtime.zmq_execution_signature import ( ZMQAuxiliaryExecutionParams, @@ -955,7 +958,7 @@ def _send_function_catalog_exchange( ) -class ZMQExecutionClient(FunctionCatalogExecutionClient): +class ZMQExecutionClient(ImportRootEndpoint, FunctionCatalogExecutionClient): """ZMQ client for OpenHCS pipeline execution with progress streaming.""" config: OpenHCSZMQConfig @@ -985,10 +988,6 @@ def __init__( connection_status_callback=connection_status_callback, ) - @classmethod - def code_root(cls) -> str: - return str(OpenHCSRuntimeImportAuthority.current().import_root) - @classmethod @contextmanager def with_private_server( diff --git a/openhcs/runtime/zmq_execution_server.py b/openhcs/runtime/zmq_execution_server.py index 788214bdc8..0c99f02967 100644 --- a/openhcs/runtime/zmq_execution_server.py +++ b/openhcs/runtime/zmq_execution_server.py @@ -34,7 +34,7 @@ from openhcs.core.progress import ProgressEvent, ProgressExecutionContext from openhcs.core.steps.function_step import FunctionStep from openhcs.runtime.environment_provenance import RuntimeEnvironmentSnapshot -from openhcs.runtime.import_authority import OpenHCSRuntimeImportAuthority +from openhcs.runtime.import_authority import ImportRootEndpoint from openhcs.runtime.zmq_application import OPENHCS_ENDPOINT_APPLICATION from openhcs.runtime.zmq_compilation import ( ZMQCompilationRequest, @@ -163,15 +163,11 @@ def stop(self) -> None: self._function_catalog_preparation.cancel_and_join() -class ZMQExecutionServer(FunctionCatalogExecutionServer): +class ZMQExecutionServer(ImportRootEndpoint, FunctionCatalogExecutionServer): """OpenHCS-specific execution server.""" _server_type = "execution" - @classmethod - def code_root(cls) -> str: - return str(OpenHCSRuntimeImportAuthority.current().import_root) - def __init__( self, port: int | None = None, From 6465476ab2e15ec7549fc70944b5d6e1735445e5 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 21:22:12 -0400 Subject: [PATCH 08/15] B7: the launcher records the launch on the server it builds Co-Authored-By: Claude Opus 5.5 --- external/zmqruntime | 2 +- openhcs/runtime/zmq_execution_server.py | 3 --- openhcs/runtime/zmq_execution_server_launcher.py | 3 ++- 3 files changed, 3 insertions(+), 5 deletions(-) diff --git a/external/zmqruntime b/external/zmqruntime index 3a0a24a3d7..4b54fa844a 160000 --- a/external/zmqruntime +++ b/external/zmqruntime @@ -1 +1 @@ -Subproject commit 3a0a24a3d7a30c34a72aab07d35e6921536b90b6 +Subproject commit 4b54fa844a16e444e62da64c2695ac935f3088f7 diff --git a/openhcs/runtime/zmq_execution_server.py b/openhcs/runtime/zmq_execution_server.py index 0c99f02967..7ef62bfe2d 100644 --- a/openhcs/runtime/zmq_execution_server.py +++ b/openhcs/runtime/zmq_execution_server.py @@ -15,7 +15,6 @@ from zmqruntime.config import TransportMode from zmqruntime.execution import ExecutionServer from zmqruntime.messages import ( - EndpointLaunch, ExecuteRequest, ExecutionRecord, ExecutionStatus, @@ -175,7 +174,6 @@ def __init__( log_file_path: str | None = None, transport_mode: TransportMode | None = None, config: OpenHCSZMQConfig = OPENHCS_ZMQ_CONFIG, - launch: EndpointLaunch | None = None, ): super().__init__( port=config.default_port if port is None else port, @@ -186,7 +184,6 @@ def __init__( ), config=config, application=OPENHCS_ENDPOINT_APPLICATION, - launch=launch, ) self._compile_status: str | None = None self._compile_message: str | None = None diff --git a/openhcs/runtime/zmq_execution_server_launcher.py b/openhcs/runtime/zmq_execution_server_launcher.py index b7ff3f2b14..c24a54974c 100644 --- a/openhcs/runtime/zmq_execution_server_launcher.py +++ b/openhcs/runtime/zmq_execution_server_launcher.py @@ -199,8 +199,9 @@ def main( server = execution_server_type( log_file_path=args.log_file_path, config=config, - launch=args.launch, ) + if args.launch is not None: + server.launched_by(args.launch) status_reporter.emit( EndpointStartupPhase.PREPARING_CAPABILITIES, From aadfa7145a0d636daf56671758062a76a58ab2b7 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 21:24:28 -0400 Subject: [PATCH 09/15] B7: the launch argument defaults to a server started by hand Co-Authored-By: Claude Opus 5.5 --- openhcs/runtime/zmq_execution_server_launcher.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/openhcs/runtime/zmq_execution_server_launcher.py b/openhcs/runtime/zmq_execution_server_launcher.py index c24a54974c..4e46f02f65 100644 --- a/openhcs/runtime/zmq_execution_server_launcher.py +++ b/openhcs/runtime/zmq_execution_server_launcher.py @@ -12,13 +12,15 @@ from pathlib import Path from zmqruntime.config import TransportMode -from zmqruntime.messages import EndpointLaunch +from zmqruntime.messages import AnySessionAudience, EndpointLaunch from zmqruntime.startup import ( EndpointStartupPhase, EndpointStartupStatusWriter, ) from zmqruntime.transport import TransportEndpoint +from openhcs.runtime.import_authority import ImportRootEndpoint + logger = logging.getLogger(__name__) @@ -73,8 +75,10 @@ def main( parser.add_argument( "--launch", type=EndpointLaunch.from_argument, - default=None, - help="Launch record of the client that starts this server", + default=EndpointLaunch.by_this_process( + ImportRootEndpoint.code_root(), AnySessionAudience + ), + help="Launch record of the starting client (default: started by hand)", ) parser.add_argument( "--startup-status-path", @@ -199,9 +203,7 @@ def main( server = execution_server_type( log_file_path=args.log_file_path, config=config, - ) - if args.launch is not None: - server.launched_by(args.launch) + ).launched_by(args.launch) status_reporter.emit( EndpointStartupPhase.PREPARING_CAPABILITIES, From 7276148a6cf41dd0b7b0406d38854a10aa586d9b Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 21:31:56 -0400 Subject: [PATCH 10/15] B7: launcher test fakes record the launch Co-Authored-By: Claude Opus 5.5 --- tests/unit/test_zmq_execution_server_process.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/unit/test_zmq_execution_server_process.py b/tests/unit/test_zmq_execution_server_process.py index f0925df0b0..132c390963 100644 --- a/tests/unit/test_zmq_execution_server_process.py +++ b/tests/unit/test_zmq_execution_server_process.py @@ -114,6 +114,11 @@ class _Server: def __init__(self, **_kwargs) -> None: events.append("construct") + def launched_by(self, launch): + # No --launch argument: a server started by hand owns itself. + assert launch.owner == ProcessIdentity.current() + return self + def start(self) -> None: events.append("start") @@ -149,6 +154,9 @@ class _Server: def __init__(self, **kwargs) -> None: captured.update(kwargs) + def launched_by(self, _launch): + return self + def prepare_runtime_capabilities(self, _status_callback) -> None: return None From cfe74f96a9a461d921ba969992d8cabbc5b660ec Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 21:41:41 -0400 Subject: [PATCH 11/15] B7: receipt records the result and release order Co-Authored-By: Claude Opus 5.5 --- .../consolidation-20261009/B7-server-isolation.md | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/docs/refactor/consolidation-20261009/B7-server-isolation.md b/docs/refactor/consolidation-20261009/B7-server-isolation.md index 6deb8e4786..a3d0b0189a 100644 --- a/docs/refactor/consolidation-20261009/B7-server-isolation.md +++ b/docs/refactor/consolidation-20261009/B7-server-isolation.md @@ -52,8 +52,8 @@ class EndpointLaunch(WireMessage): ``` - `ZMQServer` holds its `EndpointLaunch` and reports it in every heartbeat (`PongResponse.launch`). -- `code_root()` is a classmethod on clients and servers. Its default is the directory the class's top-level package is imported from. OpenHCS overrides it on both with `OpenHCSRuntimeImportAuthority.current().import_root`, the root its child processes are already launched from, so a client subclass defined in a script or a test is still this checkout's client. -- **A client that starts a server** hands it the launch (`owner` = the client process, audience from `persistent`) and connects only if the server that becomes ready reports exactly that launch. +- `code_root()` is a classmethod on clients and servers. Its default is the directory the class's top-level package is imported from. OpenHCS's client and server both inherit `ImportRootEndpoint` (`openhcs/runtime/import_authority.py`), which states `OpenHCSRuntimeImportAuthority.current().import_root`, the root its child processes are already launched from, so a client subclass defined in a script or a test is still this checkout's client. +- **A client that starts a server** hands it the launch (`owner` = the client process, audience from `persistent`) as the launcher's `--launch` argument, which the launcher records with `ZMQServer.launched_by`. A launcher run by hand defaults to a server its own process started. The client connects only if the server that becomes ready reports exactly that launch. - **A client that attaches** (an occupied endpoint in `connect`, or `connect_existing`) is refused with `EndpointLaunchMismatchError` unless the server reports a launch, its `code_root` equals the client's, and its audience admits the client process. The error names both code roots and the owner. - The check applies to endpoints on this host. A remote TCP host is an endpoint the user named; its paths and process ids mean nothing here. - **Namespace.** `ZMQConfig.socket_root()` owns where socket and lock files live. `OpenHCSZMQConfig` derives it from the import root: `~/.openhcs//`. Two checkouts no longer share addresses, so the refusal is the second line, not the normal case. `OPENHCS_SOCKET_ROOT` replaces the root for one process tree; it is read at each call, and child processes inherit it. An absolute `ipc_socket_dir` still selects an explicit directory. @@ -110,3 +110,12 @@ A new kind of server (a viewer, a bridge) that must not be shared across checkou | `connect` kills the listener of an unresponsive TCP address before starting its own server | `zmqruntime/client.py` `_kill_processes_on_port`; the process may be another session's | zmqruntime (L3) | | The application version is compared after connecting, and only by `ZMQClientService` and the catalog client | `authoring/session/execution_client.py:202`; a plain client attaches to a same-root server of another version | U1 | | `viewer_endpoint_discovery.py:212` names an undefined `ViewerEndpointDiscoveryResult` | ruff F821 at this head | U4 | + +## Result + +- Production (`openhcs/`): -12 +102. The additions are the socket root, the launch argument, `with_private_server` and `ImportRootEndpoint`; nothing existed to delete, because the relation was missing. zmqruntime `src`: -5 +331. +- Guards: `tests/unit/test_b7_server_isolation_guards.py`, 6 of 6 tests pass. +- Certificates: the three harness `acquire` + client pairs, the test-chosen execution port and the hand-built socket directory are **derived** (`private_endpoint`, `socket_root`). The shared default port for a measured run with no port is **contract** (this surface's rule). No **tests-only** deletion. +- Files of in-flight surfaces touched: `openhcs/agent/services/viewer_endpoint_discovery.py` (U4, one expression); `openhcs/runtime/zmq_execution_client.py`, `zmq_execution_server.py`, `zmq_execution_server_launcher.py` (near K7); `benchmark/openhcs_measured_run.py`, `well_throughput_scaling.py`, `matched_cellprofiler_batch.py` (near K1 and K8, client construction only). + +Release order: merge OpenHCSDev/ZMQRuntime#32 and release zmqruntime 0.6.2; point `external/zmqruntime` at that merge commit; merge this PR. From 7c07efe5842452af81cbd30160ef7b6fadfa87ef Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 11 Oct 2026 23:11:41 -0400 Subject: [PATCH 12/15] B7: zmqruntime 0.6.3 carries server isolation; bound and submodule follow Co-Authored-By: Claude Opus 5.5 --- docs/refactor/consolidation-20261009/B7-server-isolation.md | 2 +- external/zmqruntime | 2 +- pyproject.toml | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/refactor/consolidation-20261009/B7-server-isolation.md b/docs/refactor/consolidation-20261009/B7-server-isolation.md index a3d0b0189a..032cda5484 100644 --- a/docs/refactor/consolidation-20261009/B7-server-isolation.md +++ b/docs/refactor/consolidation-20261009/B7-server-isolation.md @@ -118,4 +118,4 @@ A new kind of server (a viewer, a bridge) that must not be shared across checkou - Certificates: the three harness `acquire` + client pairs, the test-chosen execution port and the hand-built socket directory are **derived** (`private_endpoint`, `socket_root`). The shared default port for a measured run with no port is **contract** (this surface's rule). No **tests-only** deletion. - Files of in-flight surfaces touched: `openhcs/agent/services/viewer_endpoint_discovery.py` (U4, one expression); `openhcs/runtime/zmq_execution_client.py`, `zmq_execution_server.py`, `zmq_execution_server_launcher.py` (near K7); `benchmark/openhcs_measured_run.py`, `well_throughput_scaling.py`, `matched_cellprofiler_batch.py` (near K1 and K8, client construction only). -Release order: merge OpenHCSDev/ZMQRuntime#32 and release zmqruntime 0.6.2; point `external/zmqruntime` at that merge commit; merge this PR. +Release order: merge OpenHCSDev/ZMQRuntime#32 and release zmqruntime 0.6.3 (0.6.2 went to the Windows startup-lock fix); point `external/zmqruntime` at that merge commit; merge this PR. diff --git a/external/zmqruntime b/external/zmqruntime index 4b54fa844a..20670ca0a9 160000 --- a/external/zmqruntime +++ b/external/zmqruntime @@ -1 +1 @@ -Subproject commit 4b54fa844a16e444e62da64c2695ac935f3088f7 +Subproject commit 20670ca0a92a3e09eb02c47b898371c88572a8fe diff --git a/pyproject.toml b/pyproject.toml index 96b060a695..1f18ffc9a4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -99,7 +99,7 @@ dependencies = [ "psutil>=5.9.0", # External dependencies (PyPI versions for production builds) - "zmqruntime>=0.6.2,<0.7", + "zmqruntime>=0.6.3,<0.7", "pycodify>=0.2.0,<0.3", "objectstate>=2.1,<3", "python-introspect>=0.2.8,<0.3", From 8fea7f33e2d62442642d164459f567fcfa1fe619 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 12 Oct 2026 01:54:51 -0400 Subject: [PATCH 13/15] B7: master viewer endpoint test reads the socket root instead of ~/.openhcs/ipc The test pinned the per-user ~/.openhcs/ipc directory that B7 replaced with OpenHCSZMQConfig.socket_root(). It now derives the directory from that config and checks that the viewer endpoint, viewer discovery and an execution endpoint on the same port resolve the same address. Co-Authored-By: Claude Opus 5.5 --- tests/unit/test_master_multi_plate_demo.py | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_master_multi_plate_demo.py b/tests/unit/test_master_multi_plate_demo.py index a43a7366d3..b1d0382c8f 100644 --- a/tests/unit/test_master_multi_plate_demo.py +++ b/tests/unit/test_master_multi_plate_demo.py @@ -6,7 +6,10 @@ import pytest from polystore.streaming.identity import StreamProducerIdentity +from zmqruntime.transport import get_control_url, get_zmq_transport_url +from zmqruntime.transport_modes import TransportMode +from openhcs.agent.services.viewer_endpoint_discovery import viewer_ipc_directory from openhcs.core.artifacts.kinds import ImageArtifactType, MeasurementsArtifactType from openhcs.core.steps.function_step import FunctionStep from openhcs.processing.presets.demo_contribution import PipelineDemoContribution @@ -1354,5 +1357,15 @@ def test_master_viewer_endpoint_uses_the_openhcs_transport_owner(tmp_path): endpoint = master.McpMasterDemoOperations._endpoint(item) assert endpoint.config is master.OPENHCS_ZMQ_CONFIG - assert endpoint.data_url().endswith("/.openhcs/ipc/openhcs-zmq-6200.sock") - assert endpoint.control_url().endswith("/.openhcs/ipc/openhcs-zmq-7200.sock") + config = master.OPENHCS_ZMQ_CONFIG + socket_directory = config.socket_root() / config.ipc_socket_dir + assert viewer_ipc_directory() == socket_directory + assert endpoint.data_url() == f"ipc://{socket_directory}/openhcs-zmq-6200.sock" + assert endpoint.control_url() == f"ipc://{socket_directory}/openhcs-zmq-7200.sock" + # An execution endpoint on the same port resolves the same address. + assert endpoint.data_url() == get_zmq_transport_url( + 6200, mode=TransportMode.IPC, config=config + ) + assert endpoint.control_url() == get_control_url( + 6200, TransportMode.IPC, config=config + ) From 2a3afeaa0fc2a1fe23fa89e8666cb5a00fd3d41b Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 12 Oct 2026 02:53:46 -0400 Subject: [PATCH 14/15] B7: pin released zmqruntime 0.6.3 Co-Authored-By: Claude Opus 5.5 --- external/zmqruntime | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/external/zmqruntime b/external/zmqruntime index 20670ca0a9..beea6b5e3d 160000 --- a/external/zmqruntime +++ b/external/zmqruntime @@ -1 +1 @@ -Subproject commit 20670ca0a92a3e09eb02c47b898371c88572a8fe +Subproject commit beea6b5e3d8b9c60dfe3ba773dba975c473f0036 From 08f861fb87579b306af359fdf6d6e41f2a19f5d3 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 12 Oct 2026 03:34:06 -0400 Subject: [PATCH 15/15] B7: private socket root sits beneath the checkout root, not TMPDIR; unbindable paths and failed connects end the wait The smoke check sets TMPDIR to a deep directory under the worktree. The test run's socket root came from tempfile.mkdtemp, so every socket path was 136 bytes against a limit of 107 (zmq.IPC_PATH_MAX_LEN). Each server died at bind, the client reported only 'Failed to connect', and the measured CLI waited for a compilation that no event would ever end. - use_private_socket_root() owns the test run's root: a new directory beneath checkout_socket_root(), exported as OPENHCS_SOCKET_ROOT for every child. tests/conftest.py and the unix_socket_dir fixture use it. - ZMQExecutionClient.connect raises SocketPathTooLongError (path, length, limit) before it starts a server. - A compile that cannot connect publishes CompilationFailed per dataset, so the measured CLI and other waiters end. - Guards: a child process computes its parent's socket root, code root and audience; a private root binds whatever TMPDIR is; the client refuses an unbindable path without starting a server. Co-Authored-By: Claude Opus 5.5 --- .../B7-server-isolation.md | 5 +- openhcs/authoring/session/compile_batch.py | 9 +- openhcs/runtime/zmq_config.py | 40 ++++++++ openhcs/runtime/zmq_execution_client.py | 11 ++- tests/conftest.py | 17 ++-- .../pyqt_gui/test_session_batch_workflows.py | 10 ++ tests/unit/test_b7_server_isolation_guards.py | 98 ++++++++++++++++++- 7 files changed, 175 insertions(+), 15 deletions(-) diff --git a/docs/refactor/consolidation-20261009/B7-server-isolation.md b/docs/refactor/consolidation-20261009/B7-server-isolation.md index 032cda5484..d8a35d7e2d 100644 --- a/docs/refactor/consolidation-20261009/B7-server-isolation.md +++ b/docs/refactor/consolidation-20261009/B7-server-isolation.md @@ -58,7 +58,8 @@ class EndpointLaunch(WireMessage): - The check applies to endpoints on this host. A remote TCP host is an endpoint the user named; its paths and process ids mean nothing here. - **Namespace.** `ZMQConfig.socket_root()` owns where socket and lock files live. `OpenHCSZMQConfig` derives it from the import root: `~/.openhcs//`. Two checkouts no longer share addresses, so the refusal is the second line, not the normal case. `OPENHCS_SOCKET_ROOT` replaces the root for one process tree; it is read at each call, and child processes inherit it. An absolute `ipc_socket_dir` still selects an explicit directory. - **Orphans.** The server checks `launch.owner` once a second in its message loop. When the owner is gone, an owner-only server cancels its work and stops; an any-session server stops once it has no running or queued execution. -- **Tests and harness.** `tests/conftest.py` sets `OPENHCS_SOCKET_ROOT` to a short private directory (one per pytest process, so per xdist worker). The harness uses `ZMQExecutionClient.with_private_server`, which takes its address from `TransportDeclaration.private_endpoint` (a new socket directory for IPC, the first unbound pair for TCP) and raises if the server does not start. `benchmark.openhcs_measured_run.measured_execution_client` is the one place the harness builds a client: private with no port, the named port otherwise. +- **Tests and harness.** `tests/conftest.py` calls `use_private_socket_root()`, which makes a new directory beneath the checkout's socket root and exports it as `OPENHCS_SOCKET_ROOT` (one per pytest process, so per xdist worker). The directory is not under `TMPDIR`: a socket path longer than `zmq.IPC_PATH_MAX_LEN` (107) cannot be bound, and a run with a deep `TMPDIR` put every server there. The `unix_socket_dir` fixture is a subdirectory of the same root. +- **A path no server can bind.** `ZMQExecutionClient.connect` raises `SocketPathTooLongError` (path, length, limit) before it starts a server. A session whose compile cannot connect publishes `CompilationFailed` per dataset, so a caller waiting for the compilation (the measured CLI) ends instead of waiting. The harness uses `ZMQExecutionClient.with_private_server`, which takes its address from `TransportDeclaration.private_endpoint` (a new socket directory for IPC, the first unbound pair for TCP) and raises if the server does not start. `benchmark.openhcs_measured_run.measured_execution_client` is the one place the harness builds a client: private with no port, the named port otherwise. Identity that must match: code location and owning process (through the audience). Version stays where it is: `EndpointApplication`, compared by `ZMQClientService`. A server edited after it started has the same root and version and is not detected; a source digest would be needed and no surface owns that. @@ -83,6 +84,7 @@ Runtime only. Socket and lock files move from `~/.openhcs/ipc` and `~/.openhcs/t - No module in `benchmark/` names `DataControlPortPairAuthority`. - Client and server state the import root as their code root, including a subclass defined elsewhere. - Each checkout has its own socket root, and the test run's root is private. +- A child process computes its parent's socket root, code root and audience; a private root can be bound whatever `TMPDIR` is; a client refuses a path too long to bind without starting a server. ## Tests @@ -109,6 +111,7 @@ A new kind of server (a viewer, a bridge) that must not be shared across checkou | The MCP resident-server socket restates the checkout digest and `~/.openhcs/ipc` | `openhcs/mcp/socket.py:48,61`; it should read `OPENHCS_ZMQ_CONFIG.socket_root()` | U4 | | `connect` kills the listener of an unresponsive TCP address before starting its own server | `zmqruntime/client.py` `_kill_processes_on_port`; the process may be another session's | zmqruntime (L3) | | The application version is compared after connecting, and only by `ZMQClientService` and the catalog client | `authoring/session/execution_client.py:202`; a plain client attaches to a same-root server of another version | U1 | +| `IpcTransportDeclaration.private_endpoint` makes its directory under `TMPDIR`, not under `config.socket_root()`; with a deep `TMPDIR` a private server is refused by the length check | `zmqruntime/transport_modes.py` `private_endpoint` | zmqruntime | | `viewer_endpoint_discovery.py:212` names an undefined `ViewerEndpointDiscoveryResult` | ruff F821 at this head | U4 | ## Result diff --git a/openhcs/authoring/session/compile_batch.py b/openhcs/authoring/session/compile_batch.py index 003deaf30a..ba3b2ec145 100644 --- a/openhcs/authoring/session/compile_batch.py +++ b/openhcs/authoring/session/compile_batch.py @@ -48,7 +48,14 @@ async def compile_datasets(self, scope_ids: tuple[str, ...]) -> None: session.progress_tracker.clear_execution(previous_execution_id) session.mark_compile_pending(scope_ids) try: - client = await session.connect_client() + try: + client = await session.connect_client() + except Exception as error: + # Without a server no dataset compiles: say so per dataset, so + # whoever waits for the compilation stops waiting. + for scope_id in scope_ids: + self._compile_failed(scope_id, error) + raise session.publish( StatusReported(f"Queueing compilation for {len(scope_ids)} dataset(s)...") ) diff --git a/openhcs/runtime/zmq_config.py b/openhcs/runtime/zmq_config.py index 8f2ad559fa..1f448cb667 100644 --- a/openhcs/runtime/zmq_config.py +++ b/openhcs/runtime/zmq_config.py @@ -2,11 +2,15 @@ from __future__ import annotations +import atexit import hashlib import os +import shutil +import tempfile from dataclasses import dataclass from pathlib import Path +import zmq from zmqruntime.config import NonBlankString from zmqruntime.execution.config import ExecutionTransportConfig @@ -32,6 +36,42 @@ def checkout_socket_root() -> Path: return Path.home() / ".openhcs" / digest +def use_private_socket_root() -> Path: + """Give this process and its children a socket root of their own. + + The root is a new directory beneath the checkout's socket root, so its + length does not depend on ``TMPDIR``: a socket path longer than + ``zmq.IPC_PATH_MAX_LEN`` cannot be bound. It is removed at exit. + """ + + parent = checkout_socket_root() + parent.mkdir(parents=True, exist_ok=True) + root = Path(tempfile.mkdtemp(prefix="run-", dir=parent)) + os.environ[SOCKET_ROOT_ENV] = str(root) + atexit.register(shutil.rmtree, root, ignore_errors=True) + return root + + +class SocketPathTooLongError(ValueError): + """An IPC socket path that no server can bind.""" + + +def require_bindable_address(url: str) -> None: + """Refuse an IPC address the operating system cannot bind. + + A server asked to bind such a path fails at startup, and its client would + only learn that the server never became ready. + """ + + scheme, _, path = url.partition("://") + if scheme == "ipc" and len(os.fsencode(path)) > zmq.IPC_PATH_MAX_LEN: + raise SocketPathTooLongError( + f"Socket path {path} is {len(os.fsencode(path))} bytes; the limit " + f"is {zmq.IPC_PATH_MAX_LEN}. Set {SOCKET_ROOT_ENV} to a shorter " + "directory (a private server takes its directory from TMPDIR)." + ) + + @dataclass(frozen=True, slots=True) class OpenHCSZMQConfig(ExecutionTransportConfig): """zmqruntime's execution transport under OpenHCS's application names.""" diff --git a/openhcs/runtime/zmq_execution_client.py b/openhcs/runtime/zmq_execution_client.py index c96ff66961..838ee50ab0 100644 --- a/openhcs/runtime/zmq_execution_client.py +++ b/openhcs/runtime/zmq_execution_client.py @@ -71,7 +71,11 @@ ImportRootEndpoint, OpenHCSRuntimeImportAuthority, ) -from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG, OpenHCSZMQConfig +from openhcs.runtime.zmq_config import ( + OPENHCS_ZMQ_CONFIG, + OpenHCSZMQConfig, + require_bindable_address, +) from openhcs.runtime.zmq_execution_signature import ( ZMQAuxiliaryExecutionParams, ZMQExecutionCompileControl, @@ -1025,6 +1029,11 @@ def connect( ): """Connect using the OpenHCS endpoint's declared startup deadline.""" + for url in ( + self.endpoint.data_url(self.config), + self.endpoint.control_url(self.config), + ): + require_bindable_address(url) return super().connect( timeout=( self.config.client_connect_timeout_seconds diff --git a/tests/conftest.py b/tests/conftest.py index e3de3e9047..67ef68ef62 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,8 +1,6 @@ """Global pytest configuration for OpenHCS integration tests.""" -import atexit import os -import shutil import signal import tempfile import uuid @@ -13,15 +11,12 @@ ensure_source_checkout_external_paths() -from openhcs.runtime.zmq_config import SOCKET_ROOT_ENV +from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG, use_private_socket_root # Every socket and lock file of this pytest process, and of the servers and # viewers it starts, lives under a root no other run sees. Each xdist worker -# imports this module, so each has its own. The path is short: ``sun_path`` -# holds about 104-108 bytes. -_SOCKET_ROOT = tempfile.mkdtemp(prefix="ohcs-sock-") -os.environ[SOCKET_ROOT_ENV] = _SOCKET_ROOT -atexit.register(shutil.rmtree, _SOCKET_ROOT, ignore_errors=True) +# imports this module, so each has its own. +use_private_socket_root() import psutil import pytest @@ -110,12 +105,14 @@ def cleanup_test_runtime_resources(): @pytest.fixture def unix_socket_dir(): - """A short private directory for AF_UNIX/IPC sockets. + """A private directory for AF_UNIX/IPC sockets, beneath the socket root. ``sun_path`` holds about 104-108 bytes, so a socket under a deep ``tmp_path`` (long ``--basetemp`` or ``TMPDIR``) cannot bind. """ - with tempfile.TemporaryDirectory(prefix="ohcs-sock-") as directory: + with tempfile.TemporaryDirectory( + prefix="s-", dir=OPENHCS_ZMQ_CONFIG.socket_root() + ) as directory: yield Path(directory) diff --git a/tests/unit/pyqt_gui/test_session_batch_workflows.py b/tests/unit/pyqt_gui/test_session_batch_workflows.py index a715c31784..778d9a7412 100644 --- a/tests/unit/pyqt_gui/test_session_batch_workflows.py +++ b/tests/unit/pyqt_gui/test_session_batch_workflows.py @@ -241,9 +241,19 @@ def test_compile_connection_failure_releases_only_its_reserved_dataset( async def connect(): raise RuntimeError("controlled connection failure") + failures = [] + session.subscribe( + lambda record: failures.append(record.event) + if isinstance(record.event, CompilationFailed) + else None + ) monkeypatch.setattr(session, "connect_client", connect) with pytest.raises(RuntimeError, match="controlled connection failure"): asyncio.run(session.compile_datasets((other,))) + # A caller waiting for this compilation is told it ended. + (failure,) = failures + assert failure.scope_id == other + assert "controlled connection failure" in failure.message assert not session.compile_pending assert session.batch.execution_id(running) == "original-run" diff --git a/tests/unit/test_b7_server_isolation_guards.py b/tests/unit/test_b7_server_isolation_guards.py index 7d29d8a6b7..33812b1cda 100644 --- a/tests/unit/test_b7_server_isolation_guards.py +++ b/tests/unit/test_b7_server_isolation_guards.py @@ -4,8 +4,15 @@ import ast import inspect +import json +import os +import subprocess +import sys from pathlib import Path +import pytest +import zmq + import zmqruntime.client import zmqruntime.transport_modes from zmqruntime.client import ZMQClient @@ -16,7 +23,9 @@ from openhcs.runtime.zmq_config import ( OPENHCS_ZMQ_CONFIG, SOCKET_ROOT_ENV, + SocketPathTooLongError, checkout_socket_root, + require_bindable_address, ) from openhcs.runtime.zmq_execution_client import ZMQExecutionClient from openhcs.runtime.zmq_execution_server import ZMQExecutionServer @@ -115,5 +124,90 @@ def test_each_checkout_has_its_own_socket_root(monkeypatch, tmp_path): def test_this_test_run_has_a_private_socket_root(): root = OPENHCS_ZMQ_CONFIG.socket_root() - assert root != checkout_socket_root() - assert root.name.startswith("ohcs-sock-") + assert root.parent == checkout_socket_root() + assert root.is_dir() + + +_CHILD_REPORT = """ +import json +from openhcs.runtime.zmq_execution_client import ZMQExecutionClient + +client = ZMQExecutionClient(port=65000, persistent=False) +print(json.dumps({ + "socket_root": str(client.config.socket_root()), + "code_root": client.launch().code_root, + "audience": client.launch().audience, + "control_url": client.endpoint.control_url(client.config), +})) +""" + + +def _report(environment: dict[str, str]) -> dict[str, str]: + completed = subprocess.run( + [sys.executable, "-c", _CHILD_REPORT], + env={**environment, "PYTHONPATH": str(REPOSITORY_ROOT)}, + capture_output=True, + text=True, + timeout=120, + ) + assert completed.returncode == 0, completed.stderr[-2000:] + return json.loads(completed.stdout.splitlines()[-1]) + + +def test_a_child_process_computes_its_parents_socket_root_and_launch(): + """A server or CLI started by this process looks where this process looks.""" + + client = ZMQExecutionClient(port=65000, persistent=False) + + assert _report(dict(os.environ)) == { + "socket_root": str(client.config.socket_root()), + "code_root": client.launch().code_root, + "audience": client.launch().audience, + "control_url": client.endpoint.control_url(client.config), + } + + +def test_a_private_socket_root_can_be_bound_whatever_tmpdir_is(tmp_path): + """The test run's sockets do not move with ``TMPDIR`` (a deep one cannot bind).""" + + deep_tmpdir = tmp_path / ("d" * 120) + deep_tmpdir.mkdir() + child = ( + "from openhcs.runtime.zmq_config import use_private_socket_root\n" + "use_private_socket_root()\n" + _CHILD_REPORT + ) + completed = subprocess.run( + [sys.executable, "-c", child], + env={ + **os.environ, + "PYTHONPATH": str(REPOSITORY_ROOT), + "TMPDIR": str(deep_tmpdir), + }, + capture_output=True, + text=True, + timeout=120, + ) + assert completed.returncode == 0, completed.stderr[-2000:] + report = json.loads(completed.stdout.splitlines()[-1]) + + assert Path(report["socket_root"]).parent == checkout_socket_root() + assert not Path(report["socket_root"]).exists() + require_bindable_address(report["control_url"]) + + +def test_a_client_refuses_a_socket_path_no_server_can_bind(monkeypatch, tmp_path): + """It raises before starting a server, and names the path and the limit.""" + + root = tmp_path / ("r" * 120) + monkeypatch.setenv(SOCKET_ROOT_ENV, str(root)) + client = ZMQExecutionClient(port=65000, persistent=False) + monkeypatch.setattr( + client, "_spawn_server_process", lambda: pytest.fail("a server was started") + ) + + with pytest.raises(SocketPathTooLongError) as refusal: + client.connect(timeout=5) + + assert str(root) in str(refusal.value) + assert str(zmq.IPC_PATH_MAX_LEN) in str(refusal.value) + assert SOCKET_ROOT_ENV in str(refusal.value)