Pin that a per-flow publish deploys the source it just published

The worst thing moving publish off the whole-pipeline rebuild could do
is quietly keep running the old code, and nothing was watching for it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01StpRc2C6au1WJ1EUU7fsfu

backend/tests/flow/test_runtime_control.py

@@ -332,3 +332,36 @@ def test_stopping_one_flow_does_not_touch_the_others(
         assert {name: engine.loaded[name].node for name in ("a.io", "b.io")} == before
 
     asyncio.run(asyncio.wait_for(scenario(), timeout=10))
+
+
+NEW_SOURCE = "def process():\n    return {'reading': 2.0}\n"
+
+
+def test_a_per_flow_rebuild_runs_the_source_that_was_just_published(tmp_path: Path):
+    """The one thing moving publish off the full rebuild must not break."""
+    store = FlowStore(tmp_path / "flows")
+    store.write_flow(
+        FlowDef(
+            name="probe",
+            nodes=[NodeDef(id="sensor", type="python", provides=[spec("reading")])],
+        )
+    )
+    store.write_node_source(
+        "probe", "sensor", "def process():\n    return {'reading': 1.0}\n", draft=False
+    )
+    engine = FlowController(store)
+
+    async def scenario() -> None:
+        await engine.reload()
+        engine.run_flow("probe")
+        assert engine.pipeline is not None
+        assert engine.pipeline.state["probe.reading"] == 1.0
+
+        store.write_node_source("probe", "sensor", NEW_SOURCE, draft=True)
+        store.publish_flow("probe", store.read_flow("probe", draft=True).version)
+        await engine.reload_flow("probe")
+
+        engine.run_flow("probe")
+        assert engine.pipeline.state["probe.reading"] == 2.0
+
+    asyncio.run(asyncio.wait_for(scenario(), timeout=20))