An example to evaluate: a training run, its dashboard, and two bugs it found

make seed-demo builds demo_training — prepare on the engine, a GPU-bound
train, evaluate back here — and a panel that draws the loss curve while the
training is still going. It is the session's whole argument in one flow: batch
runs with parameters and a result, a generator yielding on a declared port
rather than logging, fluksio.emit from inside a callback, artifacts carrying
the dataset and the weights between machines, and a sweep whose configs are
isolated from each other. The train node prefers its label rather than
requiring it, so it runs before a GPU box exists and says which machine and
which numeric backend it actually used.

Building it turned up two real bugs. A run waited for a worker its flow only
*preferred*, because required_labels ignored device_policy — so the example
hung on a label it did not need. And a run's seed never reached the flow, so
sweeping over seeds ran the same experiment N times; it now fills an input of
that name when the flow declares one, which is what the field looked like it
did all along.

Pressing Run on a batch flow now submits a run rather than taking the old
non-durable path — that button is the first thing anyone evaluating will press,
and it was quietly doing something else.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AD8SfVhzXBG2nAfFcVh3iD

backend/tests/flow/test_runs.py

@@ -308,3 +308,28 @@ def test_labels_come_from_the_nodes_that_ask_for_a_device():
     assert required_labels(flow) == []
     flow.nodes[0].device = "gpu"
     assert required_labels(flow) == ["gpu"]
+
+
+def test_a_preferred_device_does_not_hold_a_run_back():
+    flow = double_flow()
+    flow.nodes[0].device = "gpu"
+    flow.nodes[0].device_policy = "prefer"
+    # It runs on the engine when no such worker is attached, so waiting for
+    # one would be waiting for something the run does not need.
+    assert required_labels(flow) == []
+
+
+def test_a_runs_seed_fills_an_input_of_that_name():
+    flow = double_flow()
+    flow.inputs.append(FlowInput(spec=spec("seed", DType.INT), initial=0))
+
+    # Otherwise the field that tells two runs of one configuration apart would
+    # only look like the number the flow draws from.
+    assert seed_values(flow, {}, seed=7)["study.seed"] == 7
+    # An explicit parameter still wins.
+    assert seed_values(flow, {"seed": 3}, seed=7)["study.seed"] == 3
+
+
+def test_a_flow_without_a_seed_input_ignores_the_runs_seed():
+    flow = double_flow()
+    assert seed_values(flow, {"lr": 1.0}, seed=7) == {"study.lr": 1.0}