diff --git a/external/ObjectState b/external/ObjectState index 9fb5b7eeea..27dbe365d7 160000 --- a/external/ObjectState +++ b/external/ObjectState @@ -1 +1 @@ -Subproject commit 9fb5b7eeea96f3b7bb0579de30718bbcc0018823 +Subproject commit 27dbe365d7b858691898dac9b50c96f0a211979b diff --git a/openhcs/core/pipeline/compilation_session.py b/openhcs/core/pipeline/compilation_session.py index c86ff7e2ca..0f7f4d4e80 100644 --- a/openhcs/core/pipeline/compilation_session.py +++ b/openhcs/core/pipeline/compilation_session.py @@ -24,8 +24,6 @@ ) if TYPE_CHECKING: - from objectstate import ObjectState - from openhcs.core.config import GlobalPipelineConfig from openhcs.core.orchestrator.orchestrator import PipelineOrchestrator @@ -158,23 +156,39 @@ def resolve_declared_dataclass_paths( @dataclass(frozen=True, slots=True) class ResolvedPipelineDefinition: - """Capture enabled declarations once after ObjectState/callable resolution. + """Capture enabled saved declarations and their scope/provenance facts. Axis sessions derive metadata kwargs and provider contracts from this view; authored FunctionSteps keep their public function-pattern syntax. """ steps: Sequence[AbstractStep] - step_state_map: Mapping[int, "ObjectState"] + step_scope_ids: Mapping[int, str] + step_provenance: Mapping[int, Mapping[str, tuple[str | None, type | None]]] def __post_init__(self) -> None: - missing_states = [ - index for index in range(len(self.steps)) if index not in self.step_state_map + missing_scopes = [ + index + for index in range(len(self.steps)) + if index not in self.step_scope_ids or index not in self.step_provenance ] - if missing_states: + if missing_scopes: raise ValueError( - f"Resolved pipeline missing ObjectState entries for steps {missing_states}." + f"Resolved pipeline missing scope/provenance facts for steps {missing_scopes}." ) + object.__setattr__( + self, + "step_provenance", + { + index: dict(self.step_provenance[index]) + for index in range(len(self.steps)) + }, + ) + object.__setattr__( + self, + "step_scope_ids", + dict(self.step_scope_ids), + ) object.__setattr__( self, "steps", @@ -253,6 +267,7 @@ def from_context( def __post_init__(self) -> None: if self.plate_scope is None and self.context.plate_path is not None: self.plate_scope = CompilationPlateScope.from_context(self.context) + @property def axis_id(self) -> str: return self.context.axis_id diff --git a/openhcs/core/pipeline/compiler.py b/openhcs/core/pipeline/compiler.py index e01e1014a4..3ad38915fd 100644 --- a/openhcs/core/pipeline/compiler.py +++ b/openhcs/core/pipeline/compiler.py @@ -1,43 +1,7 @@ -""" -Pipeline module for OpenHCS. - -This module provides core pipeline compilation components for OpenHCS. -The PipelineCompiler is responsible for preparing step_plans within a ProcessingContext. - -CONFIGURATION ACCESS PATTERN: -============================ -The compiler resolves ObjectState once and replaces the submitted pipeline with -the resolved steps retained by ResolvedPipelineDefinition: - -CORRECT: - # Steps are registered in ObjectState with parent hierarchy: step → orchestrator → global - step_state = ObjectState(object_instance=step, scope_id=scope_id, parent_state=orch_state) - ObjectStateRegistry.register(step_state) - - resolved_step = step_state.to_saved_resolved_object() - var_comps = resolved_step.processing_config.variable_components - -✅ CORRECT (LIVE VALUES FOR UI): - # For UI: use get_resolved_value() to get current values with unsaved edits - current_value = step_state.get_resolved_value(field_path) - -REMOVED: - with config_context(orchestrator.pipeline_config): # REMOVED - resolved_step = resolve_lazy_configurations_for_serialization(step) # REMOVED - - # Using .parameters.get() doesn't get inheritance - current_value = step_state.parameters.get(field_path) # WRONG - no inheritance - - Compiler consumers must read semantics from the resolved step, not rebuild a - projection from ObjectState. +"""Compile saved pipeline declarations without creating editor lifecycles. -WHY: -- ObjectState.to_saved_resolved_object() provides the resolved compiler step -- ResolvedPipelineDefinition owns ordered resolved steps and their registered ObjectStates -- get_resolved_value() provides live state with unsaved edits (for UI) -- parameters.get() returns raw local value only, NO inheritance -- No cross-step pollution - each step only sees its own config hierarchy -- isinstance checks are the only type checking pattern (no hasattr) +ObjectState owns saved inheritance and provenance resolution. The resolved +pipeline retains those admission facts; live UI states remain in their registry. """ from __future__ import annotations @@ -158,7 +122,7 @@ def _compiler_step_scope_id( step: "AbstractStep", step_index: int, ) -> str: - """Build a compiler ObjectState scope id that preserves stable step tokens.""" + """Build a compiler scope id that preserves stable step tokens.""" return f"{compilation_scope}::{_step_scope_token(step, step_index)}" @@ -166,7 +130,7 @@ def _compiler_pipeline_scope_id( plate_path_str: str, pipeline_definition: Sequence["AbstractStep"], ) -> str: - """Build the compiler-owned ObjectState root for one submitted pipeline.""" + """Build the compiler declaration scope for one submitted pipeline.""" return f"{plate_path_str}::pipeline::submission_{id(pipeline_definition):x}" @@ -244,7 +208,7 @@ def initialize_step_plans_for_context( Args: context: ProcessingContext to initialize step plans for - pipeline: Ordered resolved steps and their registered ObjectStates + pipeline: Ordered resolved steps with saved provenance and scope identities orchestrator: Orchestrator instance for well filter resolution metadata_writer: If True, this well is responsible for creating OpenHCS metadata files plate_path: Path to plate root for zarr conversion detection @@ -290,41 +254,6 @@ def _assert_context_mutable_for_planning(context: ProcessingContext) -> None: if context.step_plans is None: context.step_plans = {} - @staticmethod - def _register_object_state( - object_instance, - scope_id: str, - parent_state: Optional["ObjectState"], - ) -> "ObjectState": - """Create and register an ObjectState with the compiler's step policy.""" - state = ObjectState( - object_instance=object_instance, - scope_id=scope_id, - parent_state=parent_state, - ) - ObjectStateRegistry.register(state, _skip_snapshot=True) - return state - - @staticmethod - def _get_or_register_object_state( - scope_id: str, - object_instance, - parent_state: Optional["ObjectState"], - *, - force_fresh: bool = False, - ) -> "ObjectState": - """Return an existing ObjectState unless a fresh compiler state is required.""" - state = None - if not force_fresh: - state = ObjectStateRegistry.get_by_scope(scope_id) - if state is not None: - return state - return PipelineCompiler._register_object_state( - object_instance, - scope_id, - parent_state, - ) - @staticmethod def _missing_plan_fields( plan: CompiledStepPlan, @@ -424,14 +353,14 @@ def _supplement_step_plans(session: CompilationSession) -> None: continue current_plan = session.plans[step_index] - current_plan.step_scope_id = session.pipeline.step_state_map[ - step_index - ].scope_id + current_plan.step_scope_id = session.pipeline.step_scope_ids[step_index] current_plan.step_name = step.name current_plan.step_type = type(step).__name__ current_plan.axis_id = session.axis_id current_plan.create_openhcs_metadata = session.metadata_writer - current_plan.variable_components = step.processing_config.variable_components + current_plan.variable_components = ( + step.processing_config.variable_components + ) current_plan.group_by = PathPlannerExecutionGroups.normalized_group_by( step, ) @@ -911,14 +840,13 @@ def resolve_lazy_dataclasses(session: CompilationSession) -> None: """ Resolve all lazy dataclass instances in step plans to their base configurations. - This method uses ObjectState for resolution. - All configs are already resolved via ObjectState.to_object() during compilation. + All configs are already admitted by saved declaration resolution. This method now just ensures step plans reference the resolved configs. Args: session: Axis-scoped compiler session. """ - # Configs are already resolved via ObjectState.to_object() in initialize_step_plans_for_context + # Configs are already resolved in the saved pipeline declaration. # No additional resolution needed - step plans already contain resolved configs logger.debug( f"Step plans already resolved via ObjectState for {len(session.pipeline.steps)} steps" @@ -1023,135 +951,50 @@ def _axis_values_to_process( ) @staticmethod - def _register_and_resolve_pipeline_once( + def _resolve_pipeline_once( orchestrator, pipeline_definition: List[AbstractStep], - *, - is_zmq_execution: bool, - ) -> tuple[str, "ObjectState", ResolvedPipelineDefinition]: - # Compile from the submitted pipeline definition, not from any stale UI - # ObjectState that may point at post-compile stripped step shells. - force_fresh = True - global_config_state = PipelineCompiler._compile_global_config_state( - force_fresh=force_fresh - ) - plate_path_str = str(orchestrator.plate_path) - compiler_scope_id = _compiler_pipeline_scope_id( - plate_path_str, - pipeline_definition, - ) - pipeline_config_state = ObjectStateRegistry.get_by_scope(compiler_scope_id) - if orchestrator.pipeline_config is not None: - pipeline_config_state = PipelineCompiler._get_or_register_object_state( - compiler_scope_id, - orchestrator.pipeline_config, - global_config_state, - force_fresh=force_fresh, - ) - if pipeline_config_state is None: - raise RuntimeError( - "Missing ObjectState for plate; cannot resolve pipeline config." - ) - step_state_map = PipelineCompiler._register_pipeline_step_states( - pipeline_definition, - compiler_scope_id, - pipeline_config_state, - force_fresh=force_fresh, + ) -> ResolvedPipelineDefinition: + """Admit submitted saved declarations without registering editor states.""" + if orchestrator.pipeline_config is None: + raise RuntimeError("Missing pipeline config; cannot resolve pipeline.") + compiler_scope = _compiler_pipeline_scope_id( + str(orchestrator.plate_path), pipeline_definition ) - PipelineCompiler._replace_pipeline_with_resolved_steps( - pipeline_definition, - step_state_map, + ancestors = ( + *ObjectStateRegistry.get_ancestor_objects_with_scopes( + compiler_scope, use_saved=True + ), + (compiler_scope, orchestrator.pipeline_config), ) + resolved_steps = [] + step_scopes = {} + step_provenance = {} + for index, step in enumerate(pipeline_definition): + scope = _compiler_step_scope_id(compiler_scope, step, index) + resolved_step, provenance = ObjectState.resolve_saved_object( + step, scope_id=scope, ancestor_objects_with_scopes=ancestors + ) + resolved_steps.append(resolved_step) + step_scopes[index] = scope + step_provenance[index] = provenance + pipeline_definition[:] = resolved_steps _refresh_function_objects_in_steps(pipeline_definition) - logger.debug( - "Refreshed function objects in %s steps (converted to FunctionReference)", - len(pipeline_definition), - ) - - pipeline = PipelineCompiler._filter_enabled_steps( - pipeline_definition, - step_state_map, - ) - return ( - compiler_scope_id, - pipeline_config_state, - pipeline, - ) - - @staticmethod - def _compile_global_config_state(*, force_fresh: bool) -> "ObjectState" | None: - from objectstate import get_current_global_config - from openhcs.core.config import GlobalPipelineConfig - - global_config_state = ObjectStateRegistry.get_by_scope("") - if force_fresh or global_config_state is None: - global_config = get_current_global_config( - GlobalPipelineConfig, - use_live=False, - ) - if global_config: - global_config_state = PipelineCompiler._register_object_state( - global_config, - "", - None, - ) - logger.debug("Registered global config at scope ''") - return global_config_state - - @staticmethod - def _register_pipeline_step_states( - pipeline_definition: Sequence[AbstractStep], - compiler_scope_id: str, - pipeline_config_state: "ObjectState", - *, - force_fresh: bool, - ) -> Dict[int, "ObjectState"]: - step_state_map: Dict[int, "ObjectState"] = {} - for step_index, step in enumerate(pipeline_definition): - step_scope_id = _compiler_step_scope_id( - compiler_scope_id, - step, - step_index, - ) - step_state_map[step_index] = PipelineCompiler._get_or_register_object_state( - step_scope_id, - step, - pipeline_config_state, - force_fresh=force_fresh, - ) - return step_state_map - - @staticmethod - def _replace_pipeline_with_resolved_steps( - pipeline_definition: List[AbstractStep], - step_state_map: Mapping[int, "ObjectState"], - ) -> None: - pipeline_definition.clear() - pipeline_definition.extend( - step_state.to_saved_resolved_object() - for step_state in step_state_map.values() - ) - logger.debug( - "Resolved %s steps once per pipeline (replaced original list in-place)", - len(pipeline_definition), - ) - - @staticmethod - def _filter_enabled_steps( - pipeline_definition: List[AbstractStep], - step_state_map: Mapping[int, "ObjectState"], - ) -> ResolvedPipelineDefinition: - enabled_pairs = [ - (step, step_state_map[index]) - for index, step in enumerate(pipeline_definition) - if step.enabled + enabled_indices = [ + index for index, step in enumerate(pipeline_definition) if step.enabled + ] + pipeline_definition[:] = [ + pipeline_definition[index] for index in enabled_indices ] - pipeline_definition.clear() - pipeline_definition.extend(step for step, _state in enabled_pairs) return ResolvedPipelineDefinition( steps=pipeline_definition, - step_state_map={ - index: state for index, (_step, state) in enumerate(enabled_pairs) + step_scope_ids={ + index: step_scopes[original_index] + for index, original_index in enumerate(enabled_indices) + }, + step_provenance={ + index: step_provenance[original_index] + for index, original_index in enumerate(enabled_indices) }, ) @@ -1650,14 +1493,9 @@ def _finalize_compilation( orchestrator, pipeline_definition: List[AbstractStep], compiled_contexts: Mapping[str, ProcessingContext], - compiler_scope_id: str, effective_config: GlobalPipelineConfig, ) -> None: PipelineCompiler._log_path_planning_summary(compiled_contexts) - PipelineCompiler._cleanup_compilation_object_states( - orchestrator, - compiler_scope_id, - ) logger.info("Stripping attributes from pipeline definition steps.") StepAttributeStripper.strip_step_attributes(pipeline_definition, {}) orchestrator._state = OrchestratorState.COMPILED @@ -1690,25 +1528,6 @@ def _log_path_planning_summary( plan.materialized_output.output_dir, ) - @staticmethod - def _cleanup_compilation_object_states( - orchestrator, - compiler_scope_id: str | None = None, - ) -> None: - orch_scope_id = ( - compiler_scope_id - if compiler_scope_id is not None - else f"{orchestrator.plate_path}::orchestrator" - ) - ObjectStateRegistry.unregister_scope_and_descendants( - orch_scope_id, - _skip_snapshot=True, - ) - logger.debug( - "Cleaned up compilation ObjectStates for scope: %s", - orch_scope_id, - ) - @staticmethod def _calculate_worker_assignments( wells: list[str], @@ -1752,15 +1571,13 @@ def compile_pipelines( axis_filter: Optional list of axis values to process. If None, processes all found axis values. enable_visualizer_override: If True, all steps in all compiled contexts will have their 'visualize' flag set to True. - is_zmq_execution: If True, compiler-created ObjectStates will be unregistered - after resolution to free RAM (for ZMQ server mode). + is_zmq_execution: Use the execution server's worker runtime policy. Returns: The compiler-owned bundle containing compiled contexts, worker assignments, runtime policy, and the stateless pipeline definition. """ PipelineCompiler._validate_compile_request(orchestrator, pipeline_definition) - compiler_scope_id: str | None = None try: effective_config = ( orchestrator.get_effective_config() @@ -1789,21 +1606,13 @@ def compile_pipelines( f"Starting compilation for axis values: {', '.join(axis_values_to_process)}" ) - compiler_scope_id, _pipeline_config_state, pipeline_inputs = ( - PipelineCompiler._register_and_resolve_pipeline_once( - orchestrator, - pipeline_definition, - is_zmq_execution=is_zmq_execution, - ) + pipeline_inputs = PipelineCompiler._resolve_pipeline_once( + orchestrator, pipeline_definition ) if not pipeline_definition: logger.warning( "All steps were disabled. Pipeline is empty after filtering." ) - PipelineCompiler._cleanup_compilation_object_states( - orchestrator, - compiler_scope_id, - ) return CompiledExecutionBundle.from_runtime_contexts( pipeline_definition=pipeline_definition, runtime_contexts={}, @@ -1875,7 +1684,6 @@ def compile_pipelines( orchestrator, pipeline_definition, compiled_contexts, - compiler_scope_id, effective_config, ) execution_bundle = CompiledExecutionBundle.from_runtime_contexts( @@ -1890,11 +1698,6 @@ def compile_pipelines( ) return execution_bundle except Exception as e: - if compiler_scope_id is not None: - PipelineCompiler._cleanup_compilation_object_states( - orchestrator, - compiler_scope_id, - ) orchestrator._state = OrchestratorState.COMPILE_FAILED logger.error(f"Failed to compile pipelines: {e}") raise @@ -1938,7 +1741,7 @@ def _resolve_step_axis_filters( It processes ALL WellFilterConfig instances (materialization, streaming, etc.) uniformly. Args: - pipeline: Resolved steps and ObjectState owners retaining inherited-field provenance + pipeline: Resolved steps and their same-epoch saved-field provenance context: Processing context for the current axis value orchestrator: Orchestrator instance with access to available axis values """ @@ -1952,11 +1755,12 @@ def _resolve_step_axis_filters( for step_index, step in enumerate(pipeline.steps): step_filters: dict[type[WellFilterConfig], StepAxisFilterResolution] = {} - step_state = pipeline.step_state_map[step_index] for field_name, config in vars(step).items(): if not isinstance(config, WellFilterConfig) or config.well_filter is None: continue - provenance = step_state.get_provenance(f"{field_name}.well_filter") + provenance = pipeline.step_provenance[step_index].get( + f"{field_name}.well_filter" + ) source_type = provenance[1] if provenance is not None else None if not config.accepts_well_filter_provenance(source_type): logger.debug( diff --git a/openhcs/core/pipeline/path_planner.py b/openhcs/core/pipeline/path_planner.py index 85bd258ae3..4751abeae4 100644 --- a/openhcs/core/pipeline/path_planner.py +++ b/openhcs/core/pipeline/path_planner.py @@ -2337,7 +2337,7 @@ def prime_future_artifact_inputs(self) -> None: def plan_step(self, step: AbstractStep, step_index: int) -> None: """Plan one step's directories, artifacts, and executable pattern.""" self.planner.plans[step_index].step_scope_id = ( - self.planner.session.pipeline.step_state_map[step_index].scope_id + self.planner.session.pipeline.step_scope_ids[step_index] ) self.planner.artifact_context = self.planner.artifacts.artifact_declaration_context( step, step_index diff --git a/pyproject.toml b/pyproject.toml index 608208040e..5cb8bbf773 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -101,7 +101,7 @@ dependencies = [ # External dependencies (PyPI versions for production builds) "zmqruntime>=0.3.0,<0.4", "pycodify>=0.1.3,<0.2", - "objectstate>=1.1.9,<2", + "objectstate>=1.2.0,<2", "python-introspect>=0.1.16,<0.2", "metaclass-registry>=0.2.2,<0.3", "arraybridge>=0.3.6,<0.4", diff --git a/tests/unit/test_artifact_output_ownership.py b/tests/unit/test_artifact_output_ownership.py index 04a0491855..7cef071f0f 100644 --- a/tests/unit/test_artifact_output_ownership.py +++ b/tests/unit/test_artifact_output_ownership.py @@ -402,7 +402,8 @@ def test_real_cellprofiler_declaration_compiles_without_a_table_wide_subject(): global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(step,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) provider = CellProfilerInvocationContractProviderFactory.provider_for_session( diff --git a/tests/unit/test_cellprofiler_artifact_declarations.py b/tests/unit/test_cellprofiler_artifact_declarations.py index 8331ca4488..d8df332dd4 100644 --- a/tests/unit/test_cellprofiler_artifact_declarations.py +++ b/tests/unit/test_cellprofiler_artifact_declarations.py @@ -190,7 +190,11 @@ def _compiler_contracts( ), global_config=global_config, pipeline=ResolvedPipelineDefinition( - steps=resolved_steps, step_state_map=step_states + steps=resolved_steps, + step_scope_ids={ + index: state.scope_id for index, state in step_states.items() + }, + step_provenance={index: {} for index, state in step_states.items()}, ), ) provider = PipelineInvocationContractProviderAuthority.provider_for_session( diff --git a/tests/unit/test_cellprofiler_morphology_public_contracts.py b/tests/unit/test_cellprofiler_morphology_public_contracts.py index a0aef06dad..497b9629bc 100644 --- a/tests/unit/test_cellprofiler_morphology_public_contracts.py +++ b/tests/unit/test_cellprofiler_morphology_public_contracts.py @@ -208,7 +208,10 @@ def fixture_producer(image): global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=snapshots, - step_state_map={index: object() for index in range(len(steps))}, + step_scope_ids={ + index: f"plate::step_{index}" for index in range(len(steps)) + }, + step_provenance={index: {} for index in range(len(steps))}, ), ) provider = CellProfilerInvocationContractProviderFactory.provider_for_session( diff --git a/tests/unit/test_cellprofiler_neighbors_contract.py b/tests/unit/test_cellprofiler_neighbors_contract.py index 20c2b77933..0edaecc2be 100644 --- a/tests/unit/test_cellprofiler_neighbors_contract.py +++ b/tests/unit/test_cellprofiler_neighbors_contract.py @@ -120,7 +120,8 @@ def test_relationship_module_number_is_derived_after_public_transport() -> None: global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(snapshot,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) @@ -158,7 +159,8 @@ def _compiled_neighbor_invocation(): global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(snapshot,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) provider = CellProfilerInvocationContractProviderFactory.provider_for_session( @@ -380,7 +382,10 @@ def fixture_producer(image): global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=snapshots, - step_state_map={index: object() for index in range(len(steps))}, + step_scope_ids={ + index: f"plate::step_{index}" for index in range(len(steps)) + }, + step_provenance={index: {} for index in range(len(steps))}, ), ) @@ -522,7 +527,10 @@ def fixture_producer(image): global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=snapshots, - step_state_map={index: object() for index in range(len(steps))}, + step_scope_ids={ + index: f"plate::step_{index}" for index in range(len(steps)) + }, + step_provenance={index: {} for index in range(len(steps))}, ), ) provider = CellProfilerInvocationContractProviderFactory.provider_for_session( diff --git a/tests/unit/test_cellprofiler_runtime_callable_introspection.py b/tests/unit/test_cellprofiler_runtime_callable_introspection.py index d0078ba9e5..3fc9695c11 100644 --- a/tests/unit/test_cellprofiler_runtime_callable_introspection.py +++ b/tests/unit/test_cellprofiler_runtime_callable_introspection.py @@ -71,7 +71,11 @@ def _compilation_session_for_steps( orchestrator=SimpleNamespace(pipeline_config=PipelineConfig()), global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( - steps=steps, step_state_map={index: object() for index in range(len(steps))} + steps=steps, + step_scope_ids={ + index: f"plate::step_{index}" for index in range(len(steps)) + }, + step_provenance={index: {} for index in range(len(steps))}, ), ) diff --git a/tests/unit/test_compilation_session.py b/tests/unit/test_compilation_session.py index 27a7d9f559..e9871a8cf1 100644 --- a/tests/unit/test_compilation_session.py +++ b/tests/unit/test_compilation_session.py @@ -15,6 +15,7 @@ from openhcs.core.config import ( GlobalPipelineConfig, LazyNapariStreamingConfig, + LazyProcessingConfig, LazySourceBindingsConfig, LazyStepSourceBindingsConfig, PipelineConfig, @@ -277,7 +278,9 @@ def test_compilation_session_shares_resolved_pipeline_and_owns_axis_plans(): orchestrator=_orchestrator(), global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( - steps=(_resolved_step(step),), step_state_map={0: step_state} + steps=(_resolved_step(step),), + step_scope_ids={0: step_state.scope_id}, + step_provenance={0: {}}, ), ) @@ -287,7 +290,7 @@ def test_compilation_session_shares_resolved_pipeline_and_owns_axis_plans(): captured = session.pipeline.steps[0].func assert normalize_function_pattern(captured) is captured assert next(captured.iter_items()).func is _identity - assert session.pipeline.step_state_map[0] is step_state + assert session.pipeline.step_scope_ids[0] == step_state.scope_id assert session.pipeline.steps[0].name == "step" assert session.plan(0).step_name == "step" @@ -315,7 +318,9 @@ def count_contracts(cls, func): return from_callable(cls, func) monkeypatch.setattr(CallableContract, "from_callable", classmethod(count_contracts)) - pipeline = PipelineCompiler._filter_enabled_steps(definition, {0: state}) + pipeline = ResolvedPipelineDefinition( + definition, {0: state.scope_id}, step_provenance={0: {}} + ) captured = pipeline.steps[0].func (item,) = tuple(captured.iter_items()) assert definition == [authored] and definition[0].func[1][0][1] is kwargs @@ -346,7 +351,9 @@ def count_contracts(cls, func): assert next(provider_compiled.iter_invocations()).contract is replacement kwargs["sigma"] = 7 assert item.kwargs_dict["sigma"] == 2 - recaptured = ResolvedPipelineDefinition(definition, {0: state}) + recaptured = ResolvedPipelineDefinition( + definition, {0: state.scope_id}, step_provenance={0: {}} + ) assert next(recaptured.steps[0].func.iter_items()).kwargs_dict["sigma"] == 7 assert calls == [needs_grid, needs_grid] @@ -361,7 +368,8 @@ def test_compiler_keeps_variable_components_as_stack_source(): steps=( _resolved_step(step, variable_components=(VariableComponents.CHANNEL,)), ), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) @@ -418,7 +426,9 @@ def test_path_planner_source_binding_plan_comes_from_objectstate_snapshot(): orchestrator=_orchestrator(pipeline_state.to_object()), global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( - steps=(snapshot,), step_state_map={0: step_state} + steps=(snapshot,), + step_scope_ids={0: step_state.scope_id}, + step_provenance={0: {}}, ), ) @@ -477,7 +487,9 @@ def test_compiler_streaming_config_snapshot_preserves_inherited_port(): orchestrator=_orchestrator(pipeline_state.to_object()), global_config=global_config, pipeline=ResolvedPipelineDefinition( - steps=(snapshot,), step_state_map={0: step_state} + steps=(snapshot,), + step_scope_ids={0: step_state.scope_id}, + step_provenance={0: {}}, ), ) @@ -505,7 +517,8 @@ def test_compiler_disabled_source_bindings_stay_inert_without_contract_requireme global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(snapshot,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) @@ -528,7 +541,8 @@ def test_path_planner_activates_declared_source_binding_for_pipeline_start(): global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(snapshot,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) @@ -566,7 +580,8 @@ def test_plate_export_contract_construction_projects_inputs_to_runtime_batch(): global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(snapshot,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) dependency_before_provider = session.plan(0).main_input_dependency @@ -607,7 +622,8 @@ def test_path_planner_preserves_pipeline_start_bindings_for_implicit_main_flow() global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(snapshot,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) execution_bindings, (source_binding_plan, source_universe_plan) = ( @@ -635,7 +651,8 @@ def test_path_planner_step_output_projects_only_exact_source_artifacts() -> None global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(snapshot,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) @@ -684,7 +701,8 @@ def test_path_planner_preserves_metaxpress_primary_source_order() -> None: global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(snapshot,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) contract = CallableContract.from_callable(neurite_outgrowth_metaxpress) @@ -803,7 +821,8 @@ def test_path_planner_freezes_only_contract_selected_source_bindings(): global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(snapshot,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) execution_bindings, (source_binding_plan, _source_universe_plan) = ( @@ -838,33 +857,75 @@ def test_compiler_pipeline_scope_prevents_cross_pipeline_source_binding_inherita ) first_pipeline = [FunctionStep(func=_identity, name="first")] second_pipeline = [FunctionStep(func=_identity, name="second")] - registered_scopes: list[tuple[SimpleNamespace, str]] = [] + first_resolved = PipelineCompiler._resolve_pipeline_once( + first_orchestrator, first_pipeline + ) + second_resolved = PipelineCompiler._resolve_pipeline_once( + second_orchestrator, second_pipeline + ) + first_scope = first_resolved.step_scope_ids[0] + second_scope = second_resolved.step_scope_ids[0] + + assert first_scope != second_scope + assert first_resolved.steps[0].source_bindings.bindings == (binding,) + assert second_resolved.steps[0].source_bindings.is_empty + +def test_headless_resolution_preserves_saved_ui_ancestors_and_registrations(tmp_path): + ObjectStateRegistry.clear() + ensure_global_config_context(GlobalPipelineConfig, GlobalPipelineConfig()) + plate_path = tmp_path / "plate" + saved_binding = NamedSourceBinding(alias="DNA") + live_binding = NamedSourceBinding(alias="Memb") + ui_state = ObjectState( + PipelineConfig( + source_bindings_config=LazySourceBindingsConfig(bindings=(saved_binding,)), + processing_config=LazyProcessingConfig( + variable_components=[VariableComponents.SITE], + ), + ), + scope_id=str(plate_path), + ) + ObjectStateRegistry.register(ui_state, _skip_snapshot=True) + ui_state.update_parameter("source_bindings_config.bindings", (live_binding,)) + original_states = ObjectStateRegistry.get_all() + original_parameters = dict(ui_state.parameters) + observed_registrations = [] + + def record_registration(scope, state): + observed_registrations.append((scope, state)) + + subscription = ObjectStateRegistry.add_register_callback(record_registration) + authored = [ + FunctionStep(func=_identity, name="disabled", enabled=False), + FunctionStep(func=_identity, name="enabled"), + ] + enabled_token = authored[1]._scope_token or "step_1" try: - first_scope, _first_config_state, first_resolved = ( - PipelineCompiler._register_and_resolve_pipeline_once( - first_orchestrator, - first_pipeline, - is_zmq_execution=False, - ) + resolved = PipelineCompiler._resolve_pipeline_once( + SimpleNamespace(plate_path=plate_path, pipeline_config=PipelineConfig()), + authored, ) - registered_scopes.append((first_orchestrator, first_scope)) - - second_scope, _second_config_state, second_resolved = ( - PipelineCompiler._register_and_resolve_pipeline_once( - second_orchestrator, - second_pipeline, - is_zmq_execution=False, - ) + assert len(resolved.steps) == 1 + assert resolved.steps[0].source_bindings.bindings == (saved_binding,) + assert resolved.step_scope_ids[0].endswith(f"::{enabled_token}") + source_scope, _source_type = resolved.step_provenance[0][ + "source_bindings.bindings" + ] + assert source_scope == str(plate_path) + assert ObjectStateRegistry.get_all() == original_states + assert ui_state.parameters == original_parameters + assert observed_registrations == [] + assert ui_state.get_resolved_value("source_bindings_config.bindings") == ( + live_binding, ) - registered_scopes.append((second_orchestrator, second_scope)) + saved_components = object.__getattribute__( + ui_state.object_instance.processing_config, "variable_components" + ) + saved_components.append(VariableComponents.Z_INDEX) + assert resolved.steps[0].processing_config.variable_components == [ + VariableComponents.SITE + ] finally: - for orchestrator, scope_id in registered_scopes: - PipelineCompiler._cleanup_compilation_object_states( - orchestrator, - scope_id, - ) - - assert first_scope != second_scope - assert first_resolved.steps[0].source_bindings.bindings == (binding,) - assert second_resolved.steps[0].source_bindings.is_empty + subscription.release() + ObjectStateRegistry.clear() diff --git a/tests/unit/test_invocation_contract_provider.py b/tests/unit/test_invocation_contract_provider.py index 6172ab7bbb..90cf309d07 100644 --- a/tests/unit/test_invocation_contract_provider.py +++ b/tests/unit/test_invocation_contract_provider.py @@ -698,7 +698,8 @@ def test_cellprofiler_provider_reconstructs_exact_contract_from_public_step() -> global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(snapshot,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) callable_state = vars(color_to_gray).copy() @@ -774,10 +775,8 @@ def test_cellprofiler_provider_advances_native_artifacts_through_generic_graph() global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=steps, - step_state_map={ - 0: SimpleNamespace(scope_id="plate::functionstep_0"), - 1: SimpleNamespace(scope_id="plate::functionstep_1"), - }, + step_scope_ids={0: "plate::functionstep_0", 1: "plate::functionstep_1"}, + step_provenance={0: {}, 1: {}}, ), ) @@ -833,7 +832,8 @@ def test_cellprofiler_provider_leaves_native_same_name_callable_unclaimed() -> N global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(snapshot,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) @@ -1161,7 +1161,8 @@ def test_cellprofiler_provider_rejects_under_specified_one_image_align() -> None global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=(snapshot,), - step_state_map={0: SimpleNamespace(scope_id="plate::functionstep_0")}, + step_scope_ids={0: "plate::functionstep_0"}, + step_provenance={0: {}}, ), ) @@ -1309,10 +1310,8 @@ def test_calculate_math_provider_keeps_object_identity_and_output_name_public() global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=steps, - step_state_map={ - 0: SimpleNamespace(scope_id="plate::functionstep_0"), - 1: SimpleNamespace(scope_id="plate::functionstep_1"), - }, + step_scope_ids={0: "plate::functionstep_0", 1: "plate::functionstep_1"}, + step_provenance={0: {}, 1: {}}, ), ) @@ -1404,7 +1403,10 @@ def session_for( global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( steps=steps, - step_state_map={index: object() for index in range(len(steps))}, + step_scope_ids={ + index: f"plate::step_{index}" for index in range(len(steps)) + }, + step_provenance={index: {} for index in range(len(steps))}, ), ), steps, diff --git a/tests/unit/test_materialization_filter_inheritance.py b/tests/unit/test_materialization_filter_inheritance.py index 664dd1da25..da5e84e572 100644 --- a/tests/unit/test_materialization_filter_inheritance.py +++ b/tests/unit/test_materialization_filter_inheritance.py @@ -1,6 +1,7 @@ from pathlib import Path from types import SimpleNamespace +from objectstate import ObjectState from objectstate.lazy_factory import ensure_global_config_context from objectstate.object_state_registry import ObjectStateRegistry from openhcs.constants.constants import Backend @@ -58,108 +59,88 @@ def _resolved_checkpoint( ), ) ] - scope_id, pipeline_config_state, resolved = ( - PipelineCompiler._register_and_resolve_pipeline_once( - orchestrator, - pipeline, - is_zmq_execution=False, - ) - ) + resolved = PipelineCompiler._resolve_pipeline_once(orchestrator, pipeline) + resolved_config, _provenance = ObjectState.resolve_saved_object(pipeline_config) filters = PipelineCompiler._resolve_global_step_axis_filters( - orchestrator, - resolved, - pipeline_config_state.to_saved_resolved_object(), + orchestrator, resolved, resolved_config ) - return orchestrator, scope_id, pipeline_config_state, resolved, filters + return orchestrator, resolved_config, resolved, filters def test_path_zero_keeps_inherited_step_checkpoint_independent(tmp_path) -> None: ObjectStateRegistry.clear() ensure_global_config_context(GlobalPipelineConfig, GlobalPipelineConfig()) - orchestrator, scope_id, config_state, resolved, filters = _resolved_checkpoint( + orchestrator, effective_config, resolved, filters = _resolved_checkpoint( tmp_path / "plate" ) - try: - snapshot = resolved.steps[0] - checkpoint_config = snapshot.step_materialization_config - assert checkpoint_config.well_filter == 0 - assert filters[0].allows(checkpoint_config, "A01") - assert not filters[0].allows(checkpoint_config, "B02") - - checkpoint_dir = tmp_path / "checkpoint" - materialization_stage = PathPlannerMaterializationStage( - SimpleNamespace( - ctx=SimpleNamespace(axis_id="A01", step_axis_filters=filters), - paths=SimpleNamespace( - build_output_path=lambda _config: checkpoint_dir, - ), - ) - ) - assert ( - materialization_stage.materialized_output_dir_for_step(snapshot, 0) - == checkpoint_dir - ) + snapshot = resolved.steps[0] + checkpoint_config = snapshot.step_materialization_config + assert checkpoint_config.well_filter == 0 + assert filters[0].allows(checkpoint_config, "A01") + assert not filters[0].allows(checkpoint_config, "B02") - plan = CompiledStepPlan( - step_index=0, - step_name="checkpoint", - step_type="FunctionStep", - axis_id="A01", - materialized_output=MaterializedOutputPlan( - output_dir=checkpoint_dir, - backend=Backend.MEMORY.value, - plate_root=str(tmp_path), - sub_dir="checkpoints", - analysis_results_dir=str(tmp_path / "checkpoint_results"), + checkpoint_dir = tmp_path / "checkpoint" + materialization_stage = PathPlannerMaterializationStage( + SimpleNamespace( + ctx=SimpleNamespace(axis_id="A01", step_axis_filters=filters), + paths=SimpleNamespace( + build_output_path=lambda _config: checkpoint_dir, ), ) - context = SimpleNamespace(axis_id="A01", step_plans=[plan]) - effective_config = config_state.to_object(update_delegate=False) - MaterializationFlagPlanner.prepare_pipeline_flags( - context, - [ - SimpleNamespace( - processing_config=SimpleNamespace( - input_source=InputSource.PREVIOUS_STEP, - ) + ) + assert ( + materialization_stage.materialized_output_dir_for_step(snapshot, 0) + == checkpoint_dir + ) + + plan = CompiledStepPlan( + step_index=0, + step_name="checkpoint", + step_type="FunctionStep", + axis_id="A01", + materialized_output=MaterializedOutputPlan( + output_dir=checkpoint_dir, + backend=Backend.MEMORY.value, + plate_root=str(tmp_path), + sub_dir="checkpoints", + analysis_results_dir=str(tmp_path / "checkpoint_results"), + ), + ) + context = SimpleNamespace(axis_id="A01", step_plans=[plan]) + MaterializationFlagPlanner.prepare_pipeline_flags( + context, + [ + SimpleNamespace( + processing_config=SimpleNamespace( + input_source=InputSource.PREVIOUS_STEP, ) - ], - plate_path=orchestrator.plate_path, - pipeline_config=SimpleNamespace( - vfs_config=VFSConfig( - read_backend=Backend.DISK, - materialization_backend=MaterializationBackend.DISK, - ), - path_planning_config=effective_config.path_planning_config, + ) + ], + plate_path=orchestrator.plate_path, + pipeline_config=SimpleNamespace( + vfs_config=VFSConfig( + read_backend=Backend.DISK, + materialization_backend=MaterializationBackend.DISK, ), - available_axis_values=("A01", "B02"), - ) + path_planning_config=effective_config.path_planning_config, + ), + available_axis_values=("A01", "B02"), + ) - assert plan.write_backend == Backend.MEMORY.value - assert plan.materialized_output.backend == Backend.DISK.value - finally: - PipelineCompiler._cleanup_compilation_object_states( - orchestrator, - scope_id, - ) + assert plan.write_backend == Backend.MEMORY.value + assert plan.materialized_output.backend == Backend.DISK.value def test_explicit_checkpoint_filter_overrides_inherited_workload(tmp_path) -> None: ObjectStateRegistry.clear() ensure_global_config_context(GlobalPipelineConfig, GlobalPipelineConfig()) - orchestrator, scope_id, _config_state, resolved, filters = _resolved_checkpoint( + orchestrator, _effective_config, resolved, filters = _resolved_checkpoint( tmp_path / "plate", checkpoint_filter="B02", ) - try: - checkpoint_config = resolved.steps[0].step_materialization_config - assert checkpoint_config.well_filter == "B02" - assert not filters[0].allows(checkpoint_config, "A01") - assert filters[0].allows(checkpoint_config, "B02") - finally: - PipelineCompiler._cleanup_compilation_object_states( - orchestrator, - scope_id, - ) + checkpoint_config = resolved.steps[0].step_materialization_config + assert checkpoint_config.well_filter == "B02" + assert not filters[0].allows(checkpoint_config, "A01") + assert filters[0].allows(checkpoint_config, "B02") diff --git a/tests/unit/test_path_planner_materialization.py b/tests/unit/test_path_planner_materialization.py index 9545224acb..8eafe005d9 100644 --- a/tests/unit/test_path_planner_materialization.py +++ b/tests/unit/test_path_planner_materialization.py @@ -1205,7 +1205,11 @@ def test_implicit_native_main_flow_provenance_drives_artifact_owned_scope(): orchestrator=SimpleNamespace(pipeline_config=PipelineConfig()), global_config=GlobalPipelineConfig(), pipeline=ResolvedPipelineDefinition( - steps=steps, step_state_map={index: object() for index in range(len(steps))} + steps=steps, + step_scope_ids={ + index: f"plate::step_{index}" for index in range(len(steps)) + }, + step_provenance={index: {} for index in range(len(steps))}, ), ) provider = CellProfilerInvocationContractProviderFactory.provider_for_session( diff --git a/tests/unit/test_resolved_pipeline_definition.py b/tests/unit/test_resolved_pipeline_definition.py index acdf9a8983..5df7c4edd5 100644 --- a/tests/unit/test_resolved_pipeline_definition.py +++ b/tests/unit/test_resolved_pipeline_definition.py @@ -11,20 +11,13 @@ StepSourceBindingsConfig, ) from openhcs.core.steps.function_step import FunctionStep +from openhcs.core.function_patterns import normalize_function_pattern def _identity(image): return image -class StateStub: - def __init__(self, scope_id="plate::functionstep_0"): - self.scope_id = scope_id - - def to_object(self): - raise AssertionError("Resolved pipeline must not call ObjectState.to_object()") - - def test_resolved_pipeline_reads_steps_without_object_conversion(): source_bindings = StepSourceBindingsConfig( bindings=( @@ -47,11 +40,13 @@ def test_resolved_pipeline_reads_steps_without_object_conversion(): ), step_materialization_config=StepMaterializationConfig(enabled=False), ) - state = StateStub() + scope = "plate::functionstep_0" - pipeline = ResolvedPipelineDefinition([step], {0: state}) - assert pipeline.steps[0] is step - assert pipeline.step_state_map[0] is state + pipeline = ResolvedPipelineDefinition([step], {0: scope}, step_provenance={0: {}}) + assert pipeline.steps[0].name == step.name + assert step.func is _identity + assert next(normalize_function_pattern(pipeline.steps[0].func).iter_items()).func is _identity + assert pipeline.step_scope_ids[0] == scope assert pipeline.steps[0].source_bindings is source_bindings assert ( pipeline.steps[0].processing_config.input_source is InputSource.PIPELINE_START @@ -61,8 +56,8 @@ def test_resolved_pipeline_reads_steps_without_object_conversion(): ] -def test_resolved_pipeline_requires_matching_objectstate(): +def test_resolved_pipeline_requires_matching_scope(): step = FunctionStep(func=_identity, name="missing") - with pytest.raises(ValueError, match="missing ObjectState"): - ResolvedPipelineDefinition([step], {}) + with pytest.raises(ValueError, match="missing scope/provenance facts"): + ResolvedPipelineDefinition([step], {}, step_provenance={})