Separate editing from running with a draft/publish split

Edits autosave to flow.draft.json and nodes.draft/ instead of the files the
engine reads, so the pipeline keeps running the published version until
someone publishes. Every save carries the version it was based on: a second
client editing the same flow is refused with 409 and offered the choice
between their version and its own, rather than silently overwriting.

Draft saves no longer rebuild the pipeline; validation and node status for a
draft come from a throwaway build that never touches live state.

Also fixes a latent bug where an empty state backend is falsy, so Pipeline
quietly built itself a second, private state and left message history empty.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/tests/api/routes/test_flows.py

@@ -110,6 +110,89 @@ def test_running_a_flow_produces_values(
     assert response.json()["values"]["demo.reading"]["value"] == 21.5
 
 
+def test_editing_does_not_deploy_until_published(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    saved = client.put(
+        f"{PREFIX}/staged", headers=superuser_token_headers, json=a_flow("staged")
+    ).json()
+    assert saved["has_draft"] is True
+    # Nothing is running yet, so the engine knows no nodes of this flow.
+    assert client.get(f"{PREFIX}/staged/state", headers=superuser_token_headers).json()[
+        "nodes"
+    ] == []
+
+    published = client.post(
+        f"{PREFIX}/staged/publish",
+        headers=superuser_token_headers,
+        json={"version": saved["definition"]["version"]},
+    )
+    assert published.status_code == 200
+    assert published.json()["has_draft"] is False
+    assert {
+        node["id"]
+        for node in client.get(
+            f"{PREFIX}/staged/state", headers=superuser_token_headers
+        ).json()["nodes"]
+    } == {"staged.sensor", "staged.logger"}
+
+    client.delete(f"{PREFIX}/staged", headers=superuser_token_headers)
+
+
+def test_a_stale_save_is_refused(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    saved = client.put(
+        f"{PREFIX}/contested", headers=superuser_token_headers, json=a_flow("contested")
+    ).json()
+    stale = saved["definition"]
+
+    client.put(
+        f"{PREFIX}/contested",
+        headers=superuser_token_headers,
+        json={**stale, "title": "Mine"},
+    )
+
+    response = client.put(
+        f"{PREFIX}/contested",
+        headers=superuser_token_headers,
+        json={**stale, "title": "Theirs"},
+    )
+    assert response.status_code == 409
+    assert response.json()["detail"]["current_version"] == stale["version"] + 1
+
+    client.delete(f"{PREFIX}/contested", headers=superuser_token_headers)
+
+
+def test_discarding_a_draft_restores_what_is_running(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    saved = client.put(
+        f"{PREFIX}/reverted", headers=superuser_token_headers, json=a_flow("reverted")
+    ).json()
+    client.post(
+        f"{PREFIX}/reverted/publish",
+        headers=superuser_token_headers,
+        json={"version": saved["definition"]["version"]},
+    )
+    published = client.get(f"{PREFIX}/reverted", headers=superuser_token_headers).json()
+
+    client.put(
+        f"{PREFIX}/reverted",
+        headers=superuser_token_headers,
+        json={**published["definition"], "title": "Scratch that"},
+    )
+
+    response = client.post(
+        f"{PREFIX}/reverted/discard-draft", headers=superuser_token_headers
+    )
+    assert response.status_code == 200
+    assert response.json()["has_draft"] is False
+    assert response.json()["definition"]["title"] == "Demo"
+
+    client.delete(f"{PREFIX}/reverted", headers=superuser_token_headers)
+
+
 def test_unconnected_input_is_surfaced(
     client: TestClient, superuser_token_headers: dict[str, str]
 ) -> None: