Preserve site-free aggregate source scope through publication (addresses #728)

openhcs/core/orchestrator/analysis_consolidation.py

@@ -187,7 +187,7 @@ def from_observations(
                         "Runtime observation references unknown compiled context "
                         f"{context_observation.context_key!r}."
                     )
-                inputs.append(context_observation.analysis_inputs)
+                inputs.append(context_observation.outputs.analysis_inputs)
         return cls.combine(inputs)
 
     @classmethod

openhcs/core/orchestrator/compiled_plate_execution.py

@@ -575,17 +575,9 @@ def execute_plate_scoped_steps(
                 context_key=context_key,
                 context=context,
                 records=records,
-                runtime_export_paths=tuple(
-                    dict.fromkeys(
-                        path
-                        for item in observations_by_context[context_key]
-                        for path in item.runtime_export_paths
-                    )
-                ),
                 runtime_observation_mode=RuntimeObservationMode.MERGE_INTO_PARENT,
-                analysis_inputs=RuntimeAnalysisConsolidationInputs.combine(
-                    item.analysis_inputs
-                    for item in observations_by_context[context_key]
+                outputs=StepExecutionObservation.combine(
+                    observations_by_context[context_key]
                 ),
             )
             for context_key, context in compiled_contexts.items()

openhcs/core/orchestrator/execution_result.py

@@ -8,18 +8,15 @@
 import copyreg
 from dataclasses import dataclass, field
 from enum import Enum
-from pathlib import Path
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Mapping, Optional
 
 from openhcs.core.context.processing_context import ProcessingContext
 from openhcs.core.runtime_stores import StoredRuntimeValue
+from openhcs.core.steps.abstract import StepExecutionObservation
 
 if TYPE_CHECKING:
     from openhcs.core.compiled_execution import CompiledExecutionBundle
-    from openhcs.core.orchestrator.analysis_consolidation import (
-        RuntimeAnalysisConsolidationInputs,
-    )
 
 
 class RuntimeExecutionTransportSerialization:
@@ -63,8 +60,7 @@ class RuntimeContextObservation:
 
     context_key: str
     records: tuple[StoredRuntimeValue, ...]
-    runtime_export_paths: tuple[Path, ...] = field(default_factory=tuple)
-    analysis_inputs: "RuntimeAnalysisConsolidationInputs | None" = None
+    outputs: StepExecutionObservation = field(default_factory=StepExecutionObservation.empty)
 
     @classmethod
     def from_context(
@@ -74,15 +70,13 @@ def from_context(
         context: ProcessingContext,
         records: tuple[StoredRuntimeValue, ...],
         runtime_observation_mode: "RuntimeObservationMode",
-        runtime_export_paths: tuple[Path, ...],
-        analysis_inputs: "RuntimeAnalysisConsolidationInputs | None",
+        outputs: StepExecutionObservation,
     ) -> "RuntimeContextObservation":
         """Retain requested records beside the completed output projections."""
         return cls(
             context_key=context_key,
             records=runtime_observation_mode.retain_records(records, context),
-            runtime_export_paths=tuple(dict.fromkeys(runtime_export_paths)),
-            analysis_inputs=analysis_inputs,
+            outputs=outputs,
         )
 
 

openhcs/core/orchestrator/worker_execution.py

@@ -24,9 +24,6 @@
 from openhcs.core.function_step_transport import FunctionStepTransportAuthority
 from openhcs.core.native_threading import configure_native_thread_count
 from openhcs.core.runtime_profile import RuntimeProfileLogger
-from openhcs.core.orchestrator.analysis_consolidation import (
-    RuntimeAnalysisConsolidationInputs,
-)
 from openhcs.core.orchestrator.execution_result import (
     ExecutionResult,
     RuntimeContextObservation,
@@ -806,19 +803,13 @@ def _execute_axis_with_sequential_combinations(
             observed_records = runtime_store.observed_values_after(
                 execution_observation_cursor
             )
-            runtime_export_paths = tuple(
-                path
-                for observation in result.runtime_observation.contexts
-                for path in observation.runtime_export_paths
-            )
             observation = RuntimeContextObservation.from_context(
                 context_key=context_key,
                 context=frozen_context,
                 records=observed_records,
                 runtime_observation_mode=runtime_observation_mode,
-                runtime_export_paths=runtime_export_paths,
-                analysis_inputs=RuntimeAnalysisConsolidationInputs.combine(
-                    item.analysis_inputs for item in result.runtime_observation.contexts
+                outputs=StepExecutionObservation.combine(
+                    item.outputs for item in result.runtime_observation.contexts
                 ),
             )
         finally:
@@ -830,8 +821,8 @@ def _execute_axis_with_sequential_combinations(
             frozen_context.runtime_value_store.clear()
         if (
             observation.records
-            or observation.runtime_export_paths
-            or observation.analysis_inputs
+            or observation.outputs.runtime_export_paths
+            or observation.outputs.analysis_inputs
         ):
             runtime_observations.append(observation)
         del observed_records
@@ -957,8 +948,7 @@ def _execute_single_axis_static(
     frozen_context.bind_execution_runtime(lane_context)
     lane_context.install_debug_sink(frozen_context)
     runtime_value_store = frozen_context.runtime_value_store
-    runtime_export_paths = []
-    analysis_inputs = []
+    step_observations = []
 
     for step_index, step in enumerate(pipeline_definition):
         if cancellation is not None:
@@ -993,8 +983,7 @@ def _execute_single_axis_static(
                     observed_records,
                     materialized_locations_by_address=reused_outputs.materialized_locations_by_address,
                 )
-                runtime_export_paths.extend(reused_outputs.runtime_export_paths)
-                analysis_inputs.append(reused_outputs.analysis_inputs)
+                step_observations.append(reused_outputs)
                 emit(
                     execution_id=lane_context.execution_id,
                     plate_id=lane_context.plate_id,
@@ -1028,8 +1017,7 @@ def _execute_single_axis_static(
 
         observation_cursor = runtime_value_store.observation_cursor()
         step_observation = step.process(frozen_context, step_index)
-        runtime_export_paths.extend(step_observation.runtime_export_paths)
-        analysis_inputs.append(step_observation.analysis_inputs)
+        step_observations.append(step_observation)
         observed_records = runtime_value_store.observed_values_after(observation_cursor)
         runtime_progress_context = _runtime_observation_progress_context(
             observed_records,
@@ -1063,8 +1051,7 @@ def _execute_single_axis_static(
                 RuntimeContextObservation(
                     context_key=context_key,
                     records=(),
-                    runtime_export_paths=tuple(dict.fromkeys(runtime_export_paths)),
-                    analysis_inputs=RuntimeAnalysisConsolidationInputs.combine(analysis_inputs),
+                    outputs=StepExecutionObservation.combine(step_observations),
                 ),
             )
         ),

openhcs/core/plate_image_inventory.py

@@ -322,9 +322,10 @@ def _record(
         }
         if resolved_source_metadata is not None:
             metadata.update(dict(resolved_source_metadata))
-        parsed = handler.parse_image_path(image_file)
-        if parsed is not None:
-            metadata.update(parsed.wire_mapping())
+        if resolved_source_metadata is None:
+            parsed = handler.parse_image_path(image_file)
+            if parsed is not None:
+                metadata.update(parsed.wire_mapping())
         source_file = Path(source_path)
         metadata["size"] = file_size_label(source_file)
         metadata["modified"] = file_modified_label(source_file)

openhcs/core/runtime_exports.py

@@ -4,7 +4,7 @@
 
 import csv
 from collections.abc import Mapping, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path, PurePosixPath
 from types import MappingProxyType
 from typing import TYPE_CHECKING, cast
@@ -18,6 +18,7 @@
 from openhcs.core.runtime_identifier import normalize_runtime_identifier
 from openhcs.core.runtime_measurements import MeasurementTable
 from openhcs.core.runtime_stores import StoredRuntimeValue
+from openhcs.core.steps.abstract import StepExecutionObservation
 from openhcs.processing.materialization import (
     CsvOptions,
     FileBundleOptions,
@@ -111,6 +112,7 @@ class RuntimeExportObservation:
     table_headers_by_path: Mapping[Path, tuple[str, ...]]
     table_row_counts_by_path: Mapping[Path, int]
     output_files: tuple[Path, ...] = ()
+    outputs: StepExecutionObservation = field(default_factory=StepExecutionObservation.empty)
 
     @classmethod
     def from_runtime_observations(
@@ -119,36 +121,43 @@ def from_runtime_observations(
     ) -> RuntimeExportObservation:
         """Build exports from worker-projected paths without retaining values."""
 
+        outputs = StepExecutionObservation.combine(
+            context.outputs
+            for observation in observations
+            for context in observation.contexts
+        )
         return cls.from_output_paths(
-            tuple(
-                path
-                for observation in observations
-                for context in observation.contexts
-                for path in context.runtime_export_paths
-            )
+            outputs.runtime_export_paths,
+            outputs=outputs,
         )
 
     @classmethod
     def from_output_root(
         cls,
         output_root: Path,
+        *,
+        outputs: StepExecutionObservation = StepExecutionObservation.empty(),
     ) -> "RuntimeExportObservation":
         """Build an export observation from one runtime output root."""
-        return cls.from_output_roots((Path(output_root),))
+        return cls.from_output_roots((Path(output_root),), outputs=outputs)
 
     @classmethod
     def from_output_roots(
         cls,
         output_roots: tuple[Path, ...],
+        *,
+        outputs: StepExecutionObservation = StepExecutionObservation.empty(),
     ) -> "RuntimeExportObservation":
         """Build an export observation from compiled runtime output roots."""
         roots = tuple(dict.fromkeys(Path(root) for root in output_roots))
-        return cls.from_output_paths(_output_files_from_roots(roots))
+        return cls.from_output_paths(_output_files_from_roots(roots), outputs=outputs)
 
     @classmethod
     def from_output_paths(
         cls,
         output_paths: Sequence[str | Path],
+        *,
+        outputs: StepExecutionObservation = StepExecutionObservation.empty(),
     ) -> "RuntimeExportObservation":
         """Build an export observation from exact contract-owned output paths."""
 
@@ -174,6 +183,7 @@ def from_output_paths(
             table_headers_by_path=_table_headers_by_path(table_outputs),
             table_row_counts_by_path=_table_row_counts_by_path(table_outputs),
             output_files=output_files,
+            outputs=outputs,
         )
 
     def __post_init__(self) -> None:
@@ -201,7 +211,7 @@ def with_runtime_artifact_tables(
                 path
                 for records in runtime_records_by_axis.values()
                 for record in _table_runtime_records(records)
-                for path in matching_table_outputs(record, self.table_outputs)
+                for path in self.matching_table_outputs(record)
             )
         )
         return RuntimeExportObservation(
@@ -214,8 +224,14 @@ def with_runtime_artifact_tables(
                 path: self.table_row_counts_by_path[path] for path in table_outputs
             },
             output_files=self.output_files,
+            outputs=self.outputs,
         )
 
+    def matching_table_outputs(self, record: StoredRuntimeValue) -> tuple[Path, ...]:
+        """Match tables through the original saved-output address relation."""
+        owned_paths = frozenset(self.outputs.paths_for(record))
+        return tuple(path for path in self.table_outputs if path in owned_paths)
+
 
 def runtime_export_failures(
     expectation: RuntimeExportExpectation,
@@ -305,48 +321,14 @@ def _materialized_artifact_record_failures(
     )
 
 
-def matching_table_outputs(
-    record: StoredRuntimeValue,
-    table_outputs: tuple[Path, ...],
-) -> tuple[Path, ...]:
-    """Return table output files matching one runtime artifact record."""
-    return tuple(
-        path
-        for path in table_outputs
-        if table_output_matches_artifact(
-            path,
-            record.key.name,
-            axis_id=record.key.scope.axis_id,
-        )
-    )
-
-
-def table_output_matches_artifact(
-    path: Path,
-    artifact_name: str,
-    *,
-    axis_id: str | None = None,
-) -> bool:
-    """Return whether a materialized table filename belongs to an artifact."""
-    stem = path.stem
-    if f"_{artifact_name}_step" not in stem:
-        return False
-    if axis_id is None:
-        return True
-    return stem.startswith(f"{axis_id}_")
-
-
 def _table_artifact_failures(
     output_specs: tuple[ArtifactSpec, ...],
     observation: RuntimeExportObservation,
     runtime_records_by_axis: Mapping[str, tuple[StoredRuntimeValue, ...]],
 ) -> tuple[str, ...]:
     failures: list[str] = []
     for record in _runtime_records_for_specs(output_specs, runtime_records_by_axis):
-        matching_outputs = matching_table_outputs(
-            record,
-            observation.table_outputs,
-        )
+        matching_outputs = observation.matching_table_outputs(record)
         if not matching_outputs:
             failures.append(
                 f"axis {record.key.scope.axis_id!r} produced table artifact "

openhcs/core/source_image_provenance.py

@@ -967,6 +967,39 @@ def scalar_source_identity(self) -> SourceImageIdentity:
             )
         return self.source_identity
 
+    def required_scalar_components(
+        self, component_order: Iterable[str]
+    ) -> tuple[str, ...]:
+        """Exclude only coordinates proven reduced into pixel contributors.
+
+        Contributors do not own a projectable runtime axis. A component that
+        varies across their complete source addresses therefore has no scalar
+        address on the derived image. Absent or incomplete contributor evidence
+        cannot exempt an acquired plane from its required coordinates.
+        """
+        contributors = self.source_image_provenance_planes.contributors
+        scalar_components = SourceMetadataFields.component_domains(
+            self.scalar_source_identity.component_metadata or {}
+        )
+        contributor_components = tuple(
+            SourceMetadataFields.component_domains(contributor.component_metadata or {})
+            for contributor in contributors
+        )
+        reduced = set()
+        for component in AllComponents:
+            if component in scalar_components:
+                continue
+            if len(contributor_components) < 2 or not all(
+                len(values.get(component, ())) == 1 for values in contributor_components
+            ):
+                continue
+            values = frozenset(
+                value for values in contributor_components for value in values[component]
+            )
+            if len(values) > 1:
+                reduced.add(component.value)
+        return tuple(component for component in component_order if component not in reduced)
+
     def image_set_identities(
         self,
         policy: SourceImageSetIdentityPolicy,

openhcs/core/steps/abstract.py

@@ -34,6 +34,7 @@
 from openhcs.core.runtime_stores import (
     RuntimeArtifactAddress,
     RuntimeArtifactLocation,
+    StoredRuntimeValue,
 )
 
 # ProcessingContext is used in type hints
@@ -60,6 +61,14 @@ class StepExecutionObservation:
     def empty(cls) -> "StepExecutionObservation":
         return cls(MappingProxyType({}))
 
+    def paths_for(self, record: StoredRuntimeValue) -> tuple[Path, ...]:
+        """Return saved paths for this exact producer value, never its filename."""
+        address = RuntimeArtifactAddress.from_record(record)
+        return tuple(
+            Path(location.path)
+            for location in self.materialized_locations_by_address.get(address, ())
+        )
+
     @classmethod
     def combine(
         cls, observations: Iterable["StepExecutionObservation"]