From 82b1422b53958e86dce61c0055372dc90d3e0760 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 4 Oct 2026 13:06:35 -0400 Subject: [PATCH 01/10] Integrate supplied runtime performance plan with current owners --- plans/runtime_plumbing/IMPLEMENTATION.md | 18 +++++++ .../P0-trustworthy-profile.md | 17 +++++++ .../P1-compiled-invocation.md | 28 +++++++++++ plans/runtime_plumbing/P2-parse-once.md | 15 ++++++ plans/runtime_plumbing/P3-manifest-store.md | 17 +++++++ plans/runtime_plumbing/P4-held-and-lazy.md | 19 ++++++++ plans/runtime_plumbing/P5-memory-plan.md | 15 ++++++ plans/runtime_plumbing/README.md | 47 +++++++++++++++++++ 8 files changed, 176 insertions(+) create mode 100644 plans/runtime_plumbing/IMPLEMENTATION.md create mode 100644 plans/runtime_plumbing/P0-trustworthy-profile.md create mode 100644 plans/runtime_plumbing/P1-compiled-invocation.md create mode 100644 plans/runtime_plumbing/P2-parse-once.md create mode 100644 plans/runtime_plumbing/P3-manifest-store.md create mode 100644 plans/runtime_plumbing/P4-held-and-lazy.md create mode 100644 plans/runtime_plumbing/P5-memory-plan.md create mode 100644 plans/runtime_plumbing/README.md diff --git a/plans/runtime_plumbing/IMPLEMENTATION.md b/plans/runtime_plumbing/IMPLEMENTATION.md new file mode 100644 index 0000000000..97e09c1cdb --- /dev/null +++ b/plans/runtime_plumbing/IMPLEMENTATION.md @@ -0,0 +1,18 @@ +# Runtime performance plan integration + +User input: `plans/perofrmance_plan.zip`, SHA256 `52949085276b0974243b54c3a2766e2ffae77c691e40a3796aa072eb34bd061b`. The seven supplied documents in this directory are unchanged. Their reference source is main `3f76f5e1`; implementation starts from main `175262f0` after #641. + +The active performance objective remains execution and total pipeline latency, with CellProfiler parity. Server startup is excluded. Refactoring completion and measured runtime improvement are separate claims. + +| Surface | Existing determining owners | Integration status and obligations | +|---|---|---| +| P0 | RuntimeProfileLogger; worker/run lifecycle | In progress first. Consolidate all profile emitters into a run-owned raw event buffer; flush outside callable/chain timed regions, retain failure cleanup and independent worker/run lifetimes. Capture baseline before execution refactors. | +| P1 | CompiledFunctionInvocation; CallableContract; CompiledStepPlan | Compile fixed memory, ABI, bindings and return-layout decisions. Adapter requests/current pixels/active edges remain invocation-owned. Build debug parameters only for debug or the existing declaration-error diagnostics. Preserve eager adapter binding and error epochs. | +| P2 | Plate inventory and declared source bindings | Parse physical source filenames once and index components. Produced filename aliases remain live; do not reintroduce the stale alias cache removed by #638. | +| P3 | StepOutputManifestStore; compiled pattern/step declaration | Replace whole-cohort rebuilds and global invalidation with incremental per-producer records. Preserve inherited-domain replacement, insertion order, last-write replacement, begin-step epochs and live aliases. | +| P4 | ProcessingContext; source projection authority; manifest | Hold context-owned projections, lazy log formatting, producer order derived from P3. Preserve ambiguity/cohort admission before cached payload return. | +| P5 | CompiledStepPlan; compiled invocation; device scope | Plan memory/device boundaries across steps. Keep preprocessing InputConversionPlan distinct from runtime placement. Validate mixed CPU/GPU semantics separately; a CPU-only profile cannot prove a GPU gain. | + +#641 is merged and removes post-compilation kernel warming; normal registry READY is the only child-cache consumer. Full catalog READY qualification passed. #630 (runtime ownership refactor), #635 (intensity maxima) and #597 (saved compiler resolution) remain drafts with negative whole-pipeline observations. They are not performance acceptance for this plan. + +Required final evidence: corrected baseline and integrated execution profile on the same real pipeline; production benchmark before/after execution and total time; affected parity, MCP and GUI behavior; mixed-device receiving for P5. Each implementation PR must close its corresponding issue and declare any persisted format changes. diff --git a/plans/runtime_plumbing/P0-trustworthy-profile.md b/plans/runtime_plumbing/P0-trustworthy-profile.md new file mode 100644 index 0000000000..191a7c6f54 --- /dev/null +++ b/plans/runtime_plumbing/P0-trustworthy-profile.md @@ -0,0 +1,17 @@ +# P0: The profiler stops measuring itself + +**Index:** [README.md](README.md). **First.** + +## What is wrong + +`RuntimeProfileSink.record` (`openhcs/core/steps/function_runtime.py:188`), when profiling is enabled, formats its fields, emits a `logger.info` record, then **opens the profile file, appends one line and closes it, on every record**. The runtime has 16 record sites, and some are nested inside other timed windows: `function_call` and `runtime_adapter_factory` are recorded inside `pattern_execute_chain`'s window. Plumbing computed as chain time minus callable time therefore includes the nested records' logging and file I/O. + +## Target + +- Records go into an in-memory buffer owned by the profile sink: label, seconds, and the typed fields, with no formatting at record time. +- The buffer is written once, when the run ends (or per well, if a run must survive a crash), and logging the summary happens then too. +- No I/O, logging or string formatting inside any timed region. + +## Done when + +With profiling enabled, a timed region contains no file operation or log emission, and the profile file is written once per run. Record the baseline split on a real pipeline with it. diff --git a/plans/runtime_plumbing/P1-compiled-invocation.md b/plans/runtime_plumbing/P1-compiled-invocation.md new file mode 100644 index 0000000000..73a11c6d85 --- /dev/null +++ b/plans/runtime_plumbing/P1-compiled-invocation.md @@ -0,0 +1,28 @@ +# P1: A compiled invocation + +**Index:** [README.md](README.md). **Coordinate with S4's owner.** + +## What is wrong + +Every function call re-decides how to call the function: + +- `execute` (`openhcs/core/steps/function_runtime.py:1058`): `contract.require_memory_types()` and `execution_plan.device_id_for(...)`; the main-flow projection through `contract.main_flow_call_argument`; `final_kwargs = dict(self.base_kwargs)` plus the compiled bindings copied in. +- `bind_runtime_adapter` (`:1139`): **a new adapter from `runtime_adapter.factory(...)` for every call.** +- `invoke` (`:1159`): `logger.info(f"Executing function: {self.function_name}")` formatted at any log level; `bound_parameters = dict(final_kwargs)`, used only by the debug sink and built without one; a device scope entered and left per call. +- `save_artifact_outputs` (`:1243`): a new `RuntimeReturnedOutputMatcher` per call to map returned values to declared outputs, a mapping fixed by the contract and the output plans. + +## Target + +An immutable compiled invocation, built once per function invocation and group shape, owned by what the compiler produces (S4), holding: + +- the input and execution memory types, and the target device; +- the compiled parameter bindings, frozen; +- the resolved runtime adapter, or the adapter itself when its inputs are fixed for the group; +- the chosen main-flow projection; +- the mapping from return slots to output plans. + +Each call then does data work only: convert when needed, project with the chosen strategy, call, save through the resolved mapping. Debug copies exist only when a debug sink is attached; logging is lazy. + +## Done when + +None of the decisions above is computed inside `execute`, `invoke` or `save_artifact_outputs`, and P0's profile shows the per-call plumbing before and after on the same pipeline. diff --git a/plans/runtime_plumbing/P2-parse-once.md b/plans/runtime_plumbing/P2-parse-once.md new file mode 100644 index 0000000000..23ffbe709e --- /dev/null +++ b/plans/runtime_plumbing/P2-parse-once.md @@ -0,0 +1,15 @@ +# P2: Filenames parsed once + +**Index:** [README.md](README.md). + +## What is wrong + +`_filter_matching_files_for_group` (`openhcs/core/steps/function_runtime.py:1761`) runs `parser.parse_filename(Path(filename).name)` over the matching files and `parser.component_for_name(group_component)` for every pattern group, and every later step does it again. A file's name and its metadata never change during a run. + +## Target + +The plate's file inventory owns parsed filename metadata: each file parsed once, and files indexed by component value. A pattern group selects its files with an index lookup. The source-binding rules already place metadata extraction there; this makes the runtime use it rather than re-derive it. + +## Done when + +`parse_filename` runs once per file per run, and no pattern group filters files by parsing names. diff --git a/plans/runtime_plumbing/P3-manifest-store.md b/plans/runtime_plumbing/P3-manifest-store.md new file mode 100644 index 0000000000..2d2a1811c5 --- /dev/null +++ b/plans/runtime_plumbing/P3-manifest-store.md @@ -0,0 +1,17 @@ +# P3: The output manifest grows linearly + +**Index:** [README.md](README.md). + +## What is wrong + +`StepOutputManifestStore.record_outputs` (`openhcs/core/steps/function_output_manifest.py:283`) runs once per pattern group. Each time it rebuilds a dict and a tuple of **every** record the step has produced for that well, `{record.main_flow_address: record for record in (*existing, *current_outputs)}`, stores the whole tuple again, re-walks `plan.compiled_function_pattern.iter_invocations()`, and calls `_invalidate_record_selection_caches` (`:323`), which clears every plan's selection and filtered-path caches. The work per step and well grows with the square of the group count, and consumers recompute selections after every group. + +## Target + +- Records per key in an append-only structure with an incrementally maintained address index; a group's outputs are added, never the whole set rebuilt. +- The per-plan facts `record_outputs` derives from `iter_invocations()` (composed consumption, grouping) are computed once with the plan. +- Invalidation is per key, by version: only selections that read the changed key are recomputed. + +## Done when + +Recording a group's outputs costs time proportional to that group's outputs, and recording for one key leaves other keys' cached selections intact. diff --git a/plans/runtime_plumbing/P4-held-and-lazy.md b/plans/runtime_plumbing/P4-held-and-lazy.md new file mode 100644 index 0000000000..377bcc1506 --- /dev/null +++ b/plans/runtime_plumbing/P4-held-and-lazy.md @@ -0,0 +1,19 @@ +# P4: Held objects and lazy logging + +**Index:** [README.md](README.md). + +## What is wrong + +- `source_workspace_projection_authority` (`openhcs/core/steps/function_runtime.py:1420`) constructs a new `VirtualWorkspaceSourceProjectionAuthority` for every pattern group; its cache is per context, but the authority around it is rebuilt each time. +- `_load_input_stack` (`:1586`) re-sorts and compares producer records per group, and its debug message at line 1651 builds `[Path(f).name for f in matching_files]` **for every group, with debug logging off.** +- `run` (`:1454`) and the loader format `logger.debug(f"…")` messages eagerly. + +## Target + +- The projection authority is held per context, like its cache. +- Producer records arrive sorted from P3's store, so no group re-sorts them. +- Logging in the runtime uses lazy `%`-style arguments, and anything that builds a collection for a message is guarded by `logger.isEnabledFor`. + +## Done when + +No object that depends only on the context is constructed per group, and no log message does work when its level is disabled. diff --git a/plans/runtime_plumbing/P5-memory-plan.md b/plans/runtime_plumbing/P5-memory-plan.md new file mode 100644 index 0000000000..bee59d9994 --- /dev/null +++ b/plans/runtime_plumbing/P5-memory-plan.md @@ -0,0 +1,15 @@ +# P5: Conversions planned across steps + +**Index:** [README.md](README.md). **Last; coordinate with S4's owner.** + +## What is wrong + +`convert_memory` (from `openhcs.core.memory`) runs per function call in `execute` (`openhcs/core/steps/function_runtime.py:1058`), converting from the payload's memory type to the callable's. Consecutive steps on the same device can convert data away and back, and the decision is made per call, though the whole step sequence and every step's memory types are known when the pipeline compiles. + +## Target + +The compiler plans memory placement across the step sequence: a conversion is scheduled only where consecutive steps' memory types differ, and data stays on the GPU across consecutive GPU steps. The compiled invocation (P1) carries whether its input needs converting, and the runtime never decides it per call. + +## Done when + +Converting between memory types happens only at planned boundaries, and P0's profile shows conversion time before and after on a pipeline that mixes CPU and GPU steps. diff --git a/plans/runtime_plumbing/README.md b/plans/runtime_plumbing/README.md new file mode 100644 index 0000000000..854a3bb9fb --- /dev/null +++ b/plans/runtime_plumbing/README.md @@ -0,0 +1,47 @@ +# OpenHCS runtime plumbing + +**Head:** OpenHCS `main` at `3f76f5e1`. **Rules:** `docs/refactor/00-RULES.md`. Line numbers refer to that head. + +## The question + +Generic runtime plumbing takes too long relative to the callables it runs. Reading the execution path from `PatternGroupRuntime.run` (`openhcs/core/steps/function_runtime.py:1454`) through `execute` (`:1058`) and `invoke` (`:1159`) gives one answer: **facts that are fixed for a step, a plate or a run are re-derived per pattern group or per function call, because no object owns them.** The `PipelineCompiler` exists so the runtime can be thin; these are compile-time facts the runtime recomputes. It's the same defect as duplicated authority, appearing as cost. + +## What runs, and how often + +| Work | Recurs per | Depends only on | +|---|---|---| +| Memory types, target device | function call | contract, execution plan | +| A newly constructed runtime adapter | function call | plan, plus group-invariant payload metadata | +| Main-flow projection through the contract | function call | contract's declared processing and raw ABI | +| A new `RuntimeReturnedOutputMatcher` | function call | contract, output plans | +| Two copies of the arguments, one only for debugging | function call | nothing | +| `logger.info(f"Executing function: …")`, formatted at any level | function call | nothing | +| Filename parsing and component lookup | pattern group, every step | the file name | +| A new `VirtualWorkspaceSourceProjectionAuthority` | pattern group | the context | +| Rebuilding all of a step's output records, and clearing every selection cache | pattern group | grows with groups done: quadratic per step and well | +| Memory conversion | function call | the step sequence's memory types | + +## Surfaces + +| ID | Fix | Recurs per | +|---|---|---| +| [P0](P0-trustworthy-profile.md) | The profiler stops measuring itself | every record | +| [P1](P1-compiled-invocation.md) | A compiled invocation | function call | +| [P2](P2-parse-once.md) | Filenames parsed once, files indexed by component | pattern group and step | +| [P3](P3-manifest-store.md) | The output manifest grows linearly | pattern group | +| [P4](P4-held-and-lazy.md) | Held objects and lazy logging | pattern group and call | +| [P5](P5-memory-plan.md) | Conversions planned across steps | function call | + +## Order + +1. **P0 first,** so every measurement after it is trustworthy. Record the baseline split between plumbing and callable time on a real pipeline with it. +2. **P1 and P2,** the most frequent costs, in parallel. +3. **P3 and P4.** +4. **P5,** which needs compiler work, last. +5. Re-run the corrected profiler on the same pipeline and post the before-and-after split. + +## Crossings + +- **#394** (`perf/shared-runtime-plumbing`) added `CallableContractRuntimeCache` and an identity-bound process cache. P1 builds on that intent; a cache still pays a key and a lookup per call, while a compiled invocation held by the runtime pays nothing. +- **S4** (pipeline compilation) in `docs/refactor/` owns `PipelineCompiler`; P1 and P5 extend what it produces, so whoever takes P1 coordinates with S4's owner. +- **P2** gives the plate's file inventory ownership of filename metadata, which the source-binding rules already require: no filename parsing outside the declared source bindings. From 88522fb0889521f9be75706d085f0d299b40aa2f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 4 Oct 2026 13:34:31 -0400 Subject: [PATCH 02/10] Publish stable output slots and update producer cohorts incrementally --- openhcs/core/function_patterns.py | 14 ++ .../core/steps/function_output_manifest.py | 144 +++++++++--------- ...st_function_output_manifest_input_scope.py | 35 +++++ 3 files changed, 125 insertions(+), 68 deletions(-) diff --git a/openhcs/core/function_patterns.py b/openhcs/core/function_patterns.py index 78ed0fdbd9..37ca4910d5 100644 --- a/openhcs/core/function_patterns.py +++ b/openhcs/core/function_patterns.py @@ -35,6 +35,7 @@ from openhcs.core.callable_contract import ( CallableContract, FunctionStepExecutionScope, + ImagePayloadConsumption, PrimaryImageCarrierRequirement, ) from openhcs.core.component_group_scope import ComponentGroupScope @@ -1079,6 +1080,19 @@ class CompiledFunctionPattern: groups: tuple[CompiledFunctionGroup, ...] is_grouped: bool + replaces_inherited_main_flow_domain: bool = field(init=False) + + def __post_init__(self) -> None: + object.__setattr__( + self, + "replaces_inherited_main_flow_domain", + self.is_grouped + or any( + invocation.contract.image_payload_consumption + is ImagePayloadConsumption.COMPOSED + for invocation in self.iter_invocations() + ), + ) @property def runtime_domain(self) -> RuntimeInvocationDomain: diff --git a/openhcs/core/steps/function_output_manifest.py b/openhcs/core/steps/function_output_manifest.py index f70662622c..ba0e610998 100644 --- a/openhcs/core/steps/function_output_manifest.py +++ b/openhcs/core/steps/function_output_manifest.py @@ -14,7 +14,6 @@ source_path_relative_to, ) from openhcs.core.artifacts import ArtifactOutputPlan, ArtifactType, ImageArtifactType -from openhcs.core.callable_contract import ImagePayloadConsumption from openhcs.core.aligned_image_payload import AlignedImageSliceContext from openhcs.core.context.processing_context import ProcessingContext from openhcs.core.path_pattern_matching import PathPatternTemplateMatcher @@ -24,7 +23,7 @@ image_payload_metadata, ) from openhcs.core.runtime_plane_projection import RuntimePlaneAxis -from openhcs.core.source_metadata import SourceMetadataValue +from openhcs.core.source_metadata import DurableSourceMetadata, SourceMetadataValue from openhcs.core.step_dependencies import StepInputDependencyKind from openhcs.core.steps.function_output_identity import ( @@ -58,6 +57,16 @@ class ProducedOutputSemantics(FunctionOutputIdentity): image_metadata: ImagePayloadMetadata | None = None main_flow_plane_axis: RuntimePlaneAxis | None = RuntimePlaneAxis.RUNTIME_SLICE + def published(self) -> "ProducedOutputSemantics": + """Fix the saved slot while retaining live provenance and storage aliases.""" + if isinstance(self.component_values, DurableSourceMetadata): + return self + return replace( + self, + component_values=DurableSourceMetadata.from_mapping(self.component_values), + filename_component_values=self.filename_values, + ) + def path_tokens(self, parser: FilenameParser) -> frozenset[str]: """Expose this occurrence's exact storage and unqualified address aliases.""" tokens = { @@ -258,15 +267,20 @@ def from_existing_main_flow_path( class StepOutputManifestStore: """Execution-local main-flow output lineage for shared VFS directories.""" - records_by_key: dict[StepOutputManifestKey, tuple[ProducedOutputSemantics, ...]] = ( - field(default_factory=dict) - ) - records_revision: int = 0 + records_by_key: dict[ + StepOutputManifestKey, + dict[ + tuple[str, str, str | None, tuple[tuple[str, str | int], ...]], + ProducedOutputSemantics, + ], + ] = field(default_factory=dict) + own_output_counts: dict[StepOutputManifestKey, int] = field(default_factory=dict) selected_records_by_source: dict[ - tuple[ - int, StepOutputManifestKey | None, frozenset[tuple[str, str, str | None]] + StepOutputManifestKey, + dict[ + tuple[frozenset[tuple[str, str, str | None]], bool], + tuple[ProducedOutputSemantics, ...], ], - tuple[ProducedOutputSemantics, ...] | None, ] = field(default_factory=dict) def begin_step( @@ -277,8 +291,17 @@ def begin_step( key = self.key_for_producer(plan) if key is None: return - self.records_by_key[key] = tuple(input_records) - self._invalidate_record_selection_caches() + records = { + published.main_flow_address: published + for record in input_records + for published in (record.published(),) + } + self.records_by_key[key] = records + self.own_output_counts[key] = sum( + record.producer_identity.step_scope_id == key.step_scope_id + for record in records.values() + ) + self._invalidate_record_selection_caches(key) def record_outputs( self, @@ -290,39 +313,32 @@ def record_outputs( key = self.key_for_producer(plan) if key is None: return - existing = self.records_for_key(key) - current_outputs = tuple(output_records) - has_current_step_output = any( - record.producer_identity.step_scope_id == plan.step_scope_id - for record in existing - ) - if existing and current_outputs and not has_current_step_output: - inherited_addresses = frozenset( - record.main_flow_address for record in existing - ) - output_addresses = frozenset( - record.main_flow_address for record in current_outputs - ) + if not output_records: + return + records = self.records_by_key.setdefault(key, {}) + current_outputs = tuple(record.published() for record in output_records) + own_output_count = self.own_output_counts.get(key, 0) + if records and not own_output_count: if ( collapsed_input_domain - or any( - invocation.contract.image_payload_consumption - is ImagePayloadConsumption.COMPOSED - for invocation in plan.compiled_function_pattern.iter_invocations() - ) - or plan.compiled_function_pattern.is_grouped - or not output_addresses.issubset(inherited_addresses) + or plan.compiled_function_pattern.replaces_inherited_main_flow_domain + or any(record.main_flow_address not in records for record in current_outputs) ): - existing = () - records_by_address = { - record.main_flow_address: record for record in (*existing, *current_outputs) - } - self.records_by_key[key] = tuple(records_by_address.values()) - self._invalidate_record_selection_caches() + records.clear() + for record in current_outputs: + address = record.main_flow_address + previous = records.get(address) + if previous is not None: + own_output_count -= ( + previous.producer_identity.step_scope_id == key.step_scope_id + ) + records[address] = record + own_output_count += record.producer_identity.step_scope_id == key.step_scope_id + self.own_output_counts[key] = own_output_count + self._invalidate_record_selection_caches(key) - def _invalidate_record_selection_caches(self) -> None: - self.records_revision += 1 - self.selected_records_by_source.clear() + def _invalidate_record_selection_caches(self, key: StepOutputManifestKey) -> None: + self.selected_records_by_source.pop(key, None) def producer_records_for( self, @@ -440,7 +456,7 @@ def producer_record_index_for( self, plan: CompiledStepPlan, parser: FilenameParser, ) -> ProducedPathRecordIndex | None: """Admit one current producer cohort with its correlated address aliases.""" - records = self._selected_unique_producer_records_for(plan) + records = self._selected_unique_producer_records_for(plan, path_ordered=True) return ( None if records is None else ProducedPathRecordIndex.from_records(records, parser) @@ -462,36 +478,30 @@ def filter_to_producer_paths( raise NoStepOutputManifestMatch return [] - def _producer_selection_key( - self, - plan: CompiledStepPlan, - ) -> tuple[ - int, StepOutputManifestKey | None, frozenset[tuple[str, str, str | None]] - ]: - """Select by current producer declarations, never a temporary plan address.""" - return ( - self.records_revision, - self._main_input_producer_key(plan), - self._requested_producer_outputs(plan), - ) - def _selected_unique_producer_records_for( self, plan: CompiledStepPlan, + *, + path_ordered: bool = False, ) -> tuple[ProducedOutputSemantics, ...] | None: - cache_key = self._producer_selection_key(plan) - if cache_key in self.selected_records_by_source: - return self.selected_records_by_source[cache_key] - - producer_key, requested = cache_key[1:] + producer_key = self._main_input_producer_key(plan) if producer_key is None: - self.selected_records_by_source[cache_key] = None return None - producer_records = self.records_for_key(producer_key) - selected = self._select_requested_producer_records(requested, producer_records) - selected = self._unique_output_path_records(selected) - self.selected_records_by_source[cache_key] = selected - return selected + requested = self._requested_producer_outputs(plan) + selections = self.selected_records_by_source.setdefault(producer_key, {}) + insertion_key = (requested, False) + if insertion_key not in selections: + producer_records = self.records_for_key(producer_key) + selected = self._select_requested_producer_records(requested, producer_records) + selections[insertion_key] = self._unique_output_path_records(selected) + if not path_ordered: + return selections[insertion_key] + path_key = (requested, True) + if path_key not in selections: + selections[path_key] = tuple( + sorted(selections[insertion_key], key=lambda record: record.output_path) + ) + return selections[path_key] @staticmethod def _unique_output_path_records( @@ -574,9 +584,7 @@ def records_for_key( self, key: StepOutputManifestKey, ) -> tuple[ProducedOutputSemantics, ...]: - if key not in self.records_by_key: - return () - return self.records_by_key[key] + return tuple(self.records_by_key.get(key, {}).values()) @staticmethod def key_for_producer( diff --git a/tests/unit/test_function_output_manifest_input_scope.py b/tests/unit/test_function_output_manifest_input_scope.py index 0613070e19..c650e229e5 100644 --- a/tests/unit/test_function_output_manifest_input_scope.py +++ b/tests/unit/test_function_output_manifest_input_scope.py @@ -124,6 +124,41 @@ def test_produced_occurrence_owns_memory_path_and_current_source_projection( record.source_metadata_for_projection(metadata, "/disk/output.tif") +def test_published_slots_stay_fixed_with_live_source_metadata_and_filename_aliases(): + plan = CompiledStepPlan( + step_index=1, step_type="FunctionStep", step_name="Producer", axis_id="A01", + step_scope_id="producer", pipeline_position=1, output_dir=Path("/memory"), + ) + coordinates = [{"channel": 1}, {"channel": 2}] + metadata = ImagePayloadMetadata(source_component_metadata={"acquisition": "first"}) + records = tuple( + ProducedOutputSemantics.from_output( + plan, plan.output_dir / f"channel{channel}.tif", + FunctionOutputIdentity(values, ".tif", "test"), + image_metadata=metadata, + ) + for channel, values in enumerate(coordinates, 1) + ) + manifest = StepOutputManifestStore() + manifest.begin_step(plan) + manifest.record_outputs(plan, records) + coordinates[0]["channel"] = 2 + metadata.source_component_metadata = {"acquisition": "second"} + manifest.record_outputs(plan, ()) + published = manifest.produced_records_for(plan) + assert tuple(record.component_values["channel"] for record in published) == (1, 2) + assert published[0].filename_values["channel"] == 2 + assert records[0].component_values["channel"] == 2 + projection = published[0].source_metadata_for_projection(metadata, "/saved/image.tif") + assert projection["channel"] == "1" + assert projection["acquisition"] == "second" + replacement = replace(records[0], component_values={"channel": 1}, output_path="new.tif") + manifest.record_outputs(plan, (replacement,)) + assert tuple(record.output_path for record in manifest.produced_records_for(plan)) == ( + "new.tif", "/memory/channel2.tif", + ) + + def _compiled_pattern_with_input_edges( specs_with_scopes: tuple[tuple[ArtifactSpec, str], ...], ): From c1650d851214dad201a06209fdf3ad0b9b532373 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 4 Oct 2026 13:37:02 -0400 Subject: [PATCH 03/10] perf: own filename facts and source authority by execution context Share parsed filename results and derived component membership through the existing RuntimePatternDiscoveryCache. Physical discovery, matching and grouped selection reuse the same parser semantic snapshot; grouped selection does not parse names. Retain the context source authority and rebind against strong owners. Observe optional workspace file eligibility and document epochs live while retaining its provider. Consume the producer store ordered load view without re-sorting or re-comparing known cohorts. Make runtime logging lazy and remove duplicate traceback formatting. Existing source/provenance/ambiguity controls: 153 passed. Production consumer probe verifies parse-once behavior and live ownership/document epochs. Whole-pipeline parity and performance remain integration gates; no speedup claim. External persisted result formats unchanged; internal execution caches derive from current declarations. --- openhcs/core/context/processing_context.py | 18 ++++ openhcs/core/pipeline/compiler.py | 8 +- openhcs/core/runtime_pattern_cache.py | 47 +++++++++ openhcs/core/source_workspace_projection.py | 50 ++++++---- openhcs/core/steps/function_execution.py | 19 ++-- openhcs/core/steps/function_runtime.py | 97 +++++++++---------- openhcs/formats/pattern/pattern_discovery.py | 89 +++++------------ openhcs/microscopes/microscope_base.py | 12 +-- openhcs/microscopes/openhcs.py | 17 ++-- .../test_bioformats_microscope_handler.py | 3 +- ...st_function_execution_lifecycle_anchors.py | 6 ++ ...test_function_runtime_source_projection.py | 12 ++- ..._source_bound_matched_image_set_anchors.py | 9 +- 13 files changed, 212 insertions(+), 175 deletions(-) diff --git a/openhcs/core/context/processing_context.py b/openhcs/core/context/processing_context.py index 187565b391..93a3997f42 100644 --- a/openhcs/core/context/processing_context.py +++ b/openhcs/core/context/processing_context.py @@ -34,6 +34,7 @@ from openhcs.core.runtime_stack_cache import RuntimeImageStackCache from openhcs.core.runtime_source_binding_cache import RuntimeSourceBindingContextCache from openhcs.core.source_workspace_projection import ( + VirtualWorkspaceSourceProjectionAuthority, VirtualWorkspaceSourceProjectionCache, ) from openhcs.core.source_matching import SourceImageSetIdentityPolicy @@ -134,6 +135,9 @@ def __init__( self.runtime_source_workspace_projection_cache = ( VirtualWorkspaceSourceProjectionCache() ) + self._runtime_source_workspace_projection_authority: ( + VirtualWorkspaceSourceProjectionAuthority | None + ) = None self.source_image_set_identity_policy = SourceImageSetIdentityPolicy() self.axis_id = axis_id self.filemanager = filemanager @@ -168,6 +172,20 @@ def __init__( self.pipeline_sequential_combinations = None self.current_sequential_combination = None + @property + def runtime_source_workspace_projection_authority( + self, + ) -> VirtualWorkspaceSourceProjectionAuthority: + """Hold the context's metadata owners; projection documents remain live.""" + authority = self._runtime_source_workspace_projection_authority + if authority is None or not authority.is_bound_to_context(self): + authority = VirtualWorkspaceSourceProjectionAuthority.from_context( + self, + cache=self.runtime_source_workspace_projection_cache, + ) + self._runtime_source_workspace_projection_authority = authority + return authority + def bind_execution_runtime( self, runtime: WorkerLaneExecutionContext, diff --git a/openhcs/core/pipeline/compiler.py b/openhcs/core/pipeline/compiler.py index 90b10228f8..82d151418a 100644 --- a/openhcs/core/pipeline/compiler.py +++ b/openhcs/core/pipeline/compiler.py @@ -113,9 +113,6 @@ PipelinePathPlanner, ) from openhcs.core.source_matching import SourceImageSetIdentityPolicy -from openhcs.core.source_workspace_projection import ( - VirtualWorkspaceSourceProjectionAuthority, -) from openhcs.core.source_load_plan import SourceLoadPlan from openhcs.core.invocation_artifacts import ( PipelineInvocationContractProviderAuthority, @@ -1338,10 +1335,7 @@ def _run_post_plan_compile_stages( def validate_source_workspace_projection(session: CompilationSession) -> None: """Validate source-workspace metadata before runtime image loading.""" - projection = VirtualWorkspaceSourceProjectionAuthority.from_context( - session.context, - cache=session.context.runtime_source_workspace_projection_cache, - ).projection_if_available() + projection = session.context.runtime_source_workspace_projection_authority.projection_if_available() if projection is None: return projection.validate_runtime_metadata_projection(axis_id=session.axis_id) diff --git a/openhcs/core/runtime_pattern_cache.py b/openhcs/core/runtime_pattern_cache.py index 1c2f0c35aa..f791fe7873 100644 --- a/openhcs/core/runtime_pattern_cache.py +++ b/openhcs/core/runtime_pattern_cache.py @@ -4,10 +4,17 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass, field +from pathlib import Path +from typing import Hashable, TYPE_CHECKING +from openhcs.constants.constants import AllComponents +from openhcs.core.components.parser_metaprogramming import FilenameParseResult from openhcs.core.function_patterns import FunctionGroupKey from openhcs.core.source_binding_selection import SourceCandidatePath +if TYPE_CHECKING: + from openhcs.microscopes.microscope_interfaces import FilenameParser + DiscoveredPatternCollection = ( Sequence[SourceCandidatePath] | Mapping[FunctionGroupKey, Sequence[SourceCandidatePath]] @@ -115,6 +122,46 @@ class RuntimePatternDiscoveryCache: FrozenPatternDiscoveryResult, ] = field(default_factory=dict) + filename_metadata: dict[ + tuple["FilenameParser", tuple[Hashable, ...], str], + FilenameParseResult | None, + ] = field(default_factory=dict) + filenames_by_component: dict[ + tuple["FilenameParser", tuple[Hashable, ...], AllComponents, str], set[str], + ] = field(default_factory=dict) + + def metadata_for_filename( + self, parser: "FilenameParser", filename: str, + ) -> FilenameParseResult | None: + """Own each filename result under its actual parser semantics.""" + semantics = parser.semantic_identity() + key = (parser, semantics, filename) + if key not in self.filename_metadata: + metadata = parser.parse_filename(filename) + self.filename_metadata[key] = metadata + if metadata is not None: + for component, value in metadata.declared_values(): + self.filenames_by_component.setdefault( + (parser, semantics, component, str(value)), set(), + ).add(filename) + return self.filename_metadata[key] + + def files_for_component( + self, + parser: "FilenameParser", + files: Sequence[str], + component: AllComponents, + value: object, + ) -> list[str]: + """Select component membership from the admitted filename inventory.""" + names = tuple(Path(path).name for path in files) + members = self.filenames_by_component.get( + (parser, parser.semantic_identity(), component, str(value)), (), + ) + return [ + path for path, name in zip(files, names, strict=True) if name in members + ] + def get( self, key: RuntimePatternDiscoveryCacheKey, diff --git a/openhcs/core/source_workspace_projection.py b/openhcs/core/source_workspace_projection.py index 677f214007..0fcfa88111 100644 --- a/openhcs/core/source_workspace_projection.py +++ b/openhcs/core/source_workspace_projection.py @@ -50,6 +50,7 @@ from openhcs.microscopes.microscope_interfaces import MetadataHandler from openhcs.core.vfs_protocol import FileManagerLike from polystore.filemanager import FileManager + from openhcs.microscopes.openhcs import OpenHCSMetadataHandler LookupValueT = TypeVar("LookupValueT") @@ -673,8 +674,12 @@ class VirtualWorkspaceSourceProjectionAuthority: """Projection authority for source-workspace metadata owned by a plate handler.""" plate_path: Path - metadata_handlers: tuple["MetadataHandler", ...] + metadata_handler: "MetadataHandler" + filemanager: "FileManager" cache: VirtualWorkspaceSourceProjectionCache | None = None + _workspace_metadata_handler: "OpenHCSMetadataHandler | None" = field( + default=None, init=False, compare=False, repr=False, + ) @classmethod def from_context( @@ -703,35 +708,38 @@ def from_plate_metadata( return cls( plate_path=plate_path, - metadata_handlers=cls._plate_metadata_handlers( - plate_path, - metadata_handler, - filemanager, - ), + metadata_handler=metadata_handler, + filemanager=filemanager, cache=DEFAULT_SOURCE_PROJECTION_CACHE if cache is None else cache, ) - @staticmethod - def _plate_metadata_handlers( - plate_path: Path, - metadata_handler: "MetadataHandler", - filemanager: "FileManager", - ) -> tuple["MetadataHandler", ...]: - """Return metadata handlers that can own source-workspace metadata.""" + def is_bound_to_context(self, context: "ProcessingContext") -> bool: + """Compare actual owners, never identities of already-released objects.""" + return ( + self.plate_path == Path(context.plate_path) + and self.metadata_handler is context.microscope_handler.metadata_handler + and self.filemanager is context.filemanager + ) + def metadata_handlers(self) -> tuple["MetadataHandler", ...]: + """Observe workspace eligibility live while retaining admitted providers.""" from openhcs.microscopes.openhcs import OpenHCSMetadataHandler - handlers: list["MetadataHandler"] = [metadata_handler] - metadata_path = plate_path / OpenHCSMetadataHandler.METADATA_FILENAME - if not isinstance(handlers[0], OpenHCSMetadataHandler) and filemanager.exists( - str(metadata_path), Backend.DISK.value - ): - handlers.append(OpenHCSMetadataHandler(filemanager)) - return tuple(handlers) + if isinstance(self.metadata_handler, OpenHCSMetadataHandler): + return (self.metadata_handler,) + metadata_path = self.plate_path / OpenHCSMetadataHandler.METADATA_FILENAME + if not self.filemanager.exists(str(metadata_path), Backend.DISK.value): + return (self.metadata_handler,) + workspace_handler = self._workspace_metadata_handler + if workspace_handler is None: + workspace_handler = OpenHCSMetadataHandler(self.filemanager) + object.__setattr__(self, "_workspace_metadata_handler", workspace_handler) + workspace_handler.invalidate_metadata_cache() + return (self.metadata_handler, workspace_handler) def metadata_documents(self) -> tuple[OpenHCSMetadataPayload, ...]: documents: list[OpenHCSMetadataPayload] = [] - for metadata_handler in self.metadata_handlers: + for metadata_handler in self.metadata_handlers(): metadata = metadata_handler.source_workspace_metadata_document( self.plate_path ) diff --git a/openhcs/core/steps/function_execution.py b/openhcs/core/steps/function_execution.py index ae9b13d3dd..5d0348180d 100644 --- a/openhcs/core/steps/function_execution.py +++ b/openhcs/core/steps/function_execution.py @@ -40,9 +40,6 @@ SourcePatternResolutionContext, ) from openhcs.core.source_bindings import CompiledSourceBindingPlan -from openhcs.core.source_workspace_projection import ( - VirtualWorkspaceSourceProjectionAuthority, -) from openhcs.core.step_dependencies import StepInputDependencyKind from openhcs.core.steps.function_io import ( generate_materialized_paths, @@ -623,10 +620,9 @@ def apply( def source_pattern_context(self) -> SourcePatternResolutionContext: """Return source-path context used to filter source-bound anchors.""" - projection = VirtualWorkspaceSourceProjectionAuthority.from_context( - self.context, - cache=self.context.runtime_source_workspace_projection_cache, - ).projection_or_empty() + projection = ( + self.context.runtime_source_workspace_projection_authority.projection_or_empty() + ) return self.context.runtime_source_binding_context_cache.source_pattern_context( parser=self.context.microscope_handler.parser, projection=self.context.runtime_source_workspace_projection_cache.filtered_by_axis( @@ -903,10 +899,9 @@ def _detect_patterns(self) -> dict[str, DiscoveredPatternCollection]: axis_filter = {f"{axis_name}_filter": [plan.axis_id]} source_files = step_output_manifest(self.context).producer_paths_for(plan) if source_files is None: - source_projection = VirtualWorkspaceSourceProjectionAuthority.from_context( - self.context, - cache=self.context.runtime_source_workspace_projection_cache, - ).projection_if_available() + source_projection = ( + self.context.runtime_source_workspace_projection_authority.projection_if_available() + ) if ( plan.main_input_dependency.kind is StepInputDependencyKind.PIPELINE_START @@ -932,6 +927,7 @@ def _detect_patterns(self) -> dict[str, DiscoveredPatternCollection]: patterns_by_axis = PatternDiscoveryEngine( self.context.microscope_handler.parser, self.context.filemanager, + self.context.runtime_pattern_discovery_cache, ).auto_detect_patterns_from_axis_files( list(source_files), axis_id=plan.axis_id, @@ -949,6 +945,7 @@ def _detect_patterns(self) -> dict[str, DiscoveredPatternCollection]: extensions=LOADABLE_IMAGE_EXTENSIONS, group_by=plan.group_by, variable_components=plan.variable_component_values, + pattern_cache=self.context.runtime_pattern_discovery_cache, **axis_filter, ) diff --git a/openhcs/core/steps/function_runtime.py b/openhcs/core/steps/function_runtime.py index a7b436b5ee..b9a709001f 100644 --- a/openhcs/core/steps/function_runtime.py +++ b/openhcs/core/steps/function_runtime.py @@ -860,7 +860,7 @@ def load_artifact_inputs( if not self.should_load_artifact_inputs(): return {} logger.info( - f"Artifact inputs for {self.function_name}: {self.artifacts.inputs}" + "Artifact inputs for %s: %s", self.function_name, self.artifacts.inputs ) loaded_artifact_payloads: dict[ArtifactSpecRef, RuntimePayload] = {} parameter_values: dict[str, list[RuntimeValue]] = {} @@ -911,8 +911,9 @@ def load_artifact_input( if storage_plan is None: raise ValueError("Artifact input loading requires a storage-backed edge.") logger.info( - f"Loading artifact input '{arg_name}' from path '{storage_plan.path}' " - "(memory backend)" + "Loading artifact input '%s' from path '%s' (memory backend)", + arg_name, + storage_plan.path, ) load_started_at = time.perf_counter() try: @@ -924,8 +925,10 @@ def load_artifact_input( ).projected_values(self.group_data.context.runtime_value_store) except Exception as exc: logger.error( - f"Failed to load artifact input '{arg_name}' from " - f"'{storage_plan.path}': {exc}", + "Failed to load artifact input '%s' from '%s': %s", + arg_name, + storage_plan.path, + exc, exc_info=True, ) raise @@ -1113,7 +1116,7 @@ def invoke( loaded_artifact_payloads: Mapping[ArtifactSpecRef, RuntimePayload], debug_sink: DebugEventSink | None, ) -> RuntimeFunctionOutput: - logger.info(f"Executing function: {self.function_name}") + logger.info("Executing function: %s", self.function_name) func_callable = self.func_callable contract = self.invocation.contract primary_parameter = contract.primary_input_parameter_name @@ -1291,8 +1294,9 @@ def save_artifact_output( loaded_artifact_payloads: Mapping[ArtifactSpecRef, RuntimePayload], ) -> RuntimePayload: logger.info( - f"Saving artifact output '{output_key}' to VFS path '{output_plan.path}' " - "(memory backend)" + "Saving artifact output '%s' to VFS path '%s' (memory backend)", + output_key, + output_plan.path, ) save_started_at = time.perf_counter() artifact_source_payload = self.artifact_output_source_payload( @@ -1369,10 +1373,7 @@ def source_workspace_projection_cache( def source_workspace_projection_authority( self, ) -> VirtualWorkspaceSourceProjectionAuthority: - return VirtualWorkspaceSourceProjectionAuthority.from_context( - self.request.context, - cache=self.source_workspace_projection_cache(), - ) + return self.request.context.runtime_source_workspace_projection_authority @staticmethod def _is_relative_to(path: Path, root: Path) -> bool: @@ -1401,9 +1402,9 @@ def _input_relative_path(cls, input_dir: Path, matched_path: str) -> Path: return path def run(self) -> None: - start_time = time.time() + start_time = time.time() if logger.isEnabledFor(logging.DEBUG) else None plan = self.request.execution_plan - logger.debug(f"Processing pattern {self.pattern_repr} for axis {plan.axis_id}") + logger.debug("Processing pattern %s for axis %s", self.pattern_repr, plan.axis_id) try: load_started_at = time.perf_counter() @@ -1477,20 +1478,17 @@ def run(self) -> None: step_name=plan.step_name, pattern=self.pattern_repr, ) - logger.debug( - f"Finished pattern group {self.pattern_repr} in {(time.time() - start_time):.2f}s." - ) + if start_time is not None and logger.isEnabledFor(logging.DEBUG): + logger.debug( + "Finished pattern group %s in %.2fs.", + self.pattern_repr, + time.time() - start_time, + ) except Exception as e: - import traceback - - full_traceback = traceback.format_exc() logger.error( - f"Error processing pattern group {self.pattern_repr}: {e}", + "Error processing pattern group %s: %s", self.pattern_repr, e, exc_info=True, ) - logger.error( - f"Full traceback for pattern group {self.pattern_repr}:\n{full_traceback}" - ) raise ValueError( f"Failed to process pattern group {self.pattern_repr}: {e}" ) from e @@ -1568,6 +1566,7 @@ def _load_input_stack( if plan.variable_components else None ), + pattern_cache=context.runtime_pattern_discovery_cache, ) if producer_index is not None and not producer_matching_files: selected_paths = [ @@ -1588,17 +1587,22 @@ def _load_input_stack( matching_files = self._filter_matching_files_for_group(matching_files) - logger.debug( - "Pattern %s matched %d files: %s", - self.pattern_repr, - len(matching_files), - [Path(f).name for f in matching_files], - ) + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + "Pattern %s matched %d files: %s", + self.pattern_repr, + len(matching_files), + [Path(f).name for f in matching_files], + ) - matching_files.sort() - logger.debug( - f"Pattern {self.pattern_repr} sorted files: {[Path(f).name for f in matching_files]}" - ) + if not producer_matching_files: + matching_files.sort() + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + "Pattern %s sorted files: %s", + self.pattern_repr, + [Path(f).name for f in matching_files], + ) matching_files = self._filter_matching_files_for_source_bindings(matching_files) full_file_paths = [ @@ -1625,13 +1629,10 @@ def _load_input_stack( else () ) if producer_index is not None: - producer_records = tuple(sorted( - producer_records, key=lambda record: record.output_path, - )) - if matching_files != [record.output_path for record in producer_records]: - producer_records = producer_index.records_for_paths(matching_files) - else: + if producer_matching_files: producer_index.validate_input_records(producer_records) + else: + producer_records = producer_index.records_for_paths(matching_files) ImagePayloadStackComposition.validate_main_flow_cohort(producer_records) cached_stack = context.runtime_image_stack_cache.get( tuple(full_file_paths), @@ -1728,14 +1729,12 @@ def _filter_matching_files_for_group( return matching_files parser = self.request.context.microscope_handler.parser - group_component_declaration = parser.component_for_name(group_component) - filtered = [ - filename - for filename in matching_files - if (metadata := parser.parse_filename(Path(filename).name)) - and str(metadata.value_for(group_component_declaration)) - == str(component_value) - ] + filtered = self.request.context.runtime_pattern_discovery_cache.files_for_component( + parser, + matching_files, + parser.component_for_name(group_component), + component_value, + ) if not filtered: raise ValueError( f"Pattern group {self.pattern_repr} for {group_component}=" @@ -2127,7 +2126,7 @@ def _project_output_slices( output_shape = np.shape(processed_data) output_ndim = np.ndim(processed_data) logger.error("Function output is not an OpenHCS image stack.") - logger.error(f"Output type: {type(processed_stack)}") + logger.error("Output type: %s", type(processed_stack)) logger.error("Output shape: %s", output_shape) logger.error("Output ndim: %s", output_ndim) raise ValueError( diff --git a/openhcs/formats/pattern/pattern_discovery.py b/openhcs/formats/pattern/pattern_discovery.py index 24d23d8076..79ffd2d958 100644 --- a/openhcs/formats/pattern/pattern_discovery.py +++ b/openhcs/formats/pattern/pattern_discovery.py @@ -16,6 +16,7 @@ from openhcs.constants.constants import DEFAULT_IMAGE_EXTENSION from openhcs.core.components.parser_metaprogramming import FilenameParseResult +from openhcs.core.runtime_pattern_cache import RuntimePatternDiscoveryCache # Core OpenHCS Interfaces from openhcs.microscopes.microscope_interfaces import FilenameParser @@ -48,16 +49,18 @@ class PatternDiscoveryEngine: # Constants PLACEHOLDER_PATTERN = "{iii}" - def __init__(self, parser: FilenameParser, filemanager: FileManager): - """ - Initialize the pattern discovery engine. - - Args: - parser: Parser for microscopy filenames - filemanager: FileManager for file system operations - """ + def __init__( + self, + parser: FilenameParser, + filemanager: FileManager, + pattern_cache: RuntimePatternDiscoveryCache | None = None, + ): + """Initialize the pattern discovery engine.""" self.parser = parser self.filemanager = filemanager + self.pattern_cache = ( + RuntimePatternDiscoveryCache() if pattern_cache is None else pattern_cache + ) def path_list_from_pattern( self, @@ -66,21 +69,7 @@ def path_list_from_pattern( backend: str, variable_components: Optional[List[str]] = None, ) -> List[str]: - """ - Get a list of filenames matching a pattern in a directory. - - Args: - directory: Directory to search (string or Path object) - pattern: Pattern to match (string with optional {iii} placeholders) - backend: Backend to use for file operations (required) - variable_components: List of components that can vary (will be ignored during matching) - - Returns: - List of matching filenames - - Raises: - ValueError: If directory does not exist - """ + """Get a list of filenames matching a pattern in a directory.""" directory_path = str(directory) # Keep as string for FileManager consistency if not self.filemanager.is_dir(directory_path, backend): raise FileNotFoundError(f"Directory not found: {directory_path}") @@ -95,6 +84,7 @@ def path_list_from_pattern( ) # Use os.path.join instead of / file_exists = self.filemanager.exists(file_path, backend) if file_exists: + self.pattern_cache.metadata_for_filename(self.parser, pattern_str) return [pattern_str] return [] @@ -102,7 +92,7 @@ def path_list_from_pattern( logger.debug("Using pattern template: %s", pattern_str) # Parse pattern template to get expected structure - pattern_metadata = self.parser.parse_filename(pattern_str) + pattern_metadata = self.pattern_cache.metadata_for_filename(self.parser, pattern_str) if not pattern_metadata: logger.error("Failed to parse pattern template: %s", pattern_str) return [] @@ -122,7 +112,7 @@ def path_list_from_pattern( continue # Parse the actual filename - file_metadata = self.parser.parse_filename(filename) + file_metadata = self.pattern_cache.metadata_for_filename(self.parser, filename) if not file_metadata: continue @@ -140,17 +130,7 @@ def _matches_pattern_structure( pattern_metadata: FilenameParseResult, variable_components: List[str], ) -> bool: - """ - Check if a file's metadata matches a pattern's structure. - - Args: - file_metadata: Metadata extracted from actual filename - pattern_metadata: Metadata extracted from pattern template - variable_components: List of components that can vary - - Returns: - True if file matches pattern structure, False otherwise - """ + """Check if a file's metadata matches a pattern's structure.""" # Check all components in the pattern variable_declarations = { self.parser.component_for_name(component) @@ -205,7 +185,7 @@ def group_patterns_by_component( # The has_placeholders() check is only relevant when using patterns as concrete filenames # For pattern discovery and grouping, we WANT patterns with placeholders - metadata = self.parser.parse_filename(pattern_str) + metadata = self.pattern_cache.metadata_for_filename(self.parser, pattern_str) component_declaration = self.parser.component_for_name(component) if metadata is None or metadata.value_for(component_declaration) is None: @@ -221,17 +201,7 @@ def group_patterns_by_component( def subdivide_patterns_by_components( self, patterns: List[str], components: List[str] ) -> Dict[tuple, List[str]]: - """ - Subdivide patterns by multiple component values. - - Args: - patterns: List of pattern strings - components: List of component names to subdivide by - - Returns: - Dictionary mapping component value tuples to pattern lists - Example: {('001', '1'): [...], ('001', '2'): [...]} - """ + """Subdivide patterns by multiple component values.""" if not components: return {(): patterns} @@ -240,7 +210,7 @@ def subdivide_patterns_by_components( self.parser.component_for_name(component) for component in components ) for pattern in patterns: - metadata = self.parser.parse_filename(str(pattern)) + metadata = self.pattern_cache.metadata_for_filename(self.parser, str(pattern)) if not metadata: raise ValueError(f"Failed to parse pattern: {pattern}") key = tuple( @@ -419,7 +389,7 @@ def _filter_images_by_axis( logger.warning(f"Unexpected file path type: {type(img_path).__name__}") continue - metadata = self.parser.parse_filename(filename) + metadata = self.pattern_cache.metadata_for_filename(self.parser, filename) if not metadata: continue @@ -444,20 +414,7 @@ def _filter_images_by_axis( def _generate_patterns_for_files( self, files: List[Any], variable_components: List[str], axis_value: str ) -> List[str]: - """ - Generate patterns for a list of files. - - Args: - files: List of file path objects representing files - variable_components: List of components that can vary in the pattern - - Returns: - List of pattern strings - - Raises: - TypeError: If files list is not a list - ValueError: If pattern templates cannot be instantiated - """ + """Generate patterns for a list of files.""" # Validate input parameters if not isinstance(files, list): raise TypeError( @@ -483,7 +440,7 @@ def _generate_patterns_for_files( logger.warning(f"Unexpected file path type: {type(file_path).__name__}") continue - metadata = self.parser.parse_filename(filename) + metadata = self.pattern_cache.metadata_for_filename(self.parser, filename) if not metadata: continue @@ -521,7 +478,7 @@ def _generate_patterns_for_files( ) # Validate that the pattern can be instantiated - if not self.parser.parse_filename(pattern_str): + if not self.pattern_cache.metadata_for_filename(self.parser, pattern_str): raise ValueError( f"Clause 93 Violation: Pattern template '{pattern_str}' cannot be instantiated" ) diff --git a/openhcs/microscopes/microscope_base.py b/openhcs/microscopes/microscope_base.py index 2c09e82dc0..500d4ea1d8 100644 --- a/openhcs/microscopes/microscope_base.py +++ b/openhcs/microscopes/microscope_base.py @@ -42,6 +42,7 @@ if TYPE_CHECKING: from openhcs.core.config import MaterializationBackend, PipelineConfig from openhcs.core.source_bindings import SourceBindingsConfig + from openhcs.core.runtime_pattern_cache import RuntimePatternDiscoveryCache # Dictionary to store registered metadata handlers for auto-detection # This will be auto-wrapped with SecondaryRegistryDict by the metaclass @@ -755,6 +756,7 @@ def auto_detect_patterns( extensions=None, group_by=None, variable_components=None, + pattern_cache: "RuntimePatternDiscoveryCache | None" = None, **kwargs, ): """ @@ -784,12 +786,10 @@ def auto_detect_patterns( if not filemanager.exists(str(folder_path), backend): raise ValueError(f"Folder path does not exist: {folder_path}") - # Create pattern engine on demand with the provided filemanager from openhcs.formats.pattern.pattern_discovery import PatternDiscoveryEngine - pattern_engine = PatternDiscoveryEngine(self.parser, filemanager) + pattern_engine = PatternDiscoveryEngine(self.parser, filemanager, pattern_cache) - # Get patterns from the pattern engine patterns_by_well = pattern_engine.auto_detect_patterns( folder_path, extensions=extensions, @@ -813,6 +813,8 @@ def path_list_from_pattern( filemanager: FileManager, backend: str, variable_components: Optional[List[str]] = None, + *, + pattern_cache: "RuntimePatternDiscoveryCache | None" = None, ): """ Delegate to pattern engine. @@ -848,12 +850,10 @@ def path_list_from_pattern( # Allow string patterns with braces - they are used for template matching # The pattern engine will handle template expansion to find matching files - # Create pattern engine on demand with the provided filemanager from openhcs.formats.pattern.pattern_discovery import PatternDiscoveryEngine - pattern_engine = PatternDiscoveryEngine(self.parser, filemanager) + pattern_engine = PatternDiscoveryEngine(self.parser, filemanager, pattern_cache) - # Delegate to the pattern engine return pattern_engine.path_list_from_pattern( directory_path, pattern, diff --git a/openhcs/microscopes/openhcs.py b/openhcs/microscopes/openhcs.py index c9392894c8..4af2a07d22 100644 --- a/openhcs/microscopes/openhcs.py +++ b/openhcs/microscopes/openhcs.py @@ -109,14 +109,13 @@ class OpenHCSMetadataHandler(MetadataHandler, OpenHCSMetadataBase): METADATA_FILENAME = METADATA_CONFIG.METADATA_FILENAME def __init__(self, filemanager: FileManager): - """ - Initialize the metadata handler. - - Args: - filemanager: FileManager instance for file operations. - """ + """Bind the file owner and initialize derived metadata views.""" MetadataHandler.__init__(self) OpenHCSMetadataBase.__init__(self, filemanager) + self.invalidate_metadata_cache() + + def invalidate_metadata_cache(self) -> None: + """Release derived metadata views before a new source observation.""" self._metadata_cache: Optional[Dict[str, Any]] = None self._plate_path_cache: Optional[Path] = None self._metadata_dict_cache: Optional[Dict[str, Any]] = None @@ -814,11 +813,7 @@ def update_available_backends( self.atomic_writer.update_available_backends( metadata_file_path, available_backends ) - # Clear cache to force reload on next access - self._metadata_cache = None - self._plate_path_cache = None - self._metadata_dict_cache = None - self._metadata_dict_plate_path_cache = None + self.invalidate_metadata_cache() logger.info( f"Updated available backends to {available_backends} in {metadata_file_path}" ) diff --git a/tests/unit/test_bioformats_microscope_handler.py b/tests/unit/test_bioformats_microscope_handler.py index 4b304185c4..49c3942b03 100644 --- a/tests/unit/test_bioformats_microscope_handler.py +++ b/tests/unit/test_bioformats_microscope_handler.py @@ -163,7 +163,8 @@ def test_bioformats_structured_refs_project_inside_pattern_runtime( ) authority = VirtualWorkspaceSourceProjectionAuthority( plate_path=tmp_path, - metadata_handlers=(OpenHCSMetadataHandler(filemanager),), + metadata_handler=OpenHCSMetadataHandler(filemanager), + filemanager=filemanager, cache=VirtualWorkspaceSourceProjectionCache(), ) diff --git a/tests/unit/test_function_execution_lifecycle_anchors.py b/tests/unit/test_function_execution_lifecycle_anchors.py index d6b2693dba..5875dab3a8 100644 --- a/tests/unit/test_function_execution_lifecycle_anchors.py +++ b/tests/unit/test_function_execution_lifecycle_anchors.py @@ -24,6 +24,7 @@ from openhcs.core.pipeline.path_planner import PathPlanner, PathPlannerArtifactStage from openhcs.core.runtime_adapters import runtime_adapter from openhcs.core.runtime_source_binding_cache import RuntimeSourceBindingContextCache +from openhcs.core.source_workspace_projection import VirtualWorkspaceSourceProjectionAuthority from openhcs.core.steps.function_output_manifest import _STEP_OUTPUT_MANIFESTS from openhcs.core.source_bindings import CompiledSourceBindingPlan, NamedSourceBinding from openhcs.core.step_dependencies import StepInputDependency @@ -47,6 +48,11 @@ def _anchor_executor( runtime_source_workspace_projection_cache=source_workspace_projection_cache, runtime_source_binding_context_cache=RuntimeSourceBindingContextCache(), ) + executor.context.runtime_source_workspace_projection_authority = ( + VirtualWorkspaceSourceProjectionAuthority.from_context( + executor.context, cache=source_workspace_projection_cache, + ) + ) if output_manifest is not None: _STEP_OUTPUT_MANIFESTS[executor.context] = output_manifest return executor diff --git a/tests/unit/test_function_runtime_source_projection.py b/tests/unit/test_function_runtime_source_projection.py index f2e158aaa2..b82a18cef3 100644 --- a/tests/unit/test_function_runtime_source_projection.py +++ b/tests/unit/test_function_runtime_source_projection.py @@ -65,6 +65,7 @@ from openhcs.core.runtime_stack_cache import RuntimeImageStackCache from openhcs.core.source_binding_selection import SourcePatternResolutionContext from openhcs.core.runtime_source_binding_cache import RuntimeSourceBindingContextCache +from openhcs.core.runtime_pattern_cache import RuntimePatternDiscoveryCache from openhcs.core.steps.function_output_manifest import _STEP_OUTPUT_MANIFESTS from openhcs.core.source_bindings import ( SOURCE_BINDING_ALIAS_METADATA_FIELD, @@ -92,6 +93,7 @@ SourceProjectionSet, ) from openhcs.core.source_workspace_projection import ( + VirtualWorkspaceSourceProjectionAuthority, VirtualWorkspacePathLookup, VirtualWorkspaceSourceProjection, VirtualWorkspaceSourceProjectionCache, @@ -132,6 +134,11 @@ def _anchor_executor( runtime_source_workspace_projection_cache=source_workspace_projection_cache, runtime_source_binding_context_cache=RuntimeSourceBindingContextCache(), ) + executor.context.runtime_source_workspace_projection_authority = ( + VirtualWorkspaceSourceProjectionAuthority.from_context( + executor.context, cache=source_workspace_projection_cache, + ) + ) if output_manifest is not None: _STEP_OUTPUT_MANIFESTS[executor.context] = output_manifest return executor @@ -3943,10 +3950,11 @@ def resolve_address(backend_address, backend, *, base_path): context = SimpleNamespace( microscope_handler=SimpleNamespace( parser=SourceSchemaFilenameParser(), - path_list_from_pattern=lambda *_args: list(virtual_paths), + path_list_from_pattern=lambda *_args, **_kwargs: list(virtual_paths), ), filemanager=SourceFileManager(), runtime_image_stack_cache=RuntimeImageStackCache(), + runtime_pattern_discovery_cache=RuntimePatternDiscoveryCache(), runtime_source_binding_context_cache=RuntimeSourceBindingContextCache(), runtime_source_workspace_projection_cache=( VirtualWorkspaceSourceProjectionCache() @@ -5627,7 +5635,7 @@ def test_producer_admission_rederives_live_storage_aliases_each_epoch( ): store, producer, consumer, records, parser = qualified_producer_manifest storage_components = dict(records[0].filename_values) - record = replace(records[0], filename_component_values=storage_components) + record = replace(records[0], filename_component_values=storage_components).published() store.record_outputs(producer, (record,)) old_index = store.producer_record_index_for(consumer, parser) old_alias = record.without_filename_qualifier().filename(parser) diff --git a/tests/unit/test_source_bound_matched_image_set_anchors.py b/tests/unit/test_source_bound_matched_image_set_anchors.py index 12f5e916f1..628263bdc4 100644 --- a/tests/unit/test_source_bound_matched_image_set_anchors.py +++ b/tests/unit/test_source_bound_matched_image_set_anchors.py @@ -39,7 +39,9 @@ SourceSelector, ) from openhcs.core.source_projection import OpenHCSPlaneAddress, SourcePlaneProjection -from openhcs.core.source_workspace_projection import VirtualWorkspaceSourceProjection +from openhcs.core.source_workspace_projection import ( + VirtualWorkspaceSourceProjection, VirtualWorkspaceSourceProjectionAuthority, +) from openhcs.core.step_dependencies import StepInputDependency from openhcs.core.steps.function_execution import ( FunctionStepExecutor, @@ -72,6 +74,11 @@ def _anchor_executor( runtime_source_workspace_projection_cache=source_workspace_projection_cache, runtime_source_binding_context_cache=RuntimeSourceBindingContextCache(), ) + executor.context.runtime_source_workspace_projection_authority = ( + VirtualWorkspaceSourceProjectionAuthority.from_context( + executor.context, cache=source_workspace_projection_cache, + ) + ) if output_manifest is not None: _STEP_OUTPUT_MANIFESTS[executor.context] = output_manifest return executor From 12e67a668b27738b411ade7f7f578346d4294be5 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 4 Oct 2026 13:40:59 -0400 Subject: [PATCH 04/10] Record runtime plan integration and remaining performance gates --- plans/runtime_plumbing/IMPLEMENTATION.md | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/plans/runtime_plumbing/IMPLEMENTATION.md b/plans/runtime_plumbing/IMPLEMENTATION.md index 97e09c1cdb..e32fafd729 100644 --- a/plans/runtime_plumbing/IMPLEMENTATION.md +++ b/plans/runtime_plumbing/IMPLEMENTATION.md @@ -6,13 +6,15 @@ The active performance objective remains execution and total pipeline latency, w | Surface | Existing determining owners | Integration status and obligations | |---|---|---| -| P0 | RuntimeProfileLogger; worker/run lifecycle | In progress first. Consolidate all profile emitters into a run-owned raw event buffer; flush outside callable/chain timed regions, retain failure cleanup and independent worker/run lifetimes. Capture baseline before execution refactors. | +| P0 | RuntimeProfileLogger; worker/run lifecycle | Implemented in #645 and integrated here. One raw worker-run buffer and final flush; independent failure/fork/thread lifetimes. Real 3D diagnostic completed, with strict comparison against both retained native outputs. Observer disposition is separate from ordinary latency. | | P1 | CompiledFunctionInvocation; CallableContract; CompiledStepPlan | Compile fixed memory, ABI, bindings and return-layout decisions. Adapter requests/current pixels/active edges remain invocation-owned. Build debug parameters only for debug or the existing declaration-error diagnostics. Preserve eager adapter binding and error epochs. | -| P2 | Plate inventory and declared source bindings | Parse physical source filenames once and index components. Produced filename aliases remain live; do not reintroduce the stale alias cache removed by #638. | -| P3 | StepOutputManifestStore; compiled pattern/step declaration | Replace whole-cohort rebuilds and global invalidation with incremental per-producer records. Preserve inherited-domain replacement, insertion order, last-write replacement, begin-step epochs and live aliases. | -| P4 | ProcessingContext; source projection authority; manifest | Hold context-owned projections, lazy log formatting, producer order derived from P3. Preserve ambiguity/cohort admission before cached payload return. | +| P2 | RuntimePatternDiscoveryCache; declared source bindings | Implemented in c1650d8. Production consumer probe confirms one physical filename parse across discovery, matching and indexed group selection, and re-admission when parser semantics change. Produced filename aliases remain live. | +| P3 | StepOutputManifestStore; CompiledFunctionPattern | Implemented in 88522fb. Ordered address maps update only new records; producer-scoped invalidation eliminates revision counters. Existing immutable DurableSourceMetadata fixes publication slots; source metadata and filename aliases remain live. Publication policy explicitly approved by user. Derived path order is retained without changing insertion order. | +| P4 | ProcessingContext; source projection authority; manifest | Implemented in c1650d8. Held authority rebinds strong owners; optional file eligibility and document changes remain live. Lazy logging and retained P3 order remove per-group formatting/sorting. Ambiguity admission still precedes cached payload return. | | P5 | CompiledStepPlan; compiled invocation; device scope | Plan memory/device boundaries across steps. Keep preprocessing InputConversionPlan distinct from runtime placement. Validate mixed CPU/GPU semantics separately; a CPU-only profile cannot prove a GPU gain. | #641 is merged and removes post-compilation kernel warming; normal registry READY is the only child-cache consumer. Full catalog READY qualification passed. #630 (runtime ownership refactor), #635 (intensity maxima) and #597 (saved compiler resolution) remain drafts with negative whole-pipeline observations. They are not performance acceptance for this plan. Required final evidence: corrected baseline and integrated execution profile on the same real pipeline; production benchmark before/after execution and total time; affected parity, MCP and GUI behavior; mixed-device receiving for P5. Each implementation PR must close its corresponding issue and declare any persisted format changes. + +Implementation is tracked by #647 and closing draft #648. Current source merges P0 head 1236f744 and main 602d577. The integrated source/projection/manifest/profile gate passes 127 controls in 2.51s; this verifies behavior and metadata pickle/cloudpickle transport, not runtime improvement. Actual parse/rebind/optional-document production receiving is retained under the runtime-file-inventory receipts in the maintenance namespace. P1/P5 remain active and no integrated speedup is accepted yet. From c1d872807203706dc13060656b7b97e971496150 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 4 Oct 2026 13:53:48 -0400 Subject: [PATCH 05/10] perf: keep memory checkpoints on their declared backend --- openhcs/core/image_file_serialization.py | 20 ++++++--- openhcs/core/memory/__init__.py | 42 ++++--------------- openhcs/core/pipeline/compiler.py | 20 --------- .../unit/test_aligned_image_payload_device.py | 3 ++ 4 files changed, 25 insertions(+), 60 deletions(-) diff --git a/openhcs/core/image_file_serialization.py b/openhcs/core/image_file_serialization.py index cb49cfa1c0..d62b9ed9d8 100644 --- a/openhcs/core/image_file_serialization.py +++ b/openhcs/core/image_file_serialization.py @@ -10,6 +10,7 @@ from typing import Any, Callable, ClassVar, Sequence import numpy as np +from arraybridge import MemoryType, detect_memory_type from metaclass_registry import AutoRegisterMeta from polystore.config import TiffConfig, TiffPhotometric, TiffPlanarConfig @@ -249,9 +250,16 @@ def require_path(cls, path: str | Path) -> "ImageFileFormat": f"No image serialization format is registered for suffix {suffix!r}." ) - @abstractmethod def prepare(self, payload: Any) -> Any: - """Return a payload suitable for this file format.""" + """Project runtime pixels onto the host before format-specific encoding.""" + pixels = image_payload_data(payload) + host_pixels = MemoryType(detect_memory_type(pixels)).to_numpy(pixels) + host_payload = image_payload_metadata(payload).payload_with(host_pixels) + return self.prepare_host_payload(host_payload) + + @abstractmethod + def prepare_host_payload(self, payload: Any) -> Any: + """Encode host pixels while respecting this format's image semantics.""" def storage_config( self, payload: Any, configured: TiffConfig | None @@ -413,7 +421,7 @@ def preserves_pixel_values(self, source_dtype: Any) -> bool: del source_dtype return True - def prepare(self, payload: Any) -> Any: + def prepare_host_payload(self, payload: Any) -> Any: return image_payload_data(payload) def read(self, path: str | Path) -> np.ndarray: @@ -448,7 +456,7 @@ def preserves_pixel_values(self, source_dtype: Any) -> bool: del source_dtype return True - def prepare(self, payload: Any) -> Any: + def prepare_host_payload(self, payload: Any) -> Any: return image_payload_data(payload) def storage_config( @@ -568,7 +576,7 @@ class EightBitRasterImageFileFormat(ImageFileFormat): format_key = None suffixes = () - def prepare(self, payload: Any) -> Any: + def prepare_host_payload(self, payload: Any) -> Any: return image_payload_as_uint8(require_single_image_payload(payload)) def preserves_pixel_values(self, source_dtype: Any) -> bool: @@ -604,7 +612,7 @@ class PngImageFileFormat(ImageFileFormat): format_key = "png" suffixes = (".png",) - def prepare(self, payload: Any) -> Any: + def prepare_host_payload(self, payload: Any) -> Any: array = require_single_image_payload(payload) if array.dtype == np.uint8 or array.dtype == np.uint16: return array diff --git a/openhcs/core/memory/__init__.py b/openhcs/core/memory/__init__.py index d8d694c0c8..51ce4be807 100644 --- a/openhcs/core/memory/__init__.py +++ b/openhcs/core/memory/__init__.py @@ -7,8 +7,6 @@ from collections.abc import Sequence from typing import Any -import numpy as np - # Re-export from arraybridge from arraybridge import ( # Converters @@ -54,34 +52,18 @@ def stack_runtime_slices( slice_values = tuple(slices) if not slice_values: raise ValueError("Runtime-slice stacking requires at least one slice.") - numpy_slices = tuple( - np.asarray( - value - if detect_memory_type(value) == MEMORY_TYPE_NUMPY - else convert_memory( - data=value, - source_type=detect_memory_type(value), - target_type=MEMORY_TYPE_NUMPY, - gpu_id=gpu_id, - ) - ) + target = MemoryType(memory_type) + prepared_slices = [ + MemoryType(detect_memory_type(value)).convert_to(value, target, gpu_id) for value in slice_values - ) - shapes = tuple(tuple(value.shape) for value in numpy_slices) + ] + shapes = tuple(tuple(value.shape) for value in prepared_slices) if any(shape != shapes[0] for shape in shapes[1:]): raise ValueError( "Runtime slices must have one exact shape before stacking; " f"got {shapes!r}." ) - stacked = np.stack(numpy_slices, axis=0) - if memory_type == MEMORY_TYPE_NUMPY: - return stacked - return convert_memory( - data=stacked, - source_type=MEMORY_TYPE_NUMPY, - target_type=memory_type, - gpu_id=gpu_id, - ) + return target.stack_arrays(prepared_slices, gpu_id) def unstack_runtime_slices( @@ -93,16 +75,8 @@ def unstack_runtime_slices( ) -> tuple[Any, ...]: """Split an explicitly declared leading runtime-slice axis.""" - source_type = detect_memory_type(stack) - converted = ( - stack - if source_type == memory_type - else convert_memory( - data=stack, - source_type=source_type, - target_type=memory_type, - gpu_id=gpu_id, - ) + converted = MemoryType(detect_memory_type(stack)).convert_to( + stack, MemoryType(memory_type), gpu_id ) shape = tuple(int(value) for value in converted.shape) if not shape: diff --git a/openhcs/core/pipeline/compiler.py b/openhcs/core/pipeline/compiler.py index 82d151418a..605598da20 100644 --- a/openhcs/core/pipeline/compiler.py +++ b/openhcs/core/pipeline/compiler.py @@ -863,26 +863,6 @@ def validate_memory_contracts(session: CompilationSession) -> None: f"Memory validation must set {missing_fields} for FunctionStep {step_plan.step_name} (index: {step_index})." ) - # Apply memory type override: Any step with disk output must use numpy for disk writing - for step_index, step in enumerate(session.pipeline.steps): - if isinstance(step, FunctionStep): - if step_index in context.step_plans: - step_plan = context.step_plans[step_index] - compiled_pattern = step_plan.compiled_function_pattern - if ( - compiled_pattern is not None - and compiled_pattern.execution_scope - is FunctionStepExecutionScope.PLATE - ): - continue - write_backend = step_plan.write_backend - - if write_backend == "disk": - logger.debug( - f"Step {session.pipeline.steps[step_index].name} has disk output, overriding output_memory_type to numpy" - ) - step_plan.output_memory_type = "numpy" - @staticmethod def assign_framework_device_resources(session: CompilationSession) -> None: """Resolve each step's declaration-derived framework device footprint.""" diff --git a/tests/unit/test_aligned_image_payload_device.py b/tests/unit/test_aligned_image_payload_device.py index 69ec625c23..b4335dd7c9 100644 --- a/tests/unit/test_aligned_image_payload_device.py +++ b/tests/unit/test_aligned_image_payload_device.py @@ -74,6 +74,9 @@ def upload(values): runtime=SimpleNamespace(getDeviceCount=lambda: 2), Device=DeviceScope, ), array=upload, + stack=lambda values, axis=0: DeviceArray( + np.stack([value.values for value in values], axis=axis), state['device'], + ), logical_and=lambda left, right: DeviceArray( np.logical_and(left.values, right.values), state['device'], ), From cbf1857f7b449108c9c5916c2b874cf8b7e462c6 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 4 Oct 2026 14:02:37 -0400 Subject: [PATCH 06/10] perf: compile invocation ABI and device placement Implement P1 of plans/runtime_plumbing using the existing callable contract and compiled invocation. Bind declared return slots, executable callables, parameter bindings and framework devices before runtime; keep live adapter requests and selected artifact admission. Delete the per-call return matcher and Core projection forwarding. Rebuild process-local executable state from declaration-only pickle transport. Preserve authored kwargs, plate calls without image domains, callable-qualified memory errors and debug/error diagnostics. --- openhcs/core/callable_contract.py | 291 ++++++++++++++++-- openhcs/core/function_patterns.py | 154 +++++++++ .../pipeline/framework_device_assignment.py | 19 ++ .../pipeline/funcstep_contract_validator.py | 8 +- openhcs/core/runtime_output_matching.py | 187 ----------- openhcs/core/steps/function_runtime.py | 89 ++---- .../runtime/function_contract_execution.py | 7 +- .../runtime/measurement_execution_support.py | 11 +- .../cellprofiler/runtime/module_execution.py | 9 +- tests/unit/test_count_cells_simple.py | 3 +- tests/unit/test_framework_execution_scope.py | 27 +- tests/unit/test_function_patterns.py | 29 ++ .../unit/test_function_step_execution_plan.py | 35 ++- tests/unit/test_function_step_transport.py | 14 + tests/unit/test_multi_template_matching.py | 3 +- ...est_registered_object_output_boundaries.py | 5 +- tests/unit/test_returned_output_abi.py | 16 +- tests/unit/test_runtime_output_matching.py | 85 ++--- 18 files changed, 605 insertions(+), 387 deletions(-) diff --git a/openhcs/core/callable_contract.py b/openhcs/core/callable_contract.py index b2d61b24fc..3237773ce1 100644 --- a/openhcs/core/callable_contract.py +++ b/openhcs/core/callable_contract.py @@ -50,6 +50,7 @@ NativeReturnArtifactOutputPolicy, ) from openhcs.core.artifacts import ( + ArtifactOutputPlan, ArtifactSpec, ArtifactSpecCollection, ArtifactSpecRef, @@ -71,6 +72,8 @@ from openhcs.core.pipeline.compilation_session import CompilationPathResolver from openhcs.core.processing_preparation import PreparationOperation from openhcs.core.runtime_adapters import RuntimeAdapterSpec + from openhcs.core.runtime_output_matching import RuntimeMatchedOutput + from openhcs.core.runtime_plane_projection import RuntimePlaneAxisValueProjection from openhcs.core.runtime_batch_contracts import RuntimeBatchExecutionDomain from openhcs.core.vfs_protocol import PlatePathDeclaration from openhcs.processing.backends.lib_registry.unified_registry import ( @@ -535,7 +538,11 @@ def __post_init__(self) -> None: ) if not normalized: normalized = spec_parameter_names - object.__setattr__(self, "artifact_input_parameter_names", normalized) + object.__setattr__( + self, + "artifact_input_parameter_names", + normalized, + ) _validate_optional_enum( self.primary_image_carrier_requirement, PrimaryImageCarrierRequirement, @@ -779,6 +786,93 @@ class CallableContract(ArtifactPlanKeySelector): None ) + artifact_outputs: ArtifactSpecCollection = dataclasses.field( + init=False, + repr=False, + compare=False, + ) + canonical_return_output_specs: ArtifactSpecCollection = dataclasses.field( + init=False, + repr=False, + compare=False, + ) + trailing_return_output_specs: ArtifactSpecCollection = dataclasses.field( + init=False, + repr=False, + compare=False, + ) + _return_specs_by_ref: Mapping[ArtifactSpecRef, ArtifactSpec] = dataclasses.field( + init=False, + repr=False, + compare=False, + ) + canonical_return_output_refs: tuple[ArtifactSpecRef, ...] = dataclasses.field( + init=False, + repr=False, + compare=False, + ) + _trailing_return_output_refs: tuple[ArtifactSpecRef, ...] = dataclasses.field( + init=False, + repr=False, + compare=False, + ) + + def __post_init__(self) -> None: + """Capture the declared positional ABI once, before any returned values.""" + outputs = ArtifactSpecCollection(self.metadata.artifact_outputs) + by_ref: dict[ArtifactSpecRef, ArtifactSpec] = {} + for spec in outputs: + ref = spec.ref() + if ref in by_ref: + raise ValueError( + f"Callable output ABI contains duplicate artifact ref {ref!r}." + ) + by_ref[ref] = spec + canonical_count = 0 + if self.execution_scope is FunctionStepExecutionScope.PLATE: + canonical_count = min(1, len(outputs)) + elif outputs and outputs[0].participates_in_main_flow: + canonical_count = 1 + if outputs[0].artifact_type is ImageArtifactType: + for spec in outputs[1:]: + if ( + not spec.participates_in_main_flow + or spec.artifact_type is not ImageArtifactType + ): + break + canonical_count += 1 + refs = tuple(by_ref) + object.__setattr__( + self, + "artifact_outputs", + outputs, + ) + object.__setattr__( + self, + "canonical_return_output_specs", + ArtifactSpecCollection(outputs[:canonical_count]), + ) + object.__setattr__( + self, + "trailing_return_output_specs", + ArtifactSpecCollection(outputs[canonical_count:]), + ) + object.__setattr__( + self, + "_return_specs_by_ref", + MappingProxyType(by_ref), + ) + object.__setattr__( + self, + "canonical_return_output_refs", + refs[:canonical_count], + ) + object.__setattr__( + self, + "_trailing_return_output_refs", + refs[canonical_count:], + ) + def __reduce__( self, ) -> tuple[Callable[..., "CallableContract"], tuple[Any, ...]]: @@ -865,11 +959,6 @@ def artifact_input_parameter_names(self) -> tuple[str, ...]: return self.metadata.artifact_input_parameter_names - @property - def artifact_outputs(self) -> ArtifactSpecCollection: - """Return effective compiled output declarations in contract order.""" - return ArtifactSpecCollection(self.metadata.artifact_outputs) - @property def main_flow_outputs(self) -> ArtifactSpecCollection: """Return declared outputs eligible for the canonical image flow.""" @@ -878,32 +967,159 @@ def main_flow_outputs(self) -> ArtifactSpecCollection: spec for spec in self.artifact_outputs if spec.participates_in_main_flow ) - @property - def canonical_return_output_specs(self) -> ArtifactSpecCollection: - """Return named outputs carried by the callable's first return slot.""" + def contextualize_returned_canonical_output( + self, + returned_output: Any, + *, + plane_projection: RuntimePlaneAxisValueProjection | None = None, + ) -> Any: + """Attach the compiled canonical ABI to one exact returned image axis.""" + + from openhcs.core.aligned_image_payload import ( + AlignedImageSliceContext, + AlignedImageStack, + pack_aligned_image_outputs, + ) + from openhcs.core.runtime_image_values import image_payload_metadata + from openhcs.core.runtime_output_matching import split_runtime_output + from openhcs.core.runtime_slice_projection import ( + RuntimeSliceProjection, + RuntimeSliceProjectionDeclarationError, + ) + + canonical_specs = self.canonical_return_output_specs.specs + if len(canonical_specs) <= 1: + return returned_output + canonical_output, trailing_outputs = split_runtime_output(returned_output) + if isinstance(canonical_output, AlignedImageStack): + if canonical_output.slice_contexts: + return returned_output + output_values = canonical_output.slices + else: + projection = plane_projection + function_name = self.function_name + if projection is None: + raise RuntimeSliceProjectionDeclarationError( + f"{function_name} declares {len(canonical_specs)} canonical " + "outputs but returned a non-aligned payload without a " + "compiled plane projection." + ) + if projection.plane_index is not None: + raise RuntimeSliceProjectionDeclarationError( + f"{function_name} declares {len(canonical_specs)} canonical " + "outputs after the compiled plane projection already selected " + f"plane {projection.plane_index}." + ) + if projection.axis_size != len(canonical_specs): + raise ValueError( + f"{function_name} declares {len(canonical_specs)} canonical " + "outputs but its compiled plane projection declares " + f"{projection.axis_size} value(s)." + ) + output_axis = image_payload_metadata(canonical_output).plane_axis + if output_axis is not projection.axis: + raise RuntimeSliceProjectionDeclarationError( + f"{function_name} declares {len(canonical_specs)} canonical " + "outputs but its returned payload does not declare the " + f"compiled {projection.axis.value!r} plane axis; got " + f"{output_axis!r}." + ) + output_values = tuple( + RuntimeSliceProjection.value_for_slice( + canonical_output, + projection.selected_plane(output_index), + ) + for output_index in range(projection.axis_size) + ) + if len(output_values) != len(canonical_specs): + raise ValueError( + f"{self.function_name} returned {len(output_values)} " + "canonical output value(s) for " + f"{len(canonical_specs)} compiled output spec(s)." + ) + contextualized_output = pack_aligned_image_outputs( + output_values, + slice_contexts=AlignedImageSliceContext.main_flow_for_artifact_specs( + canonical_specs + ), + ) + return ( + (contextualized_output, *trailing_outputs) + if trailing_outputs + else contextualized_output + ) - declared_outputs = self.artifact_outputs - if self.execution_scope is FunctionStepExecutionScope.PLATE: - return ArtifactSpecCollection(declared_outputs[:1]) - if not declared_outputs or not declared_outputs[0].participates_in_main_flow: - return ArtifactSpecCollection(()) - canonical_count = 1 - if declared_outputs[0].artifact_type is ImageArtifactType: - for spec in declared_outputs[1:]: - if ( - not spec.participates_in_main_flow - or spec.artifact_type is not ImageArtifactType - ): - break - canonical_count += 1 - return ArtifactSpecCollection(declared_outputs[:canonical_count]) + def resolve_returned_output( + self, returned_output: Any + ) -> dict[ArtifactSpecRef, Any]: + """Resolve every output in the callable ABI.""" - @property - def trailing_return_output_specs(self) -> ArtifactSpecCollection: - """Return named outputs carried by trailing positional return slots.""" + from openhcs.core.aligned_image_payload import AlignedImageStack + from openhcs.core.runtime_output_matching import split_runtime_output - canonical_count = len(self.canonical_return_output_specs) - return ArtifactSpecCollection(self.artifact_outputs[canonical_count:]) + canonical_output, trailing_values = split_runtime_output(returned_output) + trailing_specs = self.trailing_return_output_specs.specs + if len(trailing_values) != len(trailing_specs): + raise ValueError( + "Runtime callable trailing return count does not match its " + f"declared trailing output slots: {len(trailing_values)} != " + f"{len(trailing_specs)}." + ) + resolved = { + ref: value + for ref, value in zip( + self._trailing_return_output_refs, + trailing_values, + strict=True, + ) + } + canonical_specs = self.canonical_return_output_specs.specs + if len(canonical_specs) == 1: + resolved[self.canonical_return_output_refs[0]] = canonical_output + elif canonical_specs: + if not isinstance(canonical_output, AlignedImageStack): + raise TypeError( + "Multiple canonical output specs require an AlignedImageStack with " + "one exact named slice context per output." + ) + resolved.update( + canonical_output.output_values_for_artifact_specs(canonical_specs) + ) + return resolved + + def resolve_returned_plan_values( + self, + returned_output: Any, + selected_output_plans: tuple[ArtifactOutputPlan, ...], + ) -> tuple[ + dict[ArtifactSpecRef, Any], + tuple[RuntimeMatchedOutput, ...], + ]: + """Resolve the complete ABI and bind exact selected runtime plans once.""" + + returned_values = self.resolve_returned_output(returned_output) + specs_by_ref = self._return_specs_by_ref + selected_refs: set[ArtifactSpecRef] = set() + matched_outputs: list[RuntimeMatchedOutput] = [] + for plan in selected_output_plans: + if not isinstance(plan, ArtifactOutputPlan): + raise TypeError( + "Selected runtime outputs must be ArtifactOutputPlan values, " + f"got {type(plan).__name__}." + ) + ref = plan.ref() + if ref in selected_refs: + raise ValueError( + f"Selected runtime output plans contain duplicate ref {ref!r}." + ) + selected_refs.add(ref) + spec = specs_by_ref.get(ref) + if spec is None: + raise ValueError( + f"Selected output plan {ref!r} is not declared by the callable ABI." + ) + matched_outputs.append((plan, spec, returned_values[ref])) + return returned_values, tuple(matched_outputs) @property def output_group_scope_sources(self) -> tuple[ArtifactSpecRef, ...]: @@ -1243,6 +1459,23 @@ def require_memory_types(self) -> tuple[str, str]: ) return self.input_memory_type, self.output_memory_type + def require_execution_memory_type( + self, *, step_name: str | None = None + ) -> MemoryType | None: + """Admit the declared execution framework with its callable context.""" + declaration = self.execution_memory_type + if declaration is None: + return None + valid_types = frozenset(memory_type.value for memory_type in MemoryType) + if declaration not in valid_types: + step_context = "" if step_name is None else f" in step {step_name!r}" + raise ValueError( + f"Callable {self.function_name!r}{step_context} " + f"declares invalid execution memory type {declaration!r}; " + f"valid memory types are {', '.join(sorted(valid_types))}." + ) + return MemoryType(declaration) + def resolve_runtime_callable(self) -> Callable[..., object]: """Return the executable callable for this contract.""" cache = CallableContractRuntimeCache.process_cache() diff --git a/openhcs/core/function_patterns.py b/openhcs/core/function_patterns.py index 37ca4910d5..4dbfbfeee3 100644 --- a/openhcs/core/function_patterns.py +++ b/openhcs/core/function_patterns.py @@ -10,16 +10,20 @@ from abc import ABC, abstractmethod from collections.abc import Callable, Hashable, Iterator, Mapping, Sequence +from contextlib import AbstractContextManager, nullcontext from dataclasses import dataclass, field, replace +from functools import partial from enum import Enum from pathlib import Path from typing import TYPE_CHECKING, TypeAlias if TYPE_CHECKING: from openhcs.core.aligned_image_payload import AlignedImageSliceContext + from openhcs.core.compiled_step_plan import FrameworkDeviceAssignment from openhcs.core.pipeline.compilation_session import CompilationPathResolver from openhcs.core.steps.function_runtime import FunctionCoreExecutor +from arraybridge import MemoryType from pyqt_reactive.pattern_metadata import PatternScopeToken from python_introspect import Enableable @@ -483,6 +487,43 @@ class CompiledFunctionInvocation(NormalizedFunctionItem): artifact_output_plans: tuple[ArtifactOutputPlan, ...] = () artifact_input_edges: tuple[InvocationArtifactInputEdgePlan, ...] = () runtime_parameter_bindings: tuple[RuntimeParameterBinding, ...] = () + input_device_id: int | None = None + execution_device_id: int | None = None + runtime_callable: Callable = field( + init=False, + repr=False, + compare=False, + ) + runtime_kwargs: RuntimeKwargItems = field( + init=False, + repr=False, + compare=False, + ) + primary_input_parameter_name: str | None = field( + init=False, + repr=False, + compare=False, + ) + adapter_parameter_name: str | None = field( + init=False, + repr=False, + compare=False, + ) + _main_flow_argument: Callable | None = field( + init=False, + repr=False, + compare=False, + ) + _main_flow_output_source: Callable | None = field( + init=False, + repr=False, + compare=False, + ) + _execution_memory_type: MemoryType | None = field( + init=False, + repr=False, + compare=False, + ) def __post_init__(self) -> None: expected_edge_keys = InvocationArtifactInputProjectionKey.for_input_count( @@ -504,6 +545,119 @@ def __post_init__(self) -> None: "Compiled artifact outputs must be ArtifactOutputPlan values." ) + kwargs = dict(self.kwargs) + for binding in self.runtime_parameter_bindings: + kwargs[binding.parameter_name] = binding.value + adapter = self.contract.runtime_adapter + processing = self.contract.processing_contract + declaration = ( + None + if processing is None + else self.contract.require_processing_contract().declaration + ) + object.__setattr__( + self, + "runtime_kwargs", + tuple(kwargs.items()), + ) + object.__setattr__( + self, + "runtime_callable", + self.contract.resolve_runtime_callable(), + ) + object.__setattr__( + self, + "primary_input_parameter_name", + self.contract.primary_input_parameter_name, + ) + object.__setattr__( + self, + "adapter_parameter_name", + None if adapter is None else adapter.parameter_name, + ) + object.__setattr__( + self, + "_main_flow_argument", + ( + None + if declaration is None + else partial(declaration.main_flow_call_argument, self.contract) + ), + ) + object.__setattr__( + self, + "_main_flow_output_source", + ( + None + if declaration is None + else declaration.main_flow_output_source_payload + ), + ) + object.__setattr__( + self, + "_execution_memory_type", + self.contract.require_execution_memory_type(), + ) + + def __reduce__(self) -> tuple[Callable, tuple]: + """Rebind process-local executable state from declaration-only transport.""" + return ( + type(self), + ( + self.key, + self.contract, + self.kwargs, + self.artifact_output_plans, + self.artifact_input_edges, + self.runtime_parameter_bindings, + self.input_device_id, + self.execution_device_id, + ), + ) + + def with_device_assignment( + self, + assignment: FrameworkDeviceAssignment, + ) -> "CompiledFunctionInvocation": + """Bind this immutable invocation at the compiler's placement boundary.""" + if self.contract.execution_scope is FunctionStepExecutionScope.PLATE: + input_device_id = None + else: + input_memory_type, _ = self.contract.require_memory_types() + input_device_id = assignment.device_id_for(MemoryType(input_memory_type)) + return replace( + self, + input_device_id=input_device_id, + execution_device_id=( + None + if self._execution_memory_type is None + else assignment.device_id_for(self._execution_memory_type) + ), + ) + + def main_flow_call_argument(self, source_payload: object) -> object: + """Apply the captured processing declaration to current request pixels.""" + if self._main_flow_argument is None: + self.contract.require_processing_contract() + return self._main_flow_argument(source_payload) + + def main_flow_output_source_payload(self, source_payload: object) -> object: + """Apply the captured output-domain operation at its live callback epoch.""" + if self._main_flow_output_source is None: + self.contract.require_processing_contract() + return self._main_flow_output_source(source_payload) + + def execution_device_scope(self) -> AbstractContextManager[None]: + """Enter the fixed device while retaining a fresh per-call scope lifetime.""" + memory_type = self._execution_memory_type + if memory_type is None or not memory_type.is_gpu: + return nullcontext() + if self.execution_device_id is None: + raise ValueError( + f"No compiled device assignment exists for {memory_type.value!r}." + ) + return memory_type.device_scope(self.execution_device_id) + @property def input_memory_type(self) -> str | None: """Declared input memory type from the callable contract.""" diff --git a/openhcs/core/pipeline/framework_device_assignment.py b/openhcs/core/pipeline/framework_device_assignment.py index afd5e4d79b..21d347ae33 100644 --- a/openhcs/core/pipeline/framework_device_assignment.py +++ b/openhcs/core/pipeline/framework_device_assignment.py @@ -2,6 +2,8 @@ from __future__ import annotations +from dataclasses import replace + from openhcs.core.compiled_step_plan import CompiledStepPlan from openhcs.core.framework_device_resolver import resolve_framework_devices @@ -21,3 +23,20 @@ def assign_framework_devices( step_plan.device_assignment = pipeline_assignment.select( step_plan.gpu_memory_types ) + pattern = step_plan.compiled_function_pattern + if pattern is not None: + step_plan.compiled_function_pattern = replace( + pattern, + groups=tuple( + replace( + group, + invocations=tuple( + invocation.with_device_assignment( + step_plan.device_assignment + ) + for invocation in group.invocations + ), + ) + for group in pattern.groups + ), + ) diff --git a/openhcs/core/pipeline/funcstep_contract_validator.py b/openhcs/core/pipeline/funcstep_contract_validator.py index 1aafedeaf8..0473a00064 100644 --- a/openhcs/core/pipeline/funcstep_contract_validator.py +++ b/openhcs/core/pipeline/funcstep_contract_validator.py @@ -1031,7 +1031,6 @@ def _validate_callable_memory_contract( input_type = contract.input_memory_type output_type = contract.output_memory_type - execution_type = contract.execution_memory_type if input_type is None or output_type is None: raise ValueError( missing_memory_type_error(contract.function_name, step_name) @@ -1048,12 +1047,7 @@ def _validate_callable_memory_contract( ", ".join(sorted(VALID_MEMORY_TYPES)), ) ) - if execution_type is not None and execution_type not in VALID_MEMORY_TYPES: - raise ValueError( - f"Callable {contract.function_name!r} in step {step_name!r} " - f"declares invalid execution memory type {execution_type!r}; " - f"valid memory types are {', '.join(sorted(VALID_MEMORY_TYPES))}." - ) + contract.require_execution_memory_type(step_name=step_name) return input_type, output_type @staticmethod diff --git a/openhcs/core/runtime_output_matching.py b/openhcs/core/runtime_output_matching.py index 7e6dd98b49..1762731254 100644 --- a/openhcs/core/runtime_output_matching.py +++ b/openhcs/core/runtime_output_matching.py @@ -3,25 +3,11 @@ from __future__ import annotations from abc import ABC, abstractmethod -from dataclasses import dataclass from typing import Any, TypeAlias -from openhcs.core.aligned_image_payload import ( - AlignedImageSliceContext, - AlignedImageStack, - pack_aligned_image_outputs, -) from openhcs.core.artifacts import ( ArtifactOutputPlan, ArtifactSpec, - ArtifactSpecRef, -) -from openhcs.core.callable_contract import CallableContract -from openhcs.core.runtime_image_values import image_payload_metadata -from openhcs.core.runtime_plane_projection import RuntimePlaneAxisValueProjection -from openhcs.core.runtime_slice_projection import ( - RuntimeSliceProjection, - RuntimeSliceProjectionDeclarationError, ) @@ -53,176 +39,3 @@ def split_runtime_output(value: Any) -> tuple[Any, tuple[Any, ...]]: raise ValueError("Runtime output tuples cannot be empty.") return positional[0], tuple(positional[1:]) return positional, () - - -@dataclass(frozen=True, slots=True) -class RuntimeReturnedOutputMatcher: - """Resolve one complete callable return against its declared positional ABI.""" - - callable_contract: CallableContract - returned_output: Any - - @property - def canonical_output(self) -> Any: - """Return the unmodified first position of the callable ABI.""" - - return split_runtime_output(self.returned_output)[0] - - def contextualize_canonical_output( - self, - *, - plane_projection: RuntimePlaneAxisValueProjection | None = None, - ) -> Any: - """Attach the compiled canonical ABI to one exact returned image axis.""" - - canonical_specs = self.callable_contract.canonical_return_output_specs.specs - if len(canonical_specs) <= 1: - return self.returned_output - canonical_output, trailing_outputs = split_runtime_output(self.returned_output) - if isinstance(canonical_output, AlignedImageStack): - if canonical_output.slice_contexts: - return self.returned_output - output_values = canonical_output.slices - else: - projection = plane_projection - function_name = self.callable_contract.function_name - if projection is None: - raise RuntimeSliceProjectionDeclarationError( - f"{function_name} declares {len(canonical_specs)} canonical " - "outputs but returned a non-aligned payload without a " - "compiled plane projection." - ) - if projection.plane_index is not None: - raise RuntimeSliceProjectionDeclarationError( - f"{function_name} declares {len(canonical_specs)} canonical " - "outputs after the compiled plane projection already selected " - f"plane {projection.plane_index}." - ) - if projection.axis_size != len(canonical_specs): - raise ValueError( - f"{function_name} declares {len(canonical_specs)} canonical " - "outputs but its compiled plane projection declares " - f"{projection.axis_size} value(s)." - ) - output_axis = image_payload_metadata(canonical_output).plane_axis - if output_axis is not projection.axis: - raise RuntimeSliceProjectionDeclarationError( - f"{function_name} declares {len(canonical_specs)} canonical " - "outputs but its returned payload does not declare the " - f"compiled {projection.axis.value!r} plane axis; got " - f"{output_axis!r}." - ) - output_values = tuple( - RuntimeSliceProjection.value_for_slice( - canonical_output, - projection.selected_plane(output_index), - ) - for output_index in range(projection.axis_size) - ) - if len(output_values) != len(canonical_specs): - raise ValueError( - f"{self.callable_contract.function_name} returned {len(output_values)} " - "canonical output value(s) for " - f"{len(canonical_specs)} compiled output spec(s)." - ) - contextualized_output = pack_aligned_image_outputs( - output_values, - slice_contexts=AlignedImageSliceContext.main_flow_for_artifact_specs( - canonical_specs - ), - ) - return ( - (contextualized_output, *trailing_outputs) - if trailing_outputs - else contextualized_output - ) - - def resolve(self) -> dict[ArtifactSpecRef, Any]: - """Resolve every output in the callable ABI.""" - - output_specs = self.callable_contract.artifact_outputs.specs - self._specs_by_ref( - output_specs, - contract_name="Callable output ABI", - ) - canonical_output, trailing_values = split_runtime_output(self.returned_output) - trailing_specs = self.callable_contract.trailing_return_output_specs.specs - if len(trailing_values) != len(trailing_specs): - raise ValueError( - "Runtime callable trailing return count does not match its " - f"declared trailing output slots: {len(trailing_values)} != " - f"{len(trailing_specs)}." - ) - resolved = { - spec.ref(): value - for spec, value in zip( - trailing_specs, - trailing_values, - strict=True, - ) - } - canonical_specs = self.callable_contract.canonical_return_output_specs.specs - if len(canonical_specs) == 1: - resolved[canonical_specs[0].ref()] = canonical_output - elif canonical_specs: - if not isinstance(canonical_output, AlignedImageStack): - raise TypeError( - "Multiple canonical output specs require an AlignedImageStack with " - "one exact named slice context per output." - ) - resolved.update( - canonical_output.output_values_for_artifact_specs(canonical_specs) - ) - return resolved - - def resolve_plan_values( - self, - selected_output_plans: tuple[ArtifactOutputPlan, ...], - ) -> tuple[ - dict[ArtifactSpecRef, Any], - tuple[RuntimeMatchedOutput, ...], - ]: - """Resolve the complete ABI and bind exact selected runtime plans once.""" - - returned_values = self.resolve() - specs_by_ref = self._specs_by_ref( - self.callable_contract.artifact_outputs.specs, - contract_name="Callable output ABI", - ) - selected_refs: set[ArtifactSpecRef] = set() - matched_outputs: list[RuntimeMatchedOutput] = [] - for plan in selected_output_plans: - if not isinstance(plan, ArtifactOutputPlan): - raise TypeError( - "Selected runtime outputs must be ArtifactOutputPlan values, " - f"got {type(plan).__name__}." - ) - ref = plan.ref() - if ref in selected_refs: - raise ValueError( - f"Selected runtime output plans contain duplicate ref {ref!r}." - ) - selected_refs.add(ref) - spec = specs_by_ref.get(ref) - if spec is None: - raise ValueError( - f"Selected output plan {ref!r} is not declared by the callable ABI." - ) - matched_outputs.append((plan, spec, returned_values[ref])) - return returned_values, tuple(matched_outputs) - - @staticmethod - def _specs_by_ref( - specs: tuple[ArtifactSpec, ...], - *, - contract_name: str, - ) -> dict[ArtifactSpecRef, ArtifactSpec]: - by_ref: dict[ArtifactSpecRef, ArtifactSpec] = {} - for spec in specs: - ref = spec.ref() - if ref in by_ref: - raise ValueError( - f"{contract_name} contains duplicate artifact ref {ref!r}." - ) - by_ref[ref] = spec - return by_ref diff --git a/openhcs/core/steps/function_runtime.py b/openhcs/core/steps/function_runtime.py index 63d2051204..92dd27f71d 100644 --- a/openhcs/core/steps/function_runtime.py +++ b/openhcs/core/steps/function_runtime.py @@ -11,7 +11,6 @@ from pathlib import Path from types import MappingProxyType from typing import ( - Callable, Generic, Mapping, Sequence, @@ -76,7 +75,7 @@ ) from openhcs.core.runtime_artifact_values import RuntimeValue from openhcs.core.runtime_output_matching import ( - RuntimeReturnedOutputMatcher, + split_runtime_output, ) from openhcs.core.runtime_adapters import ( RuntimeAdapterRequest, @@ -954,43 +953,21 @@ def debug_event( invocation_parameters=invocation_parameters, ) - @property - def func_callable(self) -> Callable: - return self.invocation.contract.resolve_runtime_callable() - - @property - def base_kwargs(self) -> RuntimeCallableKwargs: - return self.invocation.kwargs_dict - - def main_flow_output_source_payload( - self, - source_payload: RuntimePayload, - ) -> RuntimePayload: - """Project source context through the callable's nominal processing contract.""" - - return self.invocation.contract.require_processing_contract().declaration.main_flow_output_source_payload( - source_payload - ) - def execute( self, *, debug_sink: DebugEventSink | None = None, ) -> RuntimePayload | NoMainFlowOutput: - input_memory_type, _ = self.invocation.contract.require_memory_types() - target_device_id = self.group_data.execution_plan.device_id_for( - input_memory_type - ) + input_memory_type = self.invocation.input_memory_type converted_data = convert_memory( data=image_payload_data(self.main_data_arg), source_type=self.source_memory_type, target_type=input_memory_type, - gpu_id=target_device_id, + gpu_id=self.invocation.input_device_id, ) source_payload = with_image_payload_data(self.main_data_arg, converted_data) - main_data_arg = self.main_flow_call_argument(source_payload) - final_kwargs = dict(self.base_kwargs) - self.bind_compiled_runtime_parameters(final_kwargs) + main_data_arg = self.invocation.main_flow_call_argument(source_payload) + final_kwargs = dict(self.invocation.runtime_kwargs) loads_artifact_inputs = self.should_load_artifact_inputs() loaded_artifact_payloads: dict[ArtifactSpecRef, RuntimePayload] = {} if loads_artifact_inputs: @@ -1012,13 +989,6 @@ def execute( loaded_artifact_payloads=loaded_artifact_payloads, ) - def main_flow_call_argument( - self, source_payload: RuntimePayload - ) -> RuntimeCallableArgument: - """Project through the callable's declared processing and raw ABI owners.""" - - return self.invocation.contract.main_flow_call_argument(source_payload) - def execution_group_source_payload( self, source_payload: RuntimePayload, @@ -1046,13 +1016,6 @@ def bind_runtime_owned_parameters( if context_parameter_name is not None: final_kwargs[context_parameter_name] = self.group_data.context - def bind_compiled_runtime_parameters( - self, - final_kwargs: dict[str, RuntimeCallableArgument], - ) -> None: - for binding in self.invocation.runtime_parameter_bindings: - final_kwargs[binding.parameter_name] = binding.value - def bind_runtime_adapter( self, final_kwargs: dict[str, RuntimeCallableArgument], @@ -1061,7 +1024,7 @@ def bind_runtime_adapter( runtime_adapter = self.invocation.contract.runtime_adapter if runtime_adapter is None: return - adapter_parameter = runtime_adapter.require_parameter_name() + adapter_parameter = self.invocation.adapter_parameter_name adapter_started_at = time.perf_counter() final_kwargs[adapter_parameter] = runtime_adapter.factory( self.runtime_adapter_request(source_payload) @@ -1083,16 +1046,13 @@ def invoke( debug_sink: DebugEventSink | None, ) -> RuntimeFunctionOutput: logger.info("Executing function: %s", self.function_name) - func_callable = self.func_callable + func_callable = self.invocation.runtime_callable contract = self.invocation.contract - primary_parameter = contract.primary_input_parameter_name - bound_parameters = dict(final_kwargs) - if primary_parameter is not None: - bound_parameters[primary_parameter] = main_data_arg + primary_parameter = self.invocation.primary_input_parameter_name plane_projector: RuntimePlaneAxisProjector = self.plane_projection runtime_adapter = self.invocation.contract.runtime_adapter if runtime_adapter is not None: - adapter_parameter = runtime_adapter.require_parameter_name() + adapter_parameter = self.invocation.adapter_parameter_name adapter_value = final_kwargs[adapter_parameter] if isinstance(adapter_value, RuntimePlaneAxisProjector): plane_projector = adapter_value @@ -1102,6 +1062,8 @@ def invoke( f"Callable {self.function_name!r} has no declared primary input " "parameter for runtime invocation diagnostics." ) + bound_parameters = dict(final_kwargs) + bound_parameters[primary_parameter] = main_data_arg debug_sink.record( self.debug_event( DebugEventType.BEFORE_INVOCATION, @@ -1114,14 +1076,15 @@ def invoke( ) call_started_at = time.perf_counter() try: - with self.group_data.execution_plan.memory_device_scope( - contract.execution_memory_type - ): + with self.invocation.execution_device_scope(): raw_output = func_callable( main_data_arg, **final_kwargs, ) except RuntimeSliceProjectionDeclarationError as exc: + bound_parameters = dict(final_kwargs) + if primary_parameter is not None: + bound_parameters[primary_parameter] = main_data_arg cursor = self.debug_cursor() invocation_parameters = DebugInvocationParameter.from_kwargs( bound_parameters, @@ -1179,12 +1142,11 @@ def save_artifact_outputs( ) main_output = raw_output else: - output_matcher = RuntimeReturnedOutputMatcher( - callable_contract=self.invocation.contract, - returned_output=raw_output, - ) - _returned_values, matched_outputs = output_matcher.resolve_plan_values( - output_plans + _returned_values, matched_outputs = ( + self.invocation.contract.resolve_returned_plan_values( + raw_output, + output_plans, + ) ) saved_values = { output_plan.ref(): self.save_artifact_output( @@ -1196,10 +1158,7 @@ def save_artifact_outputs( ) for output_plan, _output_spec, output_value in matched_outputs } - canonical_refs = frozenset( - spec.ref() - for spec in self.invocation.contract.canonical_return_output_specs - ) + canonical_refs = self.invocation.contract.canonical_return_output_refs main_outputs = tuple( (output_plan, output_spec, saved_values[output_plan.ref()]) for output_plan, output_spec, _output_value in matched_outputs @@ -1223,10 +1182,10 @@ def save_artifact_outputs( ) ), ) - main_output = output_matcher.canonical_output + main_output = split_runtime_output(raw_output)[0] if isinstance(main_output, NoMainFlowOutput): return main_output - output_source_payload = self.main_flow_output_source_payload( + output_source_payload = self.invocation.main_flow_output_source_payload( self.execution_group_source_payload(source_payload) ) return ImageArtifactType.contextualize_output_from_projector( @@ -1271,7 +1230,7 @@ def save_artifact_output( source_payload, loaded_artifact_payloads=loaded_artifact_payloads, ) - output_source_payload = self.main_flow_output_source_payload( + output_source_payload = self.invocation.main_flow_output_source_payload( self.execution_group_source_payload(artifact_source_payload) ) materialization_source_metadata = None diff --git a/openhcs/interop/cellprofiler/runtime/function_contract_execution.py b/openhcs/interop/cellprofiler/runtime/function_contract_execution.py index 199f2dea38..5d744b4bb1 100644 --- a/openhcs/interop/cellprofiler/runtime/function_contract_execution.py +++ b/openhcs/interop/cellprofiler/runtime/function_contract_execution.py @@ -27,7 +27,6 @@ image_payload_mask, image_payload_metadata, ) -from openhcs.core.runtime_output_matching import RuntimeReturnedOutputMatcher from openhcs.core.runtime_plane_projection import ( RuntimePlaneAxis, RuntimePlaneAxisValueProjection, @@ -183,10 +182,10 @@ def execute( function=function_name, mode=mode.value, ) - return RuntimeReturnedOutputMatcher( - callable_contract, + return callable_contract.contextualize_returned_canonical_output( result, - ).contextualize_canonical_output(plane_projection=executor.plane_projection) + plane_projection=executor.plane_projection, + ) def execute_pure_2d_slice_batch( self, diff --git a/openhcs/interop/cellprofiler/runtime/measurement_execution_support.py b/openhcs/interop/cellprofiler/runtime/measurement_execution_support.py index 61c4adfc56..3dd2d508e4 100644 --- a/openhcs/interop/cellprofiler/runtime/measurement_execution_support.py +++ b/openhcs/interop/cellprofiler/runtime/measurement_execution_support.py @@ -39,7 +39,6 @@ from openhcs.interop.cellprofiler.runtime.measurement_rows import ( measurement_table_rows, ) -from openhcs.core.runtime_output_matching import RuntimeReturnedOutputMatcher from openhcs.core.steps.function_runtime import ( RuntimeCallableArgument, RuntimeFunctionOutput, @@ -377,10 +376,12 @@ def record( prepared_invocation: PreparedObjectMeasurementInvocation, ) -> None: split_started_at = time.perf_counter() - _returned_values, matched_outputs = RuntimeReturnedOutputMatcher( - callable_contract=self.callable_contract, - returned_output=raw_output, - ).resolve_plan_values((self.measurement_output_plan,)) + _returned_values, matched_outputs = ( + self.callable_contract.resolve_returned_plan_values( + raw_output, + (self.measurement_output_plan,), + ) + ) self.split_seconds += time.perf_counter() - split_started_at _output_plan, _output_spec, output_value = matched_outputs[0] emitted_measurement_rows = measurement_table_rows(output_value) diff --git a/openhcs/interop/cellprofiler/runtime/module_execution.py b/openhcs/interop/cellprofiler/runtime/module_execution.py index e3ea0f9637..9196ae5262 100644 --- a/openhcs/interop/cellprofiler/runtime/module_execution.py +++ b/openhcs/interop/cellprofiler/runtime/module_execution.py @@ -52,7 +52,6 @@ ) from openhcs.core.runtime_output_matching import ( RuntimeMatchedOutput, - RuntimeReturnedOutputMatcher, ) from openhcs.core.runtime_profile import RuntimeProfileTimer from openhcs.core.runtime_measurements import ( @@ -485,10 +484,10 @@ def _returned_output_values( ]: """Resolve the callable ABI against this invocation's selected plans.""" - return RuntimeReturnedOutputMatcher( - callable_contract=self.callable_contract, - returned_output=raw_output, - ).resolve_plan_values(tuple(adapter.request.artifact_outputs.values())) + return self.callable_contract.resolve_returned_plan_values( + raw_output, + tuple(adapter.request.artifact_outputs.values()), + ) def _replacement_main_flow_output( self, diff --git a/tests/unit/test_count_cells_simple.py b/tests/unit/test_count_cells_simple.py index e705888be0..088c28fa23 100644 --- a/tests/unit/test_count_cells_simple.py +++ b/tests/unit/test_count_cells_simple.py @@ -13,7 +13,6 @@ ObjectLabelsArtifactType, ) from openhcs.core.callable_contract import CallableContract -from openhcs.core.runtime_output_matching import RuntimeReturnedOutputMatcher from openhcs.processing.backends.analysis.count_cells_simple import ( Foreground, MetaXpressW2Settings, @@ -592,7 +591,7 @@ def test_round_object_diagnostics_preserve_kernel_output_and_rejection_identity( ) raw, measurements, prefilter, accepted, weak_core, adjacent_satellites = returned contract = CallableContract.from_callable(inspect_metaxpress_round_objects) - matched = RuntimeReturnedOutputMatcher(contract, returned).resolve() + matched = contract.resolve_returned_output(returned) assert ( matched[count_cells_simple_module.ROUND_OBJECT_WIDTHS_OUTPUT.ref()] is measurements diff --git a/tests/unit/test_framework_execution_scope.py b/tests/unit/test_framework_execution_scope.py index 56227eef55..228c15965c 100644 --- a/tests/unit/test_framework_execution_scope.py +++ b/tests/unit/test_framework_execution_scope.py @@ -3,6 +3,8 @@ import numpy as np +from arraybridge import MemoryType +from openhcs.core.compiled_step_plan import FrameworkDeviceAssignment from openhcs.core.function_patterns import compile_function_pattern from openhcs.core.runtime_plane_projection import RuntimePlaneProjection from openhcs.core.steps.function_runtime import ( @@ -11,7 +13,9 @@ ) -def test_function_invocation_enters_declared_execution_memory_scope() -> None: +def test_function_invocation_enters_declared_execution_memory_scope( + monkeypatch, +) -> None: events = [] def process(image): @@ -21,20 +25,23 @@ def process(image): process.input_memory_type = "numpy" process.output_memory_type = "numpy" process.execution_memory_type = "torch" - invocation = next(compile_function_pattern(process, {}, {}).iter_invocations()) + invocation = next( + compile_function_pattern(process, {}, {}).iter_invocations() + ).with_device_assignment( + FrameworkDeviceAssignment.from_mapping({MemoryType.TORCH: 4}) + ) @contextmanager - def memory_device_scope(declaration): - events.append(("enter", declaration)) + def memory_device_scope(memory_type, device_id): + assert device_id == 4 + events.append(("enter", memory_type.value)) yield - events.append(("exit", declaration)) + events.append(("exit", memory_type.value)) + + monkeypatch.setattr(MemoryType, "device_scope", memory_device_scope) executor = FunctionCoreExecutor( - group_data=SimpleNamespace( - execution_plan=SimpleNamespace( - memory_device_scope=memory_device_scope, - ) - ), + group_data=SimpleNamespace(), invocation=invocation, artifacts=ComponentArtifactPlans(inputs={}, outputs={}), group_key=None, diff --git a/tests/unit/test_function_patterns.py b/tests/unit/test_function_patterns.py index f3dc771ba0..35c3c5a481 100644 --- a/tests/unit/test_function_patterns.py +++ b/tests/unit/test_function_patterns.py @@ -949,6 +949,27 @@ def consume(*, payload): compiled = compile_function_pattern(consume, {}, {}) assert compiled.default_group.main_flow_input_refs == () + from openhcs.core.compiled_step_plan import CompiledStepPlan + from openhcs.core.pipeline.framework_device_assignment import ( + assign_framework_devices, + ) + + plan = CompiledStepPlan( + step_index=0, + step_name="consume", + step_type="FunctionStep", + axis_id="A01", + compiled_function_pattern=compiled, + ) + assign_framework_devices({0: plan}) + (placed,) = tuple(plan.compiled_function_pattern.iter_invocations()) + assert placed.input_memory_type is None + assert placed.output_memory_type is None + assert placed.input_device_id is None + assert ( + placed.artifact_input_edges + == compiled.default_group.invocations[0].artifact_input_edges + ) def test_special_input_edges_use_nominal_artifact_payload_types() -> None: @@ -1691,6 +1712,14 @@ def accepts_dtype_config(image, *, dtype_config=None, sigma=None): binding = invocation.runtime_parameter_bindings[0] assert binding.parameter_name == DtypeConversionConfig.require_parameter_name() assert binding.value is explicit_config + assert dict(invocation.runtime_kwargs) == { + "sigma": 2, + "dtype_config": explicit_config, + } + first_call_kwargs = dict(invocation.runtime_kwargs) + first_call_kwargs["sigma"] = 3 + assert dict(invocation.runtime_kwargs)["sigma"] == 2 + assert invocation.kwargs_dict == {"sigma": 2} def test_compile_function_pattern_keeps_undeclared_runtime_config_kwargs_user_owned(): diff --git a/tests/unit/test_function_step_execution_plan.py b/tests/unit/test_function_step_execution_plan.py index 07c00a468a..2e5a5604f1 100644 --- a/tests/unit/test_function_step_execution_plan.py +++ b/tests/unit/test_function_step_execution_plan.py @@ -35,7 +35,6 @@ from openhcs.core.pipeline.function_contracts import artifact_inputs, artifact_outputs from openhcs.core.artifact_key_selection import AdapterRecordedArtifactOutputPolicy from openhcs.core.runtime_adapters import RuntimeAdapterRequest, runtime_adapter -from openhcs.core.runtime_output_matching import RuntimeReturnedOutputMatcher from openhcs.core.step_dependencies import ( StepInputDependency, StepInputDependencyKind, @@ -154,7 +153,7 @@ def test_compiled_step_plan_owns_gpu_memory_classification() -> None: assert compiled_plan.device_id_for("cupy") == 2 -def test_compiled_step_plan_includes_invocation_execution_memory() -> None: +def test_compiled_step_plan_includes_invocation_execution_memory(monkeypatch) -> None: def numpy_boundary_with_torch_execution(image): return image @@ -172,6 +171,31 @@ def numpy_boundary_with_torch_execution(image): assert compiled_plan.gpu_memory_types == frozenset({MemoryType.TORCH}) assert compiled_plan.device_id_for("torch") == 4 + from contextlib import contextmanager + from openhcs.core.pipeline.framework_device_assignment import ( + assign_framework_devices, + ) + + monkeypatch.setattr( + "openhcs.core.pipeline.framework_device_assignment.resolve_framework_devices", + lambda _required: compiled_plan.device_assignment, + ) + assign_framework_devices({0: compiled_plan}) + (placed_invocation,) = tuple( + compiled_plan.compiled_function_pattern.iter_invocations() + ) + assert placed_invocation.input_device_id is None + assert placed_invocation.execution_device_id == 4 + entered = [] + + @contextmanager + def recorded_scope(memory_type, device_id): + entered.append((memory_type, device_id)) + yield + + monkeypatch.setattr(MemoryType, "device_scope", recorded_scope) + with placed_invocation.execution_device_scope(): + assert entered == [(MemoryType.TORCH, 4)] def test_compiled_step_plan_includes_intermediate_invocation_memory() -> None: @@ -549,10 +573,9 @@ def publish_shared_outputs(image, *, runtime): ) image_value = object() labels_value = object() - _returned, matched = RuntimeReturnedOutputMatcher( - callable_contract=invocation.contract, - returned_output=(image_value, labels_value), - ).resolve_plan_values(tuple(selected.values())) + _returned, matched = invocation.contract.resolve_returned_plan_values( + (image_value, labels_value), tuple(selected.values()) + ) assert selected == { image_spec.ref(): image_plan, diff --git a/tests/unit/test_function_step_transport.py b/tests/unit/test_function_step_transport.py index 47db0599aa..fb5facfc8a 100644 --- a/tests/unit/test_function_step_transport.py +++ b/tests/unit/test_function_step_transport.py @@ -499,6 +499,20 @@ def test_bundle_transport_preserves_prepared_runtime_contract_identity() -> None assert restored_item.contract.metadata.prepare is None assert restored_item.func.metadata.prepare is None assert restored_item.func.resolve() is reference.resolve() + (restored_invocation,) = tuple( + restored_plan.compiled_function_pattern.iter_invocations() + ) + assert ( + restored_invocation.runtime_callable + is restored_invocation.contract.resolve_runtime_callable() + ) + assert restored_invocation.contract is not transport_invocation.contract + assert ( + CallableContractRuntimeCache.process_cache().get_bound( + restored_invocation.contract + ) + is restored_invocation.runtime_callable + ) def test_generated_source_resolves_catalog_owned_cellprofiler_callable() -> None: diff --git a/tests/unit/test_multi_template_matching.py b/tests/unit/test_multi_template_matching.py index 08a1b22150..5572c5baad 100644 --- a/tests/unit/test_multi_template_matching.py +++ b/tests/unit/test_multi_template_matching.py @@ -18,7 +18,6 @@ from openhcs.core.pipeline.materialization_flag_planner import ( MaterializationFlagPlanner, ) -from openhcs.core.runtime_output_matching import RuntimeReturnedOutputMatcher from openhcs.processing.backends.analysis import multi_template_matching from openhcs.processing.backends.analysis.multi_template_matching import ( OpenCVTemplateMatchMethod, @@ -148,7 +147,7 @@ def test_template_crop_declares_changed_image_flow_and_materialization( [image_spec] = contract.main_flow_outputs assert image_spec.artifact_type is ImageArtifactType assert not image_spec.materialization.uses_source_identity_filename() - matched = RuntimeReturnedOutputMatcher(contract, result).resolve() + matched = contract.resolve_returned_output(result) np.testing.assert_array_equal(matched[image_spec.ref()], template[None, ...]) plans = { spec.ref(): ArtifactOutputPlan( diff --git a/tests/unit/test_registered_object_output_boundaries.py b/tests/unit/test_registered_object_output_boundaries.py index 9af17fec67..03a1c38ebf 100644 --- a/tests/unit/test_registered_object_output_boundaries.py +++ b/tests/unit/test_registered_object_output_boundaries.py @@ -24,7 +24,6 @@ ObjectLabelPayload, ObjectLabelVariantData, ) -from openhcs.core.runtime_output_matching import RuntimeReturnedOutputMatcher from openhcs.core.runtime_plane_projection import RuntimePlaneProjection from openhcs.core.runtime_tabular_values import MeasurementObjectRowIdentity from openhcs.interop.cellprofiler.measurement_dialect import ( @@ -188,12 +187,12 @@ def test_optional_removed_declaration_validates_and_matches_exact_runtime_slots( emit_removed_objects=removed, ) assert len(result) == (6 if removed else 4) - resolved = RuntimeReturnedOutputMatcher(contract, result).resolve() + resolved = contract.resolve_returned_output(result) assert len(resolved) == len(contract.artifact_outputs) assert result[-1].source_ids == (() if empty else ((2,) if removed else (7,))) assert result[-1].target_ids == (() if empty else (1,)) with pytest.raises(ValueError, match="trailing return count"): - RuntimeReturnedOutputMatcher(contract, result[:-1]).resolve() + contract.resolve_returned_output(result[:-1]) def test_optional_return_annotation_does_not_weaken_object_identity_validation(): diff --git a/tests/unit/test_returned_output_abi.py b/tests/unit/test_returned_output_abi.py index 7cac5c73e6..c4b9df0805 100644 --- a/tests/unit/test_returned_output_abi.py +++ b/tests/unit/test_returned_output_abi.py @@ -24,8 +24,8 @@ from openhcs.core.runtime_image_values import image_payload_data from openhcs.core.runtime_object_labels import ObjectLabelValue from openhcs.core.runtime_output_matching import ( + split_runtime_output, RuntimeOutputBundle, - RuntimeReturnedOutputMatcher, ) from openhcs.core.runtime_tabular_values import ColumnarRows from openhcs.interop.cellprofiler.module_declarations import CellProfilerModule @@ -171,10 +171,7 @@ def process(value: np.ndarray) -> AlignedImageStack: execution_mode=ImagePayloadExecutionMode.NATURAL, ) - resolved = RuntimeReturnedOutputMatcher( - callable_contract=contract, - returned_output=returned, - ).resolve() + resolved = contract.resolve_returned_output(returned) assert tuple(resolved) == (first.ref(), second.ref()) np.testing.assert_array_equal(resolved[first.ref()], returned.slices[0]) @@ -205,13 +202,10 @@ def test_output_matcher_does_not_duplicate_named_canonical_in_trailing_slot() -> artifact_outputs(image, measurements)(lambda value: value) ) - matcher = RuntimeReturnedOutputMatcher( - callable_contract=contract, - returned_output=("image", "measurements"), - ) + matcher_returned_output = ("image", "measurements") - assert matcher.canonical_output == "image" - assert matcher.resolve() == { + assert split_runtime_output(matcher_returned_output)[0] == "image" + assert contract.resolve_returned_output(matcher_returned_output) == { image.ref(): "image", measurements.ref(): "measurements", } diff --git a/tests/unit/test_runtime_output_matching.py b/tests/unit/test_runtime_output_matching.py index bb5cf64e9a..1670bfdf80 100644 --- a/tests/unit/test_runtime_output_matching.py +++ b/tests/unit/test_runtime_output_matching.py @@ -20,7 +20,6 @@ CallableMetadata, FunctionStepExecutionScope, ) -from openhcs.core.runtime_output_matching import RuntimeReturnedOutputMatcher from openhcs.core.runtime_image_values import ImagePayloadMetadata, image_payload_data from openhcs.core.runtime_plane_projection import ( RuntimePlaneAxis, @@ -54,10 +53,9 @@ def test_runtime_output_matcher_maps_canonical_and_trailing_slots() -> None: image = ArtifactSpec.output("Image", ImageArtifactType) measurements = ArtifactSpec.output("Measurements", MeasurementsArtifactType) - resolved = RuntimeReturnedOutputMatcher( - callable_contract=_contract(image, measurements), - returned_output=("image", "measurements"), - ).resolve() + resolved = _contract(image, measurements).resolve_returned_output( + ("image", "measurements") + ) assert resolved == { image.ref(): "image", @@ -75,14 +73,16 @@ def test_matcher_contextualizes_declared_axis_before_resolving_complete_abi() -> ).payload_with(data, None) trailing = object() contract = _contract(first, second, measurements) - matcher = RuntimeReturnedOutputMatcher(contract, (payload, trailing)) + matcher_returned_output = (payload, trailing) projection = RuntimePlaneAxisValueProjection.preserve( axis=RuntimePlaneAxis.RUNTIME_SLICE, axis_size=2, ) - returned = matcher.contextualize_canonical_output(plane_projection=projection) - resolved = RuntimeReturnedOutputMatcher(contract, returned).resolve() + returned = contract.contextualize_returned_canonical_output( + matcher_returned_output, plane_projection=projection + ) + resolved = contract.resolve_returned_output(returned) assert returned[1] is trailing assert resolved[measurements.ref()] is trailing @@ -90,30 +90,29 @@ def test_matcher_contextualizes_declared_axis_before_resolving_complete_abi() -> selected = image_payload_data(resolved[spec.ref()]) np.testing.assert_array_equal(selected, data[index]) assert np.shares_memory(selected, data) - assert ( - RuntimeReturnedOutputMatcher(contract, returned).contextualize_canonical_output() - is returned - ) + assert contract.contextualize_returned_canonical_output(returned) is returned with pytest.raises(RuntimeSliceProjectionDeclarationError, match="without a compiled"): - matcher.contextualize_canonical_output() + contract.contextualize_returned_canonical_output(matcher_returned_output) with pytest.raises(RuntimeSliceProjectionDeclarationError, match="already selected"): - matcher.contextualize_canonical_output( - plane_projection=projection.selected_plane(0) + contract.contextualize_returned_canonical_output( + matcher_returned_output, plane_projection=projection.selected_plane(0) ) with pytest.raises(ValueError, match="projection declares 3 value"): - matcher.contextualize_canonical_output( + contract.contextualize_returned_canonical_output( + matcher_returned_output, plane_projection=RuntimePlaneAxisValueProjection.preserve( axis=RuntimePlaneAxis.RUNTIME_SLICE, axis_size=3, - ) + ), ) with pytest.raises(RuntimeSliceProjectionDeclarationError, match="plane axis"): - matcher.contextualize_canonical_output( + contract.contextualize_returned_canonical_output( + matcher_returned_output, plane_projection=RuntimePlaneAxisValueProjection.preserve( axis=RuntimePlaneAxis.SOURCE_BINDING, axis_size=2, - ) + ), ) @@ -135,10 +134,9 @@ def test_runtime_output_matcher_uses_exact_multi_canonical_contexts() -> None: ), ) - resolved = RuntimeReturnedOutputMatcher( - callable_contract=_contract(first, second, measurements), - returned_output=(canonical, "measurements"), - ).resolve() + resolved = _contract(first, second, measurements).resolve_returned_output( + (canonical, "measurements") + ) assert resolved == { first.ref(): "first-value", @@ -158,10 +156,9 @@ def test_runtime_output_matcher_binds_selected_plans_after_resolving_complete_ab artifact_type=measurements.artifact_type, ) - resolved, matched_outputs = RuntimeReturnedOutputMatcher( - callable_contract=_contract(image, measurements), - returned_output=("image", "measurements"), - ).resolve_plan_values((measurement_plan,)) + resolved, matched_outputs = _contract( + image, measurements + ).resolve_returned_plan_values(("image", "measurements"), (measurement_plan,)) assert resolved == { image.ref(): "image", @@ -211,10 +208,9 @@ def test_runtime_invocation_selects_storage_without_truncating_callable_abi() -> ), ) - resolved, matched = RuntimeReturnedOutputMatcher( - callable_contract=compiled.contract, - returned_output=returned_stack, - ).resolve_plan_values(runtime_output_plans) + resolved, matched = compiled.contract.resolve_returned_plan_values( + returned_stack, runtime_output_plans + ) assert compiled.contract.canonical_return_output_specs.specs == (first, second) assert compiled.contract.artifact_inputs.specs == (first_input, second_input) @@ -239,20 +235,14 @@ def test_runtime_output_matcher_rejects_trailing_slot_count_mismatch( ValueError, match=rf"declared trailing output slots: {expected_count} != 1", ): - RuntimeReturnedOutputMatcher( - callable_contract=_contract(measurements), - returned_output=returned_output, - ).resolve() + _contract(measurements).resolve_returned_output(returned_output) def test_runtime_output_matcher_rejects_duplicate_abi_specs() -> None: objects = ArtifactSpec.output("Objects", ObjectLabelsArtifactType) with pytest.raises(ValueError, match="duplicate artifact ref"): - RuntimeReturnedOutputMatcher( - callable_contract=_contract(objects, objects), - returned_output="objects", - ).resolve() + _contract(objects, objects).resolve_returned_output("objects") def test_runtime_output_matcher_rejects_selected_plan_not_in_abi() -> None: @@ -265,10 +255,7 @@ def test_runtime_output_matcher_rejects_selected_plan_not_in_abi() -> None: ) with pytest.raises(ValueError, match="plan .* is not declared by the callable ABI"): - RuntimeReturnedOutputMatcher( - callable_contract=_contract(declared), - returned_output="declared", - ).resolve_plan_values((undeclared_plan,)) + _contract(declared).resolve_returned_plan_values("declared", (undeclared_plan,)) @pytest.mark.parametrize( @@ -338,10 +325,9 @@ def test_runtime_output_matcher_rejects_context_free_multi_canonical_stack() -> second = ArtifactSpec.output("Second", ImageArtifactType) with pytest.raises(ValueError, match="require exact AlignedImageStack"): - RuntimeReturnedOutputMatcher( - callable_contract=_contract(first, second), - returned_output=AlignedImageStack(("first", "second")), - ).resolve() + _contract(first, second).resolve_returned_output( + AlignedImageStack(("first", "second")) + ) def test_plate_scope_uses_first_declared_output_as_canonical() -> None: @@ -355,10 +341,7 @@ def test_plate_scope_uses_first_declared_output_as_canonical() -> None: assert contract.canonical_return_output_specs.specs == (measurements,) assert contract.trailing_return_output_specs.specs == (image,) - assert RuntimeReturnedOutputMatcher( - callable_contract=contract, - returned_output=("measurements", "image"), - ).resolve() == { + assert contract.resolve_returned_output(("measurements", "image")) == { measurements.ref(): "measurements", image.ref(): "image", } From 3cf75634e0179b7647a61ac14ff6ef8851559e7e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 4 Oct 2026 14:09:06 -0400 Subject: [PATCH 07/10] perf: admit immutable source identity joins once Reuse the existing runtime resolution snapshot for exact path and correlated component matching. Keep live resolver contexts on their original read path and retain source-position ambiguity and ordering. Refs #648 --- openhcs/core/runtime_source_binding_cache.py | 104 +++++++++++++++++- openhcs/core/source_binding_selection.py | 12 ++ ...test_runtime_source_resolution_snapshot.py | 36 +++++- 3 files changed, 145 insertions(+), 7 deletions(-) diff --git a/openhcs/core/runtime_source_binding_cache.py b/openhcs/core/runtime_source_binding_cache.py index 9e6a311a47..48167d65d7 100644 --- a/openhcs/core/runtime_source_binding_cache.py +++ b/openhcs/core/runtime_source_binding_cache.py @@ -2,7 +2,7 @@ from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Mapping, Sequence from dataclasses import dataclass, field, replace from types import MappingProxyType from typing import Any, Hashable, TYPE_CHECKING @@ -15,12 +15,17 @@ ResolvedSourceMetadataRecord, SourceMetadataMapping, ) +from openhcs.core.source_matching import source_component_metadata_items +from openhcs.core.source_path_identity import source_path_identity_key if TYPE_CHECKING: from openhcs.core.source_binding_selection import ( + SourceCandidatePathResolution, SourcePatternResolutionContext, SourceUniverseRuntimeState, ) + from openhcs.core.source_image_provenance import SourceImageIdentity + from openhcs.constants.constants import AllComponents from openhcs.core.source_workspace_projection import ( VirtualWorkspaceSourceProjection, ) @@ -33,6 +38,75 @@ class RuntimeSourceResolutionSnapshot: projection: "VirtualWorkspaceSourceProjection" context: "SourcePatternResolutionContext" + path_resolutions: Mapping[str, "SourceCandidatePathResolution"] + component_values: Mapping[str, tuple[Mapping["AllComponents", str], ...]] + + def context_for_matching(self) -> "SourcePatternResolutionContext": + """Expose this admitted epoch through the existing matching context.""" + return replace(self.context, resolution_snapshot=self) + + def matching_candidates_for_source_identities( + self, + identities: Sequence["SourceImageIdentity"], + candidates: Sequence[str], + ) -> tuple[tuple[str, ...], ...]: + """Join captured paths and correlated component records in query order.""" + positions = self.context.declared_positions_for_candidates(candidates) + if any(position not in self.path_resolutions for position in positions): + return self.context.matching_candidates_for_source_identities( + identities, candidates, + ) + positions_by_path: dict[str, set[str]] = {} + for position in positions: + resolution = self.path_resolutions[position] + paths = ( + resolution.mapped_source_paths[0] + if resolution.mapped_source_paths else position, + *(resolution.virtual_paths or (position,)), + ) + for path in paths: + positions_by_path.setdefault(source_path_identity_key(path), set()).add(position) + + queries = tuple( + source_component_metadata_items(identity.component_metadata or {}) + for identity in identities + ) + component_indexes = {} + for query in queries: + components = tuple(component for component, _ in query) + if not components or components in component_indexes: + continue + index: dict[tuple[str, ...], set[str]] = {} + for position in positions: + for record in self.component_values[position]: + if all(component in record for component in components): + key = tuple(record[component] for component in components) + index.setdefault(key, set()).add(position) + component_indexes[components] = index + + matches = [] + for identity, query in zip(identities, queries, strict=True): + if not identity.addressable: + matches.append(()) + continue + path_positions = ( + None if identity.path is None else positions_by_path.get( + source_path_identity_key(identity.path), set(), + ) + ) + components = tuple(component for component, _ in query) + component_positions = ( + component_indexes[components].get( + tuple(str(value) for _, value in query), set(), + ) if components else None + ) + matches.append(tuple( + position for position in positions + if (path_positions is None or position in path_positions) + and (component_positions is None or position in component_positions) + and (components or identity.path is not None) + )) + return tuple(matches) @classmethod def from_projection( @@ -45,10 +119,10 @@ def from_projection( """Resolve declared positions into one independently owned runtime view.""" from openhcs.core.source_binding_selection import ( DeclaredSourceMetadataRecord, - SourcePatternResolutionContext, + SourceIdentityResolutionContext, ) - context = SourcePatternResolutionContext.from_projection( + context = SourceIdentityResolutionContext.from_projection( parser=parser, projection=projection, metadata_rules=metadata_rules, @@ -82,7 +156,25 @@ def from_projection( {} if metadata is None else metadata ) context = replace(context, source_metadata_by_path=MappingProxyType(records)) - return cls(projection=projection, context=context) + resolutions = { + path: context._candidate_path_resolution(path) + for path in context.source_paths_by_virtual_path + } + components = { + path: tuple( + MappingProxyType({ + component: str(value) + for component, value in source_component_metadata_items(metadata) + }) + for metadata in context.metadata_for_paths(resolution.metadata_paths()) + ) + for path, resolution in resolutions.items() + } + return cls( + projection=projection, context=context, + path_resolutions=MappingProxyType(resolutions), + component_values=MappingProxyType(components), + ) @dataclass(slots=True) @@ -113,14 +205,14 @@ def source_pattern_context( key = (id(projection), id(parser), parser.semantic_identity(), metadata_rules) cached = self.source_resolution_snapshots.get(key) if cached is not None: - return cached.context + return cached.context_for_matching() snapshot = RuntimeSourceResolutionSnapshot.from_projection( parser=parser, projection=projection, metadata_rules=metadata_rules, ) self.source_resolution_snapshots[key] = snapshot - return snapshot.context + return snapshot.context_for_matching() def __reduce__(self) -> tuple[type[RuntimeSourceBindingContextCache], tuple[()]]: """Transport reconstructs all derived caches from declaration defaults.""" diff --git a/openhcs/core/source_binding_selection.py b/openhcs/core/source_binding_selection.py index 54332fb687..92e33d50b4 100644 --- a/openhcs/core/source_binding_selection.py +++ b/openhcs/core/source_binding_selection.py @@ -72,6 +72,7 @@ if TYPE_CHECKING: from openhcs.core.runtime_adapters import RuntimeAdapterRequest + from openhcs.core.runtime_source_binding_cache import RuntimeSourceResolutionSnapshot from polystore.filemanager import FileManager from openhcs.core.context.processing_context import ProcessingContext from openhcs.microscopes.microscope_interfaces import FilenameParser @@ -191,6 +192,7 @@ class SourcePatternResolutionContext: source_projections_by_virtual_path: Mapping[str, SourceProjection] = field( default_factory=dict ) + resolution_snapshot: RuntimeSourceResolutionSnapshot | None = None @classmethod def from_sources( @@ -272,6 +274,10 @@ def _candidate_path_resolution( self, pattern: SourceCandidatePath, ) -> SourceCandidatePathResolution: + if self.resolution_snapshot is not None: + admitted = self.resolution_snapshot.path_resolutions.get(pattern) + if admitted is not None: + return admitted keys = _cached_source_candidate_pattern_keys(pattern) exact_virtual_path = next( (key for key in keys if key in self.source_paths_by_virtual_path), @@ -1124,6 +1130,11 @@ def matching_candidates_for_source_identities( ) -> tuple[tuple[SourceCandidatePath, ...], ...]: """Resolve a batch without rescanning unrelated paths for each identity.""" + if self.resolution_snapshot is not None: + return self.resolution_snapshot.matching_candidates_for_source_identities( + identities, candidates, + ) + candidates = self.declared_positions_for_candidates(candidates) candidates_by_path: dict[str, list[SourceCandidatePath]] = {} for candidate in candidates: @@ -1186,6 +1197,7 @@ def from_plan( source_projections_by_virtual_path=( source_context.source_projections_by_virtual_path ), + resolution_snapshot=source_context.resolution_snapshot, bindings=tuple(bindings), match_plan=match_plan, identity_policy=identity_policy, diff --git a/tests/unit/test_runtime_source_resolution_snapshot.py b/tests/unit/test_runtime_source_resolution_snapshot.py index eaa1322d23..1e08738376 100644 --- a/tests/unit/test_runtime_source_resolution_snapshot.py +++ b/tests/unit/test_runtime_source_resolution_snapshot.py @@ -19,6 +19,7 @@ MetadataSource, ) from openhcs.core.source_matching import SourceImageSetIdentityPolicy +from openhcs.core.source_image_provenance import SourceImageIdentity from openhcs.core.source_metadata import ( ORIGINAL_SOURCE_METADATA_FIELD, ResolvedSourceMetadataRecord, @@ -104,12 +105,45 @@ def test_runtime_snapshot_owns_nested_metadata_and_unknown_fallback(): unknown = "A02_s001_w1_z001_t001.tif" assert context.metadata_for_path(unknown)["well"] == "A02" assert parser.calls[-1] == unknown - assert snapshot(cache, source_projection, parser) is context + assert ( + snapshot(cache, source_projection, parser).resolution_snapshot + is context.resolution_snapshot + ) entry = next(iter(cache.source_resolution_snapshots.values())) assert entry.projection is source_projection assert entry.context.parser is parser +def test_snapshot_identity_join_preserves_record_correlations_and_query_order(): + second = "A01_s001_w1_z002_t001.tif" + source_projection = VirtualWorkspaceSourceProjection( + source_refs_by_virtual_path={ + PATH: SourcePixelRef("disk", "/source/shared.tif"), + second: SourcePixelRef("disk", "/source/shared.tif"), + }, + source_metadata_by_path={ + PATH: {"channel": "1", "z_index": "1"}, + second: {"channel": "1", "z_index": "2"}, + "/source/shared.tif": {"channel": "2", "z_index": "2"}, + }, + ) + context = snapshot(RuntimeSourceBindingContextCache(), source_projection, CountingParser()) + identities = ( + SourceImageIdentity("/source/shared.tif", {"channel": "1", "z_index": "2"}), + SourceImageIdentity("/source/shared.tif"), + SourceImageIdentity("/source/shared.tif", {"channel": "2", "z_index": "1"}), + SourceImageIdentity(None, {"channel": "2", "z_index": "2"}), + SourceImageIdentity(), + ) + assert context.matching_candidates_for_source_identities( + identities, (second, PATH, second), + ) == ((second,), (second, PATH), (), (second, PATH), ()) + unknown = "A02_s001_w1_z001_t001.tif" + assert context.matching_candidates_for_source_identities( + (SourceImageIdentity(unknown),), (unknown,), + ) == ((unknown,),) + + def test_direct_projection_context_remains_live(): nested = {"Plate": "before"} parser = CountingParser() From b35e3bc41ac28a47414a45ac75a281dc1d514feb Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 4 Oct 2026 14:15:20 -0400 Subject: [PATCH 08/10] test: exercise compiled invocation transport rebinding --- tests/unit/test_function_step_transport.py | 27 ++++++++++++++++++---- 1 file changed, 22 insertions(+), 5 deletions(-) diff --git a/tests/unit/test_function_step_transport.py b/tests/unit/test_function_step_transport.py index fb5facfc8a..0257d3067c 100644 --- a/tests/unit/test_function_step_transport.py +++ b/tests/unit/test_function_step_transport.py @@ -23,6 +23,7 @@ PipelineConfig, ) from openhcs.core.callable_contract import CallableContract +from openhcs.core.function_patterns import CompiledFunctionInvocation, FunctionInvocationKey from openhcs.core.function_reference import ( FunctionReference, FunctionReferenceTransportAuthority, @@ -46,14 +47,16 @@ ) -def _execute_spawned_custom_contract( - contract: CallableContract, +def _execute_spawned_custom_invocation( + invocation: CompiledFunctionInvocation, image_values: list[list[int]], ) -> list[list[int]]: """Resolve one persisted callable as a fresh spawned worker would.""" import numpy as np - result = contract.resolve_runtime_callable()(np.asarray(image_values), offset=3) + result = invocation.runtime_callable( + np.asarray(image_values), **dict(invocation.runtime_kwargs) + ) return result.tolist() @@ -576,6 +579,15 @@ def codex_pickle_probe(image): ) assert restored_step.func is custom_functions.codex_pickle_probe + contract = CallableContract.from_callable(reference) + invocation = CompiledFunctionInvocation( + key=FunctionInvocationKey.from_contract(contract, "default", 0), + contract=contract, + ) + restored_invocation = pickle.loads(pickle.dumps(invocation)) + assert restored_invocation.runtime_callable is custom_func + assert restored_invocation.contract is not contract + assert restored_invocation.artifact_input_edges == invocation.artifact_input_edges assert ( CellProfilerModule.require_module("Crop").require_callable() is canonical_crop @@ -646,6 +658,11 @@ def {func_name}(image, offset=0): reference = compiler_pipeline[0].func assert isinstance(reference, FunctionReference) contract = CallableContract.from_callable(reference) + invocation = CompiledFunctionInvocation( + key=FunctionInvocationKey.from_contract(contract, "default", 0), + contract=contract, + kwargs=(("offset", 3),), + ) multiprocessing_context = multiprocessing.get_context("spawn") try: @@ -655,8 +672,8 @@ def {func_name}(image, offset=0): ) as executor: futures = tuple( executor.submit( - _execute_spawned_custom_contract, - contract, + _execute_spawned_custom_invocation, + invocation, [[task_index]], ) for task_index in (1, 2) From 09b353efa16393944c6d77faacd10ef49d08f97f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 4 Oct 2026 14:15:30 -0400 Subject: [PATCH 09/10] perf: bind input conversion to compiled invocation domains --- openhcs/core/callable_contract.py | 14 +++++- openhcs/core/function_patterns.py | 47 +++++++++++++------ openhcs/core/pipeline/compiler.py | 10 ---- .../pipeline/funcstep_contract_validator.py | 20 ++------ openhcs/core/steps/function_runtime.py | 10 ++-- 5 files changed, 52 insertions(+), 49 deletions(-) diff --git a/openhcs/core/callable_contract.py b/openhcs/core/callable_contract.py index 3237773ce1..1c4449a090 100644 --- a/openhcs/core/callable_contract.py +++ b/openhcs/core/callable_contract.py @@ -1451,12 +1451,24 @@ def runtime_batch_executor( return None return self.runtime_batch_executors.get(domain) - def require_memory_types(self) -> tuple[str, str]: + def require_memory_types( + self, *, callable_label: str | None = None + ) -> tuple[str, str]: """Admit both declared memory domains before runtime conversion.""" if self.input_memory_type is None or self.output_memory_type is None: raise ValueError( f"Callable {self.function_name!r} is missing memory types." ) + valid_types = frozenset(memory_type.value for memory_type in MemoryType) + if ( + self.input_memory_type not in valid_types + or self.output_memory_type not in valid_types + ): + raise ValueError( + f"Function '{callable_label or self.function_name}' has invalid " + f"memory types: {self.input_memory_type}/{self.output_memory_type}. " + f"Valid: {', '.join(sorted(valid_types))}" + ) return self.input_memory_type, self.output_memory_type def require_execution_memory_type( diff --git a/openhcs/core/function_patterns.py b/openhcs/core/function_patterns.py index 4dbfbfeee3..1106085d14 100644 --- a/openhcs/core/function_patterns.py +++ b/openhcs/core/function_patterns.py @@ -11,7 +11,7 @@ from abc import ABC, abstractmethod from collections.abc import Callable, Hashable, Iterator, Mapping, Sequence from contextlib import AbstractContextManager, nullcontext -from dataclasses import dataclass, field, replace +from dataclasses import dataclass, field, fields, replace from functools import partial from enum import Enum from pathlib import Path @@ -519,6 +519,11 @@ class CompiledFunctionInvocation(NormalizedFunctionItem): repr=False, compare=False, ) + _input_memory_type: MemoryType | None = field( + init=False, + repr=False, + compare=False, + ) _execution_memory_type: MemoryType | None = field( init=False, repr=False, @@ -593,6 +598,19 @@ def __post_init__(self) -> None: else declaration.main_flow_output_source_payload ), ) + input_memory_type = None + if ( + self.contract.input_memory_type is not None + and self.contract.output_memory_type is not None + ): + input_declaration, _ = self.contract.require_memory_types( + callable_label=( + f"{self.contract.function_name}" + f"[{self.key.group_key}:{self.key.position}]" + ), + ) + input_memory_type = MemoryType(input_declaration) + object.__setattr__(self, "_input_memory_type", input_memory_type) object.__setattr__( self, "_execution_memory_type", @@ -601,19 +619,12 @@ def __post_init__(self) -> None: def __reduce__(self) -> tuple[Callable, tuple]: """Rebind process-local executable state from declaration-only transport.""" - return ( - type(self), - ( - self.key, - self.contract, - self.kwargs, - self.artifact_output_plans, - self.artifact_input_edges, - self.runtime_parameter_bindings, - self.input_device_id, - self.execution_device_id, - ), - ) + constructor_values = { + declared.name: getattr(self, declared.name) + for declared in fields(self) + if declared.init + } + return partial(type(self), **constructor_values), () def with_device_assignment( self, @@ -635,6 +646,14 @@ def with_device_assignment( ), ) + def convert_input(self, data: object, source_memory_type: str) -> object: + """Place the active predecessor's pixels on the compiled input domain.""" + if self._input_memory_type is None: + self.contract.require_memory_types() + return MemoryType(source_memory_type).convert_to( + data, self._input_memory_type, self.input_device_id + ) + def main_flow_call_argument(self, source_payload: object) -> object: """Apply the captured processing declaration to current request pixels.""" if self._main_flow_argument is None: diff --git a/openhcs/core/pipeline/compiler.py b/openhcs/core/pipeline/compiler.py index 605598da20..e01e1014a4 100644 --- a/openhcs/core/pipeline/compiler.py +++ b/openhcs/core/pipeline/compiler.py @@ -1860,16 +1860,6 @@ def compile_pipelines( axis_request, axis_values_to_process, ) - # Preparation belongs to callable admission before compilation. Bind - # process-local adapters here so execution workers inherit them. - for context in compiled_contexts.values(): - if not context.step_plans: - continue - for step_plan in context.step_plans.values(): - pattern = step_plan.compiled_function_pattern - if pattern is not None: - for invocation in pattern.iter_invocations(): - invocation.contract.resolve_runtime_callable() worker_assignments = PipelineCompiler._calculate_worker_assignments( list(compiled_contexts.keys()), num_workers, diff --git a/openhcs/core/pipeline/funcstep_contract_validator.py b/openhcs/core/pipeline/funcstep_contract_validator.py index 0473a00064..ea1905ef4d 100644 --- a/openhcs/core/pipeline/funcstep_contract_validator.py +++ b/openhcs/core/pipeline/funcstep_contract_validator.py @@ -31,7 +31,6 @@ from openhcs.constants.constants import ( GroupBy, - VALID_MEMORY_TYPES, get_openhcs_config, ) from openhcs.core.callable_contract import CallableContract, FunctionStepExecutionScope @@ -111,10 +110,6 @@ def inconsistent_memory_types_error(step_name, func1, func2): return f"Functions in step '{step_name}' have different memory types: {func1} vs {func2}" -def invalid_memory_type_error(func_name, input_type, output_type, valid_types): - return f"Function '{func_name}' has invalid memory types: {input_type}/{output_type}. Valid: {valid_types}" - - def invalid_pattern_error(pattern): return f"Invalid function pattern: {pattern}" @@ -1035,18 +1030,9 @@ def _validate_callable_memory_contract( raise ValueError( missing_memory_type_error(contract.function_name, step_name) ) - if ( - input_type not in VALID_MEMORY_TYPES - or output_type not in VALID_MEMORY_TYPES - ): - raise ValueError( - invalid_memory_type_error( - callable_label or contract.function_name, - input_type, - output_type, - ", ".join(sorted(VALID_MEMORY_TYPES)), - ) - ) + input_type, output_type = contract.require_memory_types( + callable_label=callable_label, + ) contract.require_execution_memory_type(step_name=step_name) return input_type, output_type diff --git a/openhcs/core/steps/function_runtime.py b/openhcs/core/steps/function_runtime.py index 92dd27f71d..431626de24 100644 --- a/openhcs/core/steps/function_runtime.py +++ b/openhcs/core/steps/function_runtime.py @@ -65,7 +65,6 @@ unstack_image_payload_context, ) from openhcs.core.memory import ( - convert_memory, unstack_runtime_slices, ) from openhcs.core.runtime_stores import ( @@ -958,12 +957,9 @@ def execute( *, debug_sink: DebugEventSink | None = None, ) -> RuntimePayload | NoMainFlowOutput: - input_memory_type = self.invocation.input_memory_type - converted_data = convert_memory( - data=image_payload_data(self.main_data_arg), - source_type=self.source_memory_type, - target_type=input_memory_type, - gpu_id=self.invocation.input_device_id, + converted_data = self.invocation.convert_input( + image_payload_data(self.main_data_arg), + self.source_memory_type, ) source_payload = with_image_payload_data(self.main_data_arg, converted_data) main_data_arg = self.invocation.main_flow_call_argument(source_payload) From b379d3cd19e61b2afefc81349f701ab358e3a2a0 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 4 Oct 2026 14:19:40 -0400 Subject: [PATCH 10/10] Document compiled runtime and native placement integration evidence --- plans/runtime_plumbing/IMPLEMENTATION.md | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/plans/runtime_plumbing/IMPLEMENTATION.md b/plans/runtime_plumbing/IMPLEMENTATION.md index e32fafd729..cce44fe32b 100644 --- a/plans/runtime_plumbing/IMPLEMENTATION.md +++ b/plans/runtime_plumbing/IMPLEMENTATION.md @@ -7,14 +7,18 @@ The active performance objective remains execution and total pipeline latency, w | Surface | Existing determining owners | Integration status and obligations | |---|---|---| | P0 | RuntimeProfileLogger; worker/run lifecycle | Implemented in #645 and integrated here. One raw worker-run buffer and final flush; independent failure/fork/thread lifetimes. Real 3D diagnostic completed, with strict comparison against both retained native outputs. Observer disposition is separate from ordinary latency. | -| P1 | CompiledFunctionInvocation; CallableContract; CompiledStepPlan | Compile fixed memory, ABI, bindings and return-layout decisions. Adapter requests/current pixels/active edges remain invocation-owned. Build debug parameters only for debug or the existing declaration-error diagnostics. Preserve eager adapter binding and error epochs. | +| P1 | CompiledFunctionInvocation; CallableContract; CompiledStepPlan | Implemented in cbf1857 and 09b353e. Contract owns fixed return ABI; redundant matcher deleted. Invocation holds executable, authored-plus-runtime binding template, processing operations and assigned devices. Authored kwargs remain separate for declaration validation; live adapters stay per request. Derived constructor transport rebinds on actual spawn. Debug copies exist only for debug/error diagnostics. | | P2 | RuntimePatternDiscoveryCache; declared source bindings | Implemented in c1650d8. Production consumer probe confirms one physical filename parse across discovery, matching and indexed group selection, and re-admission when parser semantics change. Produced filename aliases remain live. | | P3 | StepOutputManifestStore; CompiledFunctionPattern | Implemented in 88522fb. Ordered address maps update only new records; producer-scoped invalidation eliminates revision counters. Existing immutable DurableSourceMetadata fixes publication slots; source metadata and filename aliases remain live. Publication policy explicitly approved by user. Derived path order is retained without changing insertion order. | | P4 | ProcessingContext; source projection authority; manifest | Implemented in c1650d8. Held authority rebinds strong owners; optional file eligibility and document changes remain live. Lazy logging and retained P3 order remove per-group formatting/sorting. Ambiguity admission still precedes cached payload return. | -| P5 | CompiledStepPlan; compiled invocation; device scope | Plan memory/device boundaries across steps. Keep preprocessing InputConversionPlan distinct from runtime placement. Validate mixed CPU/GPU semantics separately; a CPU-only profile cannot prove a GPU gain. | +| P5 | Compiled invocation; ArrayBridge MemoryType; ImageFileFormat | Core implemented in c1d8728 and 09b353e. Target/device are captured; conversions use the real active source, including skipped calls. Native stacking removes mandatory GPU-host-GPU traffic. Disk format parent owns host projection; memory checkpoints keep their declared framework. Actual GPU receiving covers shape/dtype/device, independent buffers, mixed NumPy/CuPy stacks/conversion and direct disk writes. Mixed intrinsic-channel promotion and mask broadcasting still use host operations and need generic ArrayBridge capabilities before P5 is fully complete. | #641 is merged and removes post-compilation kernel warming; normal registry READY is the only child-cache consumer. Full catalog READY qualification passed. #630 (runtime ownership refactor), #635 (intensity maxima) and #597 (saved compiler resolution) remain drafts with negative whole-pipeline observations. They are not performance acceptance for this plan. Required final evidence: corrected baseline and integrated execution profile on the same real pipeline; production benchmark before/after execution and total time; affected parity, MCP and GUI behavior; mixed-device receiving for P5. Each implementation PR must close its corresponding issue and declare any persisted format changes. -Implementation is tracked by #647 and closing draft #648. Current source merges P0 head 1236f744 and main 602d577. The integrated source/projection/manifest/profile gate passes 127 controls in 2.51s; this verifies behavior and metadata pickle/cloudpickle transport, not runtime improvement. Actual parse/rebind/optional-document production receiving is retained under the runtime-file-inventory receipts in the maintenance namespace. P1/P5 remain active and no integrated speedup is accepted yet. +Implementation is tracked by #647 and closing draft #648. Current source merges main e7e79cd. Integrated behavior receiving includes 127 source/projection/manifest/profile/metadata transport controls, 234 invocation/contract/projection controls and actual pickle/spawn receiving. Production consumers establish parse-once/rebind/document epochs and real GPU native memory placement; these are not whole-pipeline speedup claims. + +Additional dominant runtime closure in 3cf7563: existing RuntimeSourceResolutionSnapshot owns admitted path/component correlations, removing repeated alias scans from immutable source matching while raw callback contexts remain live. The retained real 60-plane frontier preserves every position: old matching 233.219 ms; candidate first admission plus matching 53.358 ms, including 49.301 ms admission. The baseline receipt documents recovery of exact timing fields after a helper filename overwrite. Twenty-one lifetime/alias/ambiguity/correlation controls pass. Remaining original-source restacking requires an explicit source revision and independent-buffer law before reuse. + +One ordinary 3D before/after pair against latest main is next, with no coarse/sampled observer and separate source-qualified kernel preparation before public READY. Fresh native CellProfiler timing follows successful integrated parity. Prior negative observations and observer failures remain retained; no integrated gain is accepted yet.