diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000000..afe469b493 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,7 @@ +# Tracked files are checked out as their committed bytes on every platform. +# Benchmark corpora, reference-export sidecars and documentation audits are +# compared by sha256, which Windows end-of-line conversion would change. +* -text +# Windows command scripts keep the platform's line endings. +*.cmd text +*.bat text diff --git a/benchmark/adapters/openhcs.py b/benchmark/adapters/openhcs.py index 41ae61024b..1abd59d761 100644 --- a/benchmark/adapters/openhcs.py +++ b/benchmark/adapters/openhcs.py @@ -18,6 +18,8 @@ from pathlib import Path from typing import Any +from objectstate import replace_raw + from benchmark.adapters.cppipe_source import ( CPPipeSourceRequest, CPPipeSourceResolution, @@ -308,7 +310,6 @@ def _execution_global_config( request: OpenHCSRunRequest, *, output_suffix: str, - output_dataset_root: Path, compilation_debug_config: CompilationDebugConfig, ) -> GlobalPipelineConfig: """Select benchmark output paths without replacing declared path policy.""" @@ -330,7 +331,6 @@ def _execution_global_config( ), compilation_debug_config=compilation_debug_config, materialize_runtime_artifacts=request.materialize_runtime_artifacts, - materialization_results_path=output_dataset_root / "results", ) def _run_converted_cppipe_pipeline( @@ -361,9 +361,6 @@ def _run_converted_cppipe_pipeline( ) from exc output_suffix = f"_{request.pipeline_name}_converted_cppipe" - output_dataset_root = ( - request.output_dir / f"{request.dataset_path.name}{output_suffix}" - ) generated_source_path = request.output_dir / f"{cppipe_path.stem}_openhcs.py" source_workspace_path = ( request.output_dir @@ -417,7 +414,9 @@ def _run_converted_cppipe_pipeline( "UI/ZMQ FunctionStep source before execution." ) if compilation_debug_config.enabled: - pipeline_config = replace( + # replace_raw keeps the imported config's unset fields inheriting + # from the global config; dataclasses.replace would freeze them. + pipeline_config = replace_raw( pipeline_config, compilation_debug_config=LazyCompilationDebugConfig( enabled=compilation_debug_config.enabled, @@ -430,7 +429,6 @@ def _run_converted_cppipe_pipeline( global_config = self._execution_global_config( request, output_suffix=output_suffix, - output_dataset_root=output_dataset_root, compilation_debug_config=compilation_debug_config, ) ensure_global_config_context(GlobalPipelineConfig, global_config) @@ -550,7 +548,6 @@ def _run_converted_cppipe_pipeline( raise ToolExecutionError(equivalence_failure_message) metric_results = self._metric_results(request.metrics) - output_dataset_root.mkdir(parents=True, exist_ok=True) execution_output_root.mkdir(parents=True, exist_ok=True) provenance = { diff --git a/benchmark/cellprofiler_benchmark_cli.py b/benchmark/cellprofiler_benchmark_cli.py index 1bb056d84f..b8fde1982d 100644 --- a/benchmark/cellprofiler_benchmark_cli.py +++ b/benchmark/cellprofiler_benchmark_cli.py @@ -6,6 +6,7 @@ import json import math import os +import signal import sys from abc import ABC, abstractmethod from collections.abc import Sequence @@ -1069,6 +1070,10 @@ def create_benchmark_argument_parser() -> argparse.ArgumentParser: def main(argv: Sequence[str] | None = None) -> int: """Run the registered benchmark CLI.""" + if sys.platform == "win32": + # Ctrl+Break is the Windows console interrupt a launcher can deliver + # to this process alone; it stops a run exactly as Ctrl+C does. + signal.signal(signal.SIGBREAK, signal.default_int_handler) arguments = tuple(sys.argv[1:] if argv is None else argv) parser = create_benchmark_argument_parser() parsed = parser.parse_args(arguments) diff --git a/benchmark/openhcs_measured_run.py b/benchmark/openhcs_measured_run.py index 9523226820..ef66246f87 100644 --- a/benchmark/openhcs_measured_run.py +++ b/benchmark/openhcs_measured_run.py @@ -8,10 +8,7 @@ import sys import time from collections.abc import Callable, Mapping -from dataclasses import ( - dataclass, - field, -) +from dataclasses import dataclass from pathlib import Path from typing import Any @@ -134,7 +131,12 @@ class _ZMQProgressTimingObserver: compile_completed_at: float | None = None execution_started_at: float | None = None execution_completed_at: float | None = None - last_progress_monotonic: float = field(default_factory=time.monotonic) + last_progress_monotonic: float | None = None + """Monotonic time of the latest progress; None until a run is submitted. + + Starting the execution server is not execution: its client owns the + startup readiness timeout, so silence is measured from begin_run. + """ last_progress_phase: str = "" last_progress_status: str = "" on_event: Callable[[Mapping[str, Any]], None] | None = None @@ -217,6 +219,8 @@ def _duration(started_at: float | None, ended_at: float | None) -> float | None: def inactivity_seconds(self, *, observed_at: float | None = None) -> float: """Return elapsed monotonic time since the latest server progress event.""" + if self.last_progress_monotonic is None: + return 0.0 current = time.monotonic() if observed_at is None else observed_at return max(0.0, current - self.last_progress_monotonic) diff --git a/docs/refactor/consolidation-20261009/B6d-remaining-fixes.md b/docs/refactor/consolidation-20261009/B6d-remaining-fixes.md new file mode 100644 index 0000000000..dc7ddfa8d0 --- /dev/null +++ b/docs/refactor/consolidation-20261009/B6d-remaining-fixes.md @@ -0,0 +1,37 @@ +# B6d: the fixes from B6 that main still needs + +B6 (#1195) repaired main's CI in one branch that also reformatted 412 files and re-pinned the documentation audits. #1212 took its E1 boundary fix and #1214 its trigger restructuring. This is the rest: each fix below changes behaviour. Formatting commits, the ratchet mixin moves, the documentation re-pins, the workflow edits and the pyqt-reactive pin are not carried. + +## Fixes + +Tests named here fail on main `ae6ddace5` and pass on this branch unless marked otherwise. + +| Fix | Cause | Evidence | +|---|---|---| +| Requestless UI-bridge operations pass the connection by keyword | the connection landed in the request slot | `test_capabilities.py::test_requestless_ui_bridge_capability_passes_connection_as_connection` | +| A batch stopped while compiling ends cancelled | the stop surfaced as an execution failure | `test_execution_control_stop_during_compile.py` | +| Metadata document targets follow the backend's declaration | a virtual backend (OMERO) got a local metadata document | `test_function_outputs.py::test_metadata_document_targets_follow_backend_declaration` | +| Catalog leaves out modules whose distributions the package markers exclude | Intel macOS excludes `openhcs-basicpy` but the catalog imported its module | `test_library_registry_discovery.py` (7 cases) | +| Cache preparation uses `os.cpu_count()` where there is no affinity API | macOS prepared the kernel cache serially | `test_preparation_admission.py::test_platform_without_affinity_admits_its_online_cpus` | +| Scalar components are reduced per pixel group | contributors were pooled across runtime planes, dropping `site` | `test_aggregate_source_scope.py::test_axis_varying_between_runtime_planes_is_kept_not_reduced` | +| The one-plane label projection belongs to every plane label mode | it lived only on `MatchImageStackLabels` | `test_singleton_rgb_label_invocation_477.py` (`SliceAlignedLabels` case) | +| Benchmark adapter keeps the imported config's inheritance and declared output policy | `dataclasses.replace` froze unset fields; a results path was forced | `test_openhcs_adapter.py::test_openhcs_adapter_preserves_declared_main_flow_output_policy` | +| Packaged knowledge recipes are placed by case name within a path budget | installed paths exceeded Windows MAX_PATH | `test_build_mcp_knowledge_assets.py` (does not import on main) | +| GUI smoke waits for the window's endpoint start, reports the status name, and gives the bridge connection its transport config | the catalog was read before the endpoint existed; the result held a class; U4 moved the connection spec to formbridge, where `transport_endpoint` takes the config | `test_smoke_installed_gui.py` (does not import on main); `smoke_local.sh` GUI step | +| Unit session creates its `QApplication` first; sibling test modules import through `tests` | an MCP transport created a `QCoreApplication`; bare imports fail under importlib mode | `test_ci_import_guards.py` guards the imports; it passes on main once the test files are fixed, so no production test fails on main | +| STEP_INPUT filters match stored source paths | they matched virtual workspace names | no touched local test fails on main; B6 found it in the backend-microscope CI jobs | +| Passthrough outputs keep their producer's filename qualifier | sibling surfaces merged into one file | same | +| BaSiC reads metadata through `ImagePayload.of` | `.metadata` was read off plain arrays | not shown failing on main locally; B6 found it in CI | +| Dataset rows carry the finished execution's error; demo reads `ViewerType.value` | the installed demo failed without a reason | installed demo job only | +| Measured-run silence is counted from `begin_run` | server startup time counted as execution silence | not isolated by a local test | +| `mahotas` is a core dependency | core catalog modules import it; it sat in an extra | packaging; python-314 job | +| MCP smoke sends `partition_values` | G2 renamed `wells` | wheel-integration job only | +| Windows: granularity accepts `'L'` uint32; custom-function sources are written as bytes; Ctrl+Break stops the CLI; dataset writability is probed by creating a file; `.gitattributes` keeps committed bytes; test process reaping skips process groups | each is a Windows-only difference | Windows jobs only; the POSIX paths are run by the touched tests | + +## Ratchet + +`OpenHCSMainWindow` and `PatternGroupExecutionRequest` are at their size budget. The window's four theme methods had no caller and are deleted (**unused**; guarded in `test_dead_ui_code_guards.py`). The passthrough helpers sit on `PatternGroupExecutionScope`. + +## Done when + +Touched tests pass, `smoke_local.sh` prints `smoke rc=0` with the GUI step, `guardrails_local.sh` ends `guardrails rc=0`, and the 30-workflow parity check passes. diff --git a/docs/refactor/consolidation-20261009/U1-session-interface.md b/docs/refactor/consolidation-20261009/U1-session-interface.md index 1f2929f8b5..1380f5346e 100644 --- a/docs/refactor/consolidation-20261009/U1-session-interface.md +++ b/docs/refactor/consolidation-20261009/U1-session-interface.md @@ -88,7 +88,7 @@ Measured against `origin/main` (`89b8fdbff`, Python plus the MCP contract fixtur Corrections to the plan above: - **MCP tools are derived, not subclassed.** `SessionOperation` does not subclass `AgentCapabilityDeclaration` (that import cycles through the DTOs); `capabilities.py` derives one `openhcs_` capability per `HeadlessOperation`. The guard checks the derivation is total. -- **The headless journey runs a real execution server**, not a fake client; it and the GUI journey live in `tests/integration/test_session_journey.py`. +- **The headless journey runs a real execution server**, not a fake client; it lives in `tests/integration/test_session_journey.py`; the GUI journey, which needs the GUI extras and a Qt platform, lives in `tests/pyqt_gui/test_session_gui_journey.py`. - **The pipeline editor keeps `displayed_steps`**, a render projection of `PipelineStepsView`, never written back; the guard allows it. - **500 ms polling advice remains on the UI bridge** (`recommended_poll_interval_ms`): bridge events belong to U2. The session's own event stream is push (`openhcs_session_events`). - **The session package loads PyQt6 transitively** through `pyqt_reactive.protocols`; the guard checks direct imports. Owner: U4 (library moves). diff --git a/openhcs/agent/capabilities.py b/openhcs/agent/capabilities.py index 163a5fe99a..ebdfe00409 100644 --- a/openhcs/agent/capabilities.py +++ b/openhcs/agent/capabilities.py @@ -1657,7 +1657,9 @@ class UiBridgeOperationExecution(AgentCapabilityInvocation): def execute(self, context: object, *arguments: object) -> object: *request, connection = arguments - return context.ui_bridge_service.invoke(self.operation, *request, connection) + return context.ui_bridge_service.invoke( + self.operation, *request, connection=connection + ) @dataclass(frozen=True, slots=True) diff --git a/openhcs/agent/dto/session.py b/openhcs/agent/dto/session.py index ffb953e16a..008eaa7481 100644 --- a/openhcs/agent/dto/session.py +++ b/openhcs/agent/dto/session.py @@ -160,6 +160,8 @@ class DatasetRowState: debug_session_id: str | None = None finished_execution_id: str | None = None """The dataset's latest finished ordinary execution (its evidence id).""" + finished_execution_error: str | None = None + """The server's error for that execution, when it did not complete.""" scope_accent_color: str | None = None @property diff --git a/openhcs/agent/knowledge_manifest_schema.py b/openhcs/agent/knowledge_manifest_schema.py index 9dd6751aaf..e71f892b53 100644 --- a/openhcs/agent/knowledge_manifest_schema.py +++ b/openhcs/agent/knowledge_manifest_schema.py @@ -4,6 +4,7 @@ import json import os +import re from collections.abc import Mapping from dataclasses import dataclass from enum import Enum @@ -13,6 +14,12 @@ "docs/source/development/mcp_knowledge_base_manifest.json" ) PACKAGED_KNOWLEDGE_BASE_ROOT = Path("resources/knowledge") +# Windows without long-path support rejects installed paths of 260 characters +# or more. The installer places the knowledge root at +# /env-XXXXXXXX/Lib/site-packages/openhcs/agent/resources/knowledge, +# so every resource path relative to that root must stay within this budget. +PACKAGED_KNOWLEDGE_PATH_BUDGET = 120 +_PORTABLE_CASE_NAME = re.compile(r"[A-Za-z0-9_](?:[A-Za-z0-9_ .-]*[A-Za-z0-9_])?") class KnowledgeBaseManifestField(str, Enum): @@ -145,21 +152,21 @@ def load(cls, path: Path, *, source_root: Path) -> ComparisonManifestSnapshot: return manifest def native_source_relative_path(self, case: Mapping[str, object]) -> Path: - """Derive one placement from its original manifest/root/path identity.""" + """Place one native recipe by its manifest-unique case name. + + Upstream root/path layouts are deep; keying by case name keeps every + installed resource within ``PACKAGED_KNOWLEDGE_PATH_BUDGET``. + """ path = Path(str(case["cppipe_path"])) - root = Path(str(case.get("cppipe_path_root", "unrooted"))) - if ( - path.is_absolute() - or root.is_absolute() - or ".." in path.parts - or ".." in root.parts - or path.suffix != ".cppipe" - ): + name = str(case.get("name")) + if path.is_absolute() or ".." in path.parts or path.suffix != ".cppipe": raise ValueError("Native knowledge source must be a relative .cppipe path.") - return ( - self.path.relative_to(self.source_root).with_suffix(".sources") - / root - / path + if not _PORTABLE_CASE_NAME.fullmatch(name): + raise ValueError( + f"Native knowledge case name {name!r} is not a portable file name." + ) + return self.path.relative_to(self.source_root).with_suffix(".sources") / ( + f"{name}.cppipe" ) def native_source_path(self, case: Mapping[str, object]) -> Path: diff --git a/openhcs/authoring/session/execution_control.py b/openhcs/authoring/session/execution_control.py index 59b8c2ab0a..fa0d402555 100644 --- a/openhcs/authoring/session/execution_control.py +++ b/openhcs/authoring/session/execution_control.py @@ -46,9 +46,14 @@ def check_all_completed(self) -> None: async def handle_execution_failure(self) -> None: session = self._session - for scope_id in tuple(session.batch.active_plates): - session.batch.mark_terminal(scope_id, ExecutionFailed) - session.set_dataset_state(scope_id, OrchestratorExecFailed) + if session.execution_state.stop_pending: + # The batch was stopped (for example while it was still compiling): + # the error is the stop arriving, so its datasets end cancelled. + self.cancel_active_datasets() + else: + for scope_id in tuple(session.batch.active_plates): + session.batch.mark_terminal(scope_id, ExecutionFailed) + session.set_dataset_state(scope_id, OrchestratorExecFailed) session.execution_state = ManagerExecutionState.IDLE try: await session.client.disconnect() diff --git a/openhcs/authoring/session/session.py b/openhcs/authoring/session/session.py index cd495d063d..71c9dc489d 100644 --- a/openhcs/authoring/session/session.py +++ b/openhcs/authoring/session/session.py @@ -16,6 +16,7 @@ import os import threading import time +import uuid from abc import ABC, abstractmethod from dataclasses import dataclass from collections.abc import Callable, Iterable @@ -204,6 +205,25 @@ def call(self, callback: Callable[[], ResultT]) -> ResultT: return self._dispatcher.call(callback) +def _creates_files_in(directory: Path) -> bool: + """Whether the filesystem lets this process create a file in ``directory``. + + Permission bits (``os.access``) do not see Windows ACLs or read-only + shares; an actual creation does. One attempt answers: ``tempfile`` on + Windows reads a refused creation as a name collision whenever ``os.access`` + passes and retries up to ``os.TMP_MAX`` (2**31 - 1) times. The probe file is + removed at once, so it leaves nothing behind. + """ + + probe = Path(directory) / f".openhcs-write-probe-{uuid.uuid4().hex}" + try: + os.close(os.open(probe, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)) + except OSError: + return False + probe.unlink() + return True + + class DatasetAccess(ABC): """Which dataset roots this session may read and initialize.""" @@ -218,7 +238,7 @@ def initializes_in_place(self, root: Path) -> bool: """ return all( - os.access(directory, os.W_OK) + _creates_files_in(directory) for directory in { Path(root), *(path.parent for path in METADATA_CONFIG.managed_paths(root)), diff --git a/openhcs/authoring/session/views.py b/openhcs/authoring/session/views.py index 26a967c52a..4ef5c7a48a 100644 --- a/openhcs/authoring/session/views.py +++ b/openhcs/authoring/session/views.py @@ -279,6 +279,16 @@ def row_state( terminal = activity.terminal_status runtime = activity.runtime debug_phase = activity.debug_phase + finished_id, finished = next( + ( + (execution_id, finished) + for execution_id, finished in reversed( + session.finished_executions.items() + ) + if finished.scope_id == row.scope_id + ), + (None, None), + ) return DatasetRowState( scope_id=row.scope_id, name=row.name, @@ -305,15 +315,9 @@ def row_state( source_root=relation.source_root, debug_phase=None if debug_phase is None else debug_phase.name, debug_session_id=activity.debug_session_id, - finished_execution_id=next( - ( - execution_id - for execution_id, finished in reversed( - session.finished_executions.items() - ) - if finished.scope_id == row.scope_id - ), - None, + finished_execution_id=finished_id, + finished_execution_error=( + None if finished is None else finished.record.error ), ) diff --git a/openhcs/core/pipeline/function_contracts.py b/openhcs/core/pipeline/function_contracts.py index 46f584b80d..5d1d7b2617 100644 --- a/openhcs/core/pipeline/function_contracts.py +++ b/openhcs/core/pipeline/function_contracts.py @@ -171,8 +171,22 @@ def invocation_kwargs( runtime_projection: RuntimePlaneAxisValueProjection | None, semantic_controls: Mapping[str, Any], ) -> dict[str, Any]: - """Return the invocation kwargs once the image's execution mode is known.""" - del execution_mode, image_projection, runtime_projection + """Match a scalar image to its declared singleton runtime root. + + Labels can consume a singleton root only after the image's final + execution mode and retained plane projection have been resolved. + """ + if ( + execution_mode.per_plane + and image_projection is None + and runtime_projection is not None + and runtime_projection.axis_size == 1 + ): + from openhcs.core.runtime_slice_projection import RuntimeSliceProjection + + kwargs = RuntimeSliceProjection.kwargs_for_slice( + kwargs, runtime_projection.selected_plane(0) + ) return {**kwargs, **semantic_controls} @classmethod @@ -242,16 +256,6 @@ def image_execution_mode(cls, labels, default, *, runtime_slice_count=None): del labels, default, runtime_slice_count return FullStackExecution - -class MatchImageStackLabels(ObjectLabelInputExecutionMode): - """Labels follow the image: a stack when the image needs one.""" - - key = "match_image_stack" - - @classmethod - def preserves_full_stack(cls, *, image_stack_required: bool) -> bool: - return image_stack_required - @classmethod def invocation_kwargs( cls, @@ -262,23 +266,19 @@ def invocation_kwargs( runtime_projection, semantic_controls, ): - """Match scalar images to their declared singleton runtime root. + """Full-stack label consumers keep their declared label domain.""" + del execution_mode, image_projection, runtime_projection + return {**kwargs, **semantic_controls} - Matching labels can consume a singleton root only after the image's - final execution mode and retained plane projection have been resolved. - """ - if ( - execution_mode.per_plane - and image_projection is None - and runtime_projection is not None - and runtime_projection.axis_size == 1 - ): - from openhcs.core.runtime_slice_projection import RuntimeSliceProjection - kwargs = RuntimeSliceProjection.kwargs_for_slice( - kwargs, runtime_projection.selected_plane(0) - ) - return {**kwargs, **semantic_controls} +class MatchImageStackLabels(ObjectLabelInputExecutionMode): + """Labels follow the image: a stack when the image needs one.""" + + key = "match_image_stack" + + @classmethod + def preserves_full_stack(cls, *, image_stack_required: bool) -> bool: + return image_stack_required def _artifact_spec_from_output_declaration( diff --git a/openhcs/core/processing_preparation.py b/openhcs/core/processing_preparation.py index 74c9c4d995..4ff7b08efb 100644 --- a/openhcs/core/processing_preparation.py +++ b/openhcs/core/processing_preparation.py @@ -312,6 +312,18 @@ def close(self) -> None: self.result_connection.close() +def _schedulable_cpu_count() -> int: + """CPUs this process may run on. + + Linux exposes the scheduler's affinity set. macOS has no affinity API: + every online CPU is schedulable there. + """ + try: + return len(os.sched_getaffinity(0)) + except AttributeError: + return os.cpu_count() or 1 + + @dataclass(frozen=True, slots=True) class PreparationCacheBatch: """Schedule derived cache operations without interpreting backend families.""" @@ -340,14 +352,7 @@ def populate_child_caches( raise ValueError("Preparation worker budget must be positive.") if "fork" not in multiprocessing.get_all_start_methods(): return - try: - worker_capacity = min(max_workers, len(os.sched_getaffinity(0))) - except AttributeError: - # Without an affinity witness there is no admitted parallel cache - # work. The existing parent preparation still owns readiness. - return - if worker_capacity < 1: - return + worker_capacity = min(max_workers, _schedulable_cpu_count()) operations = { operation.identity: operation for preparation in self.preparations diff --git a/openhcs/core/source_bindings.py b/openhcs/core/source_bindings.py index 256f8904c7..0270462cef 100644 --- a/openhcs/core/source_bindings.py +++ b/openhcs/core/source_bindings.py @@ -985,7 +985,7 @@ def matches_provenance(self, provenance: SourceImageProvenance) -> bool: identities = provenance.represented_source_identities return ( (not self.filters or self.path_filters_match( - tuple(identity.path for identity in identities if identity.path is not None) + tuple(path for identity in identities for path in identity.filter_paths) )) and self.metadata_candidates_match( tuple(identity.component_metadata or {} for identity in identities) diff --git a/openhcs/core/source_image_provenance.py b/openhcs/core/source_image_provenance.py index 8efcca2271..a1d5f33e17 100644 --- a/openhcs/core/source_image_provenance.py +++ b/openhcs/core/source_image_provenance.py @@ -134,6 +134,18 @@ def from_mapping(cls, values: Mapping[str, object]) -> "SourceImageIdentity": def addressable(self) -> bool: return self.path is not None or self.component_metadata is not None + @property + def filter_paths(self) -> tuple[str, ...]: + """Store-owned paths that source filters match. + + A virtual workspace path names the plane; the stored source paths it + was admitted from are what a FILE filter selects. + """ + stored = SourceMetadataFields.source_filter_paths(self.component_metadata or {}) + if stored: + return stored + return () if self.path is None else (self.path,) + @property def filename_extension(self) -> str | None: """Project the declared source filename fact, never a format default.""" @@ -629,6 +641,25 @@ def contributors(self) -> tuple[SourceImageProvenanceContributor, ...]: ) ) + @property + def contributor_groups( + self, + ) -> tuple[tuple[SourceImageProvenanceContributor, ...], ...]: + """Contributors grouped by the pixels they reduce into: top-level ones + into the whole payload, nested ones into their own runtime plane.""" + return ( + tuple( + plane + for plane in self.planes + if isinstance(plane, SourceImageProvenanceContributor) + ), + *( + plane.contributors + for plane in self.planes + if not isinstance(plane, SourceImageProvenanceContributor) + ), + ) + @property def represented_source_image_names(self) -> tuple[str, ...]: """Return every alias owned by a runtime plane or pixel contributor.""" @@ -1058,31 +1089,27 @@ def required_scalar_components( """Exclude only coordinates proven reduced into pixel contributors. Contributors do not own a projectable runtime axis. A component that - varies across their complete source addresses therefore has no scalar - address on the derived image. Absent or incomplete contributor evidence - cannot exempt an acquired plane from its required coordinates. + varies across the contributors of one pixel group therefore has no + scalar address on the derived image; one that varies only between + runtime planes is that plane axis. Absent or incomplete contributor + evidence cannot exempt an acquired plane from its required coordinates. """ - contributors = self.source_image_provenance_planes.contributors scalar_components = SourceMetadataFields.component_domains( self.scalar_source_identity.component_metadata or {} ) - contributor_components = tuple( - SourceMetadataFields.component_domains(contributor.component_metadata or {}) - for contributor in contributors - ) reduced = set() - for component in AxisFamily.active().axes: - if component in scalar_components: - continue - if len(contributor_components) < 2 or not all( - len(values.get(component, ())) == 1 for values in contributor_components - ): - continue - values = frozenset( - value for values in contributor_components for value in values[component] + for contributors in self.source_image_provenance_planes.contributor_groups: + domains = tuple( + SourceMetadataFields.component_domains(item.component_metadata or {}) + for item in contributors ) - if len(values) > 1: - reduced.add(component.name) + for component in AxisFamily.active().axes: + if component in scalar_components or len(domains) < 2: + continue + if not all(len(values.get(component, ())) == 1 for values in domains): + continue + if len({values[component][0] for values in domains}) > 1: + reduced.add(component.name) return tuple(component for component in component_order if component not in reduced) def image_set_identities( diff --git a/openhcs/core/steps/function_outputs.py b/openhcs/core/steps/function_outputs.py index 961903b0ea..bd9ebf5655 100644 --- a/openhcs/core/steps/function_outputs.py +++ b/openhcs/core/steps/function_outputs.py @@ -17,6 +17,8 @@ RegistryFamily, extract_key_from_class_name, ) +from polystore.backend_registry import STORAGE_BACKENDS +from polystore.base import VirtualBackend from polystore.streaming.identity import StreamProducerIdentity from polystore.streaming.viewer_transport import ViewerStreamProducer, ViewerStreamSourceMetadata from polystore.virtual_workspace import SourcePixelRef @@ -664,6 +666,15 @@ class OpenHCSMetadataTarget(ABC, metaclass=AutoRegisterMeta): def from_plan(cls, plan: CompiledStepPlan) -> OpenHCSMetadataTarget | None: """Resolve this declaration's independently compiled storage target.""" + @staticmethod + def hosts_metadata_document(backend: str) -> bool: + """The metadata document lives in the output root's local directory. + + A virtual backend (for example OMERO) declares that its paths name + remote objects, so its output root has no directory to hold it. + """ + return not issubclass(STORAGE_BACKENDS[backend], VirtualBackend) + @classmethod def for_plan(cls, plan: CompiledStepPlan) -> tuple[OpenHCSMetadataTarget, ...]: """Derive targets from their declarations, without a consumer roster.""" @@ -1119,12 +1130,14 @@ def from_execution( @classmethod def from_plan(cls, plan: CompiledStepPlan) -> PrimaryImageMetadataTarget | None: - if plan.write_backend in (Backend.OMERO_LOCAL.value, Backend.MEMORY.value): + if plan.write_backend == Backend.MEMORY.value: return None if plan.write_backend is None: raise ValueError( f"Step {plan.step_index} ({plan.step_name}) has no write backend." ) + if not cls.hosts_metadata_document(plan.write_backend): + return None if plan.output_dir is None: raise ValueError( f"Step {plan.step_index} ({plan.step_name}) has no output directory." @@ -1153,10 +1166,7 @@ def from_plan( cls, plan: CompiledStepPlan ) -> MaterializedImageMetadataTarget | None: output = plan.materialized_output - if output is None or output.backend in ( - Backend.OMERO_LOCAL.value, - Backend.MEMORY.value, - ): + if output is None or not cls.hosts_metadata_document(output.backend): return None return cls( output_dir=output.output_dir, @@ -1227,7 +1237,12 @@ def for_directory(self, directory: Path) -> RuntimeArtifactMetadataTarget: @classmethod def from_plan(cls, plan: CompiledStepPlan) -> RuntimeArtifactMetadataTarget | None: materialization = plan.runtime_artifact_materialization - if not materialization.has_persistent_target: + if not ( + materialization.has_persistent_target + and cls.hosts_metadata_document( + materialization.require_persistent_backend() + ) + ): return None output_dir = plan.artifact_analysis_output_dir dataset_root = plan.artifact_output_dataset_root diff --git a/openhcs/core/steps/function_runtime.py b/openhcs/core/steps/function_runtime.py index daab77bb25..1b4ce2248a 100644 --- a/openhcs/core/steps/function_runtime.py +++ b/openhcs/core/steps/function_runtime.py @@ -325,6 +325,35 @@ def selected_artifact_output_plans(self) -> ArtifactOutputPlans: self.component_key, ) + def _unwrapped_main_flow_output_context( + self, + ) -> AlignedImageSliceContext | None: + ArtifactInputPlan.require_exact_map( + self.execution_plan.artifact_inputs, + boundary="Component artifact input", + ) + return self.compiled_group.unwrapped_main_flow_output_context( + self.selected_artifact_output_plans() + ) + + def _passes_producer_outputs_through( + self, + processed_stack: ImagePayload, + output_count: int, + input_count: int, + ) -> bool: + """Whether each published slice is the producer's aligned output, passed on. + + A declared passthrough chain names no output of its own, so its slices + keep the exact producer declaration (context and filename qualifier) + of the input file they are aligned to. + """ + if processed_stack.owns_output_surfaces: + return False + if self._unwrapped_main_flow_output_context() is not None: + return False + return output_count == input_count + @dataclass(frozen=True, slots=True, kw_only=True) class PatternGroupExecutionRequest(PatternGroupExecutionScope): @@ -1010,21 +1039,13 @@ def _project_output_slices( """Split the function's main output into its published slices.""" if processed_stack.declares_whole_image_output: output_context = self._unwrapped_main_flow_output_context() - contexts = ( - (output_context,) - if output_context is not None - else ( - self._producer_output_contexts(matching_files) - if len(matching_files) == 1 - else () - ) - ) - context = ( - contexts[0] - if contexts - else AlignedImageSliceContext.anonymous_main_flow() - ) - return ((processed_stack, context),) + if output_context is None and self._passes_producer_outputs_through( + processed_stack, 1, len(matching_files) + ): + output_context = self._producer_output_contexts(matching_files)[0] + if output_context is None: + output_context = AlignedImageSliceContext.anonymous_main_flow() + return ((processed_stack, output_context),) unstack_started_at = time.perf_counter() projected = processed_stack.main_output_slices( lambda data: self._unstack_main_output(data, len(matching_files)) @@ -1040,19 +1061,15 @@ def _project_output_slices( if processed_stack.owns_output_surfaces: return projected output_payloads = tuple(payload for payload, _context in projected) - output_context = self._unwrapped_main_flow_output_context() - slice_contexts = ( - (output_context,) * len(output_payloads) - if output_context is not None - else () - ) - if not slice_contexts and len(output_payloads) == len(matching_files): + if self._passes_producer_outputs_through( + processed_stack, len(output_payloads), len(matching_files) + ): slice_contexts = self._producer_output_contexts(matching_files) - if not slice_contexts: - slice_contexts = tuple( - AlignedImageSliceContext.anonymous_main_flow() - for _payload in output_payloads - ) + else: + output_context = self._unwrapped_main_flow_output_context() + if output_context is None: + output_context = AlignedImageSliceContext.anonymous_main_flow() + slice_contexts = (output_context,) * len(output_payloads) return tuple(zip(output_payloads, slice_contexts, strict=True)) def _unstack_main_output( @@ -1079,17 +1096,6 @@ def _unstack_main_output( f"{output_shape}" ) from exc - def _unwrapped_main_flow_output_context( - self, - ) -> AlignedImageSliceContext | None: - ArtifactInputPlan.require_exact_map( - self.execution_plan.artifact_inputs, - boundary="Component artifact input", - ) - return self.compiled_group.unwrapped_main_flow_output_context( - self.selected_artifact_output_plans() - ) - def _save_outputs( self, processed_stack: RuntimeArrayData, @@ -1126,6 +1132,20 @@ def _save_outputs( output_slices = tuple(payload for payload, _context in projected_outputs) num_outputs = len(output_slices) num_inputs = len(matching_files) + passed_through_records = ( + step_output_manifest(context).producer_output_records_for_paths( + self.execution_plan, matching_files, context.microscope_handler.parser + ) + if self._passes_producer_outputs_through( + processed_stack, num_outputs, num_inputs + ) + else None + ) + passed_through_qualifiers = ( + (None,) * num_outputs + if passed_through_records is None + else tuple(record.filename_qualifier for record in passed_through_records) + ) if num_outputs < num_inputs: logger.debug( @@ -1186,6 +1206,12 @@ def _save_outputs( output_identity = output_identity.with_filename_qualifier( output_context.output_key ) + elif passed_through_qualifiers[i] is not None: + # A passed-through surface keeps the qualifier its producer + # published it under; dropping it would merge sibling surfaces. + output_identity = output_identity.with_filename_qualifier( + passed_through_qualifiers[i] + ) output_path = output_identity.path_for_request(output_path_request) output_path_text = str(output_path) img_slice = output_context.contextualize_image_payload(img_slice) diff --git a/openhcs/mcp/installed_demo.py b/openhcs/mcp/installed_demo.py index c3726b313f..3d713dc9ec 100644 --- a/openhcs/mcp/installed_demo.py +++ b/openhcs/mcp/installed_demo.py @@ -398,7 +398,8 @@ def _execute_pipeline( (row,) = (row for row in datasets.rows if row.root == str(plate_path)) if row.terminal_status != ExecutionCompleted.name: raise InstalledDemoFailure( - f"Portable neurite execution ended with {row.terminal_status}: {row}" + f"Portable neurite execution ended with {row.terminal_status}: " + f"{row.finished_execution_error}; row={row}" ) return row @@ -609,7 +610,7 @@ def run_installed_demo( viewer_type = ( None if viewer_payload is None or viewer_payload.viewer is None - else viewer_payload.viewer.viewer_type.wire_value + else viewer_payload.viewer.viewer_type.value ) result = InstalledDemoResult( openhcs_version=distribution("openhcs").version, diff --git a/openhcs/processing/backends/cellprofiler/_granularity_native.cpp b/openhcs/processing/backends/cellprofiler/_granularity_native.cpp index a49ac9118d..b875d7d539 100644 --- a/openhcs/processing/backends/cellprofiler/_granularity_native.cpp +++ b/openhcs/processing/backends/cellprofiler/_granularity_native.cpp @@ -110,9 +110,27 @@ class OwnedArrayBuffer { return acquired_; } const Py_buffer& view() const { return view_; } + // One unsigned C width has several buffer codes: NumPy exports uint32 as + // 'I' on LP64 platforms and as 'L' on LLP64 Windows. Compare by width. + static Py_ssize_t unsigned_width(const char* format) { + if (format[0] == '\0' || format[1] != '\0') return 0; + switch (format[0]) { + case 'B': return sizeof(unsigned char); + case 'H': return sizeof(unsigned short); + case 'I': return sizeof(unsigned int); + case 'L': return sizeof(unsigned long); + case 'Q': return sizeof(unsigned long long); + default: return 0; + } + } + bool has_format(const char* format) const { + if (std::strcmp(view_.format, format) == 0) return true; + const Py_ssize_t width = unsigned_width(format); + return width != 0 && unsigned_width(view_.format) == width && view_.itemsize == width; + } bool matches(const char* name, const char* format, int ndim, bool writable) const { if (view_.ndim != ndim || view_.format == nullptr || format == nullptr || - std::strcmp(view_.format, format) != 0 || + !has_format(format) || !PyBuffer_IsContiguous(&view_, 'C') || (writable && view_.readonly)) { PyErr_Format(PyExc_ValueError, "%s must be a C-contiguous %d-D array with format %s%s", diff --git a/openhcs/processing/backends/enhance/basic_processor_jax.py b/openhcs/processing/backends/enhance/basic_processor_jax.py index 42ff7f7ebf..edc3df0d2c 100644 --- a/openhcs/processing/backends/enhance/basic_processor_jax.py +++ b/openhcs/processing/backends/enhance/basic_processor_jax.py @@ -32,6 +32,10 @@ NameByArtifactName, ) +# The OpenHCS requirement markers decide where this distribution is installed; +# the function catalog leaves this module out wherever they exclude it. +REQUIRED_DISTRIBUTIONS = ("openhcs-basicpy",) + def _fitted_field_output(name: str) -> ArtifactSpec: """Project group lineage; runtime field owns aggregate contributor context.""" diff --git a/openhcs/processing/backends/enhance/flatfield.py b/openhcs/processing/backends/enhance/flatfield.py index 4f049bb880..1b95dce858 100644 --- a/openhcs/processing/backends/enhance/flatfield.py +++ b/openhcs/processing/backends/enhance/flatfield.py @@ -6,6 +6,7 @@ from openhcs.core.projected_image_output import SourceProjectedImageOutput from openhcs.core.runtime_array_values import RuntimeArrayData +from openhcs.core.runtime_image_values import ImagePayload from openhcs.core.runtime_plane_projection import RuntimePlaneAxisValueProjection from openhcs.core.axes import AxisFamily, AxisRole, TileAxis @@ -34,7 +35,7 @@ def __post_init__(self) -> None: @classmethod def validate_observation_domain(cls, source: RuntimeArrayData) -> None: """Reject mislabeled metadata-backed ensembles before fitting.""" - metadata = source.metadata + metadata = ImagePayload.of(source).metadata family = AxisFamily.active() retained_axes = tuple( family.named(name) for name in metadata.retained_plane_component_values() @@ -67,7 +68,7 @@ def resolve_source_context( if source.shape != (self.observation_count, *self.data.shape): raise ValueError("A fitted field must retain the source spatial grid.") self.validate_observation_domain(source) - metadata = source.metadata.collapse_leading_plane_axis() + metadata = ImagePayload.of(source).metadata.collapse_leading_plane_axis() # Model parameters are not unit-interval input pixels. Contributor source # dtypes/identities remain in provenance, not on the fitted field's range. metadata = metadata.replace_fields( diff --git a/openhcs/processing/backends/lib_registry/openhcs_registry.py b/openhcs/processing/backends/lib_registry/openhcs_registry.py index e89c959654..74317254d9 100644 --- a/openhcs/processing/backends/lib_registry/openhcs_registry.py +++ b/openhcs/processing/backends/lib_registry/openhcs_registry.py @@ -9,6 +9,7 @@ import ast import importlib import importlib.abc +import importlib.metadata import inspect import logging from abc import ABC, abstractmethod @@ -22,6 +23,8 @@ import numpy as np from metaclass_registry import LazyDiscoveryDict, import_module_preserving_root_logging +from packaging.requirements import Requirement +from packaging.utils import canonicalize_name from openhcs.constants import VALID_MEMORY_TYPES, MemoryType from openhcs.core.callable_contract import ( @@ -293,6 +296,18 @@ def _catalog_memory_types(func: Callable) -> frozenset[MemoryType] | None: return declared +def _module_source(module_name: str) -> tuple[str, str] | None: + """Return loader-decoded module source and origin, or None without source.""" + + spec = importlib.util.find_spec(module_name) + if spec is None or not isinstance(spec.loader, importlib.abc.InspectLoader): + return None + source = spec.loader.get_source(module_name) + if source is None: + return None + return source, spec.origin or module_name + + def _module_declares_allowed_memory_type( module_name: str, allowed_memory_types: frozenset[str], @@ -300,15 +315,135 @@ def _module_declares_allowed_memory_type( spec = importlib.util.find_spec(module_name) if spec is not None and spec.submodule_search_locations is not None: return True - if spec is None or not isinstance(spec.loader, importlib.abc.InspectLoader): + module_source = _module_source(module_name) + if module_source is None: return False - source = spec.loader.get_source(module_name) - if source is None: + return _source_declares_allowed_memory_type(*module_source, allowed_memory_types) + + +_REQUIRED_DISTRIBUTIONS_NAME = "REQUIRED_DISTRIBUTIONS" + + +def _openhcs_requirements() -> tuple[Requirement, ...]: + """Return the installed OpenHCS distribution's declared requirements.""" + + return tuple( + Requirement(value) for value in importlib.metadata.requires("openhcs") or () + ) + + +def _distribution_excluded_by_package_metadata(name: str) -> bool: + """Return whether OpenHCS's own requirement markers leave ``name`` out here. + + A distribution is legitimately absent only when no OpenHCS requirement for + it applies to this environment (markers, extras) and it is not installed. + A distribution OpenHCS requires here is never excluded, so its absence + surfaces as the module's import failure. + """ + + canonical = canonicalize_name(name) + requirements = tuple( + requirement + for requirement in _openhcs_requirements() + if canonicalize_name(requirement.name) == canonical + ) + if not requirements: + raise ValueError( + f"Distribution {name!r} is not declared by the OpenHCS package " + "requirements; a module cannot depend on an undeclared distribution." + ) + if any( + marker is None or marker.evaluate({"extra": ""}) + for marker in (requirement.marker for requirement in requirements) + ): return False - return _source_declares_allowed_memory_type( - source, - spec.origin or module_name, - allowed_memory_types, + try: + importlib.metadata.distribution(name) + except importlib.metadata.PackageNotFoundError: + return True + return False + + +@lru_cache(maxsize=1024) +def _source_required_distributions(source: str, origin: str) -> tuple[str, ...]: + """Read a module's literal ``REQUIRED_DISTRIBUTIONS`` declaration.""" + + for statement in ast.parse(source, filename=origin).body: + if isinstance(statement, ast.Assign): + targets = statement.targets + elif isinstance(statement, ast.AnnAssign): + targets = (statement.target,) + else: + continue + if not any( + isinstance(target, ast.Name) and target.id == _REQUIRED_DISTRIBUTIONS_NAME + for target in targets + ): + continue + value = ast.literal_eval(statement.value) + if not isinstance(value, tuple) or not all( + isinstance(item, str) and item for item in value + ): + raise ValueError( + f"{origin}: {_REQUIRED_DISTRIBUTIONS_NAME} must be a literal tuple " + "of distribution names." + ) + return value + return () + + +def _module_in_this_catalog( + module_name: str, + allowed_memory_types: frozenset[str], +) -> bool: + """Memory-type eligibility and the package requirement markers decide. + + A module whose ``REQUIRED_DISTRIBUTIONS`` names a distribution those + markers leave out of this environment is not in this installation's catalog. + """ + + if not _module_declares_allowed_memory_type(module_name, allowed_memory_types): + logger.debug( + "Skipping OpenHCS module %s - no allowed memory decorators", module_name + ) + return False + excluded = _module_distributions_excluded_by_package_metadata(module_name) + if excluded: + logger.info( + "Skipping OpenHCS module %s - package requirements exclude %s here", + module_name, + ", ".join(excluded), + ) + return not excluded + + +def _catalog_policy_context() -> dict[str, Any]: + """The declaration-import policy as catalogue cache identity.""" + + return { + "allowed_memory_types": sorted(_allowed_openhcs_memory_types()), + "distributions_excluded_by_package_metadata": sorted( + { + canonicalize_name(requirement.name) + for requirement in _openhcs_requirements() + if _distribution_excluded_by_package_metadata(requirement.name) + } + ), + } + + +def _module_distributions_excluded_by_package_metadata( + module_name: str, +) -> tuple[str, ...]: + """Return the module's declared distributions this installation excludes.""" + + module_source = _module_source(module_name) + if module_source is None: + return () + return tuple( + name + for name in _source_required_distributions(*module_source) + if _distribution_excluded_by_package_metadata(name) ) @@ -487,29 +622,19 @@ def _get_openhcs_modules(self) -> List[str]: return modules def get_modules_to_scan(self) -> List[Tuple[str, Any]]: - """Import every admitted native declaration before catalog publication. + """Import every native declaration in this installation's catalog. - Memory-type eligibility owns exclusions. An eligible OpenHCS module - that cannot import is a broken installation, not an optional function - inventory; its failure must prevent apparent catalog readiness. + ``_module_in_this_catalog`` owns exclusions; any other OpenHCS module + that cannot import is a broken installation and fails catalog readiness. """ self._ensure_module_inventory() assert self.MODULES_TO_SCAN is not None - modules = [] allowed_memory_types = _allowed_openhcs_memory_types() - for module_name in self.MODULES_TO_SCAN: - if not _module_declares_allowed_memory_type( - module_name, - allowed_memory_types, - ): - logger.debug( - "Skipping OpenHCS module %s - no allowed memory decorators", - module_name, - ) - continue - module = import_module_preserving_root_logging(module_name) - modules.append((module_name, module)) - return modules + return [ + (module_name, import_module_preserving_root_logging(module_name)) + for module_name in self.MODULES_TO_SCAN + if _module_in_this_catalog(module_name, allowed_memory_types) + ] # ===== ESSENTIAL ABC METHODS ===== def get_library_version(self) -> str: @@ -540,7 +665,7 @@ def cache_source_mtimes(self) -> Dict[str, float]: def cache_discovery_context(self) -> dict[str, Any]: """Project the declaration-import policy into catalogue cache identity.""" - return {"allowed_memory_types": sorted(_allowed_openhcs_memory_types())} + return _catalog_policy_context() def is_library_available(self) -> bool: """OpenHCS is always available.""" diff --git a/openhcs/processing/custom_functions/manager.py b/openhcs/processing/custom_functions/manager.py index eea25e3ec7..67ff41bda0 100644 --- a/openhcs/processing/custom_functions/manager.py +++ b/openhcs/processing/custom_functions/manager.py @@ -367,7 +367,7 @@ def list_custom_functions(self) -> list[CustomFunctionInfo]: for py_file in sorted(self.storage_dir.glob("*.py")): try: - metadata = self._prepare_source(py_file.read_text(encoding="utf-8")) + metadata = self._prepare_source(py_file.read_bytes().decode("utf-8")) contract = CallableContract.from_callable(metadata.func) functions.append( CustomFunctionInfo( @@ -401,7 +401,7 @@ def get_function_code(self, func_name: str) -> str: if not file_path.exists(): raise ValueError(f"Custom function '{func_name}' not found") - return file_path.read_text(encoding="utf-8") + return file_path.read_bytes().decode("utf-8") def update_custom_function(self, old_name: str, new_code: str) -> str: """ @@ -662,8 +662,8 @@ def _write_temporary_source(self, code: str) -> Path: dir=self.storage_dir, ) try: - with os.fdopen(file_descriptor, "w", encoding="utf-8") as stream: - stream.write(code) + with os.fdopen(file_descriptor, "wb") as stream: # the digested bytes + stream.write(code.encode("utf-8")) stream.flush() os.fsync(stream.fileno()) except BaseException: diff --git a/openhcs/pyqt_gui/main.py b/openhcs/pyqt_gui/main.py index faaac1c680..2a2c38e934 100644 --- a/openhcs/pyqt_gui/main.py +++ b/openhcs/pyqt_gui/main.py @@ -5,6 +5,7 @@ """ import logging +from concurrent.futures import Future from collections.abc import Callable from functools import singledispatchmethod from pathlib import Path @@ -19,7 +20,6 @@ from PyQt6.QtWidgets import ( QApplication, QDialog, - QFileDialog, QLabel, QMainWindow, QMessageBox, @@ -209,6 +209,8 @@ def __init__( # Core configuration self.runtime_context = runtime_context self.function_catalog_service = function_catalog_service + # Completes once the shared endpoint is attached and its catalog prewarmed. + self.execution_services_prepared: Future[None] | None = None # Create shared components self.storage_registry = storage_registry @@ -359,7 +361,9 @@ def deferred_initialization(self): def start_background_services(self) -> None: """Start non-visual services after the initialized window has painted.""" - self.service_adapter.execute_async_operation(self._prepare_execution_services) + self.execution_services_prepared = self.service_adapter.execute_async_operation( + self._prepare_execution_services + ) self._start_ui_bridge_if_enabled() self._check_for_updates_on_startup() @@ -1542,59 +1546,6 @@ def _close_object_state_subscriptions(self) -> None: self._object_state_unregister_subscription.release() self.time_travel_widget.close() - # ========== THEME MANAGEMENT METHODS ========== - - def switch_to_dark_theme(self): - """Switch to dark theme variant.""" - self.service_adapter.theme_manager.switch_to_dark_theme() - self.status_message.emit("Switched to dark theme") - - def switch_to_light_theme(self): - """Switch to light theme variant.""" - self.service_adapter.theme_manager.switch_to_light_theme() - self.status_message.emit("Switched to light theme") - - def load_theme_from_file(self): - """Load theme from JSON configuration file.""" - file_path, _ = QFileDialog.getOpenFileName( - self, "Load Theme Configuration", "", "JSON Files (*.json);;All Files (*)" - ) - - if file_path: - theme_loaded = self.service_adapter.theme_manager.load_theme_from_config( - file_path - ) - if theme_loaded: - self.status_message.emit(f"Loaded theme from {Path(file_path).name}") - else: - QMessageBox.warning( - self, - "Theme Load Error", - f"Failed to load theme from {Path(file_path).name}", - ) - - def save_theme_to_file(self): - """Save current theme to JSON configuration file.""" - file_path, _ = QFileDialog.getSaveFileName( - self, - "Save Theme Configuration", - "pyqt6_color_scheme.json", - "JSON Files (*.json);;All Files (*)", - ) - - if file_path: - theme_saved = self.service_adapter.theme_manager.save_current_theme( - file_path - ) - if theme_saved: - self.status_message.emit(f"Saved theme to {Path(file_path).name}") - else: - QMessageBox.warning( - self, - "Theme Save Error", - f"Failed to save theme to {Path(file_path).name}", - ) - @singledispatchmethod def on_session_event(self, event: SessionEvent) -> None: del event diff --git a/pyproject.toml b/pyproject.toml index c7ffa0c268..7abb8dfaab 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -63,6 +63,9 @@ dependencies = [ "numba>=0.61.0,<0.63; platform_system == 'Darwin' and platform_machine == 'x86_64'", "numba>=0.61.0; platform_system != 'Darwin' or platform_machine != 'x86_64'", "skan>=0.13.1", + # Every execution server prepares the CellProfiler watershed catalog entries, + # whose CellProfiler 4 paths run mahotas; it is a runtime requirement. + "mahotas>=1.4.18", "threadpoolctl>=3.6.0", # Image I/O and formats @@ -133,7 +136,6 @@ dev-gui = [ # Production CellProfiler compatibility dependencies cellprofiler-compat = [ - "mahotas>=1.4.18", "matplotlib>=3.10.0", ] diff --git a/scripts/build_mcp_knowledge_assets.py b/scripts/build_mcp_knowledge_assets.py index 8ed94107b1..59f8660cb7 100644 --- a/scripts/build_mcp_knowledge_assets.py +++ b/scripts/build_mcp_knowledge_assets.py @@ -32,6 +32,7 @@ def _build_module(relative_path: str) -> SimpleNamespace: ManifestRootAcquisitionKind = _UPSTREAM_SOURCES.ManifestRootAcquisitionKind ManifestRootAcquisitionSpec = _UPSTREAM_SOURCES.ManifestRootAcquisitionSpec KNOWLEDGE_MANIFEST_RELATIVE_PATH = _MANIFEST_SCHEMA.DEFAULT_KNOWLEDGE_BASE_MANIFEST_PATH +PACKAGED_KNOWLEDGE_PATH_BUDGET = _MANIFEST_SCHEMA.PACKAGED_KNOWLEDGE_PATH_BUDGET PACKAGED_KNOWLEDGE_ROOT_RELATIVE_PATH = ( Path("openhcs/agent") / _MANIFEST_SCHEMA.PACKAGED_KNOWLEDGE_BASE_ROOT ) @@ -191,6 +192,35 @@ def project_knowledge_assets( ) +def knowledge_asset_projections( + project_root: Path, + recipe_type: type[ComparisonManifestSnapshot] = ComparisonManifestSnapshot, +) -> dict[Path, Path]: + """Packaged resource path -> canonical source, within the path budget.""" + projections = knowledge_source_projections( + project_root / KNOWLEDGE_MANIFEST_RELATIVE_PATH, + source_root=project_root, + recipe_type=recipe_type, + ) + plugin_manifest = project_root / AGENT_PLUGIN_MANIFEST_PATH + if plugin_manifest.is_file(): + skill_sources = AgentSkillBundle.from_manifest(plugin_manifest).source_paths() + projections.update( + {source.relative_to(project_root): source for source in skill_sources} + ) + over_budget = tuple( + relative.as_posix() + for relative in projections + if len(relative.as_posix()) > PACKAGED_KNOWLEDGE_PATH_BUDGET + ) + if over_budget: + raise ValueError( + f"MCP knowledge resource paths exceed {PACKAGED_KNOWLEDGE_PATH_BUDGET} " + f"characters: {over_budget}" + ) + return projections + + def _project_knowledge_assets( project_root: Path, destination_root: Path, @@ -207,17 +237,7 @@ def _project_knowledge_assets( raise ValueError( f"MCP knowledge destination must not own the project root: {destination}" ) - projections = knowledge_source_projections( - root / KNOWLEDGE_MANIFEST_RELATIVE_PATH, - source_root=root, - recipe_type=native_sources.recipe_type(), - ) - plugin_manifest = root / AGENT_PLUGIN_MANIFEST_PATH - if plugin_manifest.is_file(): - skill_sources = AgentSkillBundle.from_manifest(plugin_manifest).source_paths() - projections.update( - {source.relative_to(root): source for source in skill_sources} - ) + projections = knowledge_asset_projections(root, native_sources.recipe_type()) native_sources.materialize() missing = tuple(source for source in projections.values() if not source.is_file()) if missing: diff --git a/scripts/smoke_installed_gui.py b/scripts/smoke_installed_gui.py index faf0dce9af..2171ed2c95 100644 --- a/scripts/smoke_installed_gui.py +++ b/scripts/smoke_installed_gui.py @@ -1,7 +1,6 @@ """Operate an installed OpenHCS GUI through MCP outside the source checkout.""" from __future__ import annotations -from openhcs.authoring.session.operations.datasets import ShowDatasetCode import argparse import faulthandler @@ -39,6 +38,7 @@ JsonValue = str | int | bool | None | list["JsonValue"] | dict[str, "JsonValue"] AgentDtoT = TypeVar("AgentDtoT") +ResultT = TypeVar("ResultT") @dataclass(frozen=True, slots=True) @@ -458,13 +458,14 @@ def _typed_tool_payload( return payload -def require_ui_action_accepted(action_result: UiActionInvokeResult) -> None: - """Fail the smoke unless the live GUI took the requested action.""" +def accepted_ui_action_status(action_result: UiActionInvokeResult) -> str: + """The wire name of the accepted status; fail unless the GUI took the action.""" from formbridge.bridge.dto import ActionAccepted if action_result.status is not ActionAccepted: raise AssertionError(f"Installed UI action was not accepted: {action_result}") + return action_result.status.name def _window_ids(catalog: UiWindowCatalog) -> frozenset[str]: @@ -510,20 +511,31 @@ def _capture_installed_gui_snapshot( ) -def verify_installed_function_catalog( - service: ZMQFunctionCatalogService, - *, - deadline: OperationDeadline, -) -> InstalledFunctionCatalogSmokeResult: - """Require the GUI's background endpoint catalogue and one usable reference.""" +def _result_within(future: Future[ResultT], deadline: OperationDeadline) -> ResultT: + """Wait inside the smoke deadline; an owner's own timeout stays its own.""" - future = service.prepare() try: - page = future.result(timeout=deadline.remaining_seconds()) + return future.result(timeout=deadline.remaining_seconds()) except TimeoutError as error: if not future.cancelled() and future.done() and future.exception() is error: raise raise deadline.timeout_error() from error + + +def verify_installed_function_catalog( + service: ZMQFunctionCatalogService, + *, + execution_services_prepared: Future[None], + deadline: OperationDeadline, +) -> InstalledFunctionCatalogSmokeResult: + """Require the GUI's background endpoint catalogue and one usable reference. + + The catalog attaches only to an existing endpoint, so first wait for the + GUI's own endpoint start and catalog prewarm to finish. + """ + + _result_within(execution_services_prepared, deadline) + page = _result_within(service.prepare(), deadline) if not page.revision.strip(): raise AssertionError("Installed function catalog omitted its revision.") if page.total <= 0 or not page.items: @@ -555,6 +567,7 @@ def run_installed_live_mcp_smoke( evidence_directory: Path, forbidden_root: Path, function_catalog_service: ZMQFunctionCatalogService, + execution_services_prepared: Future[None], execution_config: OpenHCSZMQConfig, deadline: OperationDeadline, journal: InstalledGuiSmokeEvidenceJournal, @@ -577,6 +590,7 @@ def run_installed_live_mcp_smoke( MainWindowWidgetIdentity, DatasetListWidgetIdentity, ) + from openhcs.authoring.session.operations.datasets import ShowDatasetCode from openhcs.mcp.dev_client import McpDevClient from openhcs.mcp.dev_client_commanding import CapabilityBackedCommandSpec from openhcs.mcp.dev_client_commands.ui import ( @@ -587,6 +601,7 @@ def run_installed_live_mcp_smoke( journal.record(InstalledGuiSmokePhase.WAITING_FOR_FUNCTION_CATALOG) function_catalog = verify_installed_function_catalog( function_catalog_service, + execution_services_prepared=execution_services_prepared, deadline=deadline, ) journal.record( @@ -703,7 +718,7 @@ def run_installed_live_mcp_smoke( capability=agent_capabilities.ui_invoke_action, payload_type=UiActionInvokeResult, ) - require_ui_action_accepted(action_result) + action_status = accepted_ui_action_status(action_result) timeout_text = f"{deadline.remaining_seconds():g}" windows_execution = client.execute( @@ -743,12 +758,14 @@ def run_installed_live_mcp_smoke( journal.record(InstalledGuiSmokePhase.LIVE_MCP_COMPLETE) return InstalledLiveMcpSmokeResult( - action_status=action_result.status, + action_status=action_status, bridge_count=len(bridge_catalog.bridges), bridge_port=bridge_status.connection.require_port("Installed UI bridge smoke"), bridge_reachable=bridge_status.reachable, bridge_transport=( - bridge_status.connection.transport_endpoint().transport_mode.value + bridge_status.connection.transport_endpoint( + execution_config + ).transport_mode.value ), execution_server_log=execution_server_log, function_catalog=function_catalog, @@ -877,6 +894,9 @@ def run_live_mcp() -> None: evidence_directory=evidence_directory, forbidden_root=forbidden_root, function_catalog_service=application.function_catalog_service, + execution_services_prepared=( + main_window.execution_services_prepared + ), execution_config=ui_config.zmq, deadline=deadline, journal=journal, diff --git a/scripts/smoke_installed_mcp.py b/scripts/smoke_installed_mcp.py index 5f3bd995f0..74ef95ebe5 100644 --- a/scripts/smoke_installed_mcp.py +++ b/scripts/smoke_installed_mcp.py @@ -280,7 +280,7 @@ async def _run_measured_execution_protocol_smoke( "wavelengths": 1, "z_stack_levels": 1, "num_cells": 2, - "wells": ["A01"], + "partition_values": ["A01"], "format": "ImageXpress", "random_seed": 7, }, diff --git a/tests/conftest.py b/tests/conftest.py index 068fef6fc7..4171a9b1cb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -8,6 +8,7 @@ from pathlib import Path from openhcs._source_dependencies import ensure_source_checkout_external_paths +from openhcs.utils.environment import OpenHCSProcessEnvironment ensure_source_checkout_external_paths() @@ -63,7 +64,7 @@ def pytest_sessionstart(session): # Conditionally import pytest-qt only when not in CPU-only mode -CPU_ONLY_MODE = os.getenv("OPENHCS_CPU_ONLY", "false").lower() == "true" +CPU_ONLY_MODE = OpenHCSProcessEnvironment.cpu_only_mode() if not CPU_ONLY_MODE: pytest_plugins = ["pytestqt"] else: @@ -151,25 +152,30 @@ def reap_spawned_processes(monkeypatch) -> Iterator[None]: marker = uuid.uuid4().hex monkeypatch.setenv(SPAWN_MARKER_VARIABLE, marker) yield - own_group = os.getpgid(0) survivors = _marked_processes(marker) for process in survivors: - try: - group = os.getpgid(process.pid) - if group != own_group: - os.killpg(group, signal.SIGTERM) - else: - process.terminate() - except (ProcessLookupError, psutil.NoSuchProcess): - continue + _stop_marked_process(process, force=False) _gone, alive = psutil.wait_procs(survivors, timeout=10) for process in alive: - try: - group = os.getpgid(process.pid) - if group != own_group: - os.killpg(group, signal.SIGKILL) - else: - process.kill() - except (ProcessLookupError, psutil.NoSuchProcess): - continue + _stop_marked_process(process, force=True) psutil.wait_procs(alive, timeout=10) + + +def _stop_marked_process(process: psutil.Process, *, force: bool) -> None: + """Signal a detached POSIX session's whole group, else the process itself. + + Windows has no process groups; every descendant carries the marker, so + stopping each marked process covers the tree there. + """ + try: + if os.name == "posix": + group = os.getpgid(process.pid) + if group != os.getpgid(0): + os.killpg(group, signal.SIGKILL if force else signal.SIGTERM) + return + if force: + process.kill() + else: + process.terminate() + except (ProcessLookupError, psutil.NoSuchProcess): + return diff --git a/tests/integration/test_agent_observation_export_zmq.py b/tests/integration/test_agent_observation_export_zmq.py index 57d1bd6252..6ebd6388b4 100644 --- a/tests/integration/test_agent_observation_export_zmq.py +++ b/tests/integration/test_agent_observation_export_zmq.py @@ -11,7 +11,7 @@ import pytest from objectstate.object_state import ObjectStateRegistry from zmqruntime import DataControlPortPairAuthority -from zmqruntime.messages import ExecutionStatus +from zmqruntime.messages import ExecutionStatus, MessageFields from benchmark.cellprofiler_benchmark_cli import create_benchmark_argument_parser from benchmark.contracts.measured_run_receipt import MeasuredPipelineRunReceipt @@ -335,7 +335,11 @@ def test_ordinary_execution_can_export_outcomes_without_value_observation( exported = ZMQRuntimeExecutionOutcomeExport.read(export_path) exported.require_successful_axes() - assert completed.results_summary["well_count"] == exported.axis_count == 1 + assert ( + completed.results_summary[MessageFields.PARTITION_COUNT] + == exported.axis_count + == 1 + ) assert completed.results_summary["runtime_observation_export_scope"] == "outcomes" assert exported.successful_axis_count == 1 assert exported.output_roots diff --git a/tests/integration/test_basicpy_real_fit.py b/tests/integration/test_basicpy_real_fit.py index 8c271094f9..4025252c49 100644 --- a/tests/integration/test_basicpy_real_fit.py +++ b/tests/integration/test_basicpy_real_fit.py @@ -17,6 +17,7 @@ def test_real_algorithm_uses_numpy_transport_and_is_cpu_catalog_eligible(monkeyp from openhcs.processing.backends.lib_registry.openhcs_registry import ( _catalog_memory_types, _module_declares_allowed_memory_type, + _module_distributions_excluded_by_package_metadata, ) monkeypatch.setenv("OPENHCS_CPU_ONLY", "true") @@ -28,6 +29,12 @@ def test_real_algorithm_uses_numpy_transport_and_is_cpu_catalog_eligible(monkeyp assert _module_declares_allowed_memory_type( basic_flatfield_correction_jax.__module__, frozenset({MemoryType.NUMPY.value}) ) + assert ( + _module_distributions_excluded_by_package_metadata( + basic_flatfield_correction_jax.__module__ + ) + == () + ) def run_fit(observations): diff --git a/tests/integration/test_benchmark_openhcs_adapter_cppipe.py b/tests/integration/test_benchmark_openhcs_adapter_cppipe.py index de09888fc7..57be0014ed 100644 --- a/tests/integration/test_benchmark_openhcs_adapter_cppipe.py +++ b/tests/integration/test_benchmark_openhcs_adapter_cppipe.py @@ -105,8 +105,9 @@ def test_openhcs_adapter_retains_declared_path_policy_and_explicit_exports( for path in result.output_path.rglob("*") if path.is_file() and path.suffix.lower() in {".tif", ".tiff"} ) - assert tiff_outputs - assert all(path.name.endswith(".labels.tif") for path in tiff_outputs) + # With automatic materialization disabled only declared exports persist: + # the spreadsheet. Retained labels and QA checkpoints stay in memory. + assert tiff_outputs == () def test_openhcs_adapter_resolves_dataset_reference_cppipe( diff --git a/tests/integration/test_cellprofiler_official30_zmq.py b/tests/integration/test_cellprofiler_official30_zmq.py index a298d1d30d..ced0397664 100644 --- a/tests/integration/test_cellprofiler_official30_zmq.py +++ b/tests/integration/test_cellprofiler_official30_zmq.py @@ -1,7 +1,5 @@ from __future__ import annotations -from itertools import product - import json import os import shutil @@ -51,7 +49,12 @@ runtime_artifact_viewer_component_identity, ) from openhcs.core.source_spatial_domain import SourceSpatialDomain -from openhcs.core.streaming_config_declarations import ViewerType +from openhcs.core.streaming_config_declarations import ( + FijiViewer, + NapariViewer, + ViewerFamily, + ViewerType, +) from openhcs.runtime.fiji_macro_runtime import FijiMacroExecutionRequest from openhcs.runtime.viewer_protocol import ( ViewerControlMessageRequest, @@ -75,20 +78,20 @@ for case_index, case in enumerate(_OFFICIAL30_CASES) ) _OFFICIAL30_FIJI_VIEWER_VARIANTS = ( - (ViewerType.FIJI,), - (ViewerType.FIJI, ViewerType.NAPARI), + (FijiViewer,), + (FijiViewer, NapariViewer), ) _OFFICIAL30_FIJI_VIEWER_PARAMS = tuple( pytest.param( - viewer_types, + viewers, case_index, case, id=( - "+".join(viewer_type.value for viewer_type in viewer_types) + "+".join(viewer.wire_value for viewer in viewers) + f"-{case_index:02d}-{case.name}" ), ) - for viewer_types in _OFFICIAL30_FIJI_VIEWER_VARIANTS + for viewers in _OFFICIAL30_FIJI_VIEWER_VARIANTS for case_index, case in enumerate(_OFFICIAL30_CASES) ) _FIJI_FRESH_PROCESS_IMAGE_PROBE = r"""#@ String Directory @@ -236,13 +239,13 @@ def _free_zmq_port_pair( def _registered_streaming_config_kwargs( - viewer_types: Sequence[ViewerType], - ports_by_viewer: Mapping[ViewerType, int], + viewers: Sequence[type[ViewerFamily]], + ports_by_viewer: Mapping[type[ViewerFamily], int], ) -> dict[str, StreamingConfig]: """Project one viewer selection through registered config owners.""" - selected_viewers = set(viewer_types) - if len(selected_viewers) != len(viewer_types): + selected_viewers = set(viewers) + if len(selected_viewers) != len(viewers): raise ValueError("Official30 viewer variants require unique declarations.") if set(ports_by_viewer) != selected_viewers: raise ValueError( @@ -250,8 +253,8 @@ def _registered_streaming_config_kwargs( ) config_kwargs: dict[str, StreamingConfig] = {} - for viewer_type, config_type in StreamingConfig.__registry__.items(): - enabled = viewer_type in selected_viewers + for viewer, config_type in StreamingConfig.__registry__.items(): + enabled = viewer in selected_viewers init_kwargs: dict[str, object] = { "enabled": enabled, "persistent": enabled, @@ -259,10 +262,10 @@ def _registered_streaming_config_kwargs( if enabled: init_kwargs.update( host="127.0.0.1", - port=ports_by_viewer[viewer_type], + port=ports_by_viewer[viewer], transport_mode=TransportMode.TCP, ) - config_kwargs[viewer_type.family.config_key] = config_type(**init_kwargs) + config_kwargs[viewer.config_key] = config_type(**init_kwargs) if selected_viewers - set(StreamingConfig.__registry__): raise ValueError("Official30 viewer variant selected an unregistered config.") @@ -399,7 +402,7 @@ def _assert_napari_state_matches_runtime( runtime_artifact_viewer_component_identity(values) for values in layer.component_values ) == Counter( - runtime_artifact_viewer_component_identity(payload["components"]) + runtime_artifact_viewer_component_identity(payload.components) for payload in layer.payload_summaries ) assert all(component in layer.axis_labels for component in layer.stack_axes) @@ -411,59 +414,44 @@ def _assert_napari_state_matches_runtime( assert layer.data_shape[axis_index] == len(component_values) for expected in runtime_observation.expectation.artifact_viewer: - # One producer identity can legitimately back both its native image - # layer and its derived shapes overlay; the payload expectation below - # compares ndarray summaries, so match the image layer. - matching_layers = tuple( + # One producer identity can back both its native image layer and its + # derived shapes overlay. The producer's expected payloads are every + # payload it streamed, so compare against all of its layers' payloads. + producer_layers = tuple( layer for layer in state.layers if expected.producer_identity in layer.producer_identities - and StreamingDataType.IMAGE.value in layer.data_types ) - assert len(matching_layers) == 1 - layer = matching_layers[0] + image_layers = tuple( + layer + for layer in producer_layers + if StreamingDataType.IMAGE.value in layer.data_types + ) + assert len(image_layers) == 1 + layer = image_layers[0] assert layer.mounted - # The presentation may aggregate a varying component axis (e.g. site) - # into one layer item. That item represents both the aggregate payload - # itself and each per-coordinate payload it covers, so the actual - # identity set is the union of the summary's own identity and every - # expanded per-coordinate identity. - actual_payloads: list[RuntimeArtifactViewerPayloadExpectation] = [] - for payload in layer.payload_summaries: - aggregates = payload.get("aggregate_component_values") or {} - varying_axes = tuple(aggregates.items()) - - def summary_identity(components) -> RuntimeArtifactViewerPayloadExpectation: - return RuntimeArtifactViewerPayloadExpectation( - components=runtime_artifact_viewer_component_identity(components), - source_spatial_domain=SourceSpatialDomain.from_viewer_wire_mapping( - payload, - source_label="official30 Napari payload summary", - ), - ) - - actual_payloads.append(summary_identity(payload["components"])) - axes_values = [values for _, values in varying_axes] - for combination in product(*axes_values): - expanded = dict(payload["components"]) - expanded.update( - { - component: value - for (component, _), value in zip( - varying_axes, combination, strict=True - ) - } - ) - actual_payloads.append(summary_identity(expanded)) + actual_payloads = tuple( + RuntimeArtifactViewerPayloadExpectation( + components=runtime_artifact_viewer_component_identity( + payload.components + ), + source_spatial_domain=SourceSpatialDomain.from_viewer_wire_mapping( + payload.to_wire_mapping(), + source_label="official30 Napari payload summary", + ), + ) + for producer_layer in producer_layers + for payload in producer_layer.payload_summaries + ) assert Counter(payload.identity_key for payload in actual_payloads) == Counter( payload.identity_key for payload in expected.payloads ) - payload_paths = tuple(payload["path"] for payload in layer.payload_summaries) + payload_paths = tuple(payload.path for payload in layer.payload_summaries) assert all(isinstance(path, str) and path for path in payload_paths) payload_data_types = tuple( - dict.fromkeys(payload["data_type"] for payload in layer.payload_summaries) + dict.fromkeys(payload.data_type for payload in layer.payload_summaries) ) assert len(payload_data_types) == 1 assert layer.data_types == payload_data_types @@ -563,13 +551,12 @@ def test_official30_fiji_variants_project_registered_viewer_configs() -> None: partition_filter_config=PartitionFilterConfig(partition_filter=1) ) - for viewer_types in _OFFICIAL30_FIJI_VIEWER_VARIANTS: + for viewers in _OFFICIAL30_FIJI_VIEWER_VARIANTS: ports_by_viewer = { - viewer_type: 20_000 + index * 100 - for index, viewer_type in enumerate(viewer_types) + viewer: 20_000 + index * 100 for index, viewer in enumerate(viewers) } configs = _registered_streaming_config_kwargs( - viewer_types, + viewers, ports_by_viewer, ) variant_config = GlobalPipelineConfig( @@ -585,11 +572,11 @@ def test_official30_fiji_variants_project_registered_viewer_configs() -> None: if field.name not in registered_keys ) assert { - config.viewer_family.viewer_type() for config in configs.values() if config.enabled - } == set(viewer_types) + config.viewer_family for config in configs.values() if config.enabled + } == set(viewers) assert all(config.persistent is config.enabled for config in configs.values()) assert { - config.viewer_family.viewer_type(): config.port + config.viewer_family: config.port for config in configs.values() if config.enabled } == ports_by_viewer @@ -742,12 +729,12 @@ def test_official30_nonpersistent_napari_isolated_per_case( @pytest.mark.integration @pytest.mark.parametrize( - ("viewer_types", "case_index", "case"), + ("viewers", "case_index", "case"), _OFFICIAL30_FIJI_VIEWER_PARAMS, ) def test_official30_persistent_fiji_variants_isolated_per_case( tmp_path: Path, - viewer_types: tuple[ViewerType, ...], + viewers: tuple[type[ViewerFamily], ...], case_index: int, case: CellProfilerComparisonCase, ) -> None: @@ -760,14 +747,14 @@ def test_official30_persistent_fiji_variants_isolated_per_case( transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, ) ports_by_viewer = { - viewer_type: _free_zmq_port_pair( + viewer: _free_zmq_port_pair( excluded_ports, transport_mode=TransportMode.TCP, ) - for viewer_type in viewer_types + for viewer in viewers } streaming_configs = _registered_streaming_config_kwargs( - viewer_types, + viewers, ports_by_viewer, ) global_config = GlobalPipelineConfig( @@ -775,15 +762,15 @@ def test_official30_persistent_fiji_variants_isolated_per_case( **streaming_configs, ) endpoints = { - viewer_type.value: ViewerRuntimeEndpoint( - streaming_configs[viewer_type.family.config_key] + viewer.wire_value: ViewerRuntimeEndpoint( + streaming_configs[viewer.config_key] .viewer_runtime_config() .transport_endpoint, OPENHCS_ZMQ_CONFIG, ) - for viewer_type in viewer_types + for viewer in viewers } - variant_id = "+".join(viewer_type.value for viewer_type in viewer_types) + variant_id = "+".join(viewer.wire_value for viewer in viewers) case_output_root = tmp_path / variant_id / f"case_{case_index:02d}" macro_path = tmp_path / f"{variant_id}_fresh_process_probe.ijm" macro_path.write_text(_FIJI_FRESH_PROCESS_IMAGE_PROBE, encoding="utf-8") @@ -821,18 +808,18 @@ def test_official30_persistent_fiji_variants_isolated_per_case( assert (case_output_root / "summary.csv").is_file() runtime_observation = _runtime_observation(observations[0]) - for viewer_type in viewer_types: - config = streaming_configs[viewer_type.family.config_key] - if viewer_type is ViewerType.NAPARI: + for viewer in viewers: + config = streaming_configs[viewer.config_key] + if viewer is NapariViewer: _assert_live_napari_state(config, runtime_observation) - elif viewer_type is ViewerType.FIJI: + elif viewer is FijiViewer: _assert_fiji_fresh_process_image( config=config, macro_path=macro_path, ) else: raise AssertionError( - f"Unhandled registered Official30 viewer {viewer_type.name!r}." + f"Unhandled registered Official30 viewer {viewer.display_name!r}." ) except Exception as error: failures.append( diff --git a/tests/integration/test_count_cells_simple_dual_channel.py b/tests/integration/test_count_cells_simple_dual_channel.py index 8804075b1c..a94fea77e4 100644 --- a/tests/integration/test_count_cells_simple_dual_channel.py +++ b/tests/integration/test_count_cells_simple_dual_channel.py @@ -55,6 +55,7 @@ SyntheticMicroscopyGenerator, ) from openhcs.domains.microscopy.axes import Microscopy +from openhcs.microscopes.imagexpress import ImageXpressHandler def _write_known_dual_channel_images(plate_dir): @@ -74,19 +75,19 @@ def _write_known_dual_channel_images(plate_dir): image_dir = plate_dir / "TimePoint_1" for well in ("A01", "B01"): tifffile.imwrite( - image_dir / f"{well}_s001_w1_z001_t001.tif", + image_dir / f"{well}_s001_w1.tif", channel_1, ) tifffile.imwrite( - image_dir / f"{well}_s001_w2_z001_t001.tif", + image_dir / f"{well}_s001_w2.tif", channel_2, ) tifffile.imwrite( - image_dir / f"{well}_s002_w1_z001_t001.tif", + image_dir / f"{well}_s002_w1.tif", channel_1, ) tifffile.imwrite( - image_dir / f"{well}_s002_w2_z001_t001.tif", + image_dir / f"{well}_s002_w2.tif", channel_2, ) @@ -186,7 +187,6 @@ def test_dual_channel_count_runs_on_synthetic_plate_with_channel_stack( from openhcs.core.materialization import CsvOptions, MaterializationSpec, ROIOptions import numpy as np -from openhcs.microscopes.imagexpress import ImageXpressHandler @numpy diff --git a/tests/integration/test_custom_source_transport.py b/tests/integration/test_custom_source_transport.py index 33162103e5..624827fdcb 100644 --- a/tests/integration/test_custom_source_transport.py +++ b/tests/integration/test_custom_source_transport.py @@ -509,7 +509,7 @@ def _spawn(path): from openhcs.core.config import MultiprocessingStartMethod from openhcs.core.orchestrator.cancellation import ExecutionCancellationSignal from openhcs.core.orchestrator.worker_execution import WorkerExecutorFactory - from openhcs.core.progress import ProgressExecutionContext + from openhcs.core.progress import ProgressExecutionContext, ProgressStatus # A future derived context can carry nontransportable local runtime state; # no generic worker-bootstrap consumer should need to learn its fields. diff --git a/tests/integration/test_measured_cli_session.py b/tests/integration/test_measured_cli_session.py index 7eb363e046..8754e3cb36 100644 --- a/tests/integration/test_measured_cli_session.py +++ b/tests/integration/test_measured_cli_session.py @@ -78,7 +78,37 @@ def _source_state(root: Path) -> dict[str, tuple]: return state +# Everyone may not create, change or delete entries; inherited by the tree. +# Write rights are named one by one: the generic W also denies SYNCHRONIZE, +# which every open (reads included) requests. +# Deny adding files and folders to the root itself. An inherited deny over the +# tree (OI)(CI) also refused reading the source files on the CI runner. +WINDOWS_READ_ONLY_ACE = "*S-1-1-0:(WD,AD)" + +if sys.platform == "win32": + # A console's Ctrl+C cannot be sent to one process; Ctrl+Break can, to a + # child started in its own process group. + INTERRUPT_LAUNCH = {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} + INTERRUPT_SIGNAL = signal.CTRL_BREAK_EVENT +else: + INTERRUPT_LAUNCH = {} + INTERRUPT_SIGNAL = signal.SIGINT + + def _set_writable(root: Path, writable: bool) -> None: + """Make ``root`` and its tree writable again, or genuinely not writable. + + Windows ignores the read-only attribute on directories (and os.access + ignores ACLs), so there a deny ACE on the root refuses creating files in it; + the before/after source state still catches any write below it. + """ + + if sys.platform == "win32": + change = ( + ("/remove:d", "*S-1-1-0") if writable else ("/deny", WINDOWS_READ_ONLY_ACE) + ) + subprocess.run(("icacls", str(root), *change), check=True, capture_output=True) + return for path in (root, *root.rglob("*")): mode = path.stat().st_mode path.chmod(mode | 0o200 if writable else mode & ~0o222) @@ -114,7 +144,9 @@ def _command( def _environment(tmp_path: Path) -> dict[str, str]: - environment = {**os.environ, "PYTHONPATH": str(REPOSITORY_ROOT)} + # The CLI child imports benchmark and openhcs from the same installation as + # this test process; a checkout path would shadow an installed candidate. + environment = dict(os.environ) environment.setdefault("XDG_DATA_HOME", str(tmp_path / "xdg")) environment.pop("DISPLAY", None) return environment @@ -176,6 +208,7 @@ def test_ctrl_c_stops_the_run_and_its_server(tmp_path: Path) -> None: stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, + **INTERRUPT_LAUNCH, ) stderr_lines: list[str] = [] try: @@ -186,7 +219,7 @@ def test_ctrl_c_stops_the_run_and_its_server(tmp_path: Path) -> None: else: pytest.fail("".join(stderr_lines)[-4000:]) time.sleep(1.0) - process.send_signal(signal.SIGINT) + process.send_signal(INTERRUPT_SIGNAL) _stdout, rest = process.communicate(timeout=RUN_DEADLINE_SECONDS) finally: if process.poll() is None: diff --git a/tests/integration/test_neurite_outgrowth_synthetic_plate.py b/tests/integration/test_neurite_outgrowth_synthetic_plate.py index 75ce69b60b..addaa8d603 100644 --- a/tests/integration/test_neurite_outgrowth_synthetic_plate.py +++ b/tests/integration/test_neurite_outgrowth_synthetic_plate.py @@ -67,6 +67,7 @@ ViewerStreamBackendCallKwargs, ) from openhcs.processing.backends.analysis.neurite_outgrowth import ( + CellProfilerNeuriteEngineProfile, MetaXpressCellBodySettings, MetaXpressNuclearSettings, MetaXpressOutgrowthSettings, @@ -94,13 +95,13 @@ def _write_known_neurite_images(plate_dir): image_dir = plate_dir / "TimePoint_1" for site in (1, 2): - tifffile.imwrite(image_dir / f"A01_s{site:03d}_w1_z001_t001.tif", neurites) - tifffile.imwrite(image_dir / f"A01_s{site:03d}_w2_z001_t001.tif", nuclei) + tifffile.imwrite(image_dir / f"A01_s{site:03d}_w1.tif", neurites) + tifffile.imwrite(image_dir / f"A01_s{site:03d}_w2.tif", nuclei) return np.stack((neurites, nuclei)) def test_neurite_outgrowth_runs_on_synthetic_plate_as_2d_channel_stack( - tmp_path, caplog, monkeypatch + tmp_path, caplog, monkeypatch, viewer_ack_return_route ): caplog.set_level(logging.CRITICAL) plate_dir = tmp_path / "synthetic_neurite_plate" @@ -311,19 +312,14 @@ def observe_dense_output(output, components): np.testing.assert_array_equal(retained, expected_labels) checkpoint_paths = [] + # Checkpoints are the declared outputs between the label artifacts and + # the trailing morphology table; the callable's output family owns them. + checkpoint_names = tuple( + spec.name + for spec in CellProfilerNeuriteEngineProfile.artifact_outputs()[6:-1] + ) for artifact_name, expected_checkpoint in zip( - ( - "neurite_candidate_mask", - "neurite_unrooted_residual", - "neurite_secondary_ownership", - "neurite_topology_dropped_trace", - "neurite_topology_added_trace", - "neurite_enhanced_response", - "neurite_threshold_support", - "neurite_retained_support", - "neurite_local_response", - "neurite_local_support", - ), + checkpoint_names, expected[7:-1], strict=True, ): @@ -336,7 +332,9 @@ def observe_dense_output(output, components): retained, np.asarray(expected_checkpoint)[0], ) - assert len({path.name for path in checkpoint_paths}) == 20 + assert len({path.name for path in checkpoint_paths}) == 2 * len( + checkpoint_names + ) assert compiled_plan.output_dataset_root is not None output_dataset_root = Path(compiled_plan.output_dataset_root) @@ -368,7 +366,8 @@ def observe_dense_output(output, components): # Exercise the real stream projection of the compiled writer outputs. # Artifact storage axes are legitimately empty; source image planes are not. - assert len(dense_outputs) == 18 + # Both sites publish four label images and every declared checkpoint. + assert len(dense_outputs) == 2 * (4 + len(checkpoint_names)) stream_kwargs = ViewerStreamBackendCallKwargs( ViewerStreamBackendKwargs( ViewerStreamRequest( @@ -424,13 +423,16 @@ def observe_dense_output(output, components): request ), display_payload_extra=backend.display_payload_extra(request), + return_route=viewer_ack_return_route, ), ) for output, item in zip(outputs, batch.batch_images, strict=True): assert output.variable_components == () - # This fixture injects physical pixel_size into the callable, - # but does not declare voxel spacing on its runtime images. - assert output.metadata.source_voxel_spacing == SourceVoxelSpacing() + # The acquisition's calibration is the same physical pixel + # size the compiler injected into the callable. + assert output.metadata.source_voxel_spacing == SourceVoxelSpacing( + (compiled_invocation.kwargs_dict["pixel_size"],) * 2 + ) if checkpoint_batch: assert "_w1_" in Path(output.path).name else: diff --git a/tests/integration/test_session_journey.py b/tests/integration/test_session_journey.py index fe7966d343..f99d4f83ec 100644 --- a/tests/integration/test_session_journey.py +++ b/tests/integration/test_session_journey.py @@ -1,8 +1,9 @@ -"""One dataset through the session, headless over MCP and in the Qt GUI. +"""One dataset through the session, headless over MCP. -Both journeys add a synthetic dataset, give it a pipeline, initialize, compile -and run it on a real execution server, following the session's events. They -use the same SessionOperation classes; only the client differs. +The journey adds a synthetic dataset, gives it a pipeline, initializes, +compiles and runs it on a real execution server, following the session's +events. tests/pyqt_gui/test_session_gui_journey.py runs the same +SessionOperation classes through the Qt widgets. """ from __future__ import annotations @@ -18,12 +19,9 @@ from openhcs.agent.path_policy import AgentPathPolicy from openhcs.authoring.session.operations.datasets import ( AddDatasets, - CompileDatasets, InitializeDatasets, - RunDatasets, ) from openhcs.authoring.session.session import CallerThread -from openhcs.authoring.session.views import DatasetListView from openhcs.core.config import PipelineConfig from openhcs.core.pipeline_document import PipelineDocumentCodec from openhcs.core.steps import FunctionStep @@ -152,91 +150,9 @@ def test_headless_mcp_journey_adds_compiles_and_runs_a_dataset(tmp_path: Path) - context.session.close() -def test_gui_journey_runs_the_same_operations_through_the_widgets( - tmp_path: Path, -) -> None: - pytest.importorskip("PyQt6") - from PyQt6.QtWidgets import QApplication - from pyqt_reactive.services.ui_thread_dispatch import UiThreadDispatcher - - from openhcs.authoring.session.session import DispatcherThread, Session - from openhcs.core.config import GlobalPipelineConfig - from openhcs.pyqt_gui.config import get_default_ui_config - from openhcs.pyqt_gui.widgets.plate_manager import PlateManagerWidget - from tests.unit.pyqt_gui.session_harness import GuiServiceStub, release_widgets - - application = QApplication.instance() or QApplication([]) - root, source = _synthetic_dataset(tmp_path) - dispatcher = UiThreadDispatcher() - session = Session( - transport_config=OpenHCSZMQConfig(default_port=_port(1), persistent=False), - global_config=GlobalPipelineConfig(), - main_thread=DispatcherThread(dispatcher), - ) - services = GuiServiceStub() - manager = PlateManagerWidget( - services, session, gui_config=get_default_ui_config() - ) - try: - assert session.invoke(AddDatasets, AddDatasets.request(roots=(str(root),))).accepted - (scope_id,) = session.dataset_scope_ids() - session.set_pipeline_source(scope_id, source) - application.processEvents() - assert [row.scope_id for row in manager.plates] == [scope_id] - manager.item_list.setCurrentRow(0) - application.processEvents() - assert manager.selection_scope_ids() == (scope_id,) - - def wait_until(condition, label) -> None: - deadline = time.monotonic() + STAGE_TIMEOUT_SECONDS - while not condition(): - application.processEvents() - assert time.monotonic() < deadline, label - time.sleep(0.05) - - for operation, finished in ( - (InitializeDatasets, lambda row: row.initialized and not row.init_pending), - (CompileDatasets, lambda row: row.compiled and not row.compile_pending), - ( - RunDatasets, - lambda row: row.terminal_status is not None and not row.execution_active, - ), - ): - if operation is CompileDatasets: - # Disconnected, the compile button is Connect; connect first. - assert manager.buttons[operation.operation_id].text() == "Connect" - manager.handle_button_action(operation.operation_id) - wait_until( - lambda: CompileDatasets.resolved(session) is CompileDatasets, - "connect", - ) - manager.update_button_states() - assert manager.buttons[operation.operation_id].isEnabled(), operation - manager.handle_button_action(operation.operation_id) - deadline = time.monotonic() + STAGE_TIMEOUT_SECONDS - while True: - application.processEvents() - (row,) = DatasetListView.state_of(session).rows - if finished(row): - break - assert time.monotonic() < deadline, (operation, row) - time.sleep(0.05) - assert row.terminal_status == "complete", row - wait_until(lambda: not session.execution_state.busy, "batch finished") - application.processEvents() - assert manager.buttons[RunDatasets.operation_id].text() == "Run" - finally: - manager.cleanup() - manager.close() - release_widgets(application, manager) - session.close() - dispatcher.close() - - def test_read_only_dataset_initializes_in_a_mirrored_workspace(tmp_path: Path) -> None: """Initialization never writes into a source it may not write.""" - from openhcs.authoring.session.operations.datasets import AddDatasets from openhcs.authoring.session.session import Session from openhcs.core.config import GlobalPipelineConfig @@ -272,7 +188,7 @@ def test_read_only_dataset_initializes_in_a_mirrored_workspace(tmp_path: Path) - assert not failures, failures assert time.monotonic() < deadline time.sleep(0.05) - assert session.orchestrator(scope_id).plate_path.is_relative_to(workspaces) + assert session.orchestrator(scope_id).dataset_root.is_relative_to(workspaces) assert _source_state(root) == before finally: session.close() diff --git a/tests/pyqt_gui/integration/test_end_to_end_workflow_foundation.py b/tests/pyqt_gui/integration/test_end_to_end_workflow_foundation.py index 454a2e3aef..61f640c85b 100644 --- a/tests/pyqt_gui/integration/test_end_to_end_workflow_foundation.py +++ b/tests/pyqt_gui/integration/test_end_to_end_workflow_foundation.py @@ -37,6 +37,12 @@ ) from pyqt_reactive.forms.parameter_form_manager import ParameterFormManager +from openhcs.authoring.session.operations.datasets import ( + AddDatasets, + EditDatasetConfig, + InitializeDatasets, +) + from openhcs.core.config import ( GlobalPipelineConfig, LazyPathPlanningConfig, @@ -630,8 +636,12 @@ def _access_plate_manager(context: WorkflowContext) -> WorkflowContext: def _add_and_select_plate(context: WorkflowContext) -> WorkflowContext: - """Add synthetic plate and select it in plate manager.""" - context.plate_manager_widget.add_plate_callback([context.synthetic_plate_dir]) + """Add synthetic plate through the session and select it in plate manager.""" + result = context.plate_manager_widget.session.invoke( + AddDatasets, + AddDatasets.request(roots=(str(context.synthetic_plate_dir),)), + ) + assert result.accepted, result.errors _wait_for_gui(TIMING.ACTION_DELAY) item_list = context.plate_manager_widget.item_list @@ -647,14 +657,21 @@ def _add_and_select_plate(context: WorkflowContext) -> WorkflowContext: def _initialize_plate(context: WorkflowContext) -> WorkflowContext: """Initialize plate using Init button.""" - init_button = context.plate_manager_widget.buttons["init_plate"] + plate_manager = context.plate_manager_widget + init_button = plate_manager.buttons[InitializeDatasets.operation_id] if not init_button.isEnabled(): raise AssertionError( "Init button is disabled - plate may not be properly added" ) init_button.click() - _wait_for_gui(TIMING.SAVE_DELAY) + scope_id = plate_manager.selected_plate_path + session = plate_manager.session + deadline = time.monotonic() + 60.0 + while not session.is_initialized(scope_id) or scope_id in session.init_pending: + if time.monotonic() >= deadline: + raise AssertionError(f"Dataset {scope_id!r} did not finish initializing") + _wait_for_gui(0.05) return context @@ -665,9 +682,8 @@ def _apply_orchestrator_config(context: WorkflowContext) -> WorkflowContext: "Test scenario must be provided for parameterized orchestrator configuration" ) - orchestrator = context.plate_manager_widget.get_selected_orchestrator() - if orchestrator is None: - raise AssertionError("The selected plate has no initialized orchestrator") + plate_manager = context.plate_manager_widget + orchestrator = plate_manager.session.orchestrator(plate_manager.selected_plate_path) orchestrator.apply_pipeline_config(context.test_scenario.pipeline_config) _wait_for_gui(TIMING.ACTION_DELAY) @@ -709,7 +725,7 @@ def _wait_for_form_build( ) def _open_config_window(context: WorkflowContext) -> WorkflowContext: """Open plate-specific configuration window.""" - edit_button = context.plate_manager_widget.buttons["edit_config"] + edit_button = context.plate_manager_widget.buttons[EditDatasetConfig.operation_id] if not edit_button.isEnabled(): raise AssertionError( "Edit button is disabled - plate may not be properly initialized" @@ -891,7 +907,7 @@ def _reopen_config_window(context: WorkflowContext) -> WorkflowContext: context = _close_config_window(context) # Validate edit button state - edit_button = context.plate_manager_widget.buttons["edit_config"] + edit_button = context.plate_manager_widget.buttons[EditDatasetConfig.operation_id] if not edit_button.isEnabled(): raise AssertionError( "LAZY CONFIG BUG: Edit button disabled after closing config window. " diff --git a/tests/pyqt_gui/test_image_browser_filter_controller.py b/tests/pyqt_gui/test_image_browser_filter_controller.py index 889f6afd78..3c83a1b970 100644 --- a/tests/pyqt_gui/test_image_browser_filter_controller.py +++ b/tests/pyqt_gui/test_image_browser_filter_controller.py @@ -24,6 +24,7 @@ from openhcs.core.dataset_sources.interfaces import ( AnalysisResultDirectory, FilenameParserCapability, + MetadataHandler, ) from openhcs.core.plate_image_inventory import PlateFileRecord from openhcs.pyqt_gui.widgets.image_browser import ( @@ -61,7 +62,8 @@ def parse_filename(self, filename: str): ) -class _InventoryMetadataHandler: +class _InventoryMetadataHandler(MetadataHandler): + """Inherits the handler defaults (workspace root, no virtual workspace).""" def __init__(self, handler_result_dir: Path) -> None: self._handler_result_dir = handler_result_dir @@ -69,11 +71,17 @@ def get_image_files(self, _plate_path: Path, *, all_subdirs: bool): assert all_subdirs is True return ("images/A01_w1.tif",) - def source_workspace_metadata_document(self, _plate_path: Path): - return None + def find_metadata_file(self, _plate_path: Path) -> Path: + raise AssertionError("Inventory does not read the metadata file.") - def source_dataset(self, _plate_path: Path): - return None + def get_grid_dimensions(self, _plate_path: Path): + raise AssertionError("Inventory does not read grid dimensions.") + + def get_pixel_size(self, _plate_path: Path): + raise AssertionError("Inventory does not read pixel size.") + + def component_value_set(self, _plate_path: Path): + raise AssertionError("Inventory does not read the component value set.") def source_workspace_root(self, plate_path: Path) -> Path: return plate_path diff --git a/tests/pyqt_gui/test_managed_windows.py b/tests/pyqt_gui/test_managed_windows.py index 0b07607d16..76f1f334c2 100644 --- a/tests/pyqt_gui/test_managed_windows.py +++ b/tests/pyqt_gui/test_managed_windows.py @@ -33,7 +33,7 @@ def __init__(self, orchestrator: _OrchestratorStub) -> None: self._orchestrator = orchestrator self.plate_selected = _Signal() - def get_selected_orchestrator(self) -> _OrchestratorStub: + def _get_current_orchestrator(self) -> _OrchestratorStub: return self._orchestrator def select(self, orchestrator: _OrchestratorStub) -> None: @@ -145,12 +145,14 @@ class PlateManagerStub(QWidget): def __init__( self, service_adapter, + session, color_scheme, *, gui_config, ) -> None: super().__init__() del service_adapter, color_scheme + captured["session"] = session captured["gui_config"] = gui_config def cleanup(self) -> None: @@ -163,11 +165,13 @@ def cleanup(self) -> None: "PlateManagerWidget", PlateManagerStub, ) - monkeypatch.setattr(PlateManagerWindow, "_setup_connections", lambda self: None) + session = object() main_window = QWidget() + main_window.session = session window = PlateManagerWindow(main_window, ServiceAdapterStub()) qtbot.addWidget(main_window) qtbot.addWidget(window) assert captured["gui_config"] is resolved_ui_config + assert captured["session"] is session diff --git a/tests/pyqt_gui/test_session_gui_journey.py b/tests/pyqt_gui/test_session_gui_journey.py new file mode 100644 index 0000000000..aad333283c --- /dev/null +++ b/tests/pyqt_gui/test_session_gui_journey.py @@ -0,0 +1,119 @@ +"""One dataset through the session, driven through the Qt widgets. + +The same SessionOperation classes as the headless MCP journey in +tests/integration/test_session_journey.py run here through the plate manager +on a real execution server; only the client differs. +""" + +from __future__ import annotations + +import time +from pathlib import Path + +import pytest +from objectstate.object_state import ObjectStateRegistry +from PyQt6.QtWidgets import QApplication +from pyqt_reactive.services.ui_thread_dispatch import UiThreadDispatcher + +from openhcs.authoring.session.operations.datasets import ( + AddDatasets, + CompileDatasets, + InitializeDatasets, + RunDatasets, +) +from openhcs.authoring.session.session import DispatcherThread, Session +from openhcs.authoring.session.views import DatasetListView +from openhcs.core.config import GlobalPipelineConfig +from openhcs.pyqt_gui.config import get_default_ui_config +from openhcs.pyqt_gui.widgets.plate_manager import PlateManagerWidget +from openhcs.runtime.zmq_config import OpenHCSZMQConfig +from tests.integration.test_session_journey import ( + STAGE_TIMEOUT_SECONDS, + _port, + _synthetic_dataset, +) +from tests.unit.pyqt_gui.session_harness import GuiServiceStub, release_widgets + + +@pytest.fixture(autouse=True) +def fresh_object_state_registry(): + """The session starts from an empty dataset list.""" + + ObjectStateRegistry.clear() + yield + ObjectStateRegistry.clear() + + +def test_gui_journey_runs_the_same_operations_through_the_widgets( + tmp_path: Path, +) -> None: + application = QApplication.instance() or QApplication([]) + root, source = _synthetic_dataset(tmp_path) + dispatcher = UiThreadDispatcher() + session = Session( + transport_config=OpenHCSZMQConfig(default_port=_port(1), persistent=False), + global_config=GlobalPipelineConfig(), + main_thread=DispatcherThread(dispatcher), + ) + services = GuiServiceStub() + manager = PlateManagerWidget(services, session, gui_config=get_default_ui_config()) + try: + assert session.invoke( + AddDatasets, AddDatasets.request(roots=(str(root),)) + ).accepted + (scope_id,) = session.dataset_scope_ids() + session.set_pipeline_source(scope_id, source) + application.processEvents() + assert [row.scope_id for row in manager.plates] == [scope_id] + manager.item_list.setCurrentRow(0) + application.processEvents() + assert manager.selection_scope_ids() == (scope_id,) + + def wait_until(condition, label) -> None: + deadline = time.monotonic() + STAGE_TIMEOUT_SECONDS + while not condition(): + application.processEvents() + assert time.monotonic() < deadline, label + time.sleep(0.05) + + for operation, finished in ( + (InitializeDatasets, lambda row: row.initialized and not row.init_pending), + (CompileDatasets, lambda row: row.compiled and not row.compile_pending), + ( + RunDatasets, + lambda row: row.terminal_status is not None + and not row.execution_active, + ), + ): + if operation is CompileDatasets: + # Disconnected, the compile button is Connect; connect first. + assert manager.buttons[operation.operation_id].text() == "Connect" + manager.handle_button_action(operation.operation_id) + wait_until( + lambda: CompileDatasets.resolved(session) is CompileDatasets, + "connect", + ) + manager.update_button_states() + # The previous stage's batch ends after its row reports done. + wait_until(lambda: not session.execution_state.busy, "batch finished") + manager.update_button_states() + assert manager.buttons[operation.operation_id].isEnabled(), operation + manager.handle_button_action(operation.operation_id) + deadline = time.monotonic() + STAGE_TIMEOUT_SECONDS + while True: + application.processEvents() + (row,) = DatasetListView.state_of(session).rows + if finished(row): + break + assert time.monotonic() < deadline, (operation, row) + time.sleep(0.05) + assert row.terminal_status == "complete", row + wait_until(lambda: not session.execution_state.busy, "batch finished") + application.processEvents() + assert manager.buttons[RunDatasets.operation_id].text() == "Run" + finally: + manager.cleanup() + manager.close() + release_widgets(application, manager) + session.close() + dispatcher.close() diff --git a/tests/smoke_tests.txt b/tests/smoke_tests.txt new file mode 100644 index 0000000000..f516d9c30b --- /dev/null +++ b/tests/smoke_tests.txt @@ -0,0 +1,11 @@ +# Fast end-to-end smoke: datasets run through the real entry points. +# One pytest path or node id per line. Read by the pull-request smoke job and +# by the local merge gate; keep the whole list under about five minutes on two +# workers. +tests/integration/test_session_journey.py::test_headless_mcp_journey_adds_compiles_and_runs_a_dataset +tests/integration/test_session_journey.py::test_read_only_dataset_initializes_in_a_mirrored_workspace +tests/pyqt_gui/test_session_gui_journey.py +tests/integration/test_agent_observation_export_zmq.py::test_headless_observation_export_uses_ordinary_execution +tests/integration/test_agent_observation_export_zmq.py::test_ordinary_execution_can_export_outcomes_without_value_observation +tests/integration/test_agent_observation_export_zmq.py::test_measured_cli_uses_ordinary_source_session_and_shared_finalizer +tests/integration/test_measured_cli_session.py::test_read_only_source_plate_completes_and_stays_byte_identical diff --git a/tests/unit/agent/test_capabilities.py b/tests/unit/agent/test_capabilities.py index eacfe31e20..119a690218 100644 --- a/tests/unit/agent/test_capabilities.py +++ b/tests/unit/agent/test_capabilities.py @@ -528,3 +528,27 @@ def test_connection_shape_is_declared_by_the_invocation_family(): ): assert issubclass(capability, RuntimeServerCapability) assert isinstance(capability.invocation, AgentFromFieldsServiceInvocation) + + +def test_requestless_ui_bridge_capability_passes_connection_as_connection(): + """A requestless operation's only argument is the connection, never a request.""" + + from types import SimpleNamespace + + from openhcs.agent.services.ui_bridge_service import UiBridgeService + + seen = [] + + class RecordingUiBridgeService(UiBridgeService): + def invoke(self, operation, request=None, connection=None): + seen.append((operation, request, connection)) + return "served" + + connection = object() + invocation = agent_capabilities.ui_bridge_status.invocation + context = SimpleNamespace( + ui_bridge_service=RecordingUiBridgeService(gateway=object()) + ) + + assert invocation.execute(context, connection) == "served" + assert seen == [(invocation.operation, None, connection)] diff --git a/tests/unit/agent/test_mcp_dev_client_sampling_results.py b/tests/unit/agent/test_mcp_dev_client_sampling_results.py index f93b39df87..ee8392d7ef 100644 --- a/tests/unit/agent/test_mcp_dev_client_sampling_results.py +++ b/tests/unit/agent/test_mcp_dev_client_sampling_results.py @@ -83,7 +83,9 @@ def command_render(response, argv): @pytest.mark.parametrize("structured", [False, True]) @pytest.mark.parametrize("selected_case", [False, True]) def test_original_session_framing_to_generated_sampling_command(structured, selected_case, monkeypatch): - from test_mcp_dev_client_pipeline_results import ControlledWireSession + from tests.unit.agent.test_mcp_dev_client_pipeline_results import ( + ControlledWireSession, + ) import openhcs.mcp.dev_client_core as ingress value = selected(sample()) if selected_case else sample() diff --git a/tests/unit/agent/test_mcp_dev_client_ui_renderers.py b/tests/unit/agent/test_mcp_dev_client_ui_renderers.py index 9c6eeb95df..21a9be8dd6 100644 --- a/tests/unit/agent/test_mcp_dev_client_ui_renderers.py +++ b/tests/unit/agent/test_mcp_dev_client_ui_renderers.py @@ -62,7 +62,7 @@ UiActionCatalogRenderOptions, WidgetTreeRenderOptions, ) -from test_mcp_dev_client_renderers import SERVER, framed, sample_dto +from tests.unit.agent.test_mcp_dev_client_renderers import SERVER, framed, sample_dto def render(payload, options=None) -> str: diff --git a/tests/unit/agent/test_mcp_plate_stream_progress.py b/tests/unit/agent/test_mcp_plate_stream_progress.py index 2dd6eb50a8..9fa6b89dac 100644 --- a/tests/unit/agent/test_mcp_plate_stream_progress.py +++ b/tests/unit/agent/test_mcp_plate_stream_progress.py @@ -29,7 +29,10 @@ from openhcs.mcp.server import build_server from openhcs.mcp.socket import McpSocketTransport, wait_for_socket -from test_persisted_output_reopen import NoRuntimeBridge, declared_output +from tests.unit.agent.test_persisted_output_reopen import ( + NoRuntimeBridge, + declared_output, +) from tests.unit.test_streaming_service import FakeViewer diff --git a/tests/unit/agent/test_selected_plane_reopen_stream.py b/tests/unit/agent/test_selected_plane_reopen_stream.py index 73f3af8cba..3eb1cfc674 100644 --- a/tests/unit/agent/test_selected_plane_reopen_stream.py +++ b/tests/unit/agent/test_selected_plane_reopen_stream.py @@ -40,7 +40,10 @@ ) from openhcs.runtime.napari_viewer_server import NapariBatchPayload, NapariImagePayload -from test_persisted_output_reopen import NoRuntimeBridge, declared_output +from tests.unit.agent.test_persisted_output_reopen import ( + NoRuntimeBridge, + declared_output, +) @pytest.fixture diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py new file mode 100644 index 0000000000..53cfa68a4f --- /dev/null +++ b/tests/unit/conftest.py @@ -0,0 +1,16 @@ +"""Unit-suite process setup.""" + +import pytest + + +@pytest.fixture(scope="session", autouse=True) +def session_qapplication(request): + """Create the unit session's one QApplication before any test runs. + + A process holds one Qt application. In-process MCP transports adopt an + existing one and otherwise create a QCoreApplication, after which no later + unit test could create a widget. Sessions without pytest-qt (CPU-only, or a + narrowed ``--confcutdir``) run no widget tests. + """ + if request.config.pluginmanager.hasplugin("pytestqt"): + request.getfixturevalue("qapp") diff --git a/tests/unit/test_aggregate_source_scope.py b/tests/unit/test_aggregate_source_scope.py index 2edded41a4..1a245b88e1 100644 --- a/tests/unit/test_aggregate_source_scope.py +++ b/tests/unit/test_aggregate_source_scope.py @@ -28,7 +28,10 @@ ) from openhcs.core.runtime_tabular_values import FieldSpec from openhcs.core.source_image_provenance import ( - RuntimeSourceImageProvenancePlane, SourceImageIdentity, SourceImageProvenancePlanes, + RuntimeSourceImageProvenancePlane, + SourceImageIdentity, + SourceImageProvenanceContributor, + SourceImageProvenancePlanes, ) from openhcs.core.source_metadata import SourceVoxelSpacing from openhcs.core.steps.abstract import StepExecutionObservation @@ -37,7 +40,7 @@ from polystore.filemanager import FileManager from polystore.memory import MemoryStorageBackend -from test_materialization_core import _viewer_stream_backend_kwargs +from tests.unit.test_materialization_core import _viewer_stream_backend_kwargs from tests.materialization_plans import output_plan as materialization_output_plan from openhcs.core.artifacts.plans import MAIN_FLOW @@ -136,6 +139,32 @@ def test_absence_one_equal_or_incomplete_contributors_cannot_exempt_acquired_pla adapter.filemanager_batches((output,)) +def test_axis_varying_between_runtime_planes_is_kept_not_reduced(): + common = {"well": "A01", "z_index": 1, "timepoint": 1} + planes = tuple( + RuntimeSourceImageProvenancePlane( + SourceImageIdentity(component_metadata={**common, "site": site}), + contributors=tuple( + SourceImageProvenanceContributor( + SourceImageIdentity( + component_metadata={**common, "site": site, "channel": channel} + ) + ) + for channel in (1, 2) + ), + ) + for site in (1, 2) + ) + metadata = ImagePayloadMetadata( + plane_axis=SourceBindingPlaneAxis, + source_component_metadata=common, + source_image_provenance_planes=SourceImageProvenancePlanes(planes), + ) + assert metadata.source_provenance.required_scalar_components( + ("site", "channel", "well") + ) == ("site", "well") + + def test_present_scalar_override_remains_required_despite_varied_contributors(): metadata = (_source((1, 3))).metadata.collapse_leading_plane_axis() metadata = metadata.replace_fields(source_component_metadata={ diff --git a/tests/unit/test_automatic_image_role_materialization_435.py b/tests/unit/test_automatic_image_role_materialization_435.py index 46267133e9..ccbe0a0c80 100644 --- a/tests/unit/test_automatic_image_role_materialization_435.py +++ b/tests/unit/test_automatic_image_role_materialization_435.py @@ -13,7 +13,11 @@ from polystore.filemanager import FileManager from polystore.memory import MemoryStorageBackend -from test_function_outputs import context_stub, function_step_plan, record_output_path +from tests.unit.test_function_outputs import ( + context_stub, + function_step_plan, + record_output_path, +) from openhcs.constants.constants import Backend from openhcs.core.aligned_image_payload import AlignedImageSliceContext, ImagePayloadBundleContext from openhcs.core.artifacts.plans import ArtifactOutputPlan diff --git a/tests/unit/test_build_mcp_knowledge_assets.py b/tests/unit/test_build_mcp_knowledge_assets.py index aa05965926..07adb739b0 100644 --- a/tests/unit/test_build_mcp_knowledge_assets.py +++ b/tests/unit/test_build_mcp_knowledge_assets.py @@ -3,13 +3,15 @@ import glob import json import tomllib -from pathlib import Path +from pathlib import Path, PureWindowsPath import pytest from scripts.build_mcp_knowledge_assets import ( KNOWLEDGE_MANIFEST_RELATIVE_PATH, + PACKAGED_KNOWLEDGE_PATH_BUDGET, PACKAGED_KNOWLEDGE_ROOT_RELATIVE_PATH, + knowledge_asset_projections, project_knowledge_assets, ) from openhcs.agent.knowledge_manifest_schema import ( @@ -294,3 +296,47 @@ def test_wheel_globs_include_the_projected_hidden_plugin_manifest(tmp_path): assert plugin_manifest in included plugin_root = plugin_manifest.parent.parent assert {path for path in projected if path.is_relative_to(plugin_root)} <= included + + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +# The installer CI proof root under the runner profile, with long paths disabled. +WINDOWS_INSTALLED_KNOWLEDGE_ROOT = PureWindowsPath( + r"C:\Users\runneradmin\AppData\Local\OpenHCS Installer Smoke", + "env-00000000", + "Lib", + "site-packages", + *PACKAGED_KNOWLEDGE_ROOT_RELATIVE_PATH.parts, +) +WINDOWS_MAX_PATH = 260 + + +def test_every_packaged_resource_fits_the_windows_installed_path_budget(): + workflow = PROJECT_ROOT / ".github" / "workflows" / "integration-tests.yml" + assert '"OpenHCS Installer Smoke"' in workflow.read_text(encoding="utf-8") + root_length = len(str(WINDOWS_INSTALLED_KNOWLEDGE_ROOT)) + 1 + assert root_length + PACKAGED_KNOWLEDGE_PATH_BUDGET < WINDOWS_MAX_PATH + projections = knowledge_asset_projections(PROJECT_ROOT) + assert any(path.suffix == ".cppipe" for path in projections) + too_long = [ + str(WINDOWS_INSTALLED_KNOWLEDGE_ROOT / relative) + for relative in projections + if len(str(WINDOWS_INSTALLED_KNOWLEDGE_ROOT / relative)) >= WINDOWS_MAX_PATH + ] + assert too_long == [] + + +@pytest.mark.parametrize( + "case_name", ["x" * PACKAGED_KNOWLEDGE_PATH_BUDGET, "nested/name", "name:stream"] +) +def test_native_placement_rejects_unbudgeted_case_names(tmp_path, case_name): + project_root, manifest_path, _ = _project_with_document(tmp_path) + recipe_path = project_root / "recipe.json" + recipe_path.write_text( + json.dumps({"cases": [{"name": case_name, "cppipe_path": "a.cppipe"}]}), + encoding="utf-8", + ) + catalog = json.loads(manifest_path.read_text(encoding="utf-8")) + catalog["documents"].append({"source_path": "recipe.json"}) + manifest_path.write_text(json.dumps(catalog), encoding="utf-8") + with pytest.raises(ValueError): + knowledge_asset_projections(project_root) diff --git a/tests/unit/test_ci_import_guards.py b/tests/unit/test_ci_import_guards.py new file mode 100644 index 0000000000..ccc9036f75 --- /dev/null +++ b/tests/unit/test_ci_import_guards.py @@ -0,0 +1,38 @@ +"""CI runs pytest with ``--import-mode=importlib`` against the installed wheel +(``scripts/run_installed_tests.py``). In that mode a test directory is not on +``sys.path``, so a test module can reach shared helpers only through the +``tests`` package, never by a sibling module's bare name.""" + +from __future__ import annotations + +import ast +from pathlib import Path + +TESTS_ROOT = Path(__file__).resolve().parents[1] + + +def _test_tree_module_names() -> frozenset[str]: + return frozenset(path.stem for path in TESTS_ROOT.rglob("*.py")) - {"__init__"} + + +def test_test_modules_import_helpers_through_the_tests_package() -> None: + local_names = _test_tree_module_names() + offenders: list[str] = [] + for path in sorted(TESTS_ROOT.rglob("*.py")): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.level == 0 and node.module: + names = (node.module,) + elif isinstance(node, ast.Import): + names = tuple(alias.name for alias in node.names) + else: + continue + for name in names: + if name.split(".")[0] in local_names: + offenders.append( + f"{path.relative_to(TESTS_ROOT.parent)}:{node.lineno}: {name}" + ) + assert not offenders, ( + "Import test helpers as tests..; bare sibling imports " + "fail under --import-mode=importlib:\n" + "\n".join(offenders) + ) diff --git a/tests/unit/test_colocalization_metric_setting_authority.py b/tests/unit/test_colocalization_metric_setting_authority.py index 59d99fede6..e5ae55a195 100644 --- a/tests/unit/test_colocalization_metric_setting_authority.py +++ b/tests/unit/test_colocalization_metric_setting_authority.py @@ -153,7 +153,7 @@ def test_recorded_source_pair_features_exclude_disabled_metrics(disabled, famili for name in enabled.columns if not any( name.startswith( - f"Correlation_{(family if family == "RWC" else family.title())}_" + f"Correlation_{(family if family == 'RWC' else family.title())}_" ) for family in families ) diff --git a/tests/unit/test_dataset_access.py b/tests/unit/test_dataset_access.py new file mode 100644 index 0000000000..1ed8f06e1b --- /dev/null +++ b/tests/unit/test_dataset_access.py @@ -0,0 +1,22 @@ +"""Whether a dataset initializes in place is the filesystem's answer. + +The non-writable case (POSIX mode bits, Windows ACLs) is exercised by the +read-only integration journeys, which make a genuinely non-writable tree. +""" + +from pathlib import Path + +from openhcs.authoring.session.session import UnrestrictedDatasetAccess + + +def test_writable_root_initializes_in_place_and_the_probe_leaves_nothing( + tmp_path: Path, +) -> None: + (tmp_path / "image.tif").write_bytes(b"pixels") + + assert UnrestrictedDatasetAccess().initializes_in_place(tmp_path) + assert [path.name for path in tmp_path.iterdir()] == ["image.tif"] + + +def test_missing_root_does_not_initialize_in_place(tmp_path: Path) -> None: + assert not UnrestrictedDatasetAccess().initializes_in_place(tmp_path / "absent") diff --git a/tests/unit/test_dead_ui_code_guards.py b/tests/unit/test_dead_ui_code_guards.py index ac47088a6e..8742c049a0 100644 --- a/tests/unit/test_dead_ui_code_guards.py +++ b/tests/unit/test_dead_ui_code_guards.py @@ -56,6 +56,14 @@ "register_theme_change_callback", } ), + "OpenHCSMainWindow": frozenset( + { + "switch_to_dark_theme", + "switch_to_light_theme", + "load_theme_from_file", + "save_theme_to_file", + } + ), } MAIN_WINDOW_SERVICE_ALIASES = frozenset( { diff --git a/tests/unit/test_derived_role_publication_435.py b/tests/unit/test_derived_role_publication_435.py index ff8d3401fc..6bc347e23b 100644 --- a/tests/unit/test_derived_role_publication_435.py +++ b/tests/unit/test_derived_role_publication_435.py @@ -13,8 +13,10 @@ from polystore.filemanager import FileManager from polystore.memory import MemoryStorageBackend -from test_function_outputs import ( - context_stub, function_step_plan, record_output_path, +from tests.unit.test_function_outputs import ( + context_stub, + function_step_plan, + record_output_path, ) from openhcs.core.dataset_sources.source_schema import SourceSchemaFilenameParser from openhcs.constants.constants import Backend diff --git a/tests/unit/test_execution_control_stop_during_compile.py b/tests/unit/test_execution_control_stop_during_compile.py new file mode 100644 index 0000000000..5af2d2094d --- /dev/null +++ b/tests/unit/test_execution_control_stop_during_compile.py @@ -0,0 +1,76 @@ +"""A batch that errors out because it was stopped ends cancelled, not failed.""" + +from __future__ import annotations + +import asyncio + +import pytest + +from openhcs.authoring.session.execution_control import ExecutionControl +from openhcs.core.execution_state import ( + ExecutionCancelled, + ExecutionFailed, + ManagerExecutionState, +) + + +class _Batch: + active_plates = ("dataset",) + + def __init__(self) -> None: + self.terminal: dict[str, type] = {} + + def cancellable_plates(self) -> tuple[str, ...]: + return self.active_plates + + def execution_id(self, scope_id: str) -> None: + return None + + def mark_terminal(self, scope_id: str, status: type) -> None: + self.terminal[scope_id] = status + + +class _Client: + async def disconnect(self) -> None: + return None + + +class _Session: + def __init__(self, state: ManagerExecutionState) -> None: + self.execution_state = state + self.batch = _Batch() + self.client = _Client() + self.finished: list[tuple[type, str]] = [] + + def set_dataset_state(self, scope_id: str, state: type) -> None: + del scope_id, state + + def finish_dataset_execution(self, completion, scope_id: str) -> None: + self.finished.append((completion.status, scope_id)) + + def refresh(self) -> None: + return None + + +@pytest.mark.parametrize( + "state", + [state for state in ManagerExecutionState if state.stop_pending], +) +def test_a_stopped_batch_that_errors_is_cancelled(state: ManagerExecutionState) -> None: + session = _Session(state) + + asyncio.run(ExecutionControl(session).handle_execution_failure()) + + assert session.finished == [(ExecutionCancelled, "dataset")] + assert session.batch.terminal == {} + assert session.execution_state is ManagerExecutionState.IDLE + + +def test_a_running_batch_that_errors_is_failed() -> None: + session = _Session(ManagerExecutionState.RUNNING) + + asyncio.run(ExecutionControl(session).handle_execution_failure()) + + assert session.finished == [] + assert session.batch.terminal == {"dataset": ExecutionFailed} + assert session.execution_state is ManagerExecutionState.IDLE diff --git a/tests/unit/test_function_outputs.py b/tests/unit/test_function_outputs.py index ac95f6b678..8a26058563 100644 --- a/tests/unit/test_function_outputs.py +++ b/tests/unit/test_function_outputs.py @@ -79,6 +79,7 @@ MemoryOutputWriter, OpenHCSMetadataTarget, MaterializedImageMetadataTarget, + PrimaryImageMetadataTarget, RuntimeArtifactMetadataTarget, StreamOutputsAuthority, StreamOutputBatch, @@ -188,6 +189,62 @@ def test_function_step_metadata_follows_runtime_artifact_persistence( ] +@pytest.mark.parametrize( + ("backend", "expected_targets"), + [ + ( + Backend.DISK.value, + { + PrimaryImageMetadataTarget, + MaterializedImageMetadataTarget, + RuntimeArtifactMetadataTarget, + }, + ), + (Backend.OMERO_LOCAL.value, set()), + ], +) +def test_metadata_document_targets_follow_backend_declaration( + monkeypatch, backend, expected_targets +) -> None: + """A virtual backend's output root holds no local metadata document.""" + root = "/omero/plate_1_outputs" + plan = CompiledStepPlan( + step_index=0, + step_name="count", + axis_id="A01", + output_dir=Path(root) / "images", + output_dataset_root=root, + sub_dir="images", + analysis_results_dir=f"{root}/results", + write_backend=backend, + materialized_output=MaterializedOutputPlan( + output_dir=Path(root) / "checkpoint", + backend=backend, + dataset_root=root, + sub_dir="checkpoint", + analysis_results_dir=f"{root}/results", + ), + runtime_artifact_materialization=RuntimeArtifactMaterializationPlan( + persistent_enabled=True, persistent_backend=backend + ), + create_openhcs_metadata=True, + ) + assert {type(t) for t in OpenHCSMetadataTarget.for_plan(plan)} == expected_targets + + documents = [] + monkeypatch.setattr( + AtomicMetadataWriter, + "reconcile_completed_dataset", + lambda _self, metadata_path, *_args, **_kwargs: documents.append(metadata_path), + ) + OpenHCSMetadataTarget.finalize_completed_dataset( + {"A01": SimpleNamespace(step_plans={0: plan})} + ) + assert documents == ( + [METADATA_CONFIG.metadata_path(root)] if expected_targets else [] + ) + + def test_memory_output_writer_rejects_payload_path_cardinality_mismatch(): with pytest.raises(ValueError, match="1 payloads for 0 paths"): MemoryOutputWriter.payloads( diff --git a/tests/unit/test_gray_to_color_binding_axis_432.py b/tests/unit/test_gray_to_color_binding_axis_432.py index 81948dbfdf..c2abb90748 100644 --- a/tests/unit/test_gray_to_color_binding_axis_432.py +++ b/tests/unit/test_gray_to_color_binding_axis_432.py @@ -31,7 +31,9 @@ CellProfilerOutputRecorder, ) from openhcs.processing.backends.cellprofiler.color import GrayToColorModule, gray_to_color -from test_cellprofiler_generic_special_input_binding import _compile_public_step +from tests.unit.test_cellprofiler_generic_special_input_binding import ( + _compile_public_step, +) from tests.unit.cellprofiler_runtime_test_support import cellprofiler_runtime_adapter_for_test from openhcs.domains.microscopy.axes import Microscopy from openhcs.core.axes import ColourAxis diff --git a/tests/unit/test_gray_to_color_payload_axis_compile_432.py b/tests/unit/test_gray_to_color_payload_axis_compile_432.py index d66fd1323b..61c79b6023 100644 --- a/tests/unit/test_gray_to_color_payload_axis_compile_432.py +++ b/tests/unit/test_gray_to_color_payload_axis_compile_432.py @@ -25,8 +25,11 @@ ColorToGrayMode, ImageChannelType, color_to_gray, gray_to_color, split_color_to_gray, ) -from test_gray_to_color_binding_axis_432 import _execute_bound_stack, _source_plane -from test_payload_axis_compile_gate import _compiled_pattern, _session +from tests.unit.test_gray_to_color_binding_axis_432 import ( + _execute_bound_stack, + _source_plane, +) +from tests.unit.test_payload_axis_compile_gate import _compiled_pattern, _session from openhcs.core.payload_axes import ColourSampleAxisSpec from openhcs.core.axes import ColourAxis diff --git a/tests/unit/test_library_registry_discovery.py b/tests/unit/test_library_registry_discovery.py index f364215d7f..73fc7c4082 100644 --- a/tests/unit/test_library_registry_discovery.py +++ b/tests/unit/test_library_registry_discovery.py @@ -50,6 +50,110 @@ def test_native_catalog_import_failures_follow_declared_memory_admission( assert registry.get_modules_to_scan() == [] +@pytest.mark.parametrize( + "requirement, outcome", + [ + # The package's own marker leaves the distribution out of this environment. + ("absent-catalog-probe>=1; platform_system == 'NoSuchSystem'", "excluded"), + # Declared only for an extra this installation did not select. + ("absent-catalog-probe>=1; extra == 'probe'", "excluded"), + # Required here, yet absent: a broken installation fails loudly. + ("absent-catalog-probe>=1", "broken"), + # A module cannot claim a distribution the package never declares. + ("unrelated-catalog-probe>=1", "undeclared"), + ], +) +def test_native_catalog_excludes_modules_whose_distribution_markers_exclude_here( + tmp_path, monkeypatch, requirement, outcome +) -> None: + from packaging.requirements import Requirement + + from openhcs.processing.backends.lib_registry import openhcs_registry + from openhcs.processing.backends.lib_registry.openhcs_registry import ( + OpenHCSRegistry, + ) + + package_name = "native_catalog_distribution_probe" + package = tmp_path / package_name + package.mkdir() + (package / "__init__.py").write_text("") + (package / "processing.py").write_text( + "from openhcs.core.memory.decorators import numpy\n" + "import absent_catalog_probe\n" + "REQUIRED_DISTRIBUTIONS = ('absent-catalog-probe',)\n" + "@numpy\n" + "def process(image):\n return image\n" + ) + monkeypatch.syspath_prepend(str(tmp_path)) + monkeypatch.setenv("OPENHCS_CPU_ONLY", "1") + monkeypatch.setattr( + openhcs_registry, + "_openhcs_requirements", + lambda: (Requirement(requirement),), + ) + registry = OpenHCSRegistry() + registry.MODULES_TO_SCAN = [f"{package_name}.processing"] + + if outcome == "excluded": + assert registry.get_modules_to_scan() == [] + assert registry.cache_discovery_context()[ + "distributions_excluded_by_package_metadata" + ] == ["absent-catalog-probe"] + elif outcome == "broken": + with pytest.raises(ModuleNotFoundError, match="absent_catalog_probe"): + registry.get_modules_to_scan() + else: + with pytest.raises(ValueError, match="absent-catalog-probe"): + registry.get_modules_to_scan() + + +@pytest.mark.parametrize( + "platform_system, platform_machine, excluded", + [ + ("Darwin", "x86_64", True), + ("Darwin", "arm64", False), + ("Linux", "x86_64", False), + ], +) +def test_basicpy_catalog_module_follows_the_package_basicpy_marker( + monkeypatch, platform_system, platform_machine, excluded +) -> None: + """Intel macOS installs omit openhcs-basicpy; its catalog module follows.""" + + import importlib.metadata + + import packaging.markers + + from openhcs.processing.backends.lib_registry.openhcs_registry import ( + _module_distributions_excluded_by_package_metadata, + ) + + environment = packaging.markers.default_environment() + monkeypatch.setattr( + packaging.markers, + "default_environment", + lambda: { + **environment, + "platform_system": platform_system, + "platform_machine": platform_machine, + }, + ) + installed_distribution = importlib.metadata.distribution + + def distribution_without_basicpy(name: str): + if name == "openhcs-basicpy": + raise importlib.metadata.PackageNotFoundError(name) + return installed_distribution(name) + + monkeypatch.setattr( + importlib.metadata, "distribution", distribution_without_basicpy + ) + + assert _module_distributions_excluded_by_package_metadata( + "openhcs.processing.backends.enhance.basic_processor_jax" + ) == (("openhcs-basicpy",) if excluded else ()) + + def test_concurrent_catalog_requests_share_one_metadata_load(monkeypatch) -> None: from openhcs.processing.backends.lib_registry.registry_service import ( RegistryService, diff --git a/tests/unit/test_named_checkpoint_publication.py b/tests/unit/test_named_checkpoint_publication.py index c71767093d..92f04d2937 100644 --- a/tests/unit/test_named_checkpoint_publication.py +++ b/tests/unit/test_named_checkpoint_publication.py @@ -12,7 +12,10 @@ import pytest import tifffile -from test_artifact_publication_journey import _plate, progress_events # noqa: F401 (fixture) +from tests.unit.test_artifact_publication_journey import ( # noqa: F401 (fixture) + _plate, + progress_events, +) from openhcs.core.artifacts.sidecars import QaCheckpoint from openhcs.core.artifacts.kinds import ImageArtifactType from openhcs.core.artifacts.specs import ( diff --git a/tests/unit/test_openhcs_adapter.py b/tests/unit/test_openhcs_adapter.py index 02b2bbf225..060e3b9deb 100644 --- a/tests/unit/test_openhcs_adapter.py +++ b/tests/unit/test_openhcs_adapter.py @@ -153,7 +153,6 @@ def test_openhcs_adapter_preserves_declared_main_flow_output_policy( configured = adapter._execution_global_config( request, output_suffix="_candidate", - output_dataset_root=tmp_path / "candidate-plate", compilation_debug_config=CompilationDebugConfig(), ) @@ -168,6 +167,8 @@ def test_openhcs_adapter_preserves_declared_main_flow_output_policy( assert ( configured.analysis_consolidation_config.output_filename == "custom-summary.csv" ) + # Analysis results resolve inside the compiled output root by default. + assert configured.materialization_results_path == Path("results") assert base_config.path_planning_config.global_output_folder is None assert base_config.analysis_consolidation_config.enabled is True @@ -486,6 +487,19 @@ def test_openhcs_progress_observer_tracks_every_server_event( assert observer.progress_description() == "artifact_transfer/running" +def test_openhcs_progress_silence_is_measured_from_run_submission( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Server startup belongs to the client's readiness timeout, not the watchdog.""" + observer = _ZMQProgressTimingObserver() + assert observer.inactivity_seconds(observed_at=500.0) == 0.0 + + monkeypatch.setattr("benchmark.openhcs_measured_run.time.monotonic", lambda: 200.0) + observer.begin_run() + + assert observer.inactivity_seconds(observed_at=321.0) == 121.0 + + def test_openhcs_watchdog_allows_execution_without_interval_timer( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/unit/test_preparation_admission.py b/tests/unit/test_preparation_admission.py index 42c4d60a8f..6f1049f93b 100644 --- a/tests/unit/test_preparation_admission.py +++ b/tests/unit/test_preparation_admission.py @@ -242,14 +242,16 @@ def prepare_selected(): assert events == ["selected", "unselected", "dynamic"] -def test_unavailable_affinity_does_not_admit_speculative_parallelism(monkeypatch): +def test_platform_without_affinity_admits_its_online_cpus(monkeypatch): events = [] - started, _, _ = simulated_workers(monkeypatch, 8) + started, peak, _ = simulated_workers(monkeypatch, 8) monkeypatch.delattr(os, "sched_getaffinity") + monkeypatch.setattr(os, "cpu_count", lambda: 2) PreparationCacheBatch( tuple(DeclaredCache(str(i), events) for i in range(3)) ).populate_child_caches(max_workers=8) - assert not started and not events + assert started == [("0",), ("1",), ("2",)] + assert max(peak) == 2 def test_affinity_query_failure_is_not_misreported_as_success(monkeypatch): diff --git a/tests/unit/test_rescale_stream_plane_declaration.py b/tests/unit/test_rescale_stream_plane_declaration.py index 80a9ee0559..31fa98754e 100644 --- a/tests/unit/test_rescale_stream_plane_declaration.py +++ b/tests/unit/test_rescale_stream_plane_declaration.py @@ -41,7 +41,9 @@ ) # Reuse the original provider-free compiler and runtime fixtures, not a second ABI. -from test_cellprofiler_generic_special_input_binding import _compile_public_step +from tests.unit.test_cellprofiler_generic_special_input_binding import ( + _compile_public_step, +) from tests.unit.cellprofiler_runtime_test_support import ( cellprofiler_runtime_adapter_for_test, ) diff --git a/tests/unit/test_selected_plane_stream_materialization.py b/tests/unit/test_selected_plane_stream_materialization.py index eda16a29e6..c53fd27638 100644 --- a/tests/unit/test_selected_plane_stream_materialization.py +++ b/tests/unit/test_selected_plane_stream_materialization.py @@ -57,12 +57,12 @@ ) from openhcs.core.materialization import materialize -from test_function_artifact_materialization import ( +from tests.unit.test_function_artifact_materialization import ( _context, _plan, streaming_config_stub, ) -from test_materialization_core import _viewer_stream_backend_kwargs +from tests.unit.test_materialization_core import _viewer_stream_backend_kwargs from openhcs.domains.microscopy.axes import Microscopy from tests.materialization_plans import output_plan as materialization_output_plan from openhcs.core.artifacts.plans import MAIN_FLOW diff --git a/tests/unit/test_singleton_rgb_label_invocation_477.py b/tests/unit/test_singleton_rgb_label_invocation_477.py index afe5734951..4de3000645 100644 --- a/tests/unit/test_singleton_rgb_label_invocation_477.py +++ b/tests/unit/test_singleton_rgb_label_invocation_477.py @@ -28,6 +28,8 @@ ) from openhcs.core.pipeline.function_contracts import ( FullStackLabels, + MatchImageStackLabels, + SliceAlignedLabels, ) @@ -71,7 +73,14 @@ def _execute(case, invocation): ) -def test_match_image_labels_consume_declared_singleton_root_for_scalar_rgb(): +@pytest.mark.parametrize("label_mode", (MatchImageStackLabels, SliceAlignedLabels)) +def test_plane_label_modes_consume_declared_singleton_root_for_scalar_rgb( + monkeypatch, label_mode +): + raw = OverlayOutlinesModule.require_callable() + monkeypatch.setattr( + raw, FunctionContractAttribute.object_label_input_execution_mode, label_mode + ) case = _invocation_case() raw, contract, image, labels, *_ = case image_before = image.data.copy() diff --git a/tests/unit/test_smoke_installed_gui.py b/tests/unit/test_smoke_installed_gui.py index 5576225472..f31694dfa1 100644 --- a/tests/unit/test_smoke_installed_gui.py +++ b/tests/unit/test_smoke_installed_gui.py @@ -35,7 +35,7 @@ InstalledGuiSnapshotEvidenceAuthority, _StartupObservation, assert_not_source_checkout_import, - require_ui_action_accepted, + accepted_ui_action_status, verify_installed_function_catalog, with_isolated_runtime_topology, ) @@ -170,12 +170,41 @@ def test_gui_smoke_allows_wheel_venv_inside_checkout(tmp_path: Path) -> None: ) +def _prepared_execution_services() -> Future[None]: + future: Future[None] = Future() + future.set_result(None) + return future + + +def test_gui_smoke_waits_for_gui_endpoint_before_reading_catalog() -> None: + class UnreachableFunctionCatalogService: + @staticmethod + def prepare() -> Future[FunctionCatalogPage]: + raise AssertionError( + "The catalog attaches only after the GUI endpoint starts." + ) + + with pytest.raises( + OperationTimeoutError, + match="Timed out waiting for test installed function catalog", + ): + verify_installed_function_catalog( + UnreachableFunctionCatalogService(), + execution_services_prepared=Future(), + deadline=OperationDeadline.after_milliseconds( + 1, + operation="test installed function catalog", + ), + ) + + def test_gui_smoke_requires_prepared_catalog_and_resolvable_reference() -> None: entry = _catalog_entry() service = _FunctionCatalogService(_catalog_page(entry)) result = verify_installed_function_catalog( service, + execution_services_prepared=_prepared_execution_services(), deadline=OperationDeadline.after_milliseconds( 2_000, operation="test installed function catalog", @@ -201,6 +230,7 @@ def prepare() -> Future[FunctionCatalogPage]: ): verify_installed_function_catalog( PendingFunctionCatalogService(), + execution_services_prepared=_prepared_execution_services(), deadline=OperationDeadline.after_milliseconds( 1, operation="test installed function catalog", @@ -221,6 +251,7 @@ def prepare() -> Future[FunctionCatalogPage]: with pytest.raises(TimeoutError, match="catalog owner timed out") as caught: verify_installed_function_catalog( FailedFunctionCatalogService(), + execution_services_prepared=_prepared_execution_services(), deadline=OperationDeadline.after_milliseconds( 2_000, operation="test installed function catalog", @@ -413,7 +444,7 @@ def test_gui_smoke_accepts_only_an_accepted_ui_action( action_result = SimpleNamespace(status=status) if status is ActionAccepted: - require_ui_action_accepted(action_result) + assert json.dumps(accepted_ui_action_status(action_result)) == '"accepted"' else: with pytest.raises(AssertionError, match="not accepted"): - require_ui_action_accepted(action_result) + accepted_ui_action_status(action_result) diff --git a/tests/unit/test_snapshot_deadline.py b/tests/unit/test_snapshot_deadline.py index b0dbb1a6da..09460a9580 100644 --- a/tests/unit/test_snapshot_deadline.py +++ b/tests/unit/test_snapshot_deadline.py @@ -8,7 +8,7 @@ import pytest -from test_napari_render_snapshot import queued_viewer +from tests.unit.test_napari_render_snapshot import queued_viewer from openhcs.agent.dto.execution import ExecutionConnectionSpec from openhcs.agent.dto.viewer import ViewerWindowSnapshotRequest from openhcs.agent.services import viewer_window_service as service_module diff --git a/tests/unit/test_viewer_spatial_sample.py b/tests/unit/test_viewer_spatial_sample.py index 12bce6de08..903e7d7b6c 100644 --- a/tests/unit/test_viewer_spatial_sample.py +++ b/tests/unit/test_viewer_spatial_sample.py @@ -5,7 +5,7 @@ import numpy as np import pytest -from test_viewer_feature_measurement import native_route +from tests.unit.test_viewer_feature_measurement import native_route from openhcs.agent.dto.execution import ExecutionConnectionSpec from openhcs.agent.dto.viewer import ViewerWindowImageSampleRequest