Restore Official30 image subject, carrier and output qualification (#330)

docs/validation/official30_qualification_330_20261001.rst

@@ -0,0 +1,112 @@
+Official30 qualification after main synchronization (#330)
+=========================================================
+
+Production source ``4c40348354a0680850ac21cdc102d00e460558ee`` includes main
+``2cbfc4a0771bf24d153b076bff4a5e321ac039f4`` and its exact dependency pins.
+All 30 native-reference comparisons pass, with zero differences. The original
+six compile failures and the subsequent eight execution/output failures are
+resolved. This correctness gate enables representative performance work;
+it is not a speedup claim.
+
+The determining owners
+----------------------
+
+``MeasurementSubject.row_identity_domain`` derives compatible row identities
+without merging exact source-qualified image subjects. Common output admission
+uses the derived domain; payload policies still enforce their own cardinality.
+Ordinary/native single-subject payloads and incompatible scopes/object domains
+retain their rejection controls.
+
+Carrier admission follows the authoritative compiled predecessor through
+inherited bindings. ``CompiledStepPlan`` owns terminal-source projection, and
+every producer transition and terminal physical header remain mandatory.
+
+``StepInputDependency`` determines which anchors originate at pipeline start;
+inherited source bindings cannot discard predecessor/artifact-only lifecycle
+anchors. ``CallableContract.group_scope_inputs`` owns execution-domain source
+selection. The planner projects preliminary source bindings through those exact
+refs, then uses its existing producer/lineage resolver. ExampleFly's auxiliary
+OrigBlue filename source cannot restrict the composed RGBImage execution domain.
+The typed output manifest remains authoritative for actual producer pixels.
+
+``OpenHCSMetadataWriter.OutputTarget`` derives production and reconciliation
+targets. Runtime artifact targets enumerate actual declared writer destinations,
+and durable typed source projections survive execution-local value-store cleanup.
+The declared root remains included so unknown image addresses still fail closed.
+Materialization's existing record factory owns the empty-output-plan case;
+callers do not add absence guards around its execution-local store.
+
+Terminal and streaming materialization policies now live in the shared
+materialization package. Internal diagnostic planes declare persistence without
+external export intent. Source serialization preserves each nominal policy
+subtype, including its options, backends and write mode. No function-name
+exceptions, ignored extra science outputs, generated-name coordinate inference
+or CSV normalization are used.
+
+Validation and limits
+---------------------
+
+* 586 local tests and 20 subtests pass; one existing optional Napari skip.
+* Original CI-pinned R0 passes for OpenHCS, scripts and benchmark, without increases.
+* Original NRA R1 passes its two policy-configured detectors with complete
+  dependency context and no increases; this is not an all-detector clean claim.
+* Original ClassDef census: 5,122 declarations, 5,110 projected, all 12 OPEN retained.
+* NRA admitted the two-class policy relocation and its import-consumer closure.
+  The initial binding-authority/annotation boundary rejection is retained.
+  Authored semantic changes have separate executed tests and native comparison.
+* Native progression: 22/30, then 29/30, then 30/30 on the synced source above.
+  All final difference counts are zero. Existing 1e-6 absolute/relative numerical
+  tolerances, exact discrete/identity checks and declared-image checks remain.
+
+Tests include real image writes into two declared destinations, cleanup and
+durable reopen with exact metadata/pixels, unknown-address rejection, implicit
+main-flow controls, stored cross-axis and direct-source payload domains, and
+three materialization-policy source roundtrips. Three stale fixtures were
+reproduced on unchanged main and repaired from authoritative declarations/source
+fields and the existing shared ACK fixture; scientific assertions remain strict.
+
+The final gate uses the production comparison suite with the retained native
+outputs from ``official30-native-complete-fresh-20260928``. Those outputs are
+scientific references. The native clocks and speedup columns in the raw reports
+are retained observations, **not fresh comparative performance evidence**. This
+gate does not regenerate native clocks or scaling figures and does not assert
+complete parity for every possible ROI/export profile outside the suite.
+
+Ready-server startup and shutdown are excluded from pipeline clocks. The suite
+uses one inline worker, one numerical thread, CPU5, and no memory observer.
+Comparison and qualification work are separate from execution timing. Tests,
+audits, builds and other replays do not overlap timed pipeline work.
+
+Receipts and reproduction
+------------------------
+
+``official30_qualification_330_evidence_20261001.tgz`` contains source and
+dependency receipts, exact task recipes, original guards/census, local test XML,
+unchanged-main controls, NRA move evidence, the saved ExampleFly anchor trace,
+all three qualification summaries/observations, and the final 30 generated
+pipeline sources and measured receipts. ``SHA256SUMS`` authenticates its files.
+Large scientific inputs/outputs remain in the benchmark-run workspace.
+
+Extract the archive, check out the validated production revision, and use its
+final qualification recipe with the configured benchmark Python and retained
+native-reference root. The original controller environment was:
+
+.. code-block:: bash
+
+   env PYTHONPATH="$PROJECT_ROOT" \
+       OPENHCS_REFERENCE_EXPORT_PIPELINES_ROOT="$PROJECT_ROOT/benchmark/reference_exports/official30_value_completion_20260914" \
+       OPENHCS_CPU_ONLY=true NUMBA_CACHE_DIR="$SHARED_NUMBA_CACHE" \
+       OMP_NUM_THREADS=1 OPENBLAS_NUM_THREADS=1 MKL_NUM_THREADS=1 \
+       NUMEXPR_NUM_THREADS=1 VECLIB_MAXIMUM_THREADS=1 PYTHONHASHSEED=0 \
+       taskset -c 5 "$BENCHMARK_PYTHON" "$QUALIFICATION_RECIPE"
+
+The recipe records its source assertion, manifest, retained native-reference
+receipt and production ``run_comparison_suite`` arguments. Its local paths must
+resolve to those retained inputs. If they are unavailable, produce fresh native
+outputs using the public comparison route and configured CellProfiler environment.
+The public throughput route remains ``scripts/benchmark_cppipe_well_throughput.py``
+with ``--manifest benchmark/manifests/official30_portable_axis1.json --mode 1w_1t``.
+
+This change fixes #330 and references #315, #214 and #324. It does not close
+#315's separate parent/live acceptance. Generic runtime ownership remains PR326,
+and execution/compilation performance work continues under #162.

openhcs/core/artifacts.py

@@ -708,9 +708,16 @@ def require_output_subject(
 
     @classmethod
     def validate_output_declaration(cls, spec: "ArtifactSpec") -> None:
-        """Explicit subject relations must agree under every recording owner."""
+        """Reject incompatible row domains before payload-owner validation.
 
-        ArtifactSpecRelation.measurement_subject_for_output(spec)
+        Source-qualified image rows share one image-set domain. Their recording
+        owner determines whether multiple subjects can occupy the output; native
+        and ordinary recorded tables still require one exact subject.
+        """
+
+        subjects = ArtifactSpecRelation.measurement_subjects_for_output(spec)
+        if len({subject.row_identity_domain for subject in subjects}) > 1:
+            ArtifactSpecRelation.measurement_subject_for_output(spec)
 
     @classmethod
     def validate_native_output_declaration(cls, spec: "ArtifactSpec") -> None:
@@ -1271,19 +1278,27 @@ def measurement_subject(self) -> "MeasurementSubject | None":
         return None
 
     @staticmethod
-    def measurement_subject_for_output(
+    def measurement_subjects_for_output(
         output: "ArtifactSpec | ArtifactOutputPlan",
-    ) -> "MeasurementSubject | None":
-        """Resolve the sole subject from an output's original relation owners."""
+    ) -> "tuple[MeasurementSubject, ...]":
+        """Derive all exact subjects from the original relation owners."""
 
-        subjects = tuple(
+        return tuple(
             dict.fromkeys(
                 subject
                 for relation in output.relations
                 for subject in (relation.measurement_subject(),)
                 if subject is not None
             )
         )
+
+    @staticmethod
+    def measurement_subject_for_output(
+        output: "ArtifactSpec | ArtifactOutputPlan",
+    ) -> "MeasurementSubject | None":
+        """Resolve one exact subject for a table-wide subject owner."""
+
+        subjects = ArtifactSpecRelation.measurement_subjects_for_output(output)
         if len(subjects) > 1:
             raise ValueError(
                 f"Artifact output {output.ref()!r} declares multiple measurement "
@@ -2008,8 +2023,7 @@ def merge_existing(
             and existing.sidecar_role is not incoming.sidecar_role
         ):
             raise ValueError(
-                f"Conflicting {self.role} artifact sidecar role for "
-                f"'{incoming.name}'."
+                f"Conflicting {self.role} artifact sidecar role for '{incoming.name}'."
             )
         sidecar_role = (
             existing.sidecar_role

openhcs/core/compiled_step_plan.py

@@ -234,7 +234,9 @@ def has_persistent_target(self) -> bool:
 
     def persists_to_backend(self, backend: str) -> bool:
         """Own admission of a storage target to persistent artifact publication."""
-        return self.has_persistent_target and self.require_persistent_backend() == backend
+        return (
+            self.has_persistent_target and self.require_persistent_backend() == backend
+        )
 
     def require_persistent_backend(self) -> str:
         if self.persistent_backend is None:
@@ -313,6 +315,14 @@ class CompiledStepPlan:
     chainbreaker: bool = False
     error: str | None = None
 
+    @property
+    def requires_terminal_source_projection(self) -> bool:
+        """Require an exact source anchor when declared bindings have no predecessor."""
+        return (
+            bool(self.source_binding_plan.binding_declarations)
+            and self.main_input_dependency.predecessor_step_index() is None
+        )
+
     def require_function_execution_ready(self) -> "CompiledStepPlan":
         """Validate the compiler-owned fields required by FunctionStep runtime."""
         if not self.axis_id:

openhcs/core/function_patterns.py

@@ -807,6 +807,11 @@ class CompiledFunctionPattern:
     groups: tuple[CompiledFunctionGroup, ...]
     is_grouped: bool
 
+    @property
+    def runtime_domain(self) -> RuntimeInvocationDomain:
+        """Derive the shared anchor domain from the compiled invocations."""
+        return RuntimeInvocationDomain.from_invocations(tuple(self.iter_invocations()))
+
     @property
     def execution_scope(self) -> FunctionStepExecutionScope:
         """Return the uniform lifecycle scope derived from its invocations."""

openhcs/core/pipeline/artifact_planning.py

@@ -33,28 +33,11 @@
     CsvOptions,
     ImageFileOptions,
     MaterializedFilenameIdentity,
-    MaterializationSpec,
     ROIOptions,
+    StreamingOnlyMaterializationSpec,
+    TerminalMaterializationSpec,
 )
 
-
-class TerminalMaterializationSpec(MaterializationSpec):
-    """Compiler-added persistence excluded from declared export comparison."""
-
-    def participates_in_runtime_export_observation(self) -> bool:
-        return False
-
-
-class StreamingOnlyMaterializationSpec(MaterializationSpec):
-    """Compiler-added viewer materialization excluded from persistent exports."""
-
-    def participates_in_runtime_export_observation(self) -> bool:
-        return False
-
-    def participates_in_persistent_materialization(self) -> bool:
-        return False
-
-
 class AutomaticArtifactOutputMaterializationStrategy(
     ArtifactTypeStrategyMatchMixin,
     MostDerivedContextStrategyMixin[type[ArtifactType]],

openhcs/core/pipeline/compiler.py

@@ -1451,7 +1451,7 @@ def validate_primary_image_carrier_requirements(
                         source_binding_plan is None
                         or not source_binding_plan.primary_plane_bindings
                     ):
-                        if plan.source_binding_plan.binding_declarations:
+                        if plan.requires_terminal_source_projection:
                             failures.append(
                                 f"{owner}: no exact primary source-binding "
                                 "projection represents this callable's main flow."
@@ -1611,7 +1611,7 @@ def _carrier_source_plan(
                 and current_source_binding_plan.primary_plane_bindings
             ):
                 return producer, current_source_binding_plan
-            if producer.source_binding_plan.binding_declarations:
+            if producer.requires_terminal_source_projection:
                 failures.append(
                     f"{owner}: carrier requirement {requirement.value!r} has no "
                     f"exact primary source-binding projection at producer step "

openhcs/core/pipeline/path_planner.py

@@ -323,6 +323,19 @@ def get_execution_groups(
             if input_component_scopes is None
             else input_component_scopes
         )
+        artifact_owned = bool(contracts) and all(
+            contract.group_scope_inputs for contract in contracts
+        )
+        if artifact_owned:
+            source_bindings = (
+                self.planner.source_bindings_for_snapshot(snapshot)
+                if source_bindings is None
+                else source_bindings
+            ).for_artifact_refs(
+                spec.ref()
+                for contract in contracts
+                for spec in contract.group_scope_inputs
+            )
         scope = component_scopes.scope_for_group_by(
             group_by,
         )
@@ -337,7 +350,7 @@ def get_execution_groups(
                 if source_scope.is_ungrouped
                 else source_scope
             )
-        if contracts and all(contract.group_scope_inputs for contract in contracts):
+        if artifact_owned:
             scope = self.artifact_owned_execution_scope(
                 snapshot,
                 contracts,

openhcs/core/runtime_measurements.py

@@ -1605,6 +1605,17 @@ def source_image_name(self) -> str | None:
             return None
         return self.name
 
+    @property
+    def row_identity_domain(self) -> tuple[MeasurementScope, str | None, str | None]:
+        """Return the row domain independently of an image's source qualifier.
+
+        Named image subjects share image-set row identity. Object and relationship
+        names identify different row domains and cannot be combined implicitly.
+        """
+
+        name = None if self.scope is MeasurementScope.IMAGE else self.name
+        return self.scope, name, self.id_field
+
     @property
     def object_name(self) -> str | None:
         """Return the concrete object set represented by this subject, if any."""

openhcs/core/step_dependencies.py

@@ -98,6 +98,10 @@ def predecessor_step_index(self) -> int | None:
             )
         return self.source_step_index
 
+    def uses_pipeline_start_anchors(self) -> bool:
+        """Only a pipeline-start main input is filtered against raw source bindings."""
+        return self.kind.is_pipeline_start
+
     def require_pipeline_start(self) -> None:
         """Prove that this terminal dependency reaches the pipeline source."""
 

openhcs/core/steps/function_artifact_materialization.py

@@ -1201,12 +1201,11 @@ def runtime_artifact_materializations(
     """Derive actual materializations from compiled outputs and runtime values."""
 
     materializations: list[RuntimeArtifactMaterialization] = []
-    store = context.runtime_value_store
     for output_plan in plan.artifact_outputs.values():
         if output_plan.materialization is None:
             continue
         records = actual_materialization_records(
-            store=store,
+            store=context.runtime_value_store,
             plan=plan,
             output_plan=output_plan,
         )

openhcs/core/steps/function_execution.py

@@ -265,8 +265,7 @@ def execution_group_anchor_patterns(
 
         pattern = self.plan.compiled_function_pattern
         source_owns_groups = (
-            RuntimeInvocationDomain.from_invocations(tuple(pattern.iter_invocations()))
-            is RuntimeInvocationDomain.SOURCE_ANCHORED
+            pattern.runtime_domain is RuntimeInvocationDomain.SOURCE_ANCHORED
         )
         if not grouped_patterns.groups:
             return grouped_patterns
@@ -316,7 +315,7 @@ def source_bound_anchor_patterns(
     ) -> PatternGroups:
         """Restrict source-bound step anchors to compatible declared sources."""
 
-        if self.plan.main_input_dependency.kind is StepInputDependencyKind.STEP_OUTPUT:
+        if not self.plan.main_input_dependency.uses_pipeline_start_anchors():
             return grouped_patterns
         if not self.plan.source_binding_plan.has_primary_content:
             return grouped_patterns