diff --git a/docs/plans/issue265_main_flow_lineage_checkpoint_20260930.md b/docs/plans/issue265_main_flow_lineage_checkpoint_20260930.md new file mode 100644 index 0000000000..760f78d346 --- /dev/null +++ b/docs/plans/issue265_main_flow_lineage_checkpoint_20260930.md @@ -0,0 +1,180 @@ +# #265: chained callable lineage binding + +Owner: Zeno. Integration/live owner: parent. This is a new follow-up, not a +continuation of closed PR205. Base normally integrated: `bd1ef8d11` (main). +Socrates owns #257/#264, `runtime_image_values`, source projection and +materialization; Lovelace owns R0. None of those implementation files changed. + +## Original failure and owning fact + +The retained MCP receipt is +`/home/ts/wt/openhcs-issue-batch-20260929/source-live-publication-262-20260930/journey.json`. +Execution `2f15f508-d405-48f1-9ce3-35ae1ae314e1` compiled and completed the first +canonical public `inspect_label_fixture` step, then failed at the second step's +`FunctionCoreExecutor.load_artifact_inputs`: a storage-backed lineage edge had +no callable parameter. Original pipeline/reference/input/outputs remain untouched; +the original job was not replayed. Both original process identities were closed +by the integration owner. + +`MainFlowStackOutputSpec.bind_main_flow_source`, through +`MainFlowArtifactContractProvider`, declares exact input-qualified output lineage. +`CallableContract.output_group_scope_sources` derives its group-source identities +from those relations. `PathPlannerArtifactStage.compile_invocation_input_edges` +previously let producer storage turn that current-primary-payload source into an +extra stored callable argument. Storage location and argument role are different +facts. The runtime guard correctly rejected the resulting edge. + +The planner now intersects the existing output-relation projection with the +current main-flow identities before selecting input storage. Only that exact +intersection takes the existing unstored source-edge path. Unrelated storage, +including an input that merely appears in the fallback group-scope projection +without an output relation, retains its original authority. + +The existing `InvocationArtifactInputEdgePlan` now owns source-edge construction +and the complete-payload versus declared-source-image projection. The planner's +former constructor/projection block is deleted, not retained as a second reader. +Its stage shrinks from 1,439 to 1,430 lines; the edge declaration grows from 55 to +88 lines. No runtime loader, binding guard, source compatibility, kind roster, +registry, adapter switch or selector store changes. + +## Focused ownership audit / new case + +- **IDEN-1 / IDEN-6:** storage presence no longer answers the primary-lineage + question. Selection uses `ArtifactSpecRef` intersections from the actual + declaration owners, not names, filename parsing or artifact-type matching. +- **BOUND-2 / IMPL-4:** source projection uses the existing typed edge owner. + Complete versus selected-source projection moves with its construction; + the planner's replacement calls that owner instead of copying its procedure. +- **IDEN-5 / IMPL-5 / IMPL-13:** no second lineage store or loader mechanism; + the relation view is invocation-local and derived. Existing source-binding + and exact-storage paths remain distinct capabilities. +- **IMPL-10 / IMPL-14:** neither edge invariants nor the runtime missing-parameter + guard is weakened. Their old/current methods compare equal by AST. No extra + anonymous validation chain, concrete type switch or string dispatch is added. + +New-case experiment: a chained image producer declaring a +`MainFlowStackOutputSpec` requires **zero** new consumer/registry/loader cases. +Changing its exact input source changes the derived intersection. A source absent +from current main flow, or a stored secondary source not declared by the output +relation, remains storage-backed. Existing +`test_storage_backed_input_keeps_exact_runtime_authority` and +`test_compiled_producer_edge_overrides_matching_source_binding` are preserved, +not rewritten to accept broader main-flow precedence. + +## Actual evidence at this source-only checkpoint + +Required interpreter: +`/home/ts/code/projects/openhcs/.venv/bin/python`; own worktree and recorded +submodule `src` directories on `PYTHONPATH`. All nine package import locations +were verified inside this worktree. Recorded submodules are initialized. +No package/environment/installed-source changes. + +- `git diff --check`: passed. +- Standard-library AST parsing: all four changed Python files passed. +- Execution of the **actual extracted planner assignments and source-edge factory + body**: eight storage/current-main-flow/declared-relation combinations passed. + Exactly one case changes from the original storage-first behavior: a stored, + explicitly declared source present in current main flow. +- Additional actual-source factory case: multiple main-flow identities retain + `DECLARED_SOURCE_IMAGE`, rather than widening to the complete payload. +- AST equality of `InvocationArtifactInputEdgePlan.__post_init__` and + `FunctionCoreExecutor.load_artifact_inputs` against integrated main: passed. +- AST caller inventory: one production caller and 26 unit/helper call sites + identified. Production supplies `step_context.main_flow_artifacts`; no caller + signatures changed. This is an inventory, not execution of all those callers. +- Existing packaged `agent_comms.debt_ratchet` source at `dbd0965d`, invoked + read-only with `--root openhcs --base bd1ef8d11 --head 3c4805473`: exit 0. + It covers seven measure families, including the complete class inventory; + it predates C0's string-dispatch/type-switch measures at the newly published + workflow pin. This is not a claim that the full current R0 suite passed. + +These are focused source checks, **not** a complete NRA/R1 scan, native proof, +pytest compiler/runtime pass, numerical parity or installed/live acceptance. +Importing the planner for source pytest currently stops at missing own-worktree +`openhcs.core._tabular_native`; no donor binary or installed-source fallback was +used. Parent reviewed `3c4805473` and integrated it into its own current-main plus +PR262 `5c0` tree with native extensions built there. The own-worker missing-native +blocker does not block that integration. Parent owns the finite unit/compiler/ +orchestrator then full dotted-source native MCP gate; results await actual +receipts. No native, MCP, JVM, GUI or surviving test handle exists for this worker. + +Published draft: . +Production/test source checkpoint: `3c4805473`. Later receipt-only updates do +not change the source tested by the focused checks. All worker check/publish +commands are terminal; no slot/lock ownership is retained by this worker. + +## Prepared behavioral regressions and remaining acceptance + +`test_compiled_source_edges_only_consume_relation_owned_main_flow` now covers +four combinations of stored primary and stored named secondary. The secondary +keeps its exact storage plan/projection/parameter; the declared primary has no +storage-loading edge. Existing exact occurrence, mismatch, ambiguity and source +projection controls stay unchanged. + +`tests/integration/test_chained_callable_lineage_journey.py` reuses the canonical +public reference fixture, rather than copying its declaration/callable. It writes +a fresh 8x8 uint16 input, round-trips the complete two-step document, compiles via +the real inspection gateway, checks the second step's producer storage and typed +main-flow edge, then executes the real orchestrator. Assertions cover both exact +step-addressed image/label/measurement records, label subject, 16-pixel rows, +saved CSV/ROI and checkpoint TIFF readback. Publication remains enabled. This +test's first parent execution is recorded below; corrected full readback remains +pending. + +Prepared focused command (not a result): + +```sh +/home/ts/code/projects/openhcs/.venv/bin/python -m pytest -q \ + --basetemp /home/ts/wt/openhcs-chained-lineage-edge-20260930/.artifacts/pytest-265-source-next \ + tests/unit/test_path_planner_materialization.py::test_compiled_source_edges_only_consume_relation_owned_main_flow \ + tests/unit/test_invocation_input_source_context_identity.py \ + tests/unit/test_artifact_input_edge_cardinality.py \ + tests/unit/test_function_patterns.py \ + tests/integration/test_chained_callable_lineage_journey.py +``` + +Run only in the authorized native-built source tree with verified submodule +`PYTHONPATH`, bounded threads, nonblocking validation lock and fresh resource +guard. Parent currently has next-start authority; this worker remains source-only. +Use a fresh persistent owned basetemp for each future attempt; do not overwrite +the failed run's outputs/receipts. + +### First integrated run and test query correction + +Parent's integrated `23e082` run completed in 5.96s: six unit controls passed, +and the actual compiler plus both-step orchestrator completed successfully. +The integration test then failed at its own readback query, passing invocation +key `default` to a compiled `CHANNEL` output scope containing `1`. Preserve +`/home/ts/wt/openhcs-issue-batch-20260929/parent-265-chained-20260930.xml` unchanged. +This is not a second-step execution failure; the output authority correctly +rejected the test's mismatched coordinate. + +The test now uses `plan.group_scope().select_runtime_key(invocation.key.group_key)`. +That existing owner selects the sole compiled static key, preserves an ungrouped +output's absent coordinate, and rejects a genuinely mismatched multi-key request. +No production code, scope guard, assertion or hardcoded channel correction. +Actual extracted `ComponentGroupScope` declaration execution passed five group +projections (including a static channel other than `1`) and one fail-closed +multi-key mismatch; corrected test syntax and `git diff --check` pass. This is +source-only verification of the correction, not a corrected real readback pass. +Parent owns the full native MCP run on unchanged production; await its receipts +and the corrected full readback rerun before claiming acceptance. + +The unchanged base lacks Socrates' pending #264 publication repair. A full +publication journey may reach that separate boundary before reaching the new +second-step case. Do not deselect/xfail publication or absorb that implementation; +parent integration will combine the reviewed checkpoints and require exact +filenames, typed addresses and complete saved inventory through fresh MCP. +Installed/live readiness and #265 closure are not claimed at this checkpoint. + +## Persisted formats / resources + +No persisted format changes. User-authored pipeline source/config, names, TIFF, +CSV and ROI formats are unchanged. Internal compiled edge state is regenerated +on compile; no legacy reader, migration or compatibility alias. + +Persistent source: `/home/ts/wt/openhcs-chained-lineage-edge-20260930`. +Owned disposable cache root: +`/home/ts/.cache/agent-scratch/openhcs-chained-lineage-edge-20260930` (Zeno, +future bounded source build/tests only). Retain receipts before cleaning owned +disposable caches after terminal validation. No heavy startup was performed. diff --git a/openhcs/core/function_patterns.py b/openhcs/core/function_patterns.py index 2f30a1fa2c..8a351dc57d 100644 --- a/openhcs/core/function_patterns.py +++ b/openhcs/core/function_patterns.py @@ -26,6 +26,7 @@ ArtifactOutputPlan, ArtifactSpec, ArtifactSpecAccumulator, + ArtifactSpecCollection, ArtifactSpecRef, ) from openhcs.core.callable_contract import ( @@ -228,6 +229,39 @@ class InvocationArtifactInputEdgePlan: consumes_main_flow: bool = False main_flow_projection: MainFlowInputProjection | None = None + @classmethod + def from_source_declarations( + cls, + *, + key: InvocationArtifactInputProjectionKey, + spec: ArtifactSpec, + main_flow_artifacts: ArtifactSpecCollection, + invocation_sources: ArtifactSpecCollection, + ) -> "InvocationArtifactInputEdgePlan": + """Project one unstored source occurrence from its exact declarations.""" + + main_flow_refs = main_flow_artifacts.ref_set() + consumes_main_flow = ( + spec.ref() in main_flow_refs + and spec.ref() in invocation_sources.ref_set() + ) + return cls( + key=key, + spec=spec, + storage_plan=None, + projection=None, + consumes_main_flow=consumes_main_flow, + main_flow_projection=( + MainFlowInputProjection.COMPLETE_PAYLOAD + if consumes_main_flow and len(main_flow_refs) == 1 + else ( + MainFlowInputProjection.DECLARED_SOURCE_IMAGE + if consumes_main_flow + else None + ) + ), + ) + def __post_init__(self) -> None: if type(self.consumes_main_flow) is not bool: raise TypeError( diff --git a/openhcs/core/pipeline/path_planner.py b/openhcs/core/pipeline/path_planner.py index 18b96f8f62..b71d5e67e8 100644 --- a/openhcs/core/pipeline/path_planner.py +++ b/openhcs/core/pipeline/path_planner.py @@ -35,7 +35,6 @@ FunctionPatternSyntax, InvocationArtifactInputEdgePlan, InvocationArtifactInputProjectionKey, - MainFlowInputProjection, RuntimeParameterBinding, compile_function_pattern, inject_artifact_input_values, @@ -1221,8 +1220,8 @@ def compile_invocation_input_edges( for group in compiled_pattern.groups: invocations: list[CompiledFunctionInvocation] = [] for invocation in group.invocations: - invocation_main_flow_refs = ( - invocation.contract.group_scope_inputs.ref_set() + relation_owned_main_flow_refs = main_flow_refs.intersection( + invocation.contract.output_group_scope_sources ) if compiled_pattern.is_grouped: if execution_group_scope.is_ungrouped: @@ -1261,29 +1260,20 @@ def compile_invocation_input_edges( invocation.contract.artifact_inputs, strict=True, ): - storage_plan = selected_plans_by_ref.get(input_spec.ref()) - consumes_main_flow = ( - storage_plan is None - and input_spec.ref() in main_flow_refs - and input_spec.ref() in invocation_main_flow_refs + # A declared lineage source already travels in the primary + # payload; its producer storage is not a second ABI argument. + storage_plan = ( + None + if input_spec.ref() in relation_owned_main_flow_refs + else selected_plans_by_ref.get(input_spec.ref()) ) if storage_plan is None: edges.append( - InvocationArtifactInputEdgePlan( + InvocationArtifactInputEdgePlan.from_source_declarations( key=input_edge_key, spec=input_spec, - storage_plan=None, - projection=None, - consumes_main_flow=consumes_main_flow, - main_flow_projection=( - MainFlowInputProjection.COMPLETE_PAYLOAD - if consumes_main_flow and len(main_flow_refs) == 1 - else ( - MainFlowInputProjection.DECLARED_SOURCE_IMAGE - if consumes_main_flow - else None - ) - ), + main_flow_artifacts=main_flow_artifacts, + invocation_sources=invocation.contract.group_scope_inputs, ) ) continue @@ -1298,7 +1288,7 @@ def compile_invocation_input_edges( consumer_variable_components=consumer_variable_components, source_bindings=source_bindings, available_artifacts=available_artifacts, - consumes_main_flow=consumes_main_flow, + consumes_main_flow=False, ) ) invocations.append(invocation.with_artifact_input_edges(tuple(edges))) diff --git a/tests/integration/test_chained_callable_lineage_journey.py b/tests/integration/test_chained_callable_lineage_journey.py new file mode 100644 index 0000000000..59c840578c --- /dev/null +++ b/tests/integration/test_chained_callable_lineage_journey.py @@ -0,0 +1,164 @@ +"""Canonical callable twice through document, compiler, execution and publication.""" + +from csv import DictReader + +import numpy as np +import tifffile + +from openhcs.agent.services.execution_session_service import ( + AgentProgressQueue, + CompileInspectionInput, + InProcessCompileInspectionGateway, +) +from openhcs.constants import AllComponents, Microscope +from openhcs.core.artifacts import ArtifactInputPlan, ImageArtifactType +from openhcs.core.config import ( + GlobalPipelineConfig, + LazyPathPlanningConfig, + LazySourceBindingsConfig, + LazyStepMaterializationConfig, + PipelineConfig, +) +from openhcs.core.function_patterns import MainFlowInputProjection +from openhcs.core.orchestrator.execution_result import RuntimeObservationMode +from openhcs.core.orchestrator.orchestrator import PipelineOrchestrator +from openhcs.core.pipeline_document import PipelineDocumentAuthority +from openhcs.core.runtime_image_values import image_payload_data +from openhcs.core.runtime_object_labels import object_label_dense_array +from openhcs.core.runtime_stores import RuntimeArtifactQuery +from openhcs.core.source_bindings import ( + ComponentSelector, + NamedSourceBinding, + SourceFilterClause, + SourceFilterMatchType, + SourceFilterSubject, + SourceSelector, +) +from openhcs.core.steps.function_step import FunctionStep +from polystore.roi import load_rois_from_zip + +# Reuse the public reference's authoritative code block, not a copied callable. +from tests.unit.agent.test_callable_artifact_reference import reference_namespace # noqa: F401 + + +def test_chained_public_callable_uses_declared_main_flow_not_storage_argument( + tmp_path, reference_namespace, +): + plate = tmp_path / "plate" + plate.mkdir() + fixture = np.zeros((8, 8), dtype=np.uint16) + fixture[2:6, 3:7] = 1 + tifffile.imwrite(plate / "A01_s001_label-fixture.tif", fixture) + source = NamedSourceBinding( + alias="label-fixture", + selector=SourceSelector(filters=( + SourceFilterClause( + subject=SourceFilterSubject.FILE, + match_type=SourceFilterMatchType.CONTAINS, + value="label-fixture", + ), + )), + component_identity=(ComponentSelector(AllComponents.CHANNEL, "1"),), + ) + document = PipelineDocumentAuthority.from_values( + pipeline_config=PipelineConfig( + num_workers=1, + use_threading=True, + microscope=Microscope.SOURCE_BINDINGS, + source_bindings_config=LazySourceBindingsConfig(bindings=(source,)), + path_planning_config=LazyPathPlanningConfig( + global_output_folder=tmp_path / "outputs", + ), + ), + pipeline_steps=[ + FunctionStep( + func=reference_namespace["inspect_label_fixture"], + name=f"PublicLabelFixture{index}", + step_materialization_config=LazyStepMaterializationConfig(enabled=True), + ) + for index in range(2) + ], + ) + document = PipelineDocumentAuthority.from_source( + PipelineDocumentAuthority.render(document) + ) + bundle = InProcessCompileInspectionGateway().compile( + CompileInspectionInput( + plate=plate, + pipeline_document=document, + axis_filter=("A01",), + global_pipeline_config=GlobalPipelineConfig(num_workers=1, use_threading=True), + progress_queue=AgentProgressQueue(), + ) + ).execution_bundle + context = bundle.runtime_contexts["A01"] + first, second = (context.step_plans[index] for index in range(2)) + first_invocation = next(first.compiled_function_pattern.iter_invocations()) + second_invocation = next(second.compiled_function_pattern.iter_invocations()) + [first_image] = ( + plan for plan in first_invocation.artifact_output_plans + if plan.artifact_type is ImageArtifactType + ) + incoming_ref = first_image.ref().for_plan_type(ArtifactInputPlan) + # Storage provenance exists, but is not a second callable argument authority. + assert second.artifact_inputs[incoming_ref].source_step_id == 0 + [edge] = second_invocation.artifact_input_edges + assert edge.spec.ref() == incoming_ref + assert edge.spec.parameter_name is None + assert second_invocation.contract.group_scope_inputs.ref_set() == {incoming_ref} + assert edge.consumes_main_flow + assert edge.main_flow_projection is MainFlowInputProjection.COMPLETE_PAYLOAD + assert edge.storage_plan is None + assert edge.projection is None + + orchestrator = PipelineOrchestrator( + plate, pipeline_config=document.pipeline_config, + ).initialize() + results = orchestrator.execute_compiled_plate( + execution_bundle=bundle, + max_workers=1, + runtime_observation_mode=RuntimeObservationMode.MERGE_INTO_PARENT, + progress_queue=AgentProgressQueue(), + progress_context={ + "execution_id": "chained-public-callable-synthetic", + "plate_id": str(plate), + "axis_id": "", + }, + ) + assert results["A01"].is_success(), results["A01"].error_message + for step_plan in (first, second): + invocation = next(step_plan.compiled_function_pattern.iter_invocations()) + image_plan, labels_plan, rows_plan = invocation.artifact_output_plans + records = [] + for plan in (image_plan, labels_plan, rows_plan): + [record] = context.runtime_value_store.find_matching( + RuntimeArtifactQuery.from_output_plan( + plan, axis_id="A01", backend="memory", + group_key=plan.group_scope().select_runtime_key( + invocation.key.group_key + ), + ) + ) + records.append(record) + image, labels, rows = (record.value.data for record in records) + np.testing.assert_array_equal(np.squeeze(image_payload_data(image)), fixture) + np.testing.assert_array_equal(np.squeeze(object_label_dense_array(labels)), fixture) + assert rows.subject.object_name == labels_plan.name + assert rows.subject.id_field == "object_label" + assert tuple(rows.rows.column_values("object_label")) == (1,) + assert tuple(rows.rows.column_values("pixel_count")) == (16,) + assert rows_plan.object_subject_binding().source == labels_plan.ref() + analysis = step_plan.artifact_analysis_output_dir + [csv_path] = analysis.glob( + f"*_{rows_plan.name}_step{step_plan.step_index}_details.csv" + ) + with csv_path.open(newline="") as stream: + [persisted] = DictReader(stream) + assert persisted["object_label"] == "1" + assert persisted["pixel_count"] == "16" + [roi_path] = analysis.glob(f"*_{labels_plan.name}_step{step_plan.step_index}*.zip") + assert len(load_rois_from_zip(roi_path)) == 1 + saved_images = tuple(step_plan.materialized_output.output_dir.glob("*.tif")) + assert saved_images + for image_path in saved_images: + np.testing.assert_array_equal(np.squeeze(tifffile.imread(image_path)), fixture) diff --git a/tests/unit/test_path_planner_materialization.py b/tests/unit/test_path_planner_materialization.py index c278406f71..b53970b980 100644 --- a/tests/unit/test_path_planner_materialization.py +++ b/tests/unit/test_path_planner_materialization.py @@ -818,10 +818,15 @@ def test_artifact_output_source_lookup_combines_repeated_main_flow_inputs(): assert outputs[output.ref()].variable_components == (AllComponents.Z_INDEX,) -def test_compiled_source_edges_only_consume_relation_owned_main_flow(): - source_specs = tuple( - ArtifactSpec.input(name, ImageArtifactType) - for name in ("DNA", "Membrane", "Mitochondria") +@pytest.mark.parametrize("stored_main_flow", [False, True]) +@pytest.mark.parametrize("stored_secondary", [False, True]) +def test_compiled_source_edges_only_consume_relation_owned_main_flow( + stored_main_flow, stored_secondary, +): + source_specs = ( + ArtifactSpec.input("DNA", ImageArtifactType), + ArtifactSpec.input("Membrane", ImageArtifactType, parameter_name="secondary"), + ArtifactSpec.input("Mitochondria", ImageArtifactType), ) output_spec = ArtifactSpec.output_preserving_source_stack_scope( "Combined", @@ -837,7 +842,7 @@ def test_compiled_source_edges_only_consume_relation_owned_main_flow(): @artifact_inputs(*source_specs) @artifact_outputs(output_spec) - def combine_sources(image): + def combine_sources(image, secondary=None): return image compiled = compile_function_pattern( @@ -845,9 +850,19 @@ def combine_sources(image): {}, {plan.ref(): plan for plan in (output_plan,)}, ) + stored_inputs = { + spec.ref(): ArtifactInputPlan( + name=spec.name, + path=f"/memory/previous-step/{spec.name}.pkl", + artifact_type=ImageArtifactType, + source_step_id=0, + ) + for spec, stored in zip(source_specs[:2], (stored_main_flow, stored_secondary)) + if stored + } compiled = _artifact_planner_stub().artifacts.compile_invocation_input_edges( compiled, - artifact_inputs={}, + artifact_inputs=stored_inputs, relation_source_scopes={}, execution_group_scope=PathPlannerGroupScope.ungrouped(), consumer_variable_components=ComponentSet((AllComponents.Z_INDEX,)), @@ -857,6 +872,11 @@ def combine_sources(image): edges = next(compiled.iter_invocations()).artifact_input_edges assert tuple(edge.spec for edge in edges) == source_specs assert tuple(edge.consumes_main_flow for edge in edges) == (True, False, False) + assert edges[0].storage_plan is None + assert edges[0].projection is None + assert edges[1].storage_plan is stored_inputs.get(source_specs[1].ref()) + assert (edges[1].projection is not None) is stored_secondary + assert edges[1].spec.parameter_name == "secondary" def test_implicit_native_main_flow_provenance_drives_artifact_owned_scope():