Allow compiled pipeline reuse with runtime observation exports

openhcs/runtime/zmq_compilation.py

@@ -75,7 +75,7 @@ class ZMQCompilationRequest:
     orchestrator: "PipelineOrchestrator"
     wells: list[str]
     compile_artifact_id: str | None
-    request_signature: str
+    compilation_signature: str
     debug_replay_signature: str
     retain_compile_artifact: bool
     compiled_artifacts: MutableMapping[str, "ZMQCompileArtifactRecord"]
@@ -91,8 +91,6 @@ def resolve(self) -> ZMQCompilationResult:
 
     def reuse_artifact(self) -> ZMQCompilationResult:
         artifact = self.compiled_artifacts.get(self.compile_artifact_id)
-        if artifact is not None and not self.retain_compile_artifact:
-            artifact = self.compiled_artifacts.pop(self.compile_artifact_id)
         if artifact is None:
             raise ValueError(
                 f"Missing compile artifact '{self.compile_artifact_id}'. "
@@ -101,7 +99,7 @@ def reuse_artifact(self) -> ZMQCompilationResult:
         expected_signature = (
             self.debug_replay_signature
             if self.retain_compile_artifact
-            else self.request_signature
+            else self.compilation_signature
         )
         artifact_signature = artifact.signature_for_retain_policy(
             self.retain_compile_artifact
@@ -133,6 +131,8 @@ def reuse_artifact(self) -> ZMQCompilationResult:
         }
         compiled_axis_ids = list(execution_bundle.axis_ids)
         compiled_step_names = extract_compiled_step_names(compiled_contexts)
+        if not self.retain_compile_artifact:
+            self.compiled_artifacts.pop(self.compile_artifact_id)
         self.progress_emitter.artifact_init_started(
             compiled_axis_ids=compiled_axis_ids,
             worker_assignments=worker_assignments,
@@ -221,12 +221,12 @@ class ZMQCompileArtifactRecord:
 
     execution_id: str
     plate_id: str
-    request_signature: str
+    compilation_signature: str
     debug_replay_signature: str
     compilation: ZMQCompilationResult
     created_at: float = field(default_factory=time.time)
 
     def signature_for_retain_policy(self, retain_compile_artifact: bool) -> str:
         if retain_compile_artifact:
             return self.debug_replay_signature
-        return self.request_signature
+        return self.compilation_signature

openhcs/runtime/zmq_execution_server.py

@@ -99,6 +99,10 @@ def compile_artifact_id(self) -> str | None:
     def request_signature(self) -> str:
         return self.request_payload.request_signature
 
+    @property
+    def compilation_signature(self) -> str:
+        return self.request_payload.compilation_signature
+
     @property
     def debug_replay_signature(self) -> str:
         return self.request_payload.debug_replay_signature
@@ -668,7 +672,7 @@ def _resolve_compilation(
             orchestrator=orchestrator,
             wells=wells,
             compile_artifact_id=request_context.compile_artifact_id,
-            request_signature=request_context.request_signature,
+            compilation_signature=request_context.compilation_signature,
             debug_replay_signature=request_context.debug_replay_signature,
             retain_compile_artifact=self._retain_compile_artifact(
                 debug_execution_config
@@ -809,7 +813,7 @@ def _store_compile_artifact(
             ZMQCompileArtifactRecord(
                 execution_id=request_context.execution_id,
                 plate_id=request_context.plate_id,
-                request_signature=request_context.request_signature,
+                compilation_signature=request_context.compilation_signature,
                 debug_replay_signature=request_context.debug_replay_signature,
                 compilation=compilation,
             )
@@ -818,7 +822,7 @@ def _store_compile_artifact(
             "[%s] Compilation-only request completed and artifact stored (artifact_id=%s sig=%s)",
             request_context.execution_id,
             request_context.execution_id,
-            request_context.request_signature[:12],
+            request_context.compilation_signature[:12],
         )
         self._set_compile_status("compiled success")
         return compilation.execution_bundle.runtime_contexts

openhcs/runtime/zmq_execution_signature.py

@@ -90,18 +90,43 @@ def to_transport(self) -> dict[str, Any]:
         if self.debug_execution_config is not None:
             params.update(self.debug_execution_config.to_config_params())
         if self.runtime_observation_export_path is not None:
-            params[ZMQAuxiliaryParamField.RUNTIME_OBSERVATION_EXPORT_PATH.value] = str(
-                self.runtime_observation_export_path
-            )
-            if (
-                self.runtime_observation_export_scope
-                is not ZMQRuntimeObservationExportScope.VALUES
-            ):
-                params[
-                    ZMQAuxiliaryParamField.RUNTIME_OBSERVATION_EXPORT_SCOPE.value
-                ] = self.runtime_observation_export_scope.value
+            params.update(self.runtime_observation_items())
         return params
 
+    def runtime_observation_items(self) -> TransportRequestItems:
+        """Project the complete observation-only field family, including defaults."""
+
+        return (
+            (
+                ZMQAuxiliaryParamField.RUNTIME_OBSERVATION_EXPORT_PATH.value,
+                (
+                    None
+                    if self.runtime_observation_export_path is None
+                    else str(self.runtime_observation_export_path)
+                ),
+            ),
+            (
+                ZMQAuxiliaryParamField.RUNTIME_OBSERVATION_EXPORT_SCOPE.value,
+                self.runtime_observation_export_scope.value,
+            ),
+        )
+
+    @classmethod
+    def compilation_config_params(
+        cls, config_params: Mapping[str, Any] | None
+    ) -> dict[str, Any] | None:
+        """Exclude only fields owned by the runtime observation projection."""
+
+        auxiliary = cls.from_transport(config_params)
+        if config_params is None:
+            return None
+        observation_fields = dict(auxiliary.runtime_observation_items())
+        return {
+            key: value
+            for key, value in config_params.items()
+            if key not in observation_fields
+        } or None
+
     @classmethod
     def from_transport(
         cls,
@@ -337,12 +362,22 @@ def compile_artifact_id(self) -> str | None:
     def request_signature(self) -> str:
         return self.signature_for_config_params(self.config_params)
 
+    @property
+    def compilation_signature(self) -> str:
+        return self.signature_for_config_params(
+            ZMQAuxiliaryExecutionParams.compilation_config_params(self.config_params)
+        )
+
     @property
     def debug_replay_signature(self) -> str:
         from openhcs.core.debug import DebugExecutionConfig
 
         return self.signature_for_config_params(
-            DebugExecutionConfig.compatibility_config_params(self.config_params)
+            DebugExecutionConfig.compatibility_config_params(
+                ZMQAuxiliaryExecutionParams.compilation_config_params(
+                    self.config_params
+                )
+            )
         )
 
     @property

paper/plans/compile_observation_identity_20260929.md

@@ -0,0 +1,71 @@
+# Compile reuse with runtime observation exports
+
+Owner: OpenHCS blind-analysis integration coordinator. Issue #179.
+Audited main ac2e10e9a190e877f6dab4964dae1e051d159ac9.
+
+The installed public MCP accepted compilation but rejected reuse of the same
+session when execution requested an observation export. The export-only path
+and scope entered the full request hash. The rejected reuse also consumed the
+valid compile artifact before compatibility validation.
+
+## Admitted answers and determining owners
+
+The authorized compile -> execute -> preserve inspectable evidence workflow
+requires observation destination and values/outcomes retention to vary without
+changing compilation compatibility. Source identity, pipeline source, global
+configuration, workload axes, debug compile inputs, and unknown transport inputs
+must still distinguish compilation. Full submission identity remains distinct.
+
+ZMQAuxiliaryExecutionParams already determines observation options and validates
+them. Its observation projection now drives both wire emission and the excluded
+compilation fields. No mirror exclusion roster is introduced. The projection
+contains defaults too, so explicit values scope and absent export path do not
+invalidate reuse. Scope is emitted explicitly with requested observation output.
+The existing transport Enum remains a value-only boundary vocabulary; it does
+not acquire a behavior dispatch table.
+
+ZMQExecutionRequestPayload owns full request identity and derives the narrower
+compilation identity through the existing canonical signature algorithm.
+DebugExecutionConfig retains ownership of debug replay normalization; it receives
+the same observation-free compilation inputs. ZMQExecutionContext is a derived
+server view. ZMQCompilationRequest and ZMQCompileArtifactRecord name and consume
+compilation_signature, not the wider request_signature. No compatibility aliases.
+
+Required: auxiliary observation declaration -> wire projection and compilation
+projection; request payload -> compile-record/reuse identity; valid artifact ->
+worker bundle. Forbidden: observation destination -> compile incompatibility;
+rejected compatibility -> artifact consumption. Distinct roles: complete request
+logs, debug replay policy, output writers, UI history and biological acceptance.
+
+## Source coverage, counterevidence and proof limits
+
+Original NRA syntax census covers execution_signature.py, compilation.py and
+execution_server.py at the pinned main: 13 original ClassDef rows. Preserve every
+row and unique compact-family join, including alternative owners. External Enum,
+ExecutionServer and imported runtime behavior are not proved by this bounded
+projection. Full global detector/raw-record/R1 and authored effect-equivalence
+proof remain OPEN. This is a scoped behavioral repair, not an equivalence claim
+or complete architecture audit.
+
+IDEN-1 and IDEN-7: a full request signature answered a narrower compile question.
+BOUND-2: derive the observation projection from its existing typed owner rather
+than adding another string-key filter in the compiler. A future observation-only
+field belongs to that owner's wire projection; compilation exclusion is derived,
+not another handwritten membership list.
+
+Runtime compile caches reset on process restart. No durable science, saved
+session, reference answer or installed shared application is rewritten.
+PR #157 owns outcome-export paths; #180 owns lazy catalog preparation; #183 owns
+completion polling. Their shared-file hunks are separate and were communicated.
+
+## Acceptance
+
+Focused tests preserve full request differences while accepting values/outcomes
+destination changes and defaults; reject source, pipeline, config, axis, debug
+and unknown changes; reject invalid observation scope. Reuse rejects signature,
+plate and missing-context differences without consuming either ordinary or
+debug-retained artifacts, and consumes ordinary artifacts only after validation.
+Installed public MCP must compile a bounded source-backed pipeline, execute
+with each observation scope using the returned artifact, and reopen actual typed
+retained observations/output files. Report installed, live and biological
+evidence separately. No timeout increase or repeated uncertain execution.

tests/integration/test_compiled_artifact_inspection.py

@@ -240,7 +240,7 @@ def produce(image):
         ZMQCompileArtifactRecord(
             execution_id="compile-1",
             plate_id="/plates/one",
-            request_signature="request",
+            compilation_signature="request",
             debug_replay_signature="debug",
             compilation=ZMQCompilationResult(
                 execution_bundle=bundle,

tests/unit/test_artifact_plan_view.py

@@ -219,7 +219,7 @@ def test_registered_control_router_reads_retained_compile_artifact() -> None:
     record = ZMQCompileArtifactRecord(
         execution_id="compile-1",
         plate_id="/plates/one",
-        request_signature="request",
+        compilation_signature="request",
         debug_replay_signature="debug",
         compilation=ZMQCompilationResult(
             execution_bundle=bundle,

tests/unit/test_zmq_compilation.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 import time
+from dataclasses import replace
+
+import pytest
 
 from openhcs.core.compiled_execution import (
     CompiledExecutionBundle,
@@ -165,7 +168,7 @@ def test_reused_compile_artifact_reads_step_names_from_compiled_plans() -> None:
         "artifact-1": ZMQCompileArtifactRecord(
             execution_id="compile-1",
             plate_id="/tmp/plate",
-            request_signature="signature",
+            compilation_signature="signature",
             debug_replay_signature="debug-signature",
             compilation=compilation,
         )
@@ -178,7 +181,7 @@ def test_reused_compile_artifact_reads_step_names_from_compiled_plans() -> None:
         orchestrator=None,
         wells=["A01"],
         compile_artifact_id="artifact-1",
-        request_signature="signature",
+        compilation_signature="signature",
         debug_replay_signature="debug-signature",
         retain_compile_artifact=False,
         compiled_artifacts=artifacts,
@@ -189,6 +192,7 @@ def test_reused_compile_artifact_reads_step_names_from_compiled_plans() -> None:
 
     result = request.reuse_artifact()
 
+    assert artifacts == {}
     assert result.execution_bundle is bundle
     assert progress_emitter.artifact_init_events == [
         {
@@ -217,7 +221,7 @@ def test_compile_fresh_emits_heartbeat_during_long_compilation() -> None:
         orchestrator=orchestrator,
         wells=["A01"],
         compile_artifact_id=None,
-        request_signature="signature",
+        compilation_signature="signature",
         debug_replay_signature="debug-signature",
         retain_compile_artifact=False,
         compiled_artifacts={},
@@ -254,3 +258,71 @@ def test_compile_fresh_emits_heartbeat_during_long_compilation() -> None:
         }
     ]
     assert progress_emitter.axis_compile_success_events == ["A01"]
+
+
+@pytest.mark.parametrize("retain", (False, True))
+@pytest.mark.parametrize("invalid", ("signature", "plate", "contexts"))
+def test_rejected_reuse_preserves_artifact_for_a_valid_request(retain, invalid):
+    bundle = CompiledExecutionBundle(
+        pipeline_definition=(),
+        runtime_contexts={"A01": _compiled_context("A01")},
+        transport_contexts={},
+        worker_assignments={"worker_0": ["A01"]},
+        runtime_environment=_runtime_environment(),
+    )
+    artifact = ZMQCompileArtifactRecord(
+        execution_id="compile-1",
+        plate_id="/plate",
+        compilation_signature="signature",
+        debug_replay_signature="debug-signature",
+        compilation=ZMQCompilationResult(
+            execution_bundle=bundle, compiled_axis_ids=["A01"]
+        ),
+    )
+    if invalid == "signature":
+        artifact = replace(
+            artifact, compilation_signature="wrong", debug_replay_signature="wrong"
+        )
+    elif invalid == "plate":
+        artifact = replace(artifact, plate_id="/another-plate")
+    else:
+        artifact = replace(
+            artifact,
+            compilation=replace(
+                artifact.compilation,
+                execution_bundle=replace(bundle, runtime_contexts={}),
+            ),
+        )
+    artifacts = {"artifact-1": artifact}
+    emitter = _ProgressEmitter()
+    request = ZMQCompilationRequest(
+        execution_id="execute-1",
+        plate_id="/plate",
+        pipeline_steps=[],
+        orchestrator=None,
+        wells=["A01"],
+        compile_artifact_id="artifact-1",
+        compilation_signature="signature",
+        debug_replay_signature="debug-signature",
+        retain_compile_artifact=retain,
+        compiled_artifacts=artifacts,
+        progress_emitter=emitter,
+        compiler_progress_queue=None,
+        debug_execution_policy=None,
+    )
+    with pytest.raises(ValueError):
+        request.reuse_artifact()
+    assert artifacts == {"artifact-1": artifact}
+    assert emitter.artifact_init_events == []
+
+    valid = replace(
+        artifact,
+        plate_id="/plate",
+        compilation_signature="signature",
+        debug_replay_signature="debug-signature",
+        compilation=replace(artifact.compilation, execution_bundle=bundle),
+    )
+    artifacts["artifact-1"] = valid
+    result = request.reuse_artifact()
+    assert result.execution_bundle is bundle
+    assert artifacts == ({"artifact-1": valid} if retain else {})