diff --git a/docs/validation/site-free-aggregate-publication-728-20261005.rst b/docs/validation/site-free-aggregate-publication-728-20261005.rst
new file mode 100644
index 0000000000..6de3d2bf52
--- /dev/null
+++ b/docs/validation/site-free-aggregate-publication-728-20261005.rst
@@ -0,0 +1,327 @@
+Site-free acquired aggregate publication: receiving checkpoint
+============================================================
+
+Singer owns issue728's aggregate publication and compiled-table relation.
+Base: main91dc36440ad3ab95673d29a2ca1cf017654468c8. Both production families
+below are implemented. Focused source controls pass; ordinary whole-installed
+default-publication acceptance remains pending.
+The previous task-transport branch4ab5096 is published and retained; the
+same isolated checkout is reused without changing eight foreign gitlinks or
+untracked evidence. No scientific input, active installation or runtime changes.
+
+Original determining evidence
+-----------------------------
+
+The complete parent receipt is retained at::
+
+  /home/ts/wt/openhcs-artifact-planning-publication-315-20260930/
+  figure-collection-20261004/P001-FRESH09-AGGREGATE-PUBLICATION-DEFECT.rst
+
+Original6022 native log1072--1103 records default finalization through
+ArtifactMaterializationTargetPlan, ViewerStreamBackendCallKwargs._output_fields and
+StreamViewerComponentMetadataProjector.project_required rejecting a site-free
+acquired mosaic. Independent candidate01t02 numerical completion records
+compiled aggregate table expectations missing site/source_image_name fields.
+These are different manifestations until their source-owner relation is proven;
+missing schema fields do not authorize fabricated scalar coordinates.
+Original failed jobs and source documents remain unchanged. No biological claim.
+
+Ownership and coordination
+--------------------------
+
+Current open727 touches introspection/registry/dependency metadata, not this
+materialization or stream family. Closed394/435/630 are not an owner queue.
+Planck explicitly released this stream/materialization scope and owns separate
+731 image-provenance spreadsheet projection. Shared row-projection hunks will
+be coordinated directly before edits. Planck also owns the distinct photometry
+fixture correction: its original failed forward job is not this aggregate issue.
+
+The NRA/refactor-audit source-first method applies: enumerate declarations,
+writes, decisions, inheritance and all consumers; read those sites; put scope
+behavior on its existing owner and delete competing decisions. Boundary,
+identity, membership and implementation catalog entries inform the review.
+No global R1 or complete behavioral proof is claimed by an AST subset.
+
+Required relation and acceptance
+--------------------------------
+
+An acquired aggregate retains an honest source frame, contributor provenance,
+calibration and applicable component domains without pretending to be a single
+acquired SITE. The original materialization/display owners must consume that
+relation. Table validation must use each original producer's own saved-output
+address and schema, not apply a preceding field record to an aggregate CSV.
+Genuinely incomplete acquired planes and strict final publication remain errors.
+No kind/string switch, second registry, metadata mirror, filename inference,
+alias, alternate codec, weakened guard or timeout increase is permitted.
+
+Source checks follow a coherent owner-and-consumer implementation. Installed
+acceptance then uses a tiny synthetic multisite assembly via ordinary compile,
+execute, default publication, public inventory and receipt-backed native reopen;
+exact pixels, contributors, calibration and table structure must agree. A new
+declaration/capability must work without generic consumer edits. Source,
+packaging, installed/public and biological readiness remain separate tiers.
+
+Working production checkpoint
+-----------------------------
+
+SourceImageProvenance.required_scalar_components owns scalar requiredness.
+Only complete, varying addresses on original nominal pixel contributors prove
+a reduced coordinate; absent evidence, incomplete contributors and acquired
+runtime planes grant no exemption. No component name has a special case.
+StreamScopedDisplayConfig consumes that owner's projection. The existing
+ViewerStreamBackendCallKwargs applies the scoped original display declaration
+consistently to single and grouped batches, retaining item metadata, producer,
+transport and display modes. The previous unscoped materialization decision is
+replaced; StreamViewerComponentMetadataProjector.project_required is unchanged.
+
+The table diagnosis is narrower than a dropped-column claim. The replaced
+matching_table_outputs used artifact-name/axis filename matching without the
+exact producer location; a field record can therefore be checked against a
+later aggregate's same-name CSV. The original report repeats aggregate header
+failures nine times. StepExecutionObservation already owns typed
+materialized_locations_by_address. RuntimeContextObservation now carries that
+original StepExecutionObservation instead of copying its export paths and
+analysis inputs while dropping its address relation. Both static/sequential
+worker paths and compiled plate execution use StepExecutionObservation.combine;
+analysis consolidation reads that same owner. Dewey explicitly reports no
+overlapping orchestrator source claim.
+
+RuntimeExportObservation carries the same original observation. Its exact
+record-to-path projection delegates to StepExecutionObservation.paths_for(),
+which uses RuntimeArtifactAddress.from_record(). Final table validation and
+artifact-scoped equivalence use that relation. The free filename matchers are
+deleted, with no fallback or alias. File-only observations remain valid file
+inventories but cannot establish table ownership. Existing schema, row-count
+and missing-export guards remain unchanged; no nullable-column waiver was added.
+
+Applicable catalog entries: BOUND-8 (owned address relation flattened at the
+worker boundary), BOUND-2 (bypassed StepExecutionObservation), IDEN-6 (filename
+matching instead of producer address), and IMPL-12 (duplicated output folding).
+The existing ancestor owns both projection algorithms; no new production type,
+registry, codec, copied procedure or ornamental inheritance was introduced.
+
+Existing refactor-audit parser invocation and byte-exact AST output are retained
+at engineering728/source01/family.jsonl and family.stderr under the issue-batch
+root. All701 production and705 test modules parsed;139 production and168 test
+modules selected by the declared family terms. Terminal0,25.34s,550360KiB peak
+RSS,0swap. This is source evidence, not global R1 or behavioral acceptance.
+Resource readback showed13.5GiB available; existing swap/disk warnings were not
+turned into invented memory caps. No imports of OpenHCS or scientific execution.
+
+The second original-parser receipt is source01/export-family.jsonl plus stderr:
+all701 production/705 test modules parsed,58/80 selected, zero omissions;
+terminal0,22.03s,550156KiB peak RSS,0swap. Complete source-family ASTs and
+semantic reads cover the original observation, writer-address creation, both
+execution paths, consolidation, final validation and equivalence consumers.
+No authenticated global R1 proof is claimed.
+
+Source qualification and preserved first failures
+------------------------------------------------
+
+All raw outputs below are retained under
+/home/ts/wt/openhcs-issue-batch-20260929/engineering728/.
+Checks reuse engineering620/source-controls01.py, its authenticated original
+engineering599 source-runtime dependencies and the existing paired Python/native
+binary. No installation, download, catalog, server or viewer was launched.
+
+source-controls01: collection failed because the original importlib-mode runner
+does not put tests/unit on the module search path. Its original stdout/stderr
+are preserved. Explicitly selecting the existing unit fixture path corrected
+only the launch environment, not product imports or algorithms.
+
+source-controls02:37PASS/2FAIL. The one/equal contributor fixtures unintentionally
+retained a valid common scalar SITE; the original provenance owner correctly
+restored it. They now declare genuinely absent scalar identity on the existing
+typed contributor carrier. This is fixture correction, not weaker assertions.
+Peak314624KiB,5.10s,0swap.
+
+source-controls03:310PASS/4FAIL/1ERROR. All new source-scope controls pass,
+including mixed acquired/reduced batches, both kwargs/batch paths, zero/one/equal
+and incomplete contributor negatives, and present scalar override. The broader
+related family exposed one old equivalence fixture that relied on filename
+ownership; it now declares its exact writer output address. Three unrelated
+server fixtures hit the borrowed dependency API: two progress-queue assertions
+and missing ObjectState.resolve_saved_object. The --noconftest runner also did
+not load the original viewer ACK fixture. These originals remain visible, not
+reported as whole-family passing. Peak482948KiB,19.29s,0swap.
+
+source-controls04:10PASS,3.52s pytest time. The independent reduced-field
+declaration composes a real cooperative ProjectionAudit hook through C3 MRO;
+both before/after hooks execute through ImageArtifactType contextualization and
+the unchanged generic stream consumer. No generic consumer edit adds this case.
+Same-named field and aggregate tables retain their exact distinct producer paths
+through ordinary pickle transport. Missing address cannot fall back to a neighbor
+filename; correct address still rejects missing site/source_image_name columns.
+The migrated equivalence fixture excludes the auxiliary CSV without name guesses.
+
+Final source checkpoint
+-----------------------
+
+Production428584584 is qualified by source-controls06:41PASS, terminal0,
+6.61s wall time,322156KiB peak RSS,0swap. This repeats only the final affected
+owner/transport/validation batch after the structural ownership correction;
+the310 earlier related passes were not repeated. source-controls05's41PASS
+precede that final projection change and remain a distinct original receipt.
+
+The original refactor-audit debt_census.py ran over all nine changed production
+files against main91dc36440. r0-01 recorded +1 foreign absence probe and +2 None
+checks. These were acted on, not waived: SourceMetadataFields.component_domains
+now owns admitted component membership; complete singleton contributor domains
+are required on at least two contributors. Output.viewer_display_config owns
+its optional payload; generic materialization no longer probes that absence.
+r0-02 has zero delta for every structural counter, no exclusions or raised
+bound, and +35 code lines. Its per-file report remains included. A separate
+historical agent-comms R0 importer --help invocation failed in that old package's
+unrelated InputDocument import; it supplied no measurement and is not substituted
+for the actual source census.
+
+source01/family-after01.jsonl and stderr record all701 production/706 test
+modules,82/151 relevant complete ASTs, zero parse omissions; terminal0,
+27.04s,549972KiB peak RSS,0swap. Final searches find table matching only at
+RuntimeExportObservation's two typed consumers; the free name matchers and
+flattened RuntimeContextObservation fields are absent. The original
+StepExecutionObservation.combine still owns its own export paths/analysis inputs.
+The added declared field needs only its own projection hook; cooperative audit
+hooks and unchanged generic materialization execute, not just inheritance checks.
+
+Normal integration of current main755d9f756 created c473a814e. Its five incoming
+files are guides/manifest/paper only: all production and test bytes equal
+qualified428584584. Eight foreign gitlinks and prior untracked evidence remain
+unchanged. Planck owns the one ordinary candidate builder and has the source
+pin/receipt; no receiving10 package overlay or scientific hot replacement.
+
+Remaining: an ordinary tiny whole-installed multisite default-publication
+journey, including actual acquired/reduced scope, persisted pixel/contributor/
+calibration checks and exact table producer-address validation. Planck's
+independent734 spreadsheet/source-column qualification is not held behind this
+receiving gap. No lane, client or viewer was launched by this source work.
+The three active scientific bundles and their original failures remain untouched.
+
+Byte-exact source and original AST/control logs are published in
+site-free-aggregate-publication-728-20261005.tar.gz (20626007bytes), SHA256
+83edab6e2af3c7fd9882ab694af7e5ca45254fb493944f63bcad429009480e79.
+The archive contains no scientific images, installed targets or UNKNOWN attempt
+replay. Original loose receipts remain retained in engineering728; no scratch
+cleanup was performed while receiving and review still need those paths.
+
+Prepared ordinary receiving packet
+----------------------------------
+
+validation/site-free-aggregate-publication-728/RECEIVING.rst now carries the
+concrete source-only next packet, complete pipeline.py and prepare_inputs.py.
+Four tiny embedded-header acquisitions (sites1/3, channels1/2) drive registered
+per-field measurements, original acquisition positions, CPU assembly and
+aggregate measurements. Default final persistence and streaming remain enabled;
+acquired/aggregate checkpoints do not bypass automatic publication. All actual
+schema names, producer addresses and paths must come from the first public plan
+and native observation, never guessed filenames or fabricated source receipts.
+
+The only added check was standard-library ast.parse on both new fixture files,
+terminal0; it imports no OpenHCS/scientific/runtime package and proves syntax
+only. No source-controls batch was repeated. The prepared packet has not been
+materialized, installed, compiled or submitted. Native6012/viewer6013 reuse is a
+proposal requiring the original receiving owner's explicit family handoff,
+not a new reservation. Planck's current734 receiving11 operation stays immutable.
+
+Current production and the source41 qualification are unchanged. This source
+packet is not part of the earlier byte-exact archive and does not alter its hash.
+Planck received a separate release of his original projected-Z message-extra
+and source-metadata producer hunks; no competing Points guard/fallback fix was
+started here. Coordinate any SourceImageProvenance/shared-scope crossing.
+
+Saved label receiving scope correction
+--------------------------------------
+
+The parent supplied the closed P001 assisted output-plate label request as a
+new read-only determining witness. Exact REPORT/journal/log/source metadata
+relationship is retained in the existing receiving packet's Saved-label section.
+Actual napari6013.log43..58 names a FIELD site5 label failing scalar CHANNEL,
+not a site-free mosaic. Saved metadata declares two channel source planes with
+empty contributors and a2D source spatial domain, while its scalar source
+metadata lacks CHANNEL. This is a producer/plane-versus-contributor route
+contract, not407 typed presentation or a transport timeout.
+
+732's current production does not alter the manual saved-image path or turn
+those declared source planes into contributors. No installed fix is claimed.
+Planck/Dewey existing152 producer ownership was contacted with the exact source
+sites and receiving requirement; Singer retains integration/control custody.
+The same tiny ordinary installed packet now explicitly requires saved-label
+reopening qualification after the existing producer owner supplies the correct
+nominal declaration. No new production patch, runtime, scientific replay,
+source-receipt fabrication or repeated test batch occurred.
+Installed case02: acquired-site CSV collision
+-------------------------------------------
+
+The original public receiving12 case02 completed execution
+``e6a9522f-4af4-4880-a4c9-68dd3a823961`` with four steps and no execution
+errors. Default and checkpoint aggregate images retain all 96 expected uint16
+pixels across two channels, SITE-free scalar metadata, two original contributor
+paths per channel, and Y/X spacing 0.25/0.5 micrometers. This positive result
+does not establish complete table-publication acceptance.
+
+``receiving12/public95/verify-case02-03.log`` proves a distinct failure: the
+acquired SITE1/channel1 record retains total intensity 0.004028381779789925,
+but its address-bound exported CSV contains SITE3's 0.004760814 value. The
+different original runtime addresses share a channel-only memory location;
+publication reused that location's filename and overwrote the first CSV.
+Original inputs, job, exports, CSVs and journals are preserved. No execution
+was repeated to obtain this observation. Earlier verifier failures concerned
+its incorrect assumptions about normalized image pixels and runtime-plane
+versus contributor metadata; they are retained separately.
+
+Planck takes the existing publication/path family from Singer. The source
+repair removes the runtime-location filename branch in
+``RuntimeArtifactMaterialization._analysis_identity``. The existing
+``_aggregate_identity`` consumes the complete ``RuntimeExecutionAxisScope``
+for an artifact-named occurrence. Actual source-filename writers still use
+their original admitted source identity. No fixture rename, duplicate writer,
+new identity store, source reload or guard relaxation repairs the failure.
+Applicable ownership pattern: IDEN-6, storage location mistaken for record
+identity. The automatic-measurement strategy, writer, observation export,
+consolidation and final consumers continue to use the existing address relation.
+
+Focused controls cover both insertion orders, original automatic terminal CSV
+declarations, explicit source-named CSV declarations, shared runtime locations,
+exact saved row values and distinct address-bound paths. Historical tests that
+asserted a memory filename as persistent identity are migrated to the existing
+scope-owned identity without weakening their row assertions. Source checks and
+corrected ordinary installed/public acceptance remain separate; the original
+receiving12 target is immutable.
+
+The same original client then reloaded the two saved default images through
+ordinary output-plate inventory. Pixels and physical calibration survived, but
+the new manual route incorrectly acquired SITE1 from the retained filename;
+the compiled route for the exact same image was SITE-free. Before/after native
+state and the original refused explicit-result-directory requests are retained.
+The existing ``PlateImageInventory._record`` was overwriting canonical source
+metadata with parsed filename fields. It now parses filenames only when no
+explicit source metadata exists. Browser records and streaming consume the
+same original metadata, including meaningful absence of a collapsed component.
+The source control loads an actual TIFF through inventory and streaming and
+checks every pixel, SITE absence, calibration and original contributor lineage.
+
+The original95 native and viewer both closed through their exact typed owners;
+client38241 reached terminal exit1. ``receiving12/public95/OWNER-TERMINAL.rst``
+and the closed original journal seals retain this disposition. No corrected
+runtime acceptance has been claimed or replayed into the frozen output.
+
+Address-family controls02: 79 passed, terminal0; controls01 retained 70 passes
+and seven failures (five old memory-name assertions, two new source-filename
+expectations). Those tests now distinguish ordinary automatic artifact-named
+CSV output from explicit source-named output; exact row/value assertions remain.
+Reload controls01 retained63 passes/three failures: a new fixture omitted its
+required execution scope, an old OME assertion expected parsed int instead of
+the canonical source string, and an unrelated old point-model fixture lacks
+the existing native ``element_identity`` capability. The last is not a claimed
+production or #732 regression, and no fractional-Z guard is weakened for it.
+
+Final reload controls02: 27 passed, terminal0, 8.41s pytest time. Together with
+the79 address-family controls02, 106 affected controls pass. The two source
+changes delete six net production lines and add no authority, registry or
+codec. Source-address-family01.ast uses the existing refactor-audit parser on
+14 related production owners/consumers before and after: 28 complete ASTs,
+zero parse omissions. Original broader source-family ASTs remain retained;
+this update is focused source evidence, not global R1 or installed acceptance.
+The failed public case remains immutable. A future ordinary package and a
+distinct public case must verify all acquired table addresses and saved
+SITE-free image reload before this PR can claim complete publication acceptance.
diff --git a/docs/validation/site-free-aggregate-publication-728-20261005.tar.gz b/docs/validation/site-free-aggregate-publication-728-20261005.tar.gz
new file mode 100644
index 0000000000..ffc08c0bf3
Binary files /dev/null and b/docs/validation/site-free-aggregate-publication-728-20261005.tar.gz differ
diff --git a/openhcs/core/orchestrator/analysis_consolidation.py b/openhcs/core/orchestrator/analysis_consolidation.py
index 13a4b4949a..2d87fc1b88 100644
--- a/openhcs/core/orchestrator/analysis_consolidation.py
+++ b/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
diff --git a/openhcs/core/orchestrator/compiled_plate_execution.py b/openhcs/core/orchestrator/compiled_plate_execution.py
index ec2a3d1170..5d7f0d6b59 100644
--- a/openhcs/core/orchestrator/compiled_plate_execution.py
+++ b/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()
diff --git a/openhcs/core/orchestrator/execution_result.py b/openhcs/core/orchestrator/execution_result.py
index 60ddd28926..a3a9273c79 100644
--- a/openhcs/core/orchestrator/execution_result.py
+++ b/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,
         )
 
 
diff --git a/openhcs/core/orchestrator/worker_execution.py b/openhcs/core/orchestrator/worker_execution.py
index e9bdca7503..aa4736ac71 100644
--- a/openhcs/core/orchestrator/worker_execution.py
+++ b/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),
                 ),
             )
         ),
diff --git a/openhcs/core/plate_image_inventory.py b/openhcs/core/plate_image_inventory.py
index cbc5e123c7..0dd247b55a 100644
--- a/openhcs/core/plate_image_inventory.py
+++ b/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)
diff --git a/openhcs/core/runtime_exports.py b/openhcs/core/runtime_exports.py
index 8712395dfd..d58db985e9 100644
--- a/openhcs/core/runtime_exports.py
+++ b/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,37 +321,6 @@ 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,
@@ -343,10 +328,7 @@ def _table_artifact_failures(
 ) -> 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 "
diff --git a/openhcs/core/source_image_provenance.py b/openhcs/core/source_image_provenance.py
index b23f9afe55..3b2736d6a1 100644
--- a/openhcs/core/source_image_provenance.py
+++ b/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,
diff --git a/openhcs/core/steps/abstract.py b/openhcs/core/steps/abstract.py
index 197a5687b4..0b1cf2e19f 100644
--- a/openhcs/core/steps/abstract.py
+++ b/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"]
diff --git a/openhcs/core/steps/function_artifact_materialization.py b/openhcs/core/steps/function_artifact_materialization.py
index b7d1712c7b..6240c0c2a3 100644
--- a/openhcs/core/steps/function_artifact_materialization.py
+++ b/openhcs/core/steps/function_artifact_materialization.py
@@ -657,15 +657,9 @@ def _analysis_identity(
             )
             if source_identity is not None and not source_identity.addressable:
                 source_identity = None
-            if artifact_path is not None and (
-                dict_key is not None
-                or cls.source_identity_for_path(context, artifact_path) is not None
-            ):
-                return (
-                    f"{Path(artifact_path).stem}.roi.zip",
-                    source_identity,
-                    None,
-                )
+            # The in-memory location is shared by records with distinct runtime
+            # scopes. Only the admitted record, not that location's filename,
+            # establishes which persistent occurrence is being published.
             return cls._aggregate_identity(
                 output_key,
                 plan,
diff --git a/openhcs/core/steps/stream_component_semantics.py b/openhcs/core/steps/stream_component_semantics.py
index fd7511e1ba..5b8bd2b168 100644
--- a/openhcs/core/steps/stream_component_semantics.py
+++ b/openhcs/core/steps/stream_component_semantics.py
@@ -43,6 +43,7 @@
 from openhcs.core.source_image_provenance import (
     SourceComponentMetadata,
     SourceImageIdentity,
+    SourceImageProvenance,
 )
 from openhcs.core.source_matching import (
     source_component_metadata_raw_value,
@@ -246,6 +247,17 @@ class StreamScopedDisplayConfig(ViewerDisplayConfigABC):
     base: ViewerDisplayConfigABC
     component_order: tuple[str, ...]
 
+    @classmethod
+    def for_source_provenance(
+        cls,
+        base: ViewerDisplayConfigABC,
+        provenance: SourceImageProvenance,
+    ) -> ViewerDisplayConfigABC:
+        """Consume the source owner's scalar-coordinate requirements."""
+        order = tuple(str(component) for component in base.COMPONENT_ORDER)
+        required = provenance.required_scalar_components(order)
+        return base if required == order else cls(base, required)
+
     @property
     def COMPONENT_ORDER(self) -> tuple[str, ...]:
         return self.component_order
diff --git a/openhcs/processing/materialization/core.py b/openhcs/processing/materialization/core.py
index e282d23497..f01b753568 100644
--- a/openhcs/processing/materialization/core.py
+++ b/openhcs/processing/materialization/core.py
@@ -27,6 +27,7 @@
 from polystore.streaming.viewer_transport import (
     PathMappedViewerStreamSourceMetadata,
     ViewerStreamBackendKwargs,
+    ViewerDisplayConfigABC,
 )
 from zmqruntime.viewer_protocol import ViewerWireValue
 
@@ -81,6 +82,7 @@
 from openhcs.core.steps.stream_component_semantics import (
     StreamImagePayloadMetadataProjector,
     StreamViewerComponentMetadataProjector,
+    StreamScopedDisplayConfig,
 )
 from openhcs.processing.materialization.constants import (
     MaterializationFormat,
@@ -319,6 +321,16 @@ def source_component_metadata(self) -> SourceComponentMetadata | None:
     def viewer_stream_requires_source_metadata(self) -> bool:
         return True
 
+    def viewer_display_config(
+        self, base: ViewerDisplayConfigABC
+    ) -> ViewerDisplayConfigABC:
+        """Project display scope from this output's own optional source payload."""
+        if self.metadata is None:
+            return base
+        return StreamScopedDisplayConfig.for_source_provenance(
+            base, self.metadata.source_provenance
+        )
+
     def require_text_content(self) -> str:
         """Return declared text content or reject a non-text output."""
 
@@ -1588,12 +1600,21 @@ def to_filemanager_kwargs(
         output_fields = self._output_fields(projected_outputs[0])
         if output_fields is None:
             return self.values.to_kwargs()
-        component_metadata, item_fields = output_fields
-        return self.values.with_single_item_source(
+        component_metadata, item_fields, display_config = output_fields
+        return self._stream_kwargs(display_config, item_fields).with_single_item_source(
             component_metadata,
             item_fields,
         ).to_kwargs()
 
+    def _stream_kwargs(
+        self,
+        display_config: ViewerDisplayConfigABC,
+        item_fields: Mapping[str, ViewerWireValue],
+    ) -> ViewerStreamBackendKwargs:
+        """Apply one source-owned display scope to the original typed request."""
+        request = self.values.with_item_fields(item_fields).stream_request
+        return ViewerStreamBackendKwargs(replace(request, display_config=display_config))
+
     def filemanager_batches(
         self,
         outputs: Sequence[Output],
@@ -1601,6 +1622,7 @@ def filemanager_batches(
         grouped: list[
             tuple[
                 dict[str, ViewerWireValue],
+                ViewerDisplayConfigABC,
                 list[Output],
                 list[dict[str, MaterializationValue]],
             ]
@@ -1611,21 +1633,22 @@ def filemanager_batches(
             if output_fields is None:
                 unprojected_outputs.append(output)
                 continue
-            component_metadata, item_fields = output_fields
+            component_metadata, item_fields, display_config = output_fields
             group = next(
-                (candidate for candidate in grouped if candidate[0] == item_fields),
+                (candidate for candidate in grouped
+                 if candidate[0] == item_fields and candidate[1] == display_config),
                 None,
             )
             if group is None:
-                group = (item_fields, [], [])
+                group = (item_fields, display_config, [], [])
                 grouped.append(group)
-            group[1].append(output)
-            group[2].append(component_metadata)
+            group[2].append(output)
+            group[3].append(component_metadata)
 
         batches: list[tuple[tuple[Output, ...], dict]] = []
         if unprojected_outputs:
             batches.append((tuple(unprojected_outputs), self.values.to_kwargs()))
-        for item_fields, batch_outputs, component_metadata in grouped:
+        for item_fields, display_config, batch_outputs, component_metadata in grouped:
             metadata_by_path = dict(
                 zip(
                     (output.path for output in batch_outputs),
@@ -1637,7 +1660,7 @@ def filemanager_batches(
                 raise ValueError(
                     "Viewer materialization batch requires unique output paths."
                 )
-            stream_request = self.values.with_item_fields(item_fields).stream_request
+            stream_request = self._stream_kwargs(display_config, item_fields).stream_request
             stream_request = replace(
                 stream_request,
                 source=replace(
@@ -1698,7 +1721,11 @@ def projected_outputs(self, outputs: Sequence[Output]) -> tuple[Output, ...]:
     def _output_fields(
         self,
         output: Output,
-    ) -> tuple[dict[str, MaterializationValue], dict[str, ViewerWireValue]] | None:
+    ) -> tuple[
+        dict[str, MaterializationValue],
+        dict[str, ViewerWireValue],
+        ViewerDisplayConfigABC,
+    ] | None:
         source_identity = output.source_identity
         if source_identity is not None:
             source_identity = source_identity.with_parsed_path_components(
@@ -1720,12 +1747,14 @@ def _output_fields(
                 output.variable_components,
             )
         )
-        display_semantics = self.values.stream_request.display_semantics
+        display_config = output.viewer_display_config(
+            self.values.stream_request.display_config
+        )
         projected_metadata = StreamViewerComponentMetadataProjector.for_item_fields(
-            display_semantics.component_order,
+            display_config.COMPONENT_ORDER,
             item_fields,
         ).project_required(index=0, metadata=component_metadata)
-        return projected_metadata, item_fields
+        return projected_metadata, item_fields, display_config
 
 
 EMPTY_BACKEND_CALL_KWARGS = RawBackendKwargs()
diff --git a/tests/unit/agent/test_plate_streaming_service.py b/tests/unit/agent/test_plate_streaming_service.py
index 98adf66c60..0ddb846cf3 100644
--- a/tests/unit/agent/test_plate_streaming_service.py
+++ b/tests/unit/agent/test_plate_streaming_service.py
@@ -18,6 +18,7 @@
 from openhcs.constants.constants import Backend, FileFormat
 from openhcs.core.plate_image_inventory import (
     PlateFileInventory,
+    PlateFileRecord,
     PlateFileKind,
     PlateImageRecord,
     PlateResultFileRecord,
@@ -58,6 +59,83 @@ def get_primary_backend(self, plate_path, filemanager):
         return "disk"
 
 
+def test_saved_site_free_image_inventory_and_loading_preserve_original_scope(tmp_path):
+    from types import SimpleNamespace
+    from openhcs.core.artifacts import ImageArtifactType
+    from openhcs.core.component_group_scope import RuntimeExecutionAxisScope
+    from openhcs.core.plate_image_inventory import PlateImageInventory
+    from openhcs.core.runtime_image_values import ImagePayloadMetadata, image_payload_metadata
+    from openhcs.core.source_image_provenance import (
+        SourceImageIdentity, SourceImageProvenance, SourceImageProvenanceContributor,
+    )
+    from openhcs.core.source_projection import SourceArtifactProjection
+    from openhcs.core.source_metadata import SourceVoxelSpacing
+    from openhcs.core.source_workspace_projection import VirtualWorkspaceSourceProjectionBuilder
+    from openhcs.core.virtual_workspace_metadata import VirtualWorkspaceSourceProjectionEntries
+    from openhcs.microscopes.source_schema import SourceSchemaFilenameParser
+
+    filename = "A01_s001_w1_z001_t001.tif"
+    pixels = np.concatenate((
+        np.full((4, 6), 11, dtype=np.uint16),
+        np.full((4, 6), 13, dtype=np.uint16),
+    ), axis=1)
+    tifffile.imwrite(tmp_path / filename, pixels)
+    components = {"well": "A01", "channel": "1", "z_index": "1", "timepoint": "1"}
+    contributors = tuple(
+        SourceImageProvenanceContributor(
+            SourceImageIdentity(
+                str(tmp_path / f"original-s{site}.tif"),
+                {**components, "site": site},
+            ),
+        ) for site in ("1", "3")
+    )
+    metadata = ImagePayloadMetadata(
+        source_provenance=SourceImageProvenance(
+            source_component_metadata=components,
+            source_image_provenance_planes=SourceImageProvenancePlanes(contributors),
+        ),
+        source_voxel_spacing=SourceVoxelSpacing((0.25, 0.5)),
+    )
+    ref = SourcePixelRef(backend="disk", backend_address=filename)
+    source = SourceArtifactProjection(
+        address=None, ref=ref, source_alias="Signal", artifact_kind=ImageArtifactType,
+        source_metadata=components, image_metadata=metadata,
+        execution_scope=RuntimeExecutionAxisScope.from_raw(
+            "A01", component="channel", value="1",
+            fixed_component_values=(("z_index", "1"), ("timepoint", "1")),
+        ),
+    )
+    builder = VirtualWorkspaceSourceProjectionBuilder(tmp_path)
+    builder.record_workspace_source_path(filename, ref)
+    builder.record_source_metadata(filename, components)
+    builder.ingest_source_projections(VirtualWorkspaceSourceProjectionEntries({filename: source}))
+    filemanager = FileManager({"disk": DiskStorageBackend()})
+    handler = SimpleNamespace(parse_image_path=SourceSchemaFilenameParser().parse_filename)
+    record = PlateImageInventory._record(
+        plate_path=tmp_path, image_file=filename, handler=handler,
+        projection=builder.projection(), filemanager=filemanager, backend="disk",
+    )
+    assert "site" not in record.metadata
+    context = PlateInspectionContext(
+        plate_path=tmp_path, filemanager=filemanager, handler=handler, parser=None,
+    )
+    projection = PlateStreamingService._inventory_source_projection(
+        (PlateFileRecord.from_image(record),), context,
+    )
+    stream_source = ViewerStreamingSource(
+        plate_path=tmp_path, filemanager=filemanager, microscope_handler=handler,
+    )
+    actual = stream_source.load_image(
+        filename, "disk", source_projection=projection, component_metadata=record.metadata,
+    )
+    np.testing.assert_array_equal(image_payload_data(actual), pixels)
+    actual_metadata = image_payload_metadata(actual)
+    assert "site" not in actual_metadata.source_component_metadata
+    assert actual_metadata.plane_axis is None
+    assert actual_metadata.source_voxel_spacing.values_zyx == (0.25, 0.5)
+    assert actual_metadata.source_provenance.source_image_provenance_planes == metadata.source_provenance.source_image_provenance_planes
+
+
 class FakeInspectionService:
     def __init__(self, inventory: PlateFileInventory) -> None:
         self.inventory = inventory
@@ -442,7 +520,7 @@ def test_inventory_source_projection_loads_exact_ome_stack_planes(tmp_path):
             component_metadata=record.metadata,
         )
         np.testing.assert_array_equal(image_payload_data(image), pixels[index])
-        assert record.metadata["z_index"] == index + 1
+        assert record.metadata["z_index"] == str(index + 1)
 
     table = MeasurementTable(
         name="centres",
diff --git a/tests/unit/test_aggregate_source_scope.py b/tests/unit/test_aggregate_source_scope.py
new file mode 100644
index 0000000000..7d94cd6b7d
--- /dev/null
+++ b/tests/unit/test_aggregate_source_scope.py
@@ -0,0 +1,197 @@
+"""Contributor scope and saved producer addresses survive their real consumers."""
+
+from dataclasses import dataclass, replace
+from pathlib import Path
+import pickle
+
+import numpy as np
+import pytest
+
+from openhcs.core.artifacts import ArtifactSpec, ImageArtifactType, MeasurementsArtifactType
+from openhcs.core.component_group_scope import RuntimeExecutionAxisScope
+from openhcs.core.measurement_row_materialization import MeasurementSparseColumnarRows
+from openhcs.core.orchestrator.execution_result import (
+    RuntimeContextObservation, RuntimeExecutionObservation,
+)
+from openhcs.core.projected_image_output import SourceProjectedImageOutput
+from openhcs.core.runtime_artifact_values import ArtifactKey, RuntimeValue
+from openhcs.core.runtime_exports import (
+    RuntimeExportExpectation, RuntimeExportObservation, runtime_export_failures,
+)
+from openhcs.core.runtime_image_values import ImagePayloadMetadata, image_payload_metadata
+from openhcs.core.runtime_measurements import MeasurementScope, MeasurementSubject, MeasurementTable
+from openhcs.core.runtime_plane_projection import RuntimePlaneAxis, RuntimePlaneAxisValueProjection
+from openhcs.core.runtime_stores import (
+    RuntimeArtifactAddress, RuntimeArtifactLocation, RuntimeValueStore,
+)
+from openhcs.core.runtime_tabular_values import FieldSpec
+from openhcs.core.source_image_provenance import (
+    RuntimeSourceImageProvenancePlane, SourceImageIdentity, SourceImageProvenancePlanes,
+)
+from openhcs.core.source_metadata import SourceVoxelSpacing
+from openhcs.core.steps.abstract import StepExecutionObservation
+from openhcs.processing.materialization import CsvOptions, MaterializationSpec
+from openhcs.processing.materialization.core import Output, materialization_outputs
+from polystore.filemanager import FileManager
+from polystore.memory import MemoryStorageBackend
+
+from test_materialization_core import _viewer_stream_backend_kwargs
+
+
+def _source(sites):
+    return ImagePayloadMetadata(
+        plane_axis=RuntimePlaneAxis.SOURCE_BINDING,
+        source_voxel_spacing=SourceVoxelSpacing((0.5, 0.5)),
+        source_image_provenance_planes=SourceImageProvenancePlanes.from_components(
+            paths=tuple(f"/input/field_{i}.tif" for i in range(len(sites))),
+            component_metadata=tuple(
+                {"well": "A01", "channel": 2, "z_index": 1, "timepoint": 1,
+                 **({"site": site} if site is not None else {})}
+                for site in sites
+            ),
+        ),
+    ).payload_with(np.arange(len(sites) * 12, dtype=np.uint16).reshape(len(sites), 3, 4))
+
+
+class ProjectionAudit:
+    def resolve_source_context(self, source, projection):
+        self.calls.append("before")
+        result = super().resolve_source_context(source, projection)
+        self.calls.append("after")
+        return result
+
+
+class ReducedDeclaration(SourceProjectedImageOutput):
+    def resolve_source_context(self, source, projection):
+        assert projection.axis_size == source.shape[0]
+        return image_payload_metadata(source).collapse_leading_plane_axis().payload_with(self.data)
+
+
+@dataclass(frozen=True)
+class DeclaredReducedField(ProjectionAudit, ReducedDeclaration):
+    data: np.ndarray
+    calls: list
+
+    def with_data(self, data):
+        return replace(self, data=data)
+
+
+def test_mixed_acquired_and_reduced_batch_uses_distinct_original_display_scopes():
+    source = _source((1, 3))
+    declaration = DeclaredReducedField(np.full((3, 4), 720, dtype=np.uint16), [])
+    reduced = ImageArtifactType.contextualize_output(
+        source, declaration, None,
+        RuntimePlaneAxisValueProjection.preserve(axis=RuntimePlaneAxis.SOURCE_BINDING, axis_size=2),
+    )
+    assert declaration.calls == ["before", "after"]
+    acquired_metadata = image_payload_metadata(source).for_leading_source_plane(0)
+    acquired = Output("acquired.tif", np.asarray(source)[0], acquired_metadata)
+    aggregate = Output("reduced.tif", np.asarray(reduced), image_payload_metadata(reduced))
+    adapter = _viewer_stream_backend_kwargs()
+    batches = adapter.filemanager_batches((acquired, aggregate))
+    assert len(batches) == 2
+    for output, (outputs, kwargs) in zip((acquired, aggregate), batches, strict=True):
+        assert outputs == (output,)
+        request = kwargs["stream_request"]
+        single = adapter.to_filemanager_kwargs(output)["stream_request"]
+        assert request.display_config == single.display_config
+        assert request.source.metadata.component_metadata_for_item(output.path, 0) == single.source.metadata.component_metadata_for_item(output.path, 0)
+        assert request.producer == adapter.values.stream_request.producer
+        assert request.viewer_transport == adapter.values.stream_request.viewer_transport
+        assert request.display_config.component_modes() == adapter.values.stream_request.display_config.component_modes()
+        assert output.metadata.source_voxel_spacing == SourceVoxelSpacing((0.5, 0.5))
+    acquired_request, reduced_request = (kwargs["stream_request"] for _, kwargs in batches)
+    assert "site" in acquired_request.display_config.COMPONENT_ORDER
+    assert "site" not in reduced_request.display_config.COMPONENT_ORDER
+    assert aggregate.metadata.source_provenance.source_image_provenance_planes.contributor_count == 2
+    np.testing.assert_array_equal(aggregate.content, declaration.data)
+
+
+@pytest.mark.parametrize("sites", ((), (None,), (1,), (1, 1), (1, None), (None, None)))
+def test_absence_one_equal_or_incomplete_contributors_cannot_exempt_acquired_plane(sites):
+    metadata = image_payload_metadata(_source(sites)).collapse_leading_plane_axis()
+    components = {
+        "well": "A01", "channel": 2, "z_index": 1, "timepoint": 1,
+    }
+    # Explicitly absent scalar identity, not a selected plane with a valid common site.
+    metadata = ImagePayloadMetadata(
+        source_component_metadata=components,
+        source_image_provenance_planes=SourceImageProvenancePlanes((
+            RuntimeSourceImageProvenancePlane(
+                SourceImageIdentity(component_metadata=components),
+                contributors=metadata.source_image_provenance_planes.contributors,
+            ),
+        )),
+    )
+    output = Output("unaddressed.tif", np.zeros((3, 4), dtype=np.uint16), metadata)
+    adapter = _viewer_stream_backend_kwargs()
+    assert "site" in metadata.source_provenance.required_scalar_components(("site",))
+    with pytest.raises(ValueError, match="site"):
+        adapter.to_filemanager_kwargs(output)
+    with pytest.raises(ValueError, match="site"):
+        adapter.filemanager_batches((output,))
+
+
+def test_present_scalar_override_remains_required_despite_varied_contributors():
+    metadata = image_payload_metadata(_source((1, 3))).collapse_leading_plane_axis()
+    metadata = metadata.replace_fields(source_component_metadata={
+        "well": "A01", "site": 7, "channel": 2, "z_index": 1, "timepoint": 1,
+    })
+    assert metadata.source_provenance.required_scalar_components(("site",)) == ("site",)
+    output = Output("override.tif", np.zeros((3, 4), dtype=np.uint16), metadata)
+    request = _viewer_stream_backend_kwargs().to_filemanager_kwargs(output)["stream_request"]
+    assert "site" in request.display_config.COMPONENT_ORDER
+    assert request.source.metadata.component_metadata_for_item(output.path, 0)["site"] == 7
+
+
+def _saved_table(tmp_path, *, producer, fields):
+    table = MeasurementTable(
+        name="Cells", subject=MeasurementSubject(MeasurementScope.ARTIFACT, "Cells"),
+        rows=MeasurementSparseColumnarRows.from_rows(
+            ({field: 1 for field in fields},), fields=tuple(FieldSpec(field) for field in fields),
+        ),
+    )
+    store = RuntimeValueStore()
+    record = store.record(RuntimeValue(
+        key=ArtifactKey(name="Cells", artifact_type=MeasurementsArtifactType,
+                        scope=RuntimeExecutionAxisScope(axis_id="A01")),
+        data=table,
+    ), path=f"/runtime/{producer}/Cells", backend="memory")
+    filemanager = FileManager({"memory": MemoryStorageBackend()})
+    outputs = materialization_outputs(
+        MaterializationSpec(CsvOptions(filename_suffix=".csv")), table,
+        str(tmp_path / f"Cells_{producer}"), filemanager,
+    )
+    for output in outputs:
+        # Persist exactly the original writer's content and declared path.
+        Path(output.path).write_text(output.require_text_content(), encoding="utf-8")
+    observation = StepExecutionObservation(
+        {RuntimeArtifactAddress.from_record(record): tuple(RuntimeArtifactLocation(output.path, "disk") for output in outputs)},
+        tuple(Path(output.path) for output in outputs),
+    )
+    return record, observation
+
+
+def test_same_named_field_and_aggregate_tables_keep_exact_producer_ownership(tmp_path):
+    acquired, acquired_outputs = _saved_table(tmp_path, producer="step1", fields=("site", "source_image_name"))
+    aggregate, aggregate_outputs = _saved_table(tmp_path, producer="step3", fields=("area",))
+    execution = RuntimeExecutionObservation((
+        RuntimeContextObservation("A01", (acquired, aggregate), StepExecutionObservation.combine((acquired_outputs, aggregate_outputs))),
+    ))
+    transported = pickle.loads(pickle.dumps(execution))
+    exports = RuntimeExportObservation.from_runtime_observations((transported,))
+    spec = ArtifactSpec.output(name="Cells", artifact_type=MeasurementsArtifactType,
+                               materialization=MaterializationSpec(CsvOptions(filename_suffix=".csv")))
+    expectation = RuntimeExportExpectation.from_output_specs((spec,))
+    assert exports.matching_table_outputs(acquired) == acquired_outputs.runtime_export_paths
+    assert exports.matching_table_outputs(aggregate) == aggregate_outputs.runtime_export_paths
+    assert runtime_export_failures(expectation, exports, {"A01": (acquired, aggregate)}) == ()
+    # Files and a same-named neighbor cannot replace the missing producer relation.
+    unowned = RuntimeExportObservation.from_output_paths(exports.output_files, outputs=aggregate_outputs)
+    failures = runtime_export_failures(expectation, unowned, {"A01": (acquired, aggregate)})
+    assert len(failures) == 1 and "no matching table output" in failures[0]
+    # A correct address does not waive its own required columns.
+    acquired_outputs.runtime_export_paths[0].write_text("area\n1\n", encoding="utf-8")
+    invalid = RuntimeExportObservation.from_runtime_observations((transported,))
+    failures = runtime_export_failures(expectation, invalid, {"A01": (acquired, aggregate)})
+    assert any("site" in failure and "source_image_name" in failure for failure in failures)
diff --git a/tests/unit/test_cppipe_execution_validation.py b/tests/unit/test_cppipe_execution_validation.py
index a734d03e22..72c25a306d 100644
--- a/tests/unit/test_cppipe_execution_validation.py
+++ b/tests/unit/test_cppipe_execution_validation.py
@@ -35,7 +35,12 @@
     MeasurementScope,
     MeasurementSubject,
 )
-from openhcs.core.runtime_stores import RuntimeValueStore
+from openhcs.core.runtime_stores import (
+    RuntimeArtifactAddress,
+    RuntimeArtifactLocation,
+    RuntimeValueStore,
+)
+from openhcs.core.steps.abstract import StepExecutionObservation
 from openhcs.processing.materialization import (
     CsvOptions,
     FileBundleOptions,
@@ -61,6 +66,7 @@ def test_cppipe_execution_validation_rejects_header_only_csv(
                 rows=({"slice_index": 0},),
             ),
             tmp_path,
+            table_path=csv_path,
         )
 
 
@@ -74,6 +80,7 @@ def test_cppipe_execution_validation_accepts_header_only_empty_csv(
         _measurement_output_specs(),
         _successful_execution_with_measurement_record(),
         tmp_path,
+        table_path=csv_path,
     )
 
     assert observation.exports.table_row_counts_by_path[csv_path] == 0
@@ -89,6 +96,7 @@ def test_cppipe_execution_validation_accepts_csv_with_data_rows(
         _measurement_output_specs(),
         _successful_execution_with_measurement_record(rows=({"slice_index": 0},)),
         tmp_path,
+        table_path=csv_path,
     )
 
     assert observation.exports.table_row_counts_by_path[csv_path] == 1
@@ -284,6 +292,8 @@ def _validate(
     output_specs: tuple[ArtifactSpec, ...],
     store: RuntimeValueStore,
     output_root: Path,
+    *,
+    table_path: Path | None = None,
 ):
     expectation = RuntimeArtifactExecutionExpectation.from_output_specs(
         output_specs,
@@ -293,7 +303,22 @@ def _validate(
         schema_version=ZMQ_RUNTIME_OBSERVATION_EXPORT_SCHEMA_VERSION,
         expectation=expectation,
         records_by_axis={"A01": tuple(store.observed_values)},
-        exports=RuntimeExportObservation.from_output_roots((output_root,)),
+        exports=RuntimeExportObservation.from_output_roots(
+            (output_root,),
+            outputs=(
+                StepExecutionObservation(
+                    {
+                        RuntimeArtifactAddress.from_record(record): (
+                            RuntimeArtifactLocation(str(table_path), "disk"),
+                        )
+                        for record in store.observed_values
+                    },
+                    (table_path,),
+                )
+                if table_path is not None
+                else StepExecutionObservation.empty()
+            ),
+        ),
         output_roots=(output_root,),
         execution_success_by_axis={"A01": True},
     ).require_valid_observation()
diff --git a/tests/unit/test_function_artifact_materialization.py b/tests/unit/test_function_artifact_materialization.py
index 8965817ed3..7d22590462 100644
--- a/tests/unit/test_function_artifact_materialization.py
+++ b/tests/unit/test_function_artifact_materialization.py
@@ -66,6 +66,9 @@
     RuntimeObservationMode,
 )
 from openhcs.core.pipeline.function_contracts import artifact_outputs
+from openhcs.core.pipeline.artifact_planning import (
+    AutomaticMeasurementsArtifactOutputMaterializationStrategy,
+)
 from openhcs.core.runtime_artifact_values import (
     RuntimeValue,
 )
@@ -1063,7 +1066,7 @@ def fake_materialize(_spec, data, path, *_args, **_kwargs):
     PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan), context)
 
     assert [(data.row_mappings(), path) for data, path in materialized] == [
-        (({"object_id": 1, "area": 42},), "/analysis/measurements_1.roi.zip")
+        (({"object_id": 1, "area": 42},), "/analysis/A01_channel-1_measurements_step7.roi.zip")
     ]
 
 
@@ -1753,6 +1756,93 @@ def record_for_site(site: str) -> None:
     assert consolidation_inputs.destination.images_dir == "/images"
 
 
+@pytest.mark.parametrize("sites", [("1", "3"), ("3", "1")])
+@pytest.mark.parametrize("automatic", [True, False])
+def test_scalar_acquired_tables_keep_fixed_site_addresses_on_shared_runtime_path(sites, automatic):
+    output_plan = ArtifactOutputPlan(
+        name="AcquiredFieldMeasurements",
+        path="/memory/A01_w1_AcquiredFieldMeasurements_step0.pkl",
+        artifact_type=MeasurementsArtifactType,
+        materialization=(
+            AutomaticMeasurementsArtifactOutputMaterializationStrategy().materialization()
+            if automatic else csv_only()
+        ),
+        variable_components=(AllComponents.Z_INDEX,),
+    )
+    context = _context(FileManagerStub())
+    plan = _plan(
+        output_plan,
+        group_by_value="channel",
+        variable_components=(VariableComponents.Z_INDEX,),
+    )
+    plan.runtime_artifact_materialization = RuntimeArtifactMaterializationPlan(
+        persistent_enabled=True,
+        persistent_backend="disk",
+    )
+    for site in sites:
+        metadata = {
+            "well": "A01", "site": site, "channel": "1",
+            "z_index": "1", "timepoint": "1",
+        }
+        context.runtime_value_store.record(
+            RuntimeValue.normalize_for_execution_scope(
+                output_plan,
+                MeasurementTable(
+                    name=output_plan.name,
+                    rows=MeasurementSparseColumnarRows.from_rows(
+                        ({"intensity": 10 + int(site)},),
+                        fields=(FieldSpec("intensity", int),),
+                    ),
+                    source_image_provenance_planes=(
+                        SourceImageProvenancePlanes.from_components(
+                            paths=(f"/input/A01_s{site}_w1_z1_t1.tif",),
+                            component_metadata=(metadata,),
+                        )
+                    ),
+                    subject=MeasurementSubject(MeasurementScope.ARTIFACT),
+                ),
+                execution_scope=RuntimeExecutionAxisScope.from_raw(
+                    "A01", component=AllComponents.CHANNEL, value="1",
+                    fixed_component_values=(
+                        (AllComponents.SITE, site),
+                        (AllComponents.TIMEPOINT, "1"),
+                    ),
+                ),
+            ),
+            path=output_plan.path,
+            backend="memory",
+        )
+
+    records = context.runtime_value_store.values()
+    assert len(records) == 2
+    assert len({record.location.path for record in records}) == 1
+    saved = PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(
+        context.filemanager, plan, context,
+    )
+    assert len(saved) == 2
+    saved_paths = set()
+    for item in saved:
+        record = item.materialization.record
+        observation = item.observation(plan, context)
+        site = record.key.scope.value_text_for_component(AllComponents.SITE)
+        paths = observation.paths_for(record)
+        expected_stem = (
+            f"A01_site-{site}_channel-1_timepoint-1"
+            if automatic else f"A01_s{int(site):03d}_w1_z001_t001"
+        )
+        assert paths == (
+            Path(
+                f"/analysis/{expected_stem}_"
+                "AcquiredFieldMeasurements_step7_details.csv"
+            ),
+        )
+        saved_paths.update(paths)
+        assert context.filemanager.load_text(str(paths[0]), "disk") == (
+            f"intensity\r\n{10 + int(site)}\r\n"
+        )
+    assert len(saved_paths) == 2
+
+
 def test_terminal_persistence_is_reported_without_becoming_declared_export() -> None:
     output_plan = ArtifactOutputPlan(
         name="cell_counts",
@@ -1912,8 +2002,7 @@ def test_completed_observation_projects_tables_before_worker_payload_release(
         context_key="A01", context=context,
         records=context.runtime_value_store.observed_values,
         runtime_observation_mode=observation_mode,
-        runtime_export_paths=step_observation.runtime_export_paths,
-        analysis_inputs=step_observation.analysis_inputs,
+        outputs=step_observation,
     )
     context.runtime_value_store.clear()
     del pixels
@@ -1923,11 +2012,11 @@ def test_completed_observation_projects_tables_before_worker_payload_release(
     assert len(observation.records) == (
         2 if observation_mode is RuntimeObservationMode.MERGE_INTO_PARENT else 0
     )
-    assert len(observation.runtime_export_paths) == 1
-    assert observation.runtime_export_paths[0].suffix == ".csv"
-    assert observation.analysis_inputs is not None
-    assert observation.analysis_inputs.destination.backend == "disk"
-    assert observation.analysis_inputs.destination.images_dir == "/images"
+    assert len(observation.outputs.runtime_export_paths) == 1
+    assert observation.outputs.runtime_export_paths[0].suffix == ".csv"
+    assert observation.outputs.analysis_inputs is not None
+    assert observation.outputs.analysis_inputs.destination.backend == "disk"
+    assert observation.outputs.analysis_inputs.destination.images_dir == "/images"
     execution_observation = RuntimeExecutionObservation(contexts=(observation,))
     transported = pickle.loads(pickle.dumps(execution_observation))
     consolidated = RuntimeAnalysisConsolidationInputs.from_observations(
@@ -1939,7 +2028,7 @@ def test_completed_observation_projects_tables_before_worker_payload_release(
     assert outputs[0].well_id == "A01"
     assert "cell_count" in outputs[0].csv_content
     assert "2" in outputs[0].csv_content
-    assert outputs[0].path == observation.runtime_export_paths[0]
+    assert outputs[0].path == observation.outputs.runtime_export_paths[0]
     with pytest.raises(KeyError, match="unknown compiled context"):
         RuntimeAnalysisConsolidationInputs.from_observations({}, (transported,))
 
@@ -2012,7 +2101,7 @@ def fake_materialize(_spec, data, path, *_args, **_kwargs):
                 {"image_area": 100.0, "source_image_name": "OrigBlue"},
                 {"object_label": 1, "area": 42.0, "object_name": "Nuclei"},
             ),
-            "/analysis/A01_w1_measurements_step7.roi.zip",
+            "/analysis/A01_channel-1_measurements_step7.roi.zip",
         )
     ]
 
@@ -2144,7 +2233,7 @@ def test_artifact_name_materialization_ignores_incomplete_source_identity() -> N
     )
 
     assert tuple(str(item.base_path) for item in materializations) == (
-        "/analysis/SavedImage.roi.zip",
+        "/analysis/A01_channel-3_z_index-1_SavedImage_step7.roi.zip",
     )
 
 
@@ -2370,7 +2459,7 @@ def fake_materialize(spec, data, path, *_args, **_kwargs):
     assert isinstance(spec.outputs[0], CsvOptions)
     assert isinstance(data, MeasurementTable)
     assert data.row_mappings() == ({"site": "1", "area": 42},)
-    assert path == "/analysis/A01_w1_measurements_step7.roi.zip"
+    assert path == "/analysis/A01_channel-1_measurements_step7.roi.zip"
 
 
 def _duplicate_scalar_image_context(
@@ -2650,7 +2739,7 @@ def fake_materialize(spec, data, path, *_args, **_kwargs):
     ]
 
 
-def test_materialize_artifact_outputs_keeps_grouped_artifact_record_path(
+def test_materialize_artifact_outputs_uses_grouped_scope_not_record_location(
     monkeypatch,
 ):
     output_plan = ArtifactOutputPlan(
@@ -2700,7 +2789,7 @@ def fake_materialize(spec, data, path, *_args, **_kwargs):
     PersistentArtifactMaterializationTargetPlan("disk").materialize_outputs(filemanager, _plan(output_plan, group_by_value="channel"), context)
 
     assert [path for _spec, _data, path in materialized] == [
-        "/analysis/A01_w2_measurements_step7.roi.zip",
+        "/analysis/A01_channel-2_measurements_step7.roi.zip",
     ]
     assert not context.runtime_value_store.values()[0].key.scope.has_fixed_components
 
diff --git a/tests/unit/test_materialization_batch_outcomes.py b/tests/unit/test_materialization_batch_outcomes.py
index ddb21fc62b..d9750fd309 100644
--- a/tests/unit/test_materialization_batch_outcomes.py
+++ b/tests/unit/test_materialization_batch_outcomes.py
@@ -194,7 +194,7 @@ def forbidden_reconstruction(*_args, **_kwargs):
 
     assert context.released
     assert transported.is_success()
-    assert transported.runtime_observation.contexts[0].runtime_export_paths == (path,)
+    assert transported.runtime_observation.contexts[0].outputs.runtime_export_paths == (path,)
     assert transported.runtime_observation.contexts[0].records == ()
     assert RuntimeExportObservation.from_runtime_observations(
         (transported.runtime_observation,)
@@ -208,10 +208,10 @@ def test_parent_and_worker_exports_share_existing_observation_authority(tmp_path
     for path in (worker_path, plate_path, unrelated):
         path.write_text("Area\n42\n")
     worker = RuntimeExecutionObservation(
-        contexts=(RuntimeContextObservation("A01", (), (worker_path,)),)
+        contexts=(RuntimeContextObservation("A01", (), StepExecutionObservation({}, (worker_path,))),)
     )
     parent = RuntimeExecutionObservation(
-        contexts=(RuntimeContextObservation("A01", (), (plate_path,)),)
+        contexts=(RuntimeContextObservation("A01", (), StepExecutionObservation({}, (plate_path,))),)
     )
     results = CompiledPlateExecutionResults(
         {"A01": ExecutionResult.success("A01", runtime_observation=worker)},
diff --git a/tests/unit/test_runtime_equivalence.py b/tests/unit/test_runtime_equivalence.py
index b00af0e456..16109d8413 100644
--- a/tests/unit/test_runtime_equivalence.py
+++ b/tests/unit/test_runtime_equivalence.py
@@ -119,10 +119,12 @@
     SpatialGrid,
 )
 from openhcs.core.runtime_stores import (
+    RuntimeArtifactAddress,
     RuntimeArtifactLocation,
     RuntimeValueStore,
     StoredRuntimeValue,
 )
+from openhcs.core.steps.abstract import StepExecutionObservation
 from openhcs.core.runtime_tabular_values import (
     ColumnarRows,
 )
@@ -1513,7 +1515,7 @@ def test_runtime_output_snapshot_from_artifact_execution_ignores_auxiliary_table
         encoding="utf-8",
     )
     store = RuntimeValueStore()
-    store.record(
+    record = store.record(
         RuntimeValue(
             key=ArtifactKey(
                 name="Measurements",
@@ -1533,7 +1535,19 @@ def test_runtime_output_snapshot_from_artifact_execution_ignores_auxiliary_table
         runtime_records_by_axis(
             {"A01": SimpleNamespace(runtime_value_store=store, step_plans={})}
         ),
-        RuntimeExportObservation.from_output_root(output_root),
+        RuntimeExportObservation.from_output_root(
+            output_root,
+            outputs=StepExecutionObservation(
+                {
+                    RuntimeArtifactAddress.from_record(record): (
+                        RuntimeArtifactLocation(
+                            str(output_root / "A01_Measurements_step1.csv"), "disk"
+                        ),
+                    ),
+                },
+                (output_root / "A01_Measurements_step1.csv",),
+            ),
+        ),
     )
 
     snapshot = RuntimeOutputSnapshot.from_artifact_execution_observation(observation)
diff --git a/tests/unit/test_runtime_execution_validation.py b/tests/unit/test_runtime_execution_validation.py
index 9906fed3bb..e749353fab 100644
--- a/tests/unit/test_runtime_execution_validation.py
+++ b/tests/unit/test_runtime_execution_validation.py
@@ -64,6 +64,7 @@
     ObjectLabelVariantData,
 )
 from openhcs.core.runtime_tabular_values import FieldSpec
+from openhcs.core.steps.abstract import StepExecutionObservation
 from openhcs.core.source_spatial_domain import SourceSpatialDomain
 from openhcs.microscopes.source_schema import SourceSchemaFilenameParser
 from openhcs.processing.materialization import (
@@ -254,7 +255,7 @@ def test_zmq_observation_exports_exact_compiler_owned_artifacts(
                 RuntimeContextObservation(
                     context_key="A01",
                     records=(),
-                    runtime_export_paths=(contracted_output,),
+                    outputs=StepExecutionObservation({}, (contracted_output,)),
                 ),
             )
         ),
@@ -538,7 +539,7 @@ def test_runtime_execution_observation_reads_plate_export_from_exact_owner(
                 RuntimeContextObservation(
                     context_key="A01",
                     records=(),
-                    runtime_export_paths=(contracted_output,),
+                    outputs=StepExecutionObservation({}, (contracted_output,)),
                 ),
             )
         ),
diff --git a/tests/unit/test_zmq_execution_server.py b/tests/unit/test_zmq_execution_server.py
index e893f9c5c3..e2cf851ada 100644
--- a/tests/unit/test_zmq_execution_server.py
+++ b/tests/unit/test_zmq_execution_server.py
@@ -23,6 +23,7 @@
     ProcessingConfig,
 )
 from openhcs.core.execution_state import ExecutionOutputPlateSummary
+from openhcs.core.steps.abstract import StepExecutionObservation
 from openhcs.core.orchestrator.compiled_plate_execution import (
     CompiledPlateExecutionResults,
 )
@@ -264,7 +265,7 @@ def test_server_exports_outcomes_without_projecting_compiled_values(
                             RuntimeContextObservation(
                                 "context",
                                 (),
-                                runtime_export_paths=(declared_output,),
+                                outputs=StepExecutionObservation({}, (declared_output,)),
                             ),
                         )
                     ),
diff --git a/tests/unit/test_zmq_runtime_outcome_export.py b/tests/unit/test_zmq_runtime_outcome_export.py
index 041116b678..8bd824f28d 100644
--- a/tests/unit/test_zmq_runtime_outcome_export.py
+++ b/tests/unit/test_zmq_runtime_outcome_export.py
@@ -12,6 +12,7 @@
     RuntimeExecutionObservation,
 )
 from openhcs.core.runtime_exports import RuntimeExportObservation
+from openhcs.core.steps.abstract import StepExecutionObservation
 from openhcs.runtime.zmq_execution_observation import (
     ZMQRuntimeExecutionObservationExport,
     ZMQRuntimeExecutionOutcomeExport,
@@ -32,7 +33,7 @@ def test_outcome_export_round_trip_excludes_runtime_values(tmp_path: Path) -> No
                     RuntimeContextObservation(
                         "context",
                         (runtime_value,),
-                        runtime_export_paths=(declared_output,),
+                        outputs=StepExecutionObservation({}, (declared_output,)),
                     ),
                 )
             ),
diff --git a/validation/site-free-aggregate-publication-728/RECEIVING.rst b/validation/site-free-aggregate-publication-728/RECEIVING.rst
new file mode 100644
index 0000000000..d57924bb1c
--- /dev/null
+++ b/validation/site-free-aggregate-publication-728/RECEIVING.rst
@@ -0,0 +1,167 @@
+732 ordinary installed multisite publication packet
+==================================================
+
+Prepared source only; NOT installed, compiled, executed or live-qualified.
+Qualified production is428584584, unchanged at published151cbfb5e. Planck owns
+the next ordinary whole candidate after his separate734 operation. Do not
+overlay/rebuild immutable receiving11 or any science bundle to include732.
+
+One existing public owner family
+--------------------------------
+
+Use the original shared builder and engineering94 launch/lifecycle/recording
+owners. No new checkout, interpreter, environment, CLI, registry or lane.
+Native6012/viewer6013 are a proposed reuse of the original family, NOT a lease.
+Pin the builder's exact whole-source/wheel/installed proof first, then require
+that family's positive closure/handoff and an actual graphical-session owner.
+This document starts no process and makes no reservation.
+
+``pipeline.py`` is one complete PipelineDocument with pipeline_config and all
+four steps. ``prepare_inputs.py`` is a one-time synthetic input preparation,
+using the original embedded acquisition header fixture, not a metadata codec.
+It needs the same source checkout's tests/unit and its qualified dependency
+backing. Do not run it against another source revision or infer that its
+imports are cheap; the original test module imports the registry family.
+
+After an exact candidate/lease is ready, the receiving owner creates the fresh
+persistent engineering728/public01 packet and calls the existing preparation
+with that candidate's authentic matching source. Never overwrite an earlier
+attempt. The input is four4x6uint16 TIFFs: sites1/3 and channels1/2, signed-safe
+embedded XY positions0/6, spacing .25Y/.5Xmicrometers. No scientific images.
+Use the already-supported public SourceBindings preparation if a writable
+workspace is required; preserve copied acquisition bytes and source hashes.
+
+Source-grounded operations
+-------------------------
+
+* Per-field and aggregate measurements use the same original registered
+  ``measure_image_intensity`` declaration, with no custom measurement algorithm.
+  SourceBindings declares the semantic Signal alias; no fabricated producer
+  identity, source receipt, CSV or runtime observation is supplied.
+* ``acquisition_tile_positions`` uses the original native source metadata and
+  emits its declared positions artifact. ``assemble_stack_cpu`` consumes that
+  exact artifact through the original input edge, TileBlendMethod.NONE.
+* Automatic final main-flow persistence remains enabled by the ordinary
+  PathPlanningConfig default. Explicit acquired/aggregate checkpoints are
+  additional evidence, NOT a named-artifacts-only bypass of default publication.
+* Napari streaming is enabled at the original proposed6013 endpoint. A
+  headless-only acceptance cannot be mislabeled as live streaming settlement.
+
+No public request has yet been submitted. The first artifact-plan response
+must confirm exact acquired and aggregate source scopes, registered invocation
+owners, positions edges, both measurement schemas/producer addresses and
+default publication. Derive actual record names/paths from that plan; do not
+invent bindings or carry an old module ordinal into the request. If a
+declaration is unsupported, preserve this new known pre-execution failure,
+repair only its declared fixture boundary, and use a distinct document/intent.
+
+Actual affected entrypoint
+-------------------------
+
+Use the existing normal public CLI: health; original owned runtime start and
+observe; original catalog preparation start/status; source artifact plan;
+one session/compile; one execution; original STATUS first terminal and repeat
+read of that SAME job; original VALUES/OUTCOMES export. No registration is
+needed. Use reflected requests and retained complete handles, not a new runner.
+Do not resend a timed-out mutation or replace an UNKNOWN disposition.
+
+At runtime terminal COMPLETE, ordinary inventory and native-resolution public
+samples must expose acquired checkpoints and the final honest aggregate.
+For each channelc the full4x12 mosaic must have24 left elements10*c+1 and24
+right elements10*c+3 (all96 mosaic elements across both channels). If an
+existing declared ingestion projection changes intensity units, retain and
+validate that original owner/transform explicitly; never infer a factor from
+observed maxima or silently loosen the expected units to match an output.
+Position artifacts must carry [(0,0),(6,0)] in admitted site order1,3.
+
+Decode saved native image metadata with the original
+VirtualWorkspaceSourceProjection. Acquired outputs keep their scalar site;
+aggregate keeps exactly its two real same-channel source contributors and
+no invented scalar site. Both preserve physical .25/.5micrometer calibration,
+wellA01/channel/Z1/T1, source file identity and unchanged acquisition hashes.
+The original public state/payload observation must prove both acquired and
+aggregate routes settle; distinguish that from capture/biological acceptance.
+
+Read VALUES/OUTCOMES using the original
+ZMQRuntimeExecutionObservationExport.read/require_valid_observation, NOT JSON
+text parsing of its compressed payload. Each measurement RuntimeArtifactRecord
+must resolve through observation.outputs and RuntimeArtifactAddress to ONLY its
+actual saved locations. Verify the acquired rows' declared site/source name
+requirements and the aggregate's own actual schema, feature values and scope.
+No same-named field record may be validated against an aggregate CSV or vice
+versa; an empty result, absent address or missing required column is a failure.
+
+Retain full original stdin/stdout, statuses, module/installed identity, schema,
+source/image/table/export hashes and any failure. Finish with one original
+native/viewer close, exact incarnation ACK/exit and ordinary CLI terminal
+recorded separately. Return the original lane to its custody owner. No
+scientific quality, scaled fleet or biological performance claim follows.
+
+Existing controls and limits
+----------------------------
+
+The source-controls06 mixed acquired/reduced batch and strict incomplete-plane
+negatives are already PASS at qualified428. Do not repeat them for ceremony.
+They include cooperative independent declaration hooks and address-specific
+same-name table negatives. This packet adds the missing installed public path;
+it does not replace those controls or manufacture a runtime failure as design.
+Planck's independent projected-Z producer work is separate; its genuine guard
+and existing source metadata owners remain intact. Coordinate any crossed
+SourceImageProvenance/StreamScopedDisplayConfig hunk before changing production.
+
+Retained saved-label reopening witness
+--------------------------------------
+
+The closed assisted technical review supplied an additional receiving case,
+not a new scientific attempt. Exact original root under the issue batch:
+next-p001-assisted-viewer-review94-after10-20261005/P001_ASSISTED_REVIEW10_94/
+author-workspace/output. REPORT.md and runtime/mcp.stdin:64 preserve the
+ordinary output-plate selected label request. The explicit result-directory
+request at stdin:17 was separately refused because its raw receipt did not bind
+that label; it is not evidence of a native mount or a receipt guard defect.
+
+runtime/data/openhcs/logs/napari_detached_port_6013.log:43..58 identifies the
+FIELD site5 ``neurons_step0.labels.tif``. It reaches
+NapariComponentAwareDisplayCoordinator -> NapariStreamLayerContext.layer_route
+-> original PolyStore build_route_key and fails required scalar CHANNEL.
+It is not the raw site-free mosaic or a failed screenshot/timeout.
+
+The saved output plate's original openhcs_metadata.json results.source_projection
+entry declares a two-dimensional source spatial domain, plane_axis SOURCE_BINDING,
+and two runtime-plane source identities DAPI/channel1 and FITC/channel2, each with
+empty contributors. Its scalar well/site5/Z1/T1 identity has no channel.
+No acquisition pixels, reference answers or scientific settings were inspected
+for this diagnosis. The metadata file and original requests were read-only.
+
+Current ViewerStreamingSource.image_source_metadata_items observes metadata's
+declared planes. StreamComponentMessageExtraAuthority keeps CHANNEL because
+those source identities supply it; the downstream scalar route key cannot
+represent both. The saved producer/plane-versus-contributor declaration is the
+boundary needing qualification. It must not be repaired by inventing a scalar
+channel, parsing a filename, dropping the guard or projecting two source
+identities as two label pixel planes without an actual output-axis declaration.
+
+732's current required_scalar_components exemption concerns actual contributor
+reduction and is used by Output's publication adapter. This manual saved-label
+path bypasses that adapter, and these retained planes have empty contributors.
+Therefore732 does NOT currently claim to repair this label reopening case.
+407 concerns typed DTO presentation and is unrelated. Historical515 is CLOSED
+and cannot serve as a current fix owner. The existing152 producer family is
+actively coordinated with Planck and Dewey; Singer retains receiving integration.
+No competing producer or Napari fallback implementation was started.
+
+Add one tiny saved-label control to the same next ordinary receiving packet
+after the original producer owner establishes its declaration: a channel-
+combined two-dimensional ObjectLabels output with exact two-channel source
+provenance, plus its matched raw routes. Reopen through ordinary output-plate
+inventory (not an invented historical source receipt). Require exact label
+pixels, producer identity, physical calibration, real source contributors and
+native settlement; never synthesize a scalar channel or duplicate label arrays.
+Retain an acquired label's valid scalar channel and malformed/missing lineage
+negative. This is an acceptance extension, NOT a claim that a new fixture or
+producer fix has already been implemented/installed. Coordinate the152 source
+patch in the shared candidate rather than starting a separate runtime/project.
+This distinct152 label qualification does not expand732's merge gate or hold
+its independently qualified default-publication fix behind another producer
+repair. Record each control's source/candidate and acceptance separately even
+when both can use the same ordinary candidate/receiving family.
diff --git a/validation/site-free-aggregate-publication-728/pipeline.py b/validation/site-free-aggregate-publication-728/pipeline.py
new file mode 100644
index 0000000000..730ec5fa24
--- /dev/null
+++ b/validation/site-free-aggregate-publication-728/pipeline.py
@@ -0,0 +1,89 @@
+"""Tiny registered multisite/default-publication receiving document.
+
+Not executed or live-qualified. The receiving owner must pin the whole installed
+candidate and release its original native6012/viewer6013 family before use.
+"""
+from pathlib import Path
+
+from zmqruntime.config import TransportMode
+
+from openhcs.constants.constants import GroupBy, Microscope, VariableComponents
+from openhcs.core.config import (
+    LazyNapariStreamingConfig,
+    LazyPathPlanningConfig,
+    LazyProcessingConfig,
+    LazySourceBindingsConfig,
+    LazyStepMaterializationConfig,
+    PipelineConfig,
+)
+from openhcs.core.source_bindings import (
+    MetadataExtractionRule,
+    MetadataSource,
+    NamedSourceBinding,
+)
+from openhcs.core.steps import FunctionStep
+from openhcs.processing.backends.assemblers.assemble_stack_cpu import assemble_stack_cpu
+from openhcs.processing.backends.assemblers.blending import TileBlendMethod
+from openhcs.processing.backends.cellprofiler.intensity import measure_image_intensity
+from openhcs.processing.backends.pos_gen.acquisition_positions import acquisition_tile_positions
+
+packet = Path(
+    "/home/ts/wt/openhcs-issue-batch-20260929/engineering728/public01"
+)
+field_processing = LazyProcessingConfig(
+    variable_components=[VariableComponents.Z_INDEX], group_by=GroupBy.CHANNEL,
+)
+site_processing = LazyProcessingConfig(
+    variable_components=[VariableComponents.SITE], group_by=GroupBy.CHANNEL,
+)
+pipeline_config = PipelineConfig(
+    microscope=Microscope.SOURCE_BINDINGS,
+    num_workers=1,
+    use_threading=True,
+    source_bindings_config=LazySourceBindingsConfig(
+        metadata_rules=(MetadataExtractionRule(
+            MetadataSource.FILE_NAME,
+            r"(?P<well>A\d+)_s(?P<site>\d+)_w(?P<channel>\d+)_z(?P<z_index>\d+)_t(?P<timepoint>\d+)\.tif",
+        ),),
+        bindings=(NamedSourceBinding(alias="Signal"),),
+        source_stack_components=(),
+    ),
+    processing_config=field_processing,
+    path_planning_config=LazyPathPlanningConfig(
+        global_output_folder=packet / "results",
+    ),
+    # This is the existing original engineering94 family, NOT a new lease.
+    # Do not submit until the sole receiving owner releases that exact family.
+    napari_streaming_config=LazyNapariStreamingConfig(
+        enabled=True, port=6013, host="127.0.0.1",
+        transport_mode=TransportMode.TCP,
+    ),
+)
+pipeline_steps = [
+    FunctionStep(
+        name="AcquiredFieldMeasurements",
+        func=measure_image_intensity,
+        processing_config=field_processing,
+        step_materialization_config=LazyStepMaterializationConfig(
+            enabled=True, sub_dir="acquired-checkpoints",
+        ),
+    ),
+    FunctionStep(
+        name="EmbeddedAcquisitionPositions",
+        func=acquisition_tile_positions,
+        processing_config=site_processing,
+    ),
+    FunctionStep(
+        name="AssembledField",
+        func=(assemble_stack_cpu, {"blend_method": TileBlendMethod.NONE}),
+        processing_config=site_processing,
+        step_materialization_config=LazyStepMaterializationConfig(
+            enabled=True, sub_dir="aggregate-checkpoints",
+        ),
+    ),
+    FunctionStep(
+        name="AggregateFieldMeasurements",
+        func=measure_image_intensity,
+        processing_config=field_processing,
+    ),
+]
diff --git a/validation/site-free-aggregate-publication-728/prepare_inputs.py b/validation/site-free-aggregate-publication-728/prepare_inputs.py
new file mode 100644
index 0000000000..0186d6db22
--- /dev/null
+++ b/validation/site-free-aggregate-publication-728/prepare_inputs.py
@@ -0,0 +1,53 @@
+"""Create fresh tiny synthetic acquisition files through the original fixture.
+
+No pipeline/runtime imports, processing, source receipts or reference images.
+Source headers use the existing test_source_tile_geometry._xml declaration;
+the packet must accompany the same pinned source as the whole installed owner.
+"""
+import hashlib
+import json
+from pathlib import Path
+import sys
+
+import numpy as np
+import tifffile
+
+
+def main():
+    checkout = Path(sys.argv[1]).resolve()
+    packet = Path(sys.argv[2]).resolve()
+    sys.path.insert(0, str(checkout / "tests/unit"))
+    from test_source_tile_geometry import _xml
+
+    # Exclusive creation preserves the first original intent and input bytes.
+    acquisition = packet / "acquisition"
+    acquisition.mkdir(parents=True, exist_ok=False)
+    records = []
+    for channel in (2, 1):
+        for site in (3, 1):
+            x_pixels = 6 if site == 3 else 0
+            path = acquisition / f"A01_s{site:03d}_w{channel}_z001_t001.tif"
+            pixels = np.full((4, 6), 10 * channel + site, dtype=np.uint16)
+            tifffile.imwrite(
+                path, pixels,
+                description=_xml(
+                    x=x_pixels * 0.5, y=0,
+                    row=1, column=2 if site == 3 else 1,
+                    sx=0.5, sy=0.25,
+                ),
+                metadata=None,
+            )
+            records.append({
+                "path": str(path),
+                "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
+                "channel": channel, "site": site,
+                "shape": list(pixels.shape), "value": int(pixels[0, 0]),
+                "positions_xy": [x_pixels, 0],
+            })
+    with (packet / "INPUTS.json").open("x") as stream:
+        json.dump(records, stream, indent=2)
+    print(json.dumps(records, indent=2))
+
+
+if __name__ == "__main__":
+    main()