Preserve grouped coordinates in scalar artifact publication (#621)

openhcs/core/component_group_scope.py

@@ -498,13 +498,17 @@ def for_group_coordinate(
             fixed_component_values=tuple(fixed_values.items()),
         )
 
-    def fixed_component_metadata(
+    def source_component_metadata(
         self,
         metadata: SourceMetadataMapping | None = None,
     ) -> SourceMetadataMapping:
-        """Merge fixed execution coordinates into source component metadata."""
+        """Project scope coordinates while retaining an explicit measured source.
 
-        from openhcs.constants.constants import get_multiprocessing_axis
+        A measurement's physical source can differ from its object's grouped
+        channel. Keep that declared source coordinate; only absent group
+        metadata is supplied by this scope. Axis and fixed coordinates must
+        agree, and unrepresented coordinates are never invented.
+        """
         from openhcs.core.source_matching import (
             source_component_metadata_value,
             with_source_component_metadata,
@@ -513,15 +517,13 @@ def fixed_component_metadata(
         merged = SourceMetadataFields.composition_snapshot(
             metadata if metadata is not None else {}
         )
-        fixed_values = (
-            (get_multiprocessing_axis(), self.axis_id),
-            *self.fixed_component_values,
-        )
-        for component, value in fixed_values:
+        for component, value in self.source_component_values:
             existing = source_component_metadata_value(merged, component)
+            if component is self.component and existing is not None:
+                continue
             if existing is not None and str(existing) != value:
                 raise ValueError(
-                    "Fixed execution scope conflicts with source component metadata "
+                    "Runtime execution scope conflicts with source component metadata "
                     f"for {component.value!r}: {existing!r} != {value!r}."
                 )
             merged = with_source_component_metadata(merged, component, value)

openhcs/core/steps/function_artifact_materialization.py

@@ -48,7 +48,6 @@
 )
 from openhcs.core.source_matching import (
     source_component_metadata_items,
-    with_source_component_metadata,
 )
 from openhcs.core.source_metadata import SourceMetadataFields
 from openhcs.core.source_projection import OpenHCSPlaneAddress
@@ -801,19 +800,9 @@ def record_metadata_with_runtime_scope(
                 metadata = metadata.with_source_provenance(
                     provenance.with_source_component_metadata(component_metadata)
                 )
-        scope = record.key.scope
-        if scope.has_fixed_components:
-            component_metadata = scope.fixed_component_metadata(
-                metadata.source_component_metadata
-            )
-        elif scope.component is not None:
-            component_metadata = with_source_component_metadata(
-                metadata.source_component_metadata or {},
-                scope.component,
-                scope.require_value_text(),
-            )
-        else:
-            return metadata
+        component_metadata = record.key.scope.source_component_metadata(
+            metadata.source_component_metadata
+        )
         return metadata.with_source_provenance(
             metadata.source_provenance.with_source_component_metadata(
                 component_metadata

tests/integration/test_source_schema_aggregate_publication.py

@@ -10,6 +10,7 @@
 from pathlib import Path
 
 import numpy as np
+import pytest
 import tifffile
 
 from openhcs.agent.services.execution_session_service import (
@@ -60,7 +61,8 @@ def volume_scalar_609(image: np.ndarray):
     return image, ({"volume_sum": int(image.sum()), "voxel_count": int(image.size)},)
 
 
-def test_real_z_aggregation_persists_scalar_and_unchanged_source_planes(tmp_path):
+@pytest.mark.parametrize("group_by", [GroupBy.NONE, GroupBy.CHANNEL])
+def test_real_z_aggregation_persists_scalar_and_unchanged_source_planes(tmp_path, group_by):
     source = tmp_path / "input"
     source.mkdir()
     pixels = np.arange(140, dtype=np.uint16).reshape(4, 5, 7)
@@ -90,10 +92,7 @@ def test_real_z_aggregation_persists_scalar_and_unchanged_source_planes(tmp_path
                 func=registered, name="Actual volume scalar",
                 processing_config=LazyProcessingConfig(
                     variable_components=[VariableComponents.Z_INDEX],
-                    # One physical channel is fixed; Z is the genuine aggregate.
-                    # Channel-grouped generic SpecialArtifact scope propagation
-                    # is independently retained in the qualification receipt.
-                    group_by=GroupBy.NONE,
+                    group_by=group_by,
                     input_source=InputSource.PIPELINE_START,
                 ),
             )],

tests/unit/test_function_step_execution_scope.py

@@ -246,7 +246,9 @@ def test_path_validation_accepts_explicit_no_main_flow_dependency(
     planner = PathPlanner.__new__(PathPlanner)
     planner.session = SimpleNamespace(
         step_count=2,
-        snapshot=lambda index: SimpleNamespace(name=("axis", "export")[index]),
+        pipeline=SimpleNamespace(
+            steps=(SimpleNamespace(name="axis"), SimpleNamespace(name="export")),
+        ),
     )
     planner.plans = {
         1: CompiledStepPlan(

tests/unit/test_runtime_execution_scope_metadata.py

@@ -0,0 +1,77 @@
+"""Complete source-coordinate projection from the original runtime scope owner."""
+from copy import deepcopy
+
+import pytest
+
+from openhcs.constants import AllComponents
+from openhcs.core.component_group_scope import RuntimeExecutionAxisScope
+from openhcs.core.source_matching import source_component_metadata_value
+
+
+@pytest.mark.parametrize("grouped,fixed", [(False, False), (True, False), (False, True), (True, True)])
+def test_scope_projects_every_owned_coordinate_without_inventing_absent_axes(grouped, fixed):
+    scope = RuntimeExecutionAxisScope.from_raw(
+        "A01",
+        component=AllComponents.CHANNEL if grouped else None,
+        value="2" if grouped else None,
+        fixed_component_values=(
+            ((AllComponents.SITE, "3"), (AllComponents.TIMEPOINT, "4")) if fixed else ()
+        ),
+    )
+    original = {"source_alias": "Synthetic", "custom": {"retained": "value"}}
+    untouched = deepcopy(original)
+    metadata = scope.source_component_metadata(original)
+    for component, value in scope.source_component_values:
+        assert source_component_metadata_value(metadata, component) == value
+    assert source_component_metadata_value(metadata, AllComponents.Z_INDEX) is None
+    assert metadata["source_alias"] == "Synthetic"
+    assert metadata["custom"] == original["custom"]
+    assert original == untouched
+
+
+@pytest.mark.parametrize("component,wrong", [
+    (AllComponents.WELL, "B01"),
+    (AllComponents.SITE, "8"),
+    (AllComponents.TIMEPOINT, "7"),
+])
+def test_scope_rejects_conflicting_axis_and_fixed_identity(component, wrong):
+    scope = RuntimeExecutionAxisScope.from_raw(
+        "A01", component=AllComponents.CHANNEL, value="2",
+        fixed_component_values=((AllComponents.SITE, "3"), (AllComponents.TIMEPOINT, "4")),
+    )
+    metadata = {component.value: wrong, "custom": "retained"}
+    untouched = dict(metadata)
+    with pytest.raises(ValueError, match="Runtime execution scope conflicts"):
+        scope.source_component_metadata(metadata)
+    assert metadata == untouched
+
+
+def test_scope_canonicalizes_matching_component_alias_without_losing_source_fields():
+    scope = RuntimeExecutionAxisScope.from_raw(
+        "A01", component=AllComponents.CHANNEL, value="2",
+        fixed_component_values=((AllComponents.SITE, "3"),),
+    )
+    metadata = scope.source_component_metadata({"Well": "A01", "Channel": "2", "custom": "retained"})
+    assert source_component_metadata_value(metadata, AllComponents.CHANNEL) == "2"
+    assert metadata["custom"] == "retained"
+
+
+@pytest.mark.parametrize("fixed", [False, True])
+def test_scope_keeps_measured_source_channel_distinct_from_object_group(fixed):
+    scope = RuntimeExecutionAxisScope.from_raw(
+        "A01", component=AllComponents.CHANNEL, value="2",
+        fixed_component_values=((AllComponents.Z_INDEX, "1"),) if fixed else (),
+    )
+    metadata = scope.source_component_metadata({"channel": "1", "custom": "retained"})
+    assert source_component_metadata_value(metadata, AllComponents.CHANNEL) == "1"
+    assert scope.value_text_for_component(AllComponents.CHANNEL) == "2"
+    assert metadata["custom"] == "retained"
+
+
+def test_scope_still_rejects_a_contradictory_declared_group_coordinate():
+    scope = RuntimeExecutionAxisScope.from_raw(
+        "A01", component=AllComponents.CHANNEL, value="2",
+        fixed_component_values=((AllComponents.Z_INDEX, "1"),),
+    )
+    with pytest.raises(ValueError, match="group coordinate conflicts"):
+        scope.for_group_coordinate(AllComponents.CHANNEL, "3")