diff --git a/openhcs/runtime/zmq_compilation.py b/openhcs/runtime/zmq_compilation.py index 20e3eed2ad..f20ef7f6d5 100644 --- a/openhcs/runtime/zmq_compilation.py +++ b/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,7 +221,7 @@ 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) @@ -229,4 +229,4 @@ class ZMQCompileArtifactRecord: 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 diff --git a/openhcs/runtime/zmq_execution_server.py b/openhcs/runtime/zmq_execution_server.py index c1b80db0a0..697c012c59 100644 --- a/openhcs/runtime/zmq_execution_server.py +++ b/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 diff --git a/openhcs/runtime/zmq_execution_signature.py b/openhcs/runtime/zmq_execution_signature.py index da95ce7055..d002f5f23e 100644 --- a/openhcs/runtime/zmq_execution_signature.py +++ b/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 diff --git a/paper/plans/compile_observation_identity_20260929.md b/paper/plans/compile_observation_identity_20260929.md new file mode 100644 index 0000000000..7f57870333 --- /dev/null +++ b/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. diff --git a/tests/integration/test_compiled_artifact_inspection.py b/tests/integration/test_compiled_artifact_inspection.py index b367322505..317bf0469a 100644 --- a/tests/integration/test_compiled_artifact_inspection.py +++ b/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, diff --git a/tests/unit/test_artifact_plan_view.py b/tests/unit/test_artifact_plan_view.py index a2557a3cb9..464d91dd1f 100644 --- a/tests/unit/test_artifact_plan_view.py +++ b/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, diff --git a/tests/unit/test_zmq_compilation.py b/tests/unit/test_zmq_compilation.py index eeb77aaf67..252c76b642 100644 --- a/tests/unit/test_zmq_compilation.py +++ b/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 {}) diff --git a/tests/unit/test_zmq_compilation_identity.py b/tests/unit/test_zmq_compilation_identity.py new file mode 100644 index 0000000000..445a09764f --- /dev/null +++ b/tests/unit/test_zmq_compilation_identity.py @@ -0,0 +1,119 @@ +"""Compilation compatibility must ignore observation, not execution semantics.""" + +from dataclasses import replace +from pathlib import Path + +import pytest + +from openhcs.core.debug import DebugExecutionConfig +from openhcs.runtime.zmq_execution_signature import ( + ZMQAuxiliaryExecutionParams, + ZMQExecutionCompileControl, + ZMQExecutionConfigTransport, + ZMQExecutionIdentity, + ZMQExecutionRequestPayload, + ZMQRuntimeObservationExportScope, +) + + +def _payload(config_params=None): + return ZMQExecutionRequestPayload( + identity=ZMQExecutionIdentity( + plate_id="/plate", + execution_plate_id="/source", + selected_pipeline_path="/pipeline", + ), + pipeline_code="pipeline_config = PipelineConfig()\npipeline_steps = []\n", + config_transport=ZMQExecutionConfigTransport(config_params=config_params), + compile_control=ZMQExecutionCompileControl(), + ) + + +@pytest.mark.parametrize("scope", tuple(ZMQRuntimeObservationExportScope)) +@pytest.mark.parametrize("path", ("/evidence/first.pkl.gz", "/evidence/second.pkl.gz")) +def test_observation_options_change_request_but_not_compilation_identity(scope, path): + options = ZMQAuxiliaryExecutionParams( + axis_filter=("A01",), + debug_execution_config=DebugExecutionConfig(debug_session_id="debug-1"), + ) + compiled = _payload({"unrelated": "keep", **options.to_transport()}) + executed = _payload( + { + "unrelated": "keep", + **replace( + options, + runtime_observation_export_path=Path(path), + runtime_observation_export_scope=scope, + ).to_transport(), + } + ) + + assert compiled.request_signature != executed.request_signature + assert compiled.compilation_signature == executed.compilation_signature + assert compiled.debug_replay_signature == executed.debug_replay_signature + + +@pytest.mark.parametrize( + "config_params", + ( + {}, + {"runtime_observation_export_path": None}, + {"runtime_observation_export_scope": "values"}, + {"runtime_observation_export_path": "/evidence/values.pkl.gz"}, + ), +) +def test_observation_defaults_do_not_distinguish_compilation(config_params): + assert ( + _payload().compilation_signature + == _payload(config_params).compilation_signature + ) + + +@pytest.mark.parametrize( + "changed", + ( + {"well_filter": ["B02"]}, + {"unrelated": "changed"}, + DebugExecutionConfig(debug_session_id="debug-2").to_config_params(), + ), +) +def test_real_execution_inputs_still_distinguish_compilation(changed): + compiled = _payload({"well_filter": ["A01"], "unrelated": "keep"}) + executed = _payload({**compiled.config_params, **changed}) + assert compiled.compilation_signature != executed.compilation_signature + + +@pytest.mark.parametrize( + "field,value", + ( + ("plate_id", "/another-plate"), + ("execution_plate_id", "/another-source"), + ("selected_pipeline_path", "/another-pipeline"), + ), +) +def test_source_identities_still_distinguish_compilation(field, value): + compiled = _payload() + executed = replace(compiled, identity=replace(compiled.identity, **{field: value})) + assert compiled.compilation_signature != executed.compilation_signature + + +def test_pipeline_and_global_configuration_still_distinguish_compilation(): + compiled = _payload() + assert ( + replace(compiled, pipeline_code="changed").compilation_signature + != compiled.compilation_signature + ) + assert ( + replace( + compiled, + config_transport=replace(compiled.config_transport, config_code="changed"), + ).compilation_signature + != compiled.compilation_signature + ) + + +def test_invalid_observation_options_are_not_hidden_by_compatibility_projection(): + with pytest.raises(ValueError, match="requires an export path"): + _payload({"runtime_observation_export_scope": "outcomes"}).compilation_signature + with pytest.raises(ValueError, match="Unknown runtime observation"): + _payload({"runtime_observation_export_scope": "invalid"}).compilation_signature diff --git a/tests/unit/test_zmq_compiled_pipeline_run.py b/tests/unit/test_zmq_compiled_pipeline_run.py index dc5122d352..68765b226e 100644 --- a/tests/unit/test_zmq_compiled_pipeline_run.py +++ b/tests/unit/test_zmq_compiled_pipeline_run.py @@ -166,6 +166,7 @@ def test_auxiliary_observation_request_is_shared_by_client_and_server(): assert execution.config_params == { "unrelated": "kept", "runtime_observation_export_path": str(path), + "runtime_observation_export_scope": "values", } assert ( ZMQAuxiliaryExecutionParams.from_transport(