CP1: CellProfiler plugin modules convert; undeclared modules stay in the pipeline

.gitignore

@@ -128,26 +128,9 @@ paper/.current-*
 # Build runs, snapshots and intermediates are local. Main publication artifacts
 # are selected explicitly; do not force-add review history or microscopy dumps.
 paper/build/
+# Generated JSON is ignored by the directory it is written to (the three rules
+# here), never by extension: JSON fixtures, manifests and records are source.
 output/
-*.json
-!examples/*.json
-!paper/figures/cellprofiler_coverage_evidence.json
-!openhcs/resources/installer_contract.json
-!website/assets/agent/cold-start-workflow-record.json
-!website/assets/agent/neurite-outgrowth-workflow-record.json
-!website/assets/gallery/release-media-record.json
-!website/assets/gallery/source-capture-evidence.json
-!docs/source/development/mcp_knowledge_base_manifest.json
-!docs/audits/*.json
-!server.json
-!packaging/codex/openhcs/.mcp.json
-!packaging/codex/openhcs/.codex-plugin/plugin.json
-!packaging/mcpb/openhcs/manifest.json
-!benchmark/manifests/official23_plus_monolayer.json
-!benchmark/manifests/extra_provided_not_in_23_axis1.json
-!benchmark/manifests/official23_plus_extra6_axis1.json
-!benchmark/manifests/annotated_validation_20260915_preparation.json
-!benchmark/native_refs/**/.cellprofiler_benchmark_reference.json
 tests/integration/tests_data/
 
 # Embedded repositories

benchmark/converter/cppipe_corpus.py

@@ -32,6 +32,7 @@ class CPPipeCorpusCase:
     cppipe_path: Path
     status: CPPipeCorpusStatus
     source_root: Path | None = None
+    expected_error: type[Exception] | None = None
     expected_error_substring: str | None = None
 
 
@@ -54,8 +55,12 @@ def in_tree_cppipe_corpus() -> tuple[CPPipeCorpusCase, ...]:
             name="ExampleFly",
             cppipe_path=pipelines_dir / "ExampleFly.cppipe",
             status=CPPipeCorpusStatus.KNOWN_INVALID,
+            # LoadData names its images in a CSV file, and this pipeline's
+            # load_data.csv is not in the tree: it imports once that file is
+            # beside it, and says which file is missing until then.
+            expected_error=FileNotFoundError,
             expected_error_substring=(
-                "No CellProfiler module declaration is registered for 'LoadData'"
+                "LoadData(1) names its images in the CSV file 'load_data.csv'"
             ),
         ),
         CPPipeCorpusCase(

docs/archive/architecture/cellprofiler_openhcs_runtime_unification_plan.md

@@ -5044,10 +5044,6 @@ AnalysisConsolidationPlan
 RuntimeImageExportSpec
 RuntimeImageExportBitDepth
 _candidate_image_snapshots_for_equivalence
-LoadDataModule
-LabelImagesModule
-CreateBatchFilesModule
-SaveCroppedObjectsModule
 CellProfilerFunctionRuntimeMetadata
 CellProfilerFunctionCatalog
 CellProfilerFunctionReferenceTransportStrategy

openhcs/core/pipeline/funcstep_contract_validator.py

@@ -453,6 +453,23 @@ def validate_external_library_installation(func: Callable, step_name: str) -> No
         Raises:
             ValueError: If the external library required by the function is not installed
         """
+        from openhcs.processing.backends.lib_registry.openhcs_registry import (
+            distributions_not_installed,
+            install_command_for_distributions,
+        )
+
+        # A file's declared distributions name the extra that installs them.
+        not_installed = distributions_not_installed(func)
+        if not_installed:
+            raise ValueError(
+                missing_external_library_error(
+                    func.__name__,
+                    step_name,
+                    ", ".join(not_installed),
+                    install_command_for_distributions(not_installed),
+                )
+            )
+
         # Get the module name from the function
         module_name = func.__module__
         if module_name is None:

openhcs/core/processing_contracts.py

@@ -846,6 +846,35 @@ class TuplePure2DAuxiliaryOutputAggregator(FlatSequencePure2DAuxiliaryOutputAggr
     value_type = tuple
 
 
+class DictPure2DAuxiliaryOutputAggregator(Pure2DAuxiliaryOutputAggregator):
+    """Unite dict auxiliary outputs (such as file bundles) across slices.
+
+    Every slice contributes its own entries. A key two slices both return has
+    no single value, so it is an error rather than a silent overwrite.
+    """
+
+    value_type = dict
+
+    def aggregate_values(
+        self,
+        values: list[dict[object, object]],
+        memory_type: str,
+        *,
+        plane_axis: type[RuntimePlaneAxis],
+    ) -> dict[object, object]:
+        del memory_type, plane_axis
+        united: dict[object, object] = {}
+        for value in values:
+            repeated = sorted(str(key) for key in united.keys() & value.keys())
+            if repeated:
+                raise ValueError(
+                    "PURE_2D dict auxiliary outputs of different slices share the "
+                    f"keys {repeated!r}; each slice must name its own entries."
+                )
+            united.update(value)
+        return united
+
+
 
 
 # ---------------------------------------------------------------------------

openhcs/core/runtime_slice_projection.py

@@ -8,6 +8,7 @@
 from collections.abc import Callable, Iterable, Mapping, Sequence
 from dataclasses import dataclass, field
 from enum import Enum
+from pathlib import PurePath
 from typing import TYPE_CHECKING, Any, TypeAlias, overload
 
 import numpy as np
@@ -307,7 +308,7 @@ class RuntimeSliceProjection:
 
     Owned runtime values answer through ``RuntimeSliceProjectableValue``.
     Tuples and lists project item by item; other foreign values (arrays,
-    mappings, primitives, enums) carry no runtime-slice axis.
+    mappings, primitives, file paths, enums) carry no runtime-slice axis.
     """
 
     FOREIGN_VALUE_TYPES: tuple[type, ...] = (
@@ -318,6 +319,7 @@ class RuntimeSliceProjection:
         bytearray,
         int,
         float,
+        PurePath,
         Enum,
         DtypeConversionConfig,
         type(None),

openhcs/core/source_binding_workspace.py

@@ -20,6 +20,8 @@
     ComponentSelector,
     ImportedMetadataTable,
     NamedSourceBinding,
+    ListedSourceFile,
+    PairAsListed,
     PairByMetadata,
     PairByOrder,
     BroadcastSource,
@@ -502,6 +504,83 @@ def _matched_source_sets(
         )
 
 
+class ListedSourceSetAssembler(SourceSetAssembler):
+    """Build each source set from the files one listed set names."""
+
+    value_type = PairAsListed
+
+    def _matched_source_sets(
+        self,
+        match_plan: SourceSetPairing,
+        candidates_by_alias: Mapping[str, tuple[SourceCandidate, ...]],
+    ) -> tuple[_SourceSet, ...]:
+        listed_sets = cast(PairAsListed, match_plan).source_sets
+        if not listed_sets:
+            raise ValueError("Pairing as listed requires at least one listed source set.")
+        source_sets: list[_SourceSet] = []
+        for index, listed_set in enumerate(listed_sets):
+            unlisted = tuple(
+                alias
+                for alias in candidates_by_alias
+                if alias not in {listed.alias for listed in listed_set.files}
+            )
+            if unlisted:
+                raise ValueError(
+                    f"Listed source set {index + 1} names no file for matched "
+                    f"source aliases {unlisted!r}."
+                )
+            source_set = _source_set(
+                index,
+                {
+                    listed.alias: self._listed_candidate(
+                        index, listed, candidates_by_alias.get(listed.alias, ())
+                    )
+                    for listed in listed_set.files
+                },
+            )
+            if listed_set.metadata:
+                source_set = source_set.with_imported_metadata(
+                    {item.field: item.value for item in listed_set.metadata},
+                    path=f"listed source set {index + 1}",
+                )
+            source_sets.append(source_set)
+        return tuple(source_sets)
+
+    @staticmethod
+    def _listed_candidate(
+        index: int,
+        listed: ListedSourceFile,
+        candidates: tuple[SourceCandidate, ...],
+    ) -> SourceCandidate:
+        """The one plate file ``listed`` names: the closest path match, if unique."""
+
+        shared_parts = tuple(
+            max(
+                listed.shared_trailing_parts(path)
+                for path in candidate.source_filter_path_identities()
+            )
+            for candidate in candidates
+        )
+        closest = max(shared_parts, default=0)
+        if closest == 0:
+            raise ValueError(
+                f"Listed source set {index + 1} names {listed.path!r} for source "
+                f"alias {listed.alias!r}, but the plate has no such file."
+            )
+        matches = tuple(
+            candidate
+            for candidate, shared in zip(candidates, shared_parts, strict=True)
+            if shared == closest
+        )
+        if len(matches) > 1:
+            raise ValueError(
+                f"Listed source set {index + 1} names {listed.path!r} for source "
+                f"alias {listed.alias!r}, which matches several plate files: "
+                f"{tuple(match.relative_path for match in matches)!r}."
+            )
+        return matches[0]
+
+
 class SourceBindingProjectionStrategy(ABC, metaclass=AutoRegisterMeta):
     """Project one binding member through its declared projection role."""
 

openhcs/core/source_bindings.py

@@ -802,7 +802,14 @@ def keyed_by(
     ) -> SourceSetPairing:
         """This pairing keyed by ``dimensions``; pairing by order takes no keys."""
 
-        return cls()
+        return cls().with_keys(dimensions)
+
+    def with_keys(
+        self, dimensions: tuple[SourceBindingMatchDimension, ...]
+    ) -> SourceSetPairing:
+        """This pairing with its shared keys replaced and the rest of it kept."""
+
+        return self
 
     @staticmethod
     def named(name: str) -> type[SourceSetPairing]:
@@ -833,11 +840,10 @@ class PairByMetadata(SourceSetPairing):
     dimensions: tuple[SourceBindingMatchDimension, ...] = ()
     """Ordered shared metadata keys; each names one field per alias."""
 
-    @classmethod
-    def keyed_by(
-        cls, dimensions: tuple[SourceBindingMatchDimension, ...]
+    def with_keys(
+        self, dimensions: tuple[SourceBindingMatchDimension, ...]
     ) -> SourceSetPairing:
-        return cls(dimensions=dimensions)
+        return replace(self, dimensions=dimensions)
 
     def __post_init__(self) -> None:
         object.__setattr__(
@@ -851,6 +857,109 @@ def __post_init__(self) -> None:
         )
 
 
+@dataclass(frozen=True, slots=True)
+class ListedSourceFile:
+    """The file one listed source set names for one alias."""
+
+    alias: str
+    """Source-binding alias the file belongs to."""
+
+    path: str
+    """File name, optionally preceded by folders, with ``/`` separators.
+
+    A plate file is the listed one when one path ends with the other, compared by
+    whole path parts: ``a.tif`` names ``images/a.tif``, and so does
+    ``D:/run/images/a.tif``.
+    """
+
+    def __post_init__(self) -> None:
+        _require_name(self.alias, "ListedSourceFile.alias")
+        if not self.path_parts:
+            raise ValueError("ListedSourceFile.path cannot be empty.")
+
+    @property
+    def path_parts(self) -> tuple[str, ...]:
+        """The path's parts, file name last."""
+
+        return tuple(part for part in self.path.split("/") if part)
+
+    def shared_trailing_parts(self, plate_path: str) -> int:
+        """How many trailing path parts ``plate_path`` shares; 0 when it is another file."""
+
+        listed = self.path_parts
+        plate = tuple(part for part in plate_path.split("/") if part)
+        shared = min(len(listed), len(plate))
+        return shared if listed[-shared:] == plate[-shared:] else 0
+
+
+@dataclass(frozen=True, slots=True)
+class ListedMetadata:
+    """One metadata value a listed source set carries."""
+
+    field: str
+    """Metadata field name."""
+
+    value: SourceMetadataScalar
+    """The field's value for this source set."""
+
+    def __post_init__(self) -> None:
+        _require_name(self.field, "ListedMetadata.field")
+        object.__setattr__(self, "value", source_metadata_scalar(self.value))
+
+
+@dataclass(frozen=True, slots=True)
+class ListedSourceSet:
+    """One source set written out in full: its file per alias and its metadata."""
+
+    files: tuple[ListedSourceFile, ...] = ()
+    """The file of each alias in this set."""
+
+    metadata: tuple[ListedMetadata, ...] = ()
+    """Metadata of the set as a whole, such as its plate, well and site."""
+
+    def __post_init__(self) -> None:
+        files = normalize_source_binding_values(
+            "ListedSourceSet.files", self.files, ListedSourceFile
+        )
+        aliases = tuple(listed.alias for listed in files)
+        if len(set(aliases)) != len(aliases):
+            raise ValueError(
+                f"ListedSourceSet names an alias more than once: {aliases!r}."
+            )
+        object.__setattr__(self, "files", files)
+        object.__setattr__(
+            self,
+            "metadata",
+            normalize_source_binding_values(
+                "ListedSourceSet.metadata", self.metadata, ListedMetadata
+            ),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class PairAsListed(PairByOrder):
+    """Pair the aliases' sources as a table lists them, one source set per row.
+
+    The order is the table's: set N holds the files row N names, whatever order
+    the plate's files come in. After assembly the sets are ordinary ordered sets,
+    which is why this is a kind of ``PairByOrder``.
+    """
+
+    name = "listed"
+
+    source_sets: tuple[ListedSourceSet, ...] = ()
+    """The source sets, in order."""
+
+    def __post_init__(self) -> None:
+        object.__setattr__(
+            self,
+            "source_sets",
+            normalize_source_binding_values(
+                "PairAsListed.source_sets", self.source_sets, ListedSourceSet
+            ),
+        )
+
+
 @dataclass(frozen=True, slots=True)
 class ComponentSelector:
     """Component-axis key/value pair used either to select sources or assign identity."""

openhcs/interop/cellprofiler/compile_time_contracts.py

@@ -217,6 +217,7 @@ def provider_for_pipeline(
                         continue
                     scope = invocation.contract.execution_scope
                     try:
+                        module_type.validate_compilation(invocation)
                         invocation_blocks, consumed_kwarg_names = (
                             module_type.module_blocks_for_invocation(
                                 invocation=invocation,