diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index ec09dfb..2e35944 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -798,7 +798,7 @@ class FlowController: nodes: list[Node] = [] loaded: dict[str, LoadedNode] = {} initial_values: dict[str, Any] = {} - # Declared flow inputs, mapped to whether they start with a value. + # Declared flow inputs, mapped to whether a value will be there. flow_inputs: dict[str, bool] = {} for flow, draft in flows: @@ -1465,14 +1465,22 @@ def with_settings( def _declared_inputs(flow: FlowDef) -> tuple[dict[str, bool], dict[str, Any]]: - """A flow's declared inputs, and the ones that start with a value.""" + """A flow's declared inputs, and the ones that start with a value. + + An input maps to whether a value will be there when the flow starts, which + for a batch flow is not the same as carrying an initial: it is started + explicitly and its inputs arrive with the run, so one without an initial is + a parameter the caller fills rather than a message nothing ever sets. + Reporting it as the latter had the health summary count an experiment as a + flow that cannot run while the Runs screen showed it running. + """ declared: dict[str, bool] = {} initial: dict[str, Any] = {} for flow_input in flow.inputs: name = qualify(flow.name, flow_input.spec.name) if not name: continue - declared[name] = flow_input.initial is not None + declared[name] = flow_input.initial is not None or flow.mode == "batch" if flow_input.initial is not None: initial[name] = flow_input.initial return declared, initial diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 3cb586d..534c61b 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -447,7 +447,8 @@ class Pipeline: """Report everything that would keep this graph from running. :param flow_inputs: Messages declared as inputs of a flow rather than - computed by it, mapped to whether they carry an initial value. + computed by it, mapped to whether a value will be there when the + flow starts — from an initial, or from the run that starts it. """ declared = flow_inputs or {} issues: list[ValidationIssue] = [] diff --git a/backend/tests/flow/test_pipeline.py b/backend/tests/flow/test_pipeline.py index 952a09d..cec2058 100644 --- a/backend/tests/flow/test_pipeline.py +++ b/backend/tests/flow/test_pipeline.py @@ -1,9 +1,11 @@ """The wiring fundamentals: name binding, fan-in, namespaces, validation.""" +from fluksio.flow.controller import _declared_inputs from fluksio.flow.events import EventBus from fluksio.flow.messages import DType, MessageSpec from fluksio.flow.nodes import Node from fluksio.flow.pipeline import Pipeline +from fluksio.flow.schemas import FlowDef, FlowInput def spec(name: str, dtype: DType = DType.FLOAT, port: str = "") -> MessageSpec: @@ -131,6 +133,25 @@ def test_a_flow_input_without_a_starting_value_is_reported(): assert [issue.code for issue in issues] == ["missing_initial_value"] +def test_a_batch_flows_input_is_a_run_parameter_not_a_missing_value(): + """A flow that only runs when asked gets its inputs from the run. + + Calling that a message nothing ever sets had the health summary count a + training flow as one that cannot run while the Runs screen showed it + running. A live flow, which nothing is going to start on its own, still + reports it. + """ + node = make_node( + "n", "f", lambda setpoint, params: None, requires=[spec("setpoint")] + ) + batch = FlowDef(name="f", mode="batch", inputs=[FlowInput(spec=spec("setpoint"))]) + live = batch.model_copy(update={"mode": "live"}) + + assert Pipeline(nodes=[node]).validate(_declared_inputs(batch)[0]) == [] + issues = Pipeline(nodes=[node]).validate(_declared_inputs(live)[0]) + assert [issue.code for issue in issues] == ["missing_initial_value"] + + def test_a_failing_node_does_not_stop_its_siblings(): ran = [] diff --git a/docs/concepts/flows.md b/docs/concepts/flows.md index 518654d..f745363 100644 --- a/docs/concepts/flows.md +++ b/docs/concepts/flows.md @@ -138,7 +138,7 @@ to: | Issue | What it means | |---|---| | `unconnected_input` | a port needs a message nothing in reach provides | -| `missing_initial_value` | the message exists but has never held a value, and nothing will give it one | +| `missing_initial_value` | the message exists but has never held a value, and nothing will give it one. Not reported on a batch flow: its inputs arrive with the run | | `cycle` | A waits for B and B waits for A — nothing could ever start | | `self_loop_needs_initial` | a node reads a message it also writes, with no starting value | | `node_error` | the node's code did not load: a syntax error, a missing import |