Archived: integrated graph ROI writer and superseded reopen receiving

docs/validation/check-persisted-roi-admission-134.py

@@ -0,0 +1,141 @@
+"""Read-only admission witness and provider-free producer regression for #134.
+
+No image loader, ROI decoder, sampler, viewer or transport is invoked. Synthetic
+archive sentinels intentionally test inventory admission, not archive validity.
+"""
+
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from polystore.filemanager import FileManager
+from polystore.disk import DiskStorageBackend
+from polystore.virtual_workspace import SourcePixelRef
+
+from openhcs.agent.dto.plate import PlateFileQueryRequest
+from openhcs.agent.path_policy import AgentPathPolicy
+from openhcs.agent.services.plate_inspection_service import PlateInspectionService
+from openhcs.constants.constants import Backend, FileFormat
+from openhcs.core.artifacts import ObjectLabelsArtifactType
+from openhcs.core.compiled_step_plan import (
+    CompiledStepPlan,
+    RuntimeArtifactMaterializationPlan,
+)
+from openhcs.core.plate_file_inventory import PlateFileKind
+from openhcs.core.plate_image_inventory import PlateResultFileInventory
+from openhcs.core.runtime_image_values import ImagePayloadMetadata
+from openhcs.core.source_projection import (
+    OpenHCSPlaneAddress,
+    SourceArtifactProjection,
+    SourceProjectionMetadataSerializer,
+)
+from openhcs.core.steps.function_outputs import RuntimeArtifactMetadataTarget
+from openhcs.core.virtual_workspace_metadata import AtomicMetadataWriter, METADATA_CONFIG
+from openhcs.microscopes.openhcs import OpenHCSMetadataHandler
+from openhcs.microscopes.source_schema import SourceSchemaFilenameParser
+
+
+RETAINED_PLATE = Path(
+    "/home/ts/wt/openhcs-issue-batch-20260929/"
+    "neurite-development-skill383-20261001/output/"
+    "assisted-2-candidate-2-sensitivity-diagnostic/input_openhcs"
+)
+RETAINED_ROI = (
+    RETAINED_PLATE / "results/A01_s001_w2_z001_t001_neurons_step0_rois.roi.zip"
+)
+
+
+def test_retained_explicit_directory_admits_exact_roi_without_content_read():
+    service = PlateInspectionService(
+        AgentPathPolicy.with_roots(readable_roots=(RETAINED_PLATE,), writable_roots=())
+    )
+    result = service.query_files(
+        PlateFileQueryRequest.from_fields(
+            plate_path=str(RETAINED_PLATE),
+            result_directory=str(RETAINED_ROI.parent),
+            kind="result",
+            include_previews=False,
+        )
+    )
+    assert result.errors == ()
+    matching = [record for record in result.records if record.full_path == str(RETAINED_ROI)]
+    assert len(matching) == 1
+    assert matching[0].file_format == "ROI"
+    assert matching[0].preview is None
+    record = service.result_directory_inventory(RETAINED_ROI.parent).require_file_record(
+        str(RETAINED_ROI)
+    )
+    assert record.kind is PlateFileKind.RESULT
+    assert record.file_format is FileFormat.ROI
+    assert record.streamable_roi_path == str(RETAINED_ROI)
+    filemanager = FileManager({Backend.DISK.value: DiskStorageBackend()})
+    ordinary = PlateResultFileInventory.from_handler(
+        plate_path=RETAINED_PLATE,
+        metadata_handler=OpenHCSMetadataHandler(filemanager),
+        parser=None,
+    )
+    assert ordinary.records == ()
+    print(f"Explicit typed inventory admits {result.total_count} files and exact ROI; no content read")
+
+
+@pytest.mark.parametrize("destination", ("declared_outputs", "nested/independent_outputs"))
+def test_runtime_result_publication_reaches_original_inventory(tmp_path, destination):
+    plate = tmp_path / "output_plate"
+    directory = plate / destination
+    directory.mkdir(parents=True)
+    roi = directory / "unrelated-name.roi.zip"
+    roi.write_bytes(b"inventory sentinel; not a scientific ROI")
+    # Mixed persisted results, like the retained run. This image is never loaded;
+    # original typed projection publication supplies its synthetic address.
+    image = directory / "unrelated-labels.tif"
+    image.write_bytes(b"inventory sentinel; not a scientific image")
+    virtual_path = str(image.relative_to(plate))
+    components = {
+        "well": "A01", "site": "1", "channel": "2",
+        "z_index": "1", "timepoint": "1", "pixel_size": 1.25,
+    }
+    projection = SourceArtifactProjection(
+        address=OpenHCSPlaneAddress.from_values("A01", "1", "2", "1", "1"),
+        ref=SourcePixelRef(Backend.DISK.value, virtual_path),
+        source_alias="synthetic labels",
+        artifact_kind=ObjectLabelsArtifactType,
+        source_metadata=components,
+        image_metadata=ImagePayloadMetadata(source_component_metadata=components),
+    )
+    AtomicMetadataWriter().merge_source_projection_metadata(
+        METADATA_CONFIG.metadata_path(plate),
+        destination,
+        SourceProjectionMetadataSerializer(SourceSchemaFilenameParser()).projection_fields(
+            ((projection, virtual_path),)
+        ),
+    )
+    filemanager = FileManager({Backend.DISK.value: DiskStorageBackend()})
+    plan = CompiledStepPlan(
+        step_index=0,
+        step_name="independent persistent result",
+        step_type="FunctionStep",
+        axis_id="A01",
+        output_plate_root=str(plate),
+        analysis_results_dir=str(directory),
+        runtime_artifact_materialization=RuntimeArtifactMaterializationPlan(
+            persistent_enabled=True, persistent_backend=Backend.DISK.value
+        ),
+    )
+    target = RuntimeArtifactMetadataTarget.from_plan(plan)
+    assert target is not None
+    context = SimpleNamespace(
+        filemanager=filemanager,
+        metadata_cache={},
+        microscope_handler=SimpleNamespace(
+            parser=SourceSchemaFilenameParser(), microscope_type="openhcsdata"
+        ),
+    )
+    # Exercise the original target's shared writer and original metadata reader,
+    # without rendering, reading the sentinel, or fabricating a source receipt.
+    target.write(context)
+    inventory = PlateResultFileInventory.from_handler(
+        plate_path=plate,
+        metadata_handler=OpenHCSMetadataHandler(filemanager),
+        parser=None,
+    )
+    assert str(roi) in tuple(record.full_path for record in inventory.records)

docs/validation/check-persisted-roi-only-proposal-134.py

@@ -0,0 +1,83 @@
+"""Select original/proposed PR394 source owners, not a second implementation.
+
+QA134_SOURCE_ROOT points to a small, explicitly retained source projection.
+The existing runner owns ABI preparation and pytest. No product method or
+assertion is replaced; normal imports select these files before collection.
+"""
+
+import importlib.abc
+import importlib.util
+import os
+from pathlib import Path
+import runpy
+import subprocess
+import sys
+
+source_root = Path(os.environ["QA134_SOURCE_ROOT"]).resolve(strict=True)
+sources = {
+    ".".join(path.relative_to(source_root).with_suffix("").parts): path
+    for path in source_root.rglob("*.py")
+}
+repository = Path(__file__).resolve().parents[2]
+revision = os.environ.get("QA134_GIT_REF")
+git_sources = {}
+if revision:
+    changed = subprocess.check_output(
+        ["git", "diff", "--name-only", "HEAD", revision, "--", "openhcs"],
+        cwd=repository,
+        text=True,
+    ).splitlines()
+    git_sources = {
+        ".".join(Path(path).with_suffix("").parts).removesuffix(".__init__"): path
+        for path in changed
+        if path.endswith(".py")
+    }
+
+
+class GitOwnerSource(importlib.abc.Loader):
+    """Import exact pinned source blobs without copying an entire checkout."""
+
+    def create_module(self, spec):
+        return None
+
+    def exec_module(self, module):
+        path = git_sources[module.__name__]
+        source = subprocess.check_output(
+            ["git", "show", f"{revision}:{path}"],
+            cwd=repository,
+        )
+        print(f"PINNED SOURCE ONLY: {module.__name__} -> {revision}:{path}")
+        exec(compile(source, f"{revision}:{path}", "exec"), module.__dict__)
+
+
+class SelectedOwnerSources(importlib.abc.MetaPathFinder):
+    def find_spec(self, fullname, path=None, target=None):
+        source = sources.get(fullname)
+        if source is not None:
+            print(f"SOURCE PROJECTION ONLY: {fullname} -> {source}")
+            return importlib.util.spec_from_file_location(fullname, source)
+        source_path = git_sources.get(fullname)
+        if source_path is not None:
+            spec = importlib.util.spec_from_loader(
+                fullname,
+                GitOwnerSource(),
+                is_package=source_path.endswith("/__init__.py"),
+            )
+            if spec.submodule_search_locations is not None:
+                spec.submodule_search_locations = [
+                    str(repository / Path(source_path).parent)
+                ]
+            return spec
+        return None
+
+
+if any(name in sys.modules for name in {*sources, *git_sources}):
+    raise RuntimeError("A selected owner was imported before source selection")
+selection = SelectedOwnerSources()
+sys.meta_path.insert(0, selection)
+try:
+    runpy.run_path(
+        str(Path(__file__).with_name("check-function-help-390.py")), run_name="__main__"
+    )
+finally:
+    sys.meta_path.remove(selection)

docs/validation/check-persisted-roi-proposal-134.py

@@ -0,0 +1,42 @@
+"""Select the explicit source proposal before provider/plugin-free collection.
+
+The existing validation runner owns ABI preparation and pytest invocation. This
+callback changes only the selected source file, not product classes or test
+assertions. The projection is not an installed package or production change.
+"""
+
+import importlib.util
+from pathlib import Path
+import runpy
+import sys
+
+import pytest
+
+
+original_main = pytest.main
+
+
+def check_proposed_source(args):
+    source = (
+        Path(__file__).resolve().parents[2]
+        / ".qa134-proposal-20261001/source/openhcs/core/steps/function_outputs.py"
+    )
+    module_name = "openhcs.core.steps.function_outputs"
+    if module_name in sys.modules:
+        raise RuntimeError("Metadata owner was imported before source selection")
+    spec = importlib.util.spec_from_file_location(module_name, source)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    spec.loader.exec_module(module)
+    print(f"SOURCE PROPOSAL ONLY: {module.__file__}")
+    pytest.main = original_main
+    return original_main(args)
+
+
+pytest.main = check_proposed_source
+try:
+    runpy.run_path(
+        str(Path(__file__).with_name("check-function-help-390.py")), run_name="__main__"
+    )
+finally:
+    pytest.main = original_main

docs/validation/graph-roi-source-roundtrip-134.patch

@@ -0,0 +1,27 @@
+diff --git a/openhcs/processing/materialization/core.py b/openhcs/processing/materialization/core.py
+index a095267a2..61340a650 100644
+--- a/openhcs/processing/materialization/core.py
++++ b/openhcs/processing/materialization/core.py
+@@ -75,0 +76 @@ from openhcs.core.source_matching import (
++from openhcs.core.source_metadata import SourceVoxelSpacing
+@@ -3286,0 +3288,2 @@ def _write_spatial_graph_roi_zip(
++    from openhcs.core.roi_source_metadata import ROIArchiveSourceMetadata
++
+@@ -3294,0 +3298,4 @@ def _write_spatial_graph_roi_zip(
++    source_metadata = ImagePayloadMetadata(
++        source_provenance=graph.source_provenance,
++        source_voxel_spacing=SourceVoxelSpacing(graph.coordinate_spacing),
++    )
+@@ -3336 +3343,2 @@ def _write_spatial_graph_roi_zip(
+-            content=rois,
++            content=ROIArchiveSourceMetadata.bind(rois, source_metadata),
++            metadata=source_metadata,
+diff --git a/tests/unit/test_spatial_graph_materialization.py b/tests/unit/test_spatial_graph_materialization.py
+index 6dc0145d2..0c0e70ef3 100644
+--- a/tests/unit/test_spatial_graph_materialization.py
++++ b/tests/unit/test_spatial_graph_materialization.py
+@@ -19,0 +20 @@ from openhcs.core.artifacts import (
++from openhcs.core.roi_source_metadata import ROIArchiveSourceMetadata
+@@ -343 +344 @@ def test_graph_roi_projection_preserves_paths_and_graph_features() -> None:
+-    first_roi = roi_output.content[0]
++    first_roi = ROIArchiveSourceMetadata.geometry(roi_output.content)[0]