Compose invocation image stacks once in their target domain

openhcs/core/function_patterns.py

@@ -39,6 +39,7 @@
     ArtifactSpecCollection,
     ArtifactSpecRef,
 )
+from openhcs.core.aligned_image_payload import ImagePayloadStackComposition
 from openhcs.core.callable_contract import (
     CallableContract,
     FunctionStepExecutionScope,
@@ -682,6 +683,11 @@ def convert_input(self, payload: object, source_memory_type: str) -> object:
         if self._input_memory_type is None:
             image_payload_data(payload)
             self.contract.require_memory_types()
+        if isinstance(payload, ImagePayloadStackComposition):
+            return payload.compose(
+                memory_type=self._input_memory_type.value,
+                device_id=self.input_device_id,
+            )
         source = MemoryType(source_memory_type)
         if source is self._input_memory_type and not source.is_gpu:
             mask = image_payload_mask(payload)

plans/runtime_plumbing/IMPLEMENTATION.md

@@ -11,7 +11,7 @@ The active performance objective remains execution and total pipeline latency, w
 | 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 | 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. #1080 completes native mixed-channel promotion, intensity reconciliation and mask geometry through ArrayBridge's existing operation owner. GTX 1080/CuPy composition receiving forbids downloads and checks exact values, masks, destination devices and independent stacks. Pyclesperanto lacks native changed-shape reshape/broadcast and boolean allocation; those operations report unsupported capability. P5 remains incomplete until a real mixed-device production pipeline has a complete before/after conversion profile, including skipped calls and disk boundaries. No CPU benchmark speedup is inferred from these GPU checks. |
+| 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. #1080 completes native mixed-channel promotion, intensity reconciliation and mask geometry through ArrayBridge's existing operation owner. GTX 1080/CuPy composition receiving forbids downloads and checks exact values, masks, destination devices and independent stacks. Pyclesperanto lacks native changed-shape reshape/broadcast and boolean allocation; those operations report unsupported capability. The real compiled CPU Gaussian → GPU sharpen → GPU percentile normalization → CPU Gaussian route now completes on saved microscopy inputs, including named mixed-intensity/masked channels, a skipped CPU invocation, a memory checkpoint and a disk roundtrip. #1083 repairs native flexible ABI admission. Invocation composition uses the existing target/device owner once: its canonical bulk conversion census falls from 10 calls / 2,621,440 bytes to 6 calls / 1,638,400 bytes, with exact numeric oracle and masks. The original pre-P5 mixed route fails before completing, so its partial counts are not a complete baseline; the plain grayscale negative control remains unchanged at two transfers. CuCIM is unavailable and installed GPU processors reject 4D colour stacks; promoted-colour full-pipeline receiving remains unqualified. The same scratch observer also emits isolated `convert_input` spans through the existing P0 RuntimeProfileLogger on exact ABI baseline 5ed8cf and composition head 53edad: retained `p5-mixed-conversion-profile-{before,after}-v1/runtime-profile.log` and receipts under the 20261006 maintenance root use identical pre-span/post-conversion CUDA synchronization. Both profiles complete the real route with exact oracle, masks, skipped-call events and checkpoint/disk checks. This is one diagnostic pair and a conversion census, not a statistical or CPU/end-to-end benchmark speedup claim. |
 
 #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), #597 (saved compiler resolution) and #608 (morphology graph intermediates) are merged following the user's explicit request to merge locally qualified work. Their earlier negative whole-pipeline observations remain documented; merge is not evidence of an end-to-end speedup.
 

tests/unit/test_callable_abi_preparation.py

@@ -399,3 +399,59 @@ def declared(*args, **kwargs):
         RuntimeBatchExecutionDomain.PURE_2D_SLICES: declared,
     })
     assert selected is declared
+
+
+def test_compiled_input_places_named_bundle_once_before_cpu_identity_shortcut(monkeypatch):
+    from openhcs.core.aligned_image_payload import (
+        AlignedImageSliceContext, ImageOutputBundle,
+    )
+    from openhcs.core.function_patterns import compile_function_pattern
+    from openhcs.core.runtime_image_values import image_payload_mask
+    from openhcs.processing.backends.processors.numpy_processor import gaussian_blur
+
+    pixels = tuple(np.full((3, 4), value, dtype=np.float32) for value in (0.25, 0.75))
+    masks = tuple(np.full((3, 4), value, dtype=bool) for value in (True, False))
+    source = ImageOutputBundle(
+        tuple(ImagePayloadMetadata().payload_with(data, mask)
+              for data, mask in zip(pixels, masks, strict=True)),
+        tuple(AlignedImageSliceContext.independent_main_flow(name)
+              for name in ("DNA", "Membrane")),
+    )
+    calls = []
+    original = ImageOutputBundle.compose
+
+    def observed(value, **destination):
+        calls.append(destination)
+        return original(value, **destination)
+
+    monkeypatch.setattr(ImageOutputBundle, "compose", observed)
+    invocation = compile_function_pattern(gaussian_blur, {}, {}).default_group.invocations[0]
+    placed = invocation.convert_input(source, "numpy")
+    invocation.main_flow_call_argument(placed)
+    image_payload_data(placed)
+    image_payload_mask(placed)
+
+    assert calls == [{"memory_type": "numpy", "device_id": None}]
+    np.testing.assert_array_equal(image_payload_data(placed), np.stack(pixels))
+    assert image_payload_data(placed).shape == (2, 3, 4)
+    assert image_payload_mask(placed).shape == (2, 3, 4)
+    assert placed.metadata.source_image_names == ("DNA", "Membrane")
+    np.testing.assert_array_equal(image_payload_mask(placed), np.stack(masks))
+
+
+def test_compiled_input_reuses_produced_stack_realization_without_copy():
+    from openhcs.core.aligned_image_payload import (
+        AlignedImageSliceContext, ProducedImageStack,
+    )
+    from openhcs.core.function_patterns import compile_function_pattern
+    from openhcs.processing.backends.processors.numpy_processor import gaussian_blur
+
+    source = ProducedImageStack(
+        tuple(np.full((3, 4), value, dtype=np.float32) for value in (1, 2)),
+        memory_type="numpy", plane_axis=RuntimePlaneAxis.RUNTIME_SLICE,
+    )
+    invocation = compile_function_pattern(gaussian_blur, {}, {}).default_group.invocations[0]
+    placed = invocation.convert_input(source, "numpy")
+    assert invocation.convert_input(source, "numpy") is placed
+    assert image_payload_data(placed) is image_payload_data(source)
+    assert np.shares_memory(source.slices[0], image_payload_data(placed))