Fix aggregate label materialization for separate viewer layers

openhcs/core/steps/stream_component_semantics.py

@@ -29,8 +29,16 @@
 )
 
 from openhcs.constants.constants import AllComponents, get_multiprocessing_axis
+from openhcs.core.config import FijiDimensionMode, NapariDimensionMode
 from openhcs.core.context.processing_context import ProcessingContext
 from openhcs.core.runtime_image_values import ImagePayloadMetadata
+from openhcs.core.runtime_plane_projection import RuntimePlaneAxisValueProjection
+from openhcs.core.runtime_slice_projection import (
+    RuntimeProjectionData,
+    RuntimeProjectedPayloadItem,
+    RuntimeProjectionSourceIdentityRequest,
+    RuntimeProjectionSourceIdentityRequirement,
+)
 
 from openhcs.core.source_image_provenance import (
     SourceComponentMetadata,
@@ -62,6 +70,47 @@
 class StreamImagePayloadMetadataProjector:
     """Project image-axis declarations into viewer batch-item fields."""
 
+    @classmethod
+    def payload_items_for_display(
+        cls,
+        payload: RuntimeProjectionData,
+        *,
+        metadata: ImagePayloadMetadata,
+        plane_components: tuple[AllComponents, ...],
+        component_modes: Mapping[str, str],
+        source_description: str,
+    ) -> tuple[RuntimeProjectedPayloadItem, ...]:
+        """Project declared planes when their display requires scalar identities."""
+        item_fields = cls.item_fields_for_plane_components(metadata, plane_components)
+        component_values = item_fields.get(
+            ViewerWireField.PLANE_COMPONENT_VALUES.value, {}
+        )
+        scalar_modes = (
+            NapariDimensionMode.LAYER.value,
+            FijiDimensionMode.WINDOW.value,
+        )
+        if not any(
+            mode in scalar_modes
+            for component, mode in component_modes.items()
+            if component in component_values
+        ):
+            return (RuntimeProjectedPayloadItem(payload, source_description),)
+        projection = RuntimePlaneAxisValueProjection.preserve(
+            axis=metadata.plane_axis,
+            axis_size=metadata.source_provenance.source_plane_count,
+        )
+        source_identity = (
+            RuntimeProjectionSourceIdentityRequirement.REQUIRED_COMPONENT_METADATA
+        )
+        return source_identity.project_payload_items(
+            RuntimeProjectionSourceIdentityRequest(
+                value=metadata.attach_to(payload),
+                source_description=source_description,
+                variable_components=plane_components,
+                plane_projection=projection,
+            )
+        )
+
     @classmethod
     def partition_indices(
         cls,

openhcs/processing/materialization/core.py

@@ -73,6 +73,10 @@
     source_metadata_value,
 )
 from openhcs.core.source_spatial_domain import SourceSpatialDomain
+from openhcs.core.steps.function_output_identity import (
+    FunctionOutputIdentityAuthority,
+    FunctionOutputPathAuthority,
+)
 from openhcs.core.steps.stream_component_semantics import (
     StreamImagePayloadMetadataProjector,
     StreamViewerComponentMetadataProjector,
@@ -1526,7 +1530,10 @@ def to_filemanager_kwargs(
         self,
         output: Output,
     ) -> dict:
-        output_fields = self._output_fields(output)
+        projected_outputs = self.projected_outputs((output,))
+        if len(projected_outputs) != 1 or projected_outputs[0] is not output:
+            raise ValueError("Projected viewer materialization requires batch saving.")
+        output_fields = self._output_fields(projected_outputs[0])
         if output_fields is None:
             return self.values.to_kwargs()
         component_metadata, item_fields = output_fields
@@ -1547,7 +1554,7 @@ def filemanager_batches(
             ]
         ] = []
         unprojected_outputs: list[Output] = []
-        for output in outputs:
+        for output in self.projected_outputs(outputs):
             output_fields = self._output_fields(output)
             if output_fields is None:
                 unprojected_outputs.append(output)
@@ -1596,6 +1603,49 @@ def filemanager_batches(
             )
         return tuple(batches)
 
+    def projected_outputs(self, outputs: Sequence[Output]) -> tuple[Output, ...]:
+        """Derive stream items without changing persisted aggregate outputs."""
+        result: list[Output] = []
+        display_semantics = self.values.stream_request.display_semantics
+        parser = self.values.stream_request.source.identity.microscope_handler.parser
+        for output in outputs:
+            if output.metadata is None:
+                result.append(output)
+                continue
+            items = StreamImagePayloadMetadataProjector.payload_items_for_display(
+                output.content,
+                metadata=output.metadata,
+                plane_components=output.variable_components,
+                component_modes=display_semantics.component_modes,
+                source_description=output.path,
+            )
+            if len(items) == 1 and items[0].value is output.content:
+                result.append(output)
+                continue
+            output_path = Path(output.path)
+            for item in items:
+                identity = FunctionOutputIdentityAuthority.identity_from_metadata(
+                    parser,
+                    item.metadata,
+                    fallback_identity_path=output.path,
+                )
+                if identity is None:
+                    raise ValueError("Projected viewer output requires a filename identity.")
+                identity = replace(identity, extension=output_path.suffix)
+                filename = FunctionOutputPathAuthority.filename_for_identity(
+                    parser,
+                    identity.with_filename_qualifier(output_path.stem),
+                )
+                result.append(
+                    replace(
+                        output,
+                        path=str(output_path.parent / filename),
+                        content=item.data,
+                        metadata=item.metadata,
+                    )
+                )
+        return tuple(result)
+
     def _output_fields(
         self,
         output: Output,