DL3: an output dataset records what produced it (draft, handed off)

openhcs/agent/capabilities.py

@@ -143,6 +143,8 @@
 from openhcs.agent.dto.plate import (
     PlateFileQueryRequest,
     PlateFileQueryResult,
+    RunRecordReadRequest,
+    RunRecordReadResult,
     PlateFileStreamRequest,
     PlateFileStreamResult,
     PlateImageSampleRequest,
@@ -2352,6 +2354,46 @@ class QueryPlateFilesCapability(MainThreadProgressCapability, PlatePathCapabilit
     )
 
 
+class ReadRunRecordCapability(PlatePathCapability):
+    name = "openhcs_read_run_record"
+    cli_command = "read-run-record"
+    title = "Read run record"
+    description = (
+        "Read-only: what produced an output dataset. Pass an output dataset root "
+        "(the directory holding openhcs_metadata.json) to get every run record "
+        "by record id, or one output file inside a dataset to get the producer "
+        "identity stored for that file, the producing step and the record of "
+        "that run. A file the dataset does not list is an error, never a guess "
+        "from its name. Each record lists what every step ran: each function "
+        "with its identity (registry key, defining and imported library "
+        "versions, sha256 of a custom function's source), every parameter with "
+        "its value and source (pipeline, function_default, run_time), declared "
+        "files with content hashes, and devices. Pass include_full_record=true "
+        "to add the complete stored record: pipeline and global config "
+        "documents, resolved configuration with the owner of every value, "
+        "custom source texts and all installed distributions."
+    )
+    exposition = PlatePathCapability.exposition.refine(
+        workflow_stage=CapabilityWorkflowStage.INSPECTION,
+        visibility=CapabilityVisibility.STANDARD,
+    )
+    service = "run_record"
+    data_exposure = (
+        "local_output_path",
+        "pipeline_source",
+        "resolved_configuration",
+        "custom_function_source",
+        "installed_distribution_versions",
+    )
+    security_requirements = ("AgentPathPolicy readable root",)
+    input_contract = RunRecordReadRequest
+    output_contract = RunRecordReadResult
+    invocation = AgentDataclassRequestServiceInvocation(
+        service=lambda context: context.run_record_service,
+        method=lambda service, request: service.read(request),
+    )
+
+
 class SamplePlateImageCapability(MainThreadProgressCapability, PlatePathCapability):
     name = "openhcs_sample_plate_image"
     cli_command = "sample-plate-image"

openhcs/agent/dto/__init__.py

@@ -130,6 +130,10 @@
             "PlatePathInspectionRequest",
             "PlatePathInspectionResult",
             "PlateWorkspacePreparationOperation",
+            "ProducedFileEntry",
+            "RunRecordEntry",
+            "RunRecordReadRequest",
+            "RunRecordReadResult",
             "SelectedPlateFileQueryTarget",
             "SelectedPlateFileQueryRequest",
             "SelectedPlateFileQueryResult",

openhcs/agent/dto/plate.py

@@ -9,10 +9,13 @@
 from typing import ClassVar
 
 from metaclass_registry import AutoRegisterMeta
+from polystore.streaming.identity import StreamProducerIdentity
 from zmqruntime.config import TransportMode
 
 from formbridge.contracts import (
     AGENT_PARAMETER_DESCRIPTION_METADATA_KEY,
+    AgentCliArgumentSpec,
+    AgentDataclassCliRequest,
     AgentError,
     AgentResultEnvelope,
     AgentResourceRef,
@@ -23,6 +26,7 @@
 from openhcs.core.component_filters import ComponentFilters
 from openhcs.core.config import NapariDisplayConfig
 from openhcs.core.dataset_sources.choice import AutoDetectedSource, DatasetSourceChoice
+from openhcs.core.run_record import RunRecord, StepCalls
 from openhcs.core.plate_file_inventory import (
     PlateFileInventoryQuery,
     PlateFileKind,
@@ -1339,3 +1343,71 @@ class SelectedPlateFileStreamResult(AgentResultEnvelope):
     selected_plate: JsonObject = field(default_factory=dict)
     target: type[SelectedPlateFileQueryTarget] = SelectedRowRoot
     stream: PlateFileStreamResult | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class RunRecordReadRequest(AgentDataclassCliRequest):
+    """Read what produced an output dataset, or one output file inside one.
+
+    ``path`` is an output dataset root (the directory that holds the OpenHCS
+    metadata document) or one output file below such a root.
+    ``include_full_record`` adds the complete stored record to each entry: the
+    pipeline and global config documents, the resolved configuration with the
+    owner of every value, custom function source texts and every installed
+    distribution.
+    """
+
+    path: str
+    include_full_record: bool = False
+
+    @classmethod
+    def agent_cli_argument_specs(cls) -> tuple[AgentCliArgumentSpec, ...]:
+        return (
+            AgentCliArgumentSpec(
+                field_name="path",
+                positional=True,
+                help="Output dataset root, or one output file inside one.",
+            ),
+            *super(RunRecordReadRequest, cls).agent_cli_argument_specs(),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class RunRecordEntry:
+    """One stored run record, addressed by the sha256 of its stored JSON.
+
+    ``steps`` says what each step ran: every function with its identity (a
+    custom function by the sha256 of its source), every parameter with its
+    value and source, and every declared file with its content hash.
+    ``record`` is the complete record, present when the request asked for it.
+    """
+
+    record_id: str
+    steps: tuple[StepCalls, ...]
+    record: RunRecord | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class ProducedFileEntry:
+    """One output file with the producer identity stored for it and its producing step."""
+
+    path: str
+    record_id: str
+    producer: StreamProducerIdentity
+    step: StepCalls
+
+
+@dataclass(frozen=True, kw_only=True, slots=True)
+class RunRecordReadResult(AgentResultEnvelope):
+    """Run records read from one output dataset.
+
+    For a dataset root ``records`` holds every record and ``produced_files`` is
+    empty; for an output file ``produced_files`` holds that file and
+    ``records`` holds the one record of the run that wrote it.
+    """
+
+    path: str
+    records: tuple[RunRecordEntry, ...] = ()
+    produced_files: tuple[ProducedFileEntry, ...] = ()
+    errors: tuple[AgentError, ...] = ()
+    warnings: tuple[AgentWarning, ...] = ()

openhcs/agent/dto/ui_bridge.py

@@ -4,7 +4,6 @@
 
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import TYPE_CHECKING
 from python_introspect import (
     JsonObject,
 )
@@ -35,10 +34,7 @@
     UiMutationRequestTokenCarrier,
     UiStateSurfaceEnvelope,
 )
-if TYPE_CHECKING:
-    from openhcs.core.progress.live_measurements import (
-        LiveMeasurementTablePreview,
-    )
+from openhcs.core.progress.live_measurements import LiveMeasurementTablePreview
 
 
 def _dataset_workflow_enum() -> type[Enum]:

openhcs/agent/services/run_record_service.py

@@ -0,0 +1,100 @@
+"""Read the record of what produced an output dataset, for agents."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from pathlib import Path
+
+from openhcs.agent.dto.common import SCHEMA_VERSION
+from openhcs.agent.dto.plate import (
+    ProducedFileEntry,
+    RunRecordEntry,
+    RunRecordReadRequest,
+    RunRecordReadResult,
+)
+from openhcs.agent.exceptions import AgentFacingErrorMixin
+from openhcs.agent.path_policy import AgentPathPolicy
+from openhcs.core.run_record import RunRecord, UnrecordedOutputFile
+from openhcs.core.virtual_workspace_metadata import get_metadata_path
+
+
+class OutputFileNotRecorded(AgentFacingErrorMixin, LookupError):
+    """The dataset's metadata does not say which step produced the requested file."""
+
+    agent_error_code = "run_record_output_file_unrecorded"
+    agent_error_hint = (
+        "Pass the output dataset root to list its run records, or a file the "
+        "dataset's metadata lists under produced_files."
+    )
+
+
+def _entry(record: RunRecord, request: RunRecordReadRequest) -> RunRecordEntry:
+    """What each step ran, with the complete stored record when the request asks for it."""
+    return RunRecordEntry(
+        record_id=record.record_id,
+        steps=tuple(step.calls for step in record.steps),
+        record=record if request.include_full_record else None,
+    )
+
+
+@dataclass(frozen=True, slots=True)
+class RunRecordTarget(ABC):
+    """A readable path a run record is asked for: a dataset root or a file inside one."""
+
+    path: Path
+
+    @classmethod
+    def at(cls, path: Path) -> "RunRecordTarget":
+        """A directory holding the metadata document is a dataset root; anything else is a file."""
+        if Path(get_metadata_path(path)).is_file():
+            return DatasetRootTarget(path)
+        return OutputFileTarget(path)
+
+    @abstractmethod
+    def read(self, request: RunRecordReadRequest) -> RunRecordReadResult:
+        """Read this target with the detail the request asks for."""
+
+
+class DatasetRootTarget(RunRecordTarget):
+    def read(self, request: RunRecordReadRequest) -> RunRecordReadResult:
+        return RunRecordReadResult(
+            schema_version=SCHEMA_VERSION,
+            path=str(self.path),
+            records=tuple(
+                _entry(record, request)
+                for record in RunRecord.for_dataset(self.path).values()
+            ),
+        )
+
+
+class OutputFileTarget(RunRecordTarget):
+    def read(self, request: RunRecordReadRequest) -> RunRecordReadResult:
+        try:
+            produced = RunRecord.for_output_file(self.path)
+        except UnrecordedOutputFile as error:
+            raise OutputFileNotRecorded(str(error), path=str(self.path)) from error
+        return RunRecordReadResult(
+            schema_version=SCHEMA_VERSION,
+            path=str(self.path),
+            records=(_entry(produced.record, request),),
+            produced_files=(
+                ProducedFileEntry(
+                    path=str(self.path),
+                    record_id=produced.record.record_id,
+                    producer=produced.producer,
+                    step=produced.step,
+                ),
+            ),
+        )
+
+
+class RunRecordService:
+    """Wrap ``RunRecord``'s readers behind the agent path policy."""
+
+    def __init__(self, path_policy: AgentPathPolicy) -> None:
+        self._path_policy = path_policy
+
+    def read(self, request: RunRecordReadRequest) -> RunRecordReadResult:
+        target = RunRecordTarget.at(self._path_policy.assert_readable(request.path))
+        return target.read(request)

openhcs/core/compiled_execution.py

@@ -10,6 +10,7 @@
 )
 from openhcs.core.config import GlobalPipelineConfig, MultiprocessingStartMethod
 from openhcs.core.context.processing_context import CompiledContext
+from openhcs.core.run_record import RunDeclaration
 from openhcs.core.steps.abstract import AbstractStep
 from openhcs.core.worker_start_policy import (
     WorkerStartDecision,
@@ -132,6 +133,8 @@ class CompiledExecutionBundle:
     transport_contexts: Mapping[str, CompiledContext]
     worker_assignments: Mapping[str, list[str]]
     runtime_environment: CompiledRuntimeEnvironmentPlan
+    run_declaration: RunDeclaration
+    """The documents and resolved configuration this bundle was compiled from."""
 
     @property
     def axis_ids(self) -> tuple[str, ...]:
@@ -156,6 +159,29 @@ def requires_parent_runtime_observation(self) -> bool:
             for plan in context.step_plans.values()
         )
 
+    @classmethod
+    def with_nothing_to_run(
+        cls,
+        pipeline_definition: Sequence[AbstractStep],
+        effective_config: GlobalPipelineConfig,
+        *,
+        server_mode: bool,
+        run_declaration: RunDeclaration,
+    ) -> "CompiledExecutionBundle":
+        """The bundle of a compilation that selected no axis value or no enabled step."""
+
+        return cls.from_runtime_contexts(
+            pipeline_definition=pipeline_definition,
+            runtime_contexts={},
+            worker_assignments={},
+            runtime_environment=CompiledRuntimeEnvironmentPlan.from_global_config(
+                effective_config,
+                compiled_contexts={},
+                server_mode=server_mode,
+            ),
+            run_declaration=run_declaration,
+        )
+
     @classmethod
     def from_runtime_contexts(
         cls,
@@ -164,6 +190,7 @@ def from_runtime_contexts(
         runtime_contexts: Mapping[str, CompiledContext],
         worker_assignments: Mapping[str, list[str]],
         runtime_environment: CompiledRuntimeEnvironmentPlan,
+        run_declaration: RunDeclaration,
     ) -> "CompiledExecutionBundle":
         from openhcs.core.function_step_transport import FunctionStepTransportAuthority
 
@@ -178,6 +205,7 @@ def from_runtime_contexts(
                 key: list(value) for key, value in worker_assignments.items()
             },
             runtime_environment=runtime_environment,
+            run_declaration=run_declaration,
         )
 
     @classmethod
@@ -187,6 +215,7 @@ def from_unassigned_runtime_contexts(
         pipeline_definition: Sequence[AbstractStep],
         runtime_contexts: Mapping[str, CompiledContext],
         runtime_environment: CompiledRuntimeEnvironmentPlan,
+        run_declaration: RunDeclaration,
     ) -> "CompiledExecutionBundle":
         """Create a bundle before worker ownership has been assigned."""
 
@@ -195,6 +224,7 @@ def from_unassigned_runtime_contexts(
             runtime_contexts=runtime_contexts,
             worker_assignments={},
             runtime_environment=runtime_environment,
+            run_declaration=run_declaration,
         )
 
     def for_transport_serialization(