Preserve ROI source metadata and reopen retained native results

openhcs/agent/capabilities.py

@@ -2132,8 +2132,13 @@ class StreamPlateFilesToViewerCapability(PlatePathCapability):
         "Resolves image or ROI result records by virtual path, source path, "
         "result path, basename, or bounded inventory query, then streams them "
         "to a managed viewer through the same core service used by the Image Browser."
+        " Set result_directory to reopen retained native results independently of "
+        "their output location; plate_path supplies the original source context. "
+        "ROI reopening requires persisted source metadata, not filename guesses."
     )
     service = "plate_streaming"
+    mutating = True
+    side_effects = ("launches_or_updates_managed_viewer",)
     data_exposure = (
         "local_plate_path",
         "plate_virtual_image_path",

openhcs/agent/dto/plate.py

@@ -383,6 +383,7 @@ class PlateFileStreamRequest:
     """Stream image or ROI files exposed by a local plate inventory to a viewer."""
 
     plate_path: str
+    result_directory: str | None = None
     context_plate_path: str | None = None
     file_paths: tuple[str, ...] = ()
     microscope_type: str = PlateInspectionDefaults.MICROSCOPE_AUTO
@@ -401,6 +402,7 @@ def from_fields(
         cls,
         *,
         plate_path: str,
+        result_directory: str | None = None,
         file_paths: list[str] | None = None,
         microscope_type: str = PlateInspectionDefaults.MICROSCOPE_AUTO,
         pattern_format: str | None = None,
@@ -418,6 +420,7 @@ def from_fields(
     ) -> "PlateFileStreamRequest":
         return cls(
             plate_path=plate_path,
+            result_directory=result_directory,
             file_paths=tuple(file_paths or ()),
             microscope_type=microscope_type,
             pattern_format=pattern_format,
@@ -439,6 +442,7 @@ def from_fields(
     def as_tool_arguments(self) -> dict[str, JsonValue]:
         return {
             "plate_path": self.plate_path,
+            "result_directory": self.result_directory,
             "file_paths": list(self.file_paths) if self.file_paths else None,
             "microscope_type": self.microscope_type,
             "pattern_format": self.pattern_format,

openhcs/agent/services/plate_inspection_service.py

@@ -1258,17 +1258,8 @@ def _query_explicit_result_directory(
             )
         if result_path is None:
             raise RuntimeError("Result directory resolution returned no path or error.")
-        from openhcs.microscopes.microscope_interfaces import AnalysisResultDirectory
-
         try:
-            result_inventory = PlateResultFileInventory.from_directory_files(
-                plate_path=result_path,
-                result_directory=AnalysisResultDirectory(
-                    subdirectory_name=result_path.name,
-                    path=result_path,
-                ),
-                file_paths=self._path_policy.iter_readable_files(result_path),
-            )
+            file_inventory = self.result_directory_inventory(result_path)
         except AgentPathPolicyError as exc:
             return self._query_files_error(
                 request,
@@ -1282,16 +1273,32 @@ def _query_explicit_result_directory(
         return self._query_files_from_inventory(
             request=request,
             plate_path=plate_path,
-            file_inventory=PlateFileInventory.from_inventories(
-                image_inventory=PlateImageInventory(plate_path=result_path, records=()),
-                result_inventory=result_inventory,
-            ),
+            file_inventory=file_inventory,
             detected_microscope_type=None,
             handler_class=None,
             parser_class=None,
             warnings=(),
         )
 
+    def result_directory_inventory(self, directory: Path) -> PlateFileInventory:
+        """Admit persisted files once for both inspection and viewer reopening."""
+        from openhcs.microscopes.microscope_interfaces import AnalysisResultDirectory
+
+        result_path = self._path_policy.assert_readable(directory)
+        if not result_path.is_dir():
+            raise ValueError(f"Result path is not a directory: {result_path}")
+        return PlateFileInventory.from_inventories(
+            image_inventory=PlateImageInventory(plate_path=result_path, records=()),
+            result_inventory=PlateResultFileInventory.from_directory_files(
+                plate_path=result_path,
+                result_directory=AnalysisResultDirectory(
+                    subdirectory_name=result_path.name,
+                    path=result_path,
+                ),
+                file_paths=self._path_policy.iter_readable_files(result_path),
+            ),
+        )
+
     @staticmethod
     def _query_files_from_inventory(
         *,

openhcs/agent/services/plate_streaming_service.py

@@ -38,6 +38,7 @@
 from openhcs.constants import AllComponents, Backend
 from openhcs.core.config import StreamingConfig
 from openhcs.core.plate_image_inventory import (
+    PlateFileKind,
     PlateFileInventoryQuery,
     PlateFileRecord,
 )
@@ -143,19 +144,42 @@ def stream_files(
                 transport_mode=config.transport_mode,
                 persistent=config.persistent,
             )
-            inventory, inventory_warnings = (
-                self._plate_inspection_service.file_inventory(
-                    stream_context,
-                    kind=None if request.file_paths else request.kind,
+            if request.result_directory is not None:
+                if request.kind is not PlateFileKind.RESULT or request.well is not None:
+                    raise ValueError(
+                        "Explicit result-directory streaming requires kind='result' "
+                        "and no acquisition-component filter."
+                    )
+                result_path = self._plate_inspection_service.resolve_readable_path(
+                    request.result_directory
+                )
+                inventory = self._plate_inspection_service.result_directory_inventory(
+                    result_path
+                )
+                inventory_warnings = context.warnings
+            else:
+                inventory, inventory_warnings = (
+                    self._plate_inspection_service.file_inventory(
+                        stream_context,
+                        kind=None if request.file_paths else request.kind,
+                    )
                 )
-            )
             resolved_records = self._resolve_records(request, inventory)
             (
                 image_paths,
                 roi_paths,
                 roi_component_metadata_by_path,
                 skipped_records,
             ) = self._streamable_paths(resolved_records)
+            if (
+                request.result_directory is not None
+                and image_paths
+                and request.source_receipt is None
+            ):
+                raise ValueError(
+                    "Explicit result images require an exact source receipt; "
+                    "artifact filenames do not establish source identity."
+                )
             all_warnings = inventory_warnings
             if skipped_records:
                 all_warnings = (
@@ -201,6 +225,8 @@ def stream_files(
             read_backend = stream_context.handler.get_primary_backend(
                 stream_context.plate_path, stream_context.filemanager
             )
+            if request.result_directory is not None:
+                read_backend = Backend.DISK.value
             source_projection, producer = self._receipt_source_projection(
                 request, resolved_records, stream_context
             )
@@ -240,7 +266,12 @@ def stream_files(
                     for record in resolved_records
                     if record.streamable_roi_path is not None
                 )
-                streaming_service.stream_rois(
+                roi_streaming_service = StreamingService(
+                    filemanager=context.filemanager,
+                    microscope_handler=context.handler,
+                    plate_path=context.plate_path,
+                )
+                roi_streaming_service.stream_rois(
                     RoiStreamingRequest(
                         viewer=viewer,
                         config=config,
@@ -249,6 +280,7 @@ def stream_files(
                         roi_filenames=roi_paths,
                         component_metadata_by_path=roi_component_metadata_by_path,
                         producer=roi_producer,
+                        require_source_metadata=request.result_directory is not None,
                     )
                 )
         except Exception as exc:

openhcs/core/roi_source_metadata.py

@@ -0,0 +1,52 @@
+"""Persist the existing image-source declaration in native ROI metadata."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import replace
+from typing import ClassVar
+
+from polystore.roi import ROI
+
+from openhcs.core.runtime_image_values import ImagePayloadMetadata
+from openhcs.serialization.json import to_jsonable
+
+
+class ROIArchiveSourceMetadata:
+    """Native ROI sidecar projection, not a second source-metadata schema."""
+
+    FIELD: ClassVar[str] = "openhcs_image_payload_metadata"
+
+    @classmethod
+    def bind(cls, rois: Sequence[ROI], metadata: ImagePayloadMetadata) -> list[ROI]:
+        encoded = to_jsonable(metadata)
+        return [
+            replace(roi, metadata={**roi.metadata, cls.FIELD: encoded}) for roi in rois
+        ]
+
+    @classmethod
+    def decode(cls, rois: Sequence[ROI]) -> ImagePayloadMetadata | None:
+        """Decode once; external ROI archives may have no OpenHCS source binding."""
+        encoded = [roi.metadata.get(cls.FIELD) for roi in rois]
+        if not any(value is not None for value in encoded):
+            return None
+        if any(value is None or value != encoded[0] for value in encoded):
+            raise ValueError(
+                "ROI archive contains missing or conflicting source metadata."
+            )
+        return ImagePayloadMetadata.from_mapping(encoded[0])
+
+    @classmethod
+    def geometry(cls, rois: Sequence[ROI]) -> list[ROI]:
+        """Keep transport-only source declarations out of the ROI feature table."""
+        return [
+            replace(
+                roi,
+                metadata={
+                    key: value
+                    for key, value in roi.metadata.items()
+                    if key != cls.FIELD
+                },
+            )
+            for roi in rois
+        ]