diff --git a/NOTEPAD.md b/NOTEPAD.md
index f3765f0..906a0bd 100644
--- a/NOTEPAD.md
+++ b/NOTEPAD.md
@@ -8,8 +8,20 @@ Deferring because out of scope is fine, but don't mention deferring than.
 
 ## Open
 
+- BUG/UI: enlarge the icon in the sidebar slightly
+- BUG/UI: clicking outside the panel does not discard the flow edit panel
+- BUG/UI: the graph showed in the node edit panel should also be shown for a specific edge inside the pop-up panel when clicking the edge
+- BUG/UI: the "run" button in the toolbar should be stateful; i.e. when a flow is running (i.e. with nodes that automatically emit data such as triggers), it should be possible to pause the flow and resume it. Also for flows without a trigger, the button should change state to allow interrupting a running node
+- BUG/UI: the enlarged panel (for code editing) should still maintain its floating style
 - BUG/UI: `SidePanel`'s mobile branch does not set `data-testid` on the `SheetContent`, so
   `[data-testid=node-panel]` does not exist on a phone. Mobile specs cannot address the panel.
+- CHORE/API: node source saves carry no version precondition, so two clients editing the same
+  node's code are last-writer-wins. The flow document is what the optimistic lock protects;
+  code files would need their own, and an exact-match one produces false conflicts against a
+  single client's own interleaved flow and source saves.
+- FEAT/UI: publishing and discarding are only reachable while no side panel is open, since the
+  floating chrome hides for the panel. Editing a node's code and publishing it is therefore
+  close-panel-then-publish.
 - FEAT/UI: no way to point a node parameter at a stored secret. The store and the
   `{"$secret": "name"}` reference both work, but the params form only produces literals, so a
   webhook secret typed there lands in `flow.json`.
diff --git a/ROADMAP.md b/ROADMAP.md
index 24998f8..7600f6a 100644
--- a/ROADMAP.md
+++ b/ROADMAP.md
@@ -54,6 +54,10 @@ external interfaces. See `docs/architecture/structure.canvas` → *Backend – M
 - [x] Dependency-loop detection and graph validation surfaced as API errors
 - [ ] MQTT broker / InfluxDB compose services for local development
 - [x] Git-based versioning of the flow store (one commit per saved change)
+- [x] Draft/publish split: edits autosave to `flow.draft.json` / `nodes.draft/`,
+      the engine runs only the published files, and publishing promotes the
+      draft. Saves carry the version they were based on, so a second client
+      editing the same flow is refused rather than overwritten
 - [ ] Import/export of a flow as human-readable code plus a JSON structure
 - [ ] Per-input/-output discretization interval setting
 - [ ] Alert / notification handler
@@ -88,6 +92,8 @@ React + Vite, primarily desktop but usable on mobile. See `docs/architecture/str
 - [x] Embedded code editor (Monaco) for node source
 - [x] Live values on the edges, with the last payload and its time on click
 - [x] Validation shown on the node it belongs to, and summarised in the dock
+- [x] Publish control and draft markers in the flow bar, discard in the flow
+      panel, and a conflict dialog when another client got there first
 - [ ] Device assignment per node, selectable from compatible devices
 - [ ] Test-node affordance on the canvas
 - [ ] User management screens
diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py
index c4a15d3..8904830 100644
--- a/backend/app/api/routes/flows.py
+++ b/backend/app/api/routes/flows.py
@@ -34,7 +34,7 @@ from app.flow.schemas import (
     NodeTypeInfo,
 )
 from app.flow.state import as_number
-from app.flow.store import FlowExists, FlowNotFound
+from app.flow.store import FlowExists, FlowNotFound, StaleVersion
 from app.models import Message
 
 router = APIRouter(
@@ -46,11 +46,17 @@ ws_router = APIRouter(prefix="/flows", tags=["flows"])
 
 
 class FlowDetail(BaseModel):
-    """A flow plus how it is currently doing."""
+    """A flow plus how it is currently doing.
+
+    ``definition`` is the working copy — the unpublished draft when there is
+    one — because that is what the editor shows. ``nodes`` reports the
+    published flow, which is what is actually running.
+    """
 
     definition: FlowDef
     nodes: list[NodeStatusPublic] = []
     issues: list[ValidationIssue] = []
+    has_draft: bool = False
 
 
 class ValidationResult(BaseModel):
@@ -61,6 +67,10 @@ class RenameRequest(BaseModel):
     new_name: str
 
 
+class PublishRequest(BaseModel):
+    version: int
+
+
 class RunRequest(BaseModel):
     inputs: dict[str, Any] = {}
 
@@ -70,16 +80,28 @@ class TriggerRequest(BaseModel):
 
 
 def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail:
+    name = definition.name
+    if controller.store.has_draft(name):
+        # Report the draft the editor is showing, not the version running
+        # underneath it — otherwise a node the author just broke looks fine.
+        preview = controller.preview(name)
+        return FlowDetail(
+            definition=definition,
+            nodes=preview.nodes,
+            issues=preview.issues,
+            has_draft=True,
+        )
     return FlowDetail(
         definition=definition,
-        nodes=controller.node_statuses(definition.name),
-        issues=controller.flow_issues(definition.name),
+        nodes=controller.node_statuses(name),
+        issues=controller.flow_issues(name),
     )
 
 
 def _read_flow(controller: FlowController, name: str) -> FlowDef:
+    """The working copy: the draft when there is one, else what is published."""
     try:
-        return controller.store.read_flow(name)
+        return controller.store.read_flow(name, draft=True)
     except FlowNotFound:
         raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
 
@@ -102,14 +124,19 @@ def _flow_state(controller: FlowController, name: str) -> FlowStatePublic:
 def read_flows(controller: FlowControllerDep) -> Any:
     """List every flow."""
     summaries = []
-    for definition in controller.store.read_all():
-        statuses = controller.node_statuses(definition.name)
+    for name in controller.store.list_flows():
+        try:
+            definition = controller.store.read_flow(name, draft=True)
+        except FlowNotFound:
+            continue
+        statuses = controller.node_statuses(name)
         summaries.append(
             FlowSummary(
                 name=definition.name,
                 title=definition.title,
                 node_count=len(definition.nodes),
                 error_count=sum(1 for s in statuses if s.status == "error"),
+                has_draft=controller.store.has_draft(name),
             )
         )
     return FlowsPublic(data=summaries, count=len(summaries))
@@ -135,7 +162,13 @@ async def save_flow(
     definition: FlowDef,
     controller: FlowControllerDep,
 ) -> Any:
-    """Create or replace a flow. Saving the same content again changes nothing."""
+    """Save unpublished changes to a flow.
+
+    This writes a draft: the running pipeline keeps the published version until
+    someone publishes. ``version`` is the one the editor last saw — a mismatch
+    means another client saved in between and answers 409 rather than throwing
+    their work away.
+    """
     if definition.name != name:
         raise HTTPException(
             status_code=400, detail="The flow name in the body must match the URL"
@@ -145,10 +178,61 @@ async def save_flow(
     if len(duplicates) != len(definition.nodes):
         raise HTTPException(status_code=400, detail="Node names must be unique")
 
-    changed = await run_in_threadpool(controller.store.write_flow, definition)
-    if changed:
-        await controller.reload()
-    return _detail(controller, definition)
+    try:
+        stored = await run_in_threadpool(
+            controller.store.write_draft, definition, definition.version
+        )
+    except StaleVersion as exc:
+        raise HTTPException(
+            status_code=409,
+            detail={"message": str(exc), "current_version": exc.current},
+        )
+    return _detail(controller, stored)
+
+
+@router.post("/{name}/publish", response_model=FlowDetail)
+async def publish_flow(
+    name: str,
+    body: PublishRequest,
+    controller: FlowControllerDep,
+) -> Any:
+    """Deploy the unpublished changes: the engine picks them up from here."""
+    _read_flow(controller, name)
+    if not controller.store.has_draft(name):
+        raise HTTPException(
+            status_code=400, detail=f"Flow '{name}' has no unpublished changes"
+        )
+    try:
+        published = await run_in_threadpool(
+            controller.store.publish_flow, name, body.version
+        )
+    except StaleVersion as exc:
+        raise HTTPException(
+            status_code=409,
+            detail={"message": str(exc), "current_version": exc.current},
+        )
+    await controller.reload()
+    return _detail(controller, published)
+
+
+@router.post("/{name}/discard-draft", response_model=FlowDetail)
+async def discard_draft(name: str, controller: FlowControllerDep) -> Any:
+    """Throw the unpublished changes away and go back to what is running."""
+    _read_flow(controller, name)
+    if not controller.store.has_draft(name):
+        raise HTTPException(
+            status_code=400, detail=f"Flow '{name}' has no unpublished changes"
+        )
+    if not controller.store.is_published(name):
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"Flow '{name}' has never been published — delete it instead of "
+                "discarding it"
+            ),
+        )
+    published = await run_in_threadpool(controller.store.discard_draft, name)
+    return _detail(controller, published)
 
 
 @router.delete("/{name}", response_model=Message)
@@ -200,9 +284,9 @@ def read_node_source(
     node_id: str,
     controller: FlowControllerDep,
 ) -> Any:
-    """Read a node's Python source."""
+    """Read a node's Python source, including unpublished edits."""
     _read_flow(controller, name)
-    return NodeSource(code=controller.store.read_node_source(name, node_id))
+    return NodeSource(code=controller.store.read_node_source(name, node_id, draft=True))
 
 
 @router.put("/{name}/nodes/{node_id}/source", response_model=NodeStatusPublic)
@@ -212,19 +296,24 @@ async def save_node_source(
     source: NodeSource,
     controller: FlowControllerDep,
 ) -> Any:
-    """Save a node's source and report whether it loads."""
-    _read_flow(controller, name)
-    changed = await run_in_threadpool(
-        controller.store.write_node_source, name, node_id, source.code
-    )
-    if changed:
-        await controller.reload()
+    """Save a node's source as an unpublished edit and report whether it loads.
 
-    status = next(
-        (s for s in controller.node_statuses(name) if s.id == f"{name}.{node_id}"),
-        None,
+    The answer comes from compiling the code rather than from the running
+    pipeline: a draft is not deployed, and compiling is both faster and more
+    precise about what the author just typed.
+    """
+    _read_flow(controller, name)
+    await run_in_threadpool(
+        controller.store.write_node_source, name, node_id, source.code, True
+    )
+    error = await run_in_threadpool(
+        controller.compile_check, name, node_id, source.code
+    )
+    return NodeStatusPublic(
+        id=f"{name}.{node_id}",
+        status="error" if error else "active",
+        error=error,
     )
-    return status or NodeStatusPublic(id=f"{name}.{node_id}")
 
 
 # -----------------------------------------------------------------------------
@@ -236,6 +325,8 @@ async def save_node_source(
 def validate_flow(name: str, controller: FlowControllerDep) -> Any:
     """Report what would keep this flow from running."""
     _read_flow(controller, name)
+    if controller.store.has_draft(name):
+        return ValidationResult(issues=controller.preview(name).issues)
     return ValidationResult(issues=controller.flow_issues(name))
 
 
@@ -245,10 +336,17 @@ async def run_flow(
     body: RunRequest,
     controller: FlowControllerDep,
 ) -> Any:
-    """Run every node of a flow once."""
+    """Run every node of a flow once.
+
+    With unpublished changes this runs the draft, so the button matches what is
+    on the canvas. Nothing is deployed by running it.
+    """
     _read_flow(controller, name)
     inputs = {qualify(name, key): value for key, value in body.inputs.items()}
-    await run_in_threadpool(controller.run_flow, name, inputs)
+    if controller.store.has_draft(name):
+        await run_in_threadpool(controller.run_preview, name, inputs)
+    else:
+        await run_in_threadpool(controller.run_flow, name, inputs)
     return _flow_state(controller, name)
 
 
diff --git a/backend/app/flow/controller.py b/backend/app/flow/controller.py
index aba7b51..8102314 100644
--- a/backend/app/flow/controller.py
+++ b/backend/app/flow/controller.py
@@ -32,10 +32,10 @@ from app.flow.nodes import (
     Node,
 )
 from app.flow.pipeline import Pipeline, ValidationIssue
-from app.flow.schemas import NodeDef, NodeStatusPublic, NodeTypeInfo
+from app.flow.schemas import FlowDef, NodeDef, NodeStatusPublic, NodeTypeInfo
 from app.flow.secrets import SecretNotFound, resolve_params
 from app.flow.state import MemoryState, StateBackend
-from app.flow.store import FlowStore
+from app.flow.store import FlowNotFound, FlowStore
 
 logger = logging.getLogger(__name__)
 
@@ -58,6 +58,14 @@ class LoadedNode:
     error: str | None = None
 
 
+@dataclass
+class Preview:
+    """A draft as it would run, without running it."""
+
+    nodes: list[NodeStatusPublic]
+    issues: list[ValidationIssue]
+
+
 @dataclass
 class NodeType:
     """A node type the editor can place on a canvas."""
@@ -172,7 +180,7 @@ class FlowController:
         fastapi_app: FastAPI | None = None,
     ) -> None:
         self.store = store
-        self.state = state or MemoryState()
+        self.state = state if state is not None else MemoryState()
         self.events = events
         self.max_workers = max_workers
         self.app = fastapi_app
@@ -197,25 +205,9 @@ class FlowController:
         async with self._lock:
             await self._teardown()
 
-            nodes: list[Node] = []
-            loaded: dict[str, LoadedNode] = {}
-            initial_values: dict[str, Any] = {}
-            # Declared flow inputs, mapped to whether they start with a value.
-            flow_inputs: dict[str, bool] = {}
-
-            for flow in self.store.read_all():
-                for node_def in flow.nodes:
-                    entry = self._build_node(flow.name, node_def)
-                    loaded[entry.id] = entry
-                    if entry.node is not None:
-                        nodes.append(entry.node)
-                for flow_input in flow.inputs:
-                    name = qualify(flow.name, flow_input.spec.name)
-                    if not name:
-                        continue
-                    flow_inputs[name] = flow_input.initial is not None
-                    if flow_input.initial is not None:
-                        initial_values[name] = flow_input.initial
+            nodes, loaded, initial_values, flow_inputs = self._build_flows(
+                [(flow, False) for flow in self.store.read_all()]
+            )
 
             self.loaded = loaded
             self.pipeline = Pipeline(
@@ -225,35 +217,7 @@ class FlowController:
                 max_workers=self.max_workers,
                 initial_values=initial_values,
             )
-            self.issues = self.pipeline.validate(flow_inputs)
-            self.issues += [
-                ValidationIssue(
-                    code="node_error",
-                    message=entry.error or "This node could not be loaded.",
-                    flow=entry.flow,
-                    node=entry.id,
-                )
-                for entry in loaded.values()
-                if entry.status is NodeStatus.ERROR
-            ]
-            # Webhooks are mounted without authentication, so a hook without a
-            # secret is open to anyone who guesses its URL. Flows written before
-            # the parameter existed keep running, but say so.
-            self.issues += [
-                ValidationIssue(
-                    code="unauthenticated_hook",
-                    message=(
-                        f"Webhook '{entry.node.local_id}' has no secret — "
-                        "anyone who knows its URL can trigger it."
-                    ),
-                    flow=entry.flow,
-                    node=entry.id,
-                )
-                for entry in loaded.values()
-                if isinstance(entry.node, HttpNode)
-                and entry.node.mode == HttpNode.Mode.TRIGGER
-                and not entry.node.secret
-            ]
+            self.issues = _collect_issues(loaded, self.pipeline, flow_inputs)
 
             await self._activate()
 
@@ -311,7 +275,35 @@ class FlowController:
     # Building
     # -------------------------------------------------------------------------
 
-    def _build_node(self, flow: str, node_def: NodeDef) -> LoadedNode:
+    def _build_flows(
+        self, flows: list[tuple[FlowDef, bool]]
+    ) -> tuple[list[Node], dict[str, LoadedNode], dict[str, Any], dict[str, bool]]:
+        """Instantiate the nodes of several flows, each published or draft."""
+        nodes: list[Node] = []
+        loaded: dict[str, LoadedNode] = {}
+        initial_values: dict[str, Any] = {}
+        # Declared flow inputs, mapped to whether they start with a value.
+        flow_inputs: dict[str, bool] = {}
+
+        for flow, draft in flows:
+            for node_def in flow.nodes:
+                entry = self._build_node(flow.name, node_def, draft=draft)
+                loaded[entry.id] = entry
+                if entry.node is not None:
+                    nodes.append(entry.node)
+            for flow_input in flow.inputs:
+                name = qualify(flow.name, flow_input.spec.name)
+                if not name:
+                    continue
+                flow_inputs[name] = flow_input.initial is not None
+                if flow_input.initial is not None:
+                    initial_values[name] = flow_input.initial
+
+        return nodes, loaded, initial_values, flow_inputs
+
+    def _build_node(
+        self, flow: str, node_def: NodeDef, draft: bool = False
+    ) -> LoadedNode:
         node_id = f"{flow}.{node_def.id}"
         entry = LoadedNode(id=node_id, flow=flow)
         try:
@@ -322,7 +314,7 @@ class FlowController:
             params = resolve_params(node_def.params)
 
             if node_type.has_source:
-                code = self.store.read_node_source(flow, node_def.id)
+                code = self.store.read_node_source(flow, node_def.id, draft=draft)
                 node = Node(
                     f=_load_function(flow, node_def.id, code),
                     requires=_bound(node_def.requires),
@@ -368,6 +360,51 @@ class FlowController:
     def flow_issues(self, flow: str) -> list[ValidationIssue]:
         return [issue for issue in self.issues if not issue.flow or issue.flow == flow]
 
+    def preview(self, name: str) -> Preview:
+        """Build a flow's unpublished draft without deploying it.
+
+        The draft stands in for its published self among the other flows, so
+        cross-flow messages resolve the way they would after publishing. The
+        graph is thrown away afterwards and seeded into a throwaway state, so
+        nothing here touches what is running.
+        """
+        flows: list[tuple[FlowDef, bool]] = [
+            (flow, False) for flow in self.store.read_all() if flow.name != name
+        ]
+        try:
+            flows.append((self.store.read_flow(name, draft=True), True))
+        except FlowNotFound:
+            return Preview(nodes=[], issues=[])
+
+        nodes, loaded, initial_values, flow_inputs = self._build_flows(flows)
+        pipeline = Pipeline(
+            nodes=nodes,
+            state=MemoryState(),
+            max_workers=self.max_workers,
+            initial_values=initial_values,
+        )
+        issues = _collect_issues(loaded, pipeline, flow_inputs)
+        return Preview(
+            nodes=[
+                NodeStatusPublic(
+                    id=entry.id, status=entry.status.value, error=entry.error
+                )
+                for entry in loaded.values()
+                if entry.flow == name
+            ],
+            issues=[
+                issue for issue in issues if not issue.flow or issue.flow == name
+            ],
+        )
+
+    def compile_check(self, flow: str, node_id: str, code: str) -> str | None:
+        """Does this source load? Returns what to show the author, or None."""
+        try:
+            _load_function(flow, node_id, code)
+        except Exception as exc:
+            return _short_error(exc)
+        return None
+
     def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]:
         return self.pipeline.values(flow) if self.pipeline else {}
 
@@ -385,6 +422,25 @@ class FlowController:
             return
         self.pipeline.run(inputs or {}, nodes=self.pipeline.flow_nodes(flow))
 
+    def run_preview(self, flow_name: str, inputs: dict[str, Any] | None = None) -> None:
+        """Run a flow's unpublished draft once, as the editor shows it.
+
+        The nodes are built and run but never activated, so a draft never
+        acquires subscriptions, schedules or webhooks — only publishing does
+        that. State and events are the live ones, so values land on the canvas
+        exactly as they do for a published run.
+        """
+        flow = self.store.read_flow(flow_name, draft=True)
+        nodes, _, initial_values, _ = self._build_flows([(flow, True)])
+        pipeline = Pipeline(
+            nodes=nodes,
+            state=self.state,
+            events=self.events,
+            max_workers=self.max_workers,
+            initial_values=initial_values,
+        )
+        pipeline.run(inputs or {})
+
     def trigger_node(self, node_id: str, values: dict[str, Any] | None = None) -> None:
         """Feed values into one node. Blocking — call from a worker thread."""
         node = self.get_node(node_id)
@@ -405,6 +461,44 @@ def _bound(specs: list[MessageSpec]) -> list[MessageSpec]:
     return [spec for spec in specs if spec.name]
 
 
+def _collect_issues(
+    loaded: dict[str, LoadedNode],
+    pipeline: Pipeline,
+    flow_inputs: dict[str, bool],
+) -> list[ValidationIssue]:
+    """Everything wrong with a built graph, whether it is running or not."""
+    issues = pipeline.validate(flow_inputs)
+    issues += [
+        ValidationIssue(
+            code="node_error",
+            message=entry.error or "This node could not be loaded.",
+            flow=entry.flow,
+            node=entry.id,
+        )
+        for entry in loaded.values()
+        if entry.status is NodeStatus.ERROR
+    ]
+    # Webhooks are mounted without authentication, so a hook without a secret is
+    # open to anyone who guesses its URL. Flows written before the parameter
+    # existed keep running, but say so.
+    issues += [
+        ValidationIssue(
+            code="unauthenticated_hook",
+            message=(
+                f"Webhook '{entry.node.local_id}' has no secret — "
+                "anyone who knows its URL can trigger it."
+            ),
+            flow=entry.flow,
+            node=entry.id,
+        )
+        for entry in loaded.values()
+        if isinstance(entry.node, HttpNode)
+        and entry.node.mode == HttpNode.Mode.TRIGGER
+        and not entry.node.secret
+    ]
+    return issues
+
+
 def _short_error(exc: Exception) -> str:
     """One line a node author can act on: what broke, and where in their code.
 
diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py
index 92498db..c8bcefb 100644
--- a/backend/app/flow/pipeline.py
+++ b/backend/app/flow/pipeline.py
@@ -66,7 +66,10 @@ class Pipeline:
         initial_values: dict[str, Any] | None = None,
     ) -> None:
         self._nodes = nodes or []
-        self._state: StateBackend = state or MemoryState()
+        # An empty state backend is falsy, so this cannot be ``state or ...``:
+        # that would quietly hand the pipeline a second, private state and
+        # leave everyone reading the shared one seeing nothing.
+        self._state: StateBackend = state if state is not None else MemoryState()
         self._events = events
         self._max_workers = max_workers
 
diff --git a/backend/app/flow/schemas.py b/backend/app/flow/schemas.py
index 20abcfd..d7fcbd2 100644
--- a/backend/app/flow/schemas.py
+++ b/backend/app/flow/schemas.py
@@ -115,6 +115,7 @@ class FlowSummary(BaseModel):
     title: str = ""
     node_count: int = 0
     error_count: int = 0
+    has_draft: bool = False
 
 
 class FlowsPublic(BaseModel):
diff --git a/backend/app/flow/store.py b/backend/app/flow/store.py
index 775d139..8081971 100644
--- a/backend/app/flow/store.py
+++ b/backend/app/flow/store.py
@@ -4,6 +4,12 @@ Every flow is a directory: ``flow.json`` for the structure, ``nodes/*.py`` for
 node logic. The whole tree is a git repository and each saved change is a
 commit, so a flow's history is readable with ordinary git tooling and two
 flows never collide in one file.
+
+Editing is separated from running. The editor writes ``flow.draft.json`` and
+``nodes.draft/*.py``; the engine reads only the published ``flow.json`` and
+``nodes/*.py``. Publishing promotes the draft over the published files and
+removes it, so a flow directory without draft files is simply one with nothing
+unpublished — which is what every flow written before this existed looks like.
 """
 
 from __future__ import annotations
@@ -11,6 +17,7 @@ from __future__ import annotations
 import logging
 import shutil
 import subprocess
+import threading
 from pathlib import Path
 
 from app.flow.schemas import FlowDef
@@ -43,12 +50,34 @@ class FlowExists(ValueError):
         return f"There is already a flow named '{self.name}'"
 
 
+class StaleVersion(ValueError):
+    """Someone else saved this flow since the version being written was read."""
+
+    def __init__(self, name: str, current: int) -> None:
+        super().__init__(name)
+        self.name = name
+        self.current = current
+
+    def __str__(self) -> str:
+        return f"Flow '{self.name}' has changed since you loaded it"
+
+
+def _same_content(left: FlowDef, right: FlowDef) -> bool:
+    """Equal but for the version counter, which the server owns."""
+    return left.model_copy(update={"version": 0}) == right.model_copy(
+        update={"version": 0}
+    )
+
+
 class FlowStore:
     """Reads and writes flows, committing every change."""
 
     def __init__(self, root: Path) -> None:
         self.root = root
         self.root.mkdir(parents=True, exist_ok=True)
+        # Draft writes are check-and-set, so two clients saving at once must not
+        # interleave between reading the current version and writing the next.
+        self._write_lock = threading.Lock()
         if not (self.root / ".git").exists():
             self._git("init", "-q")
             self._commit("Initialise flow store", allow_empty=True)
@@ -91,28 +120,57 @@ class FlowStore:
     def _flow_file(self, name: str) -> Path:
         return self._flow_dir(name) / "flow.json"
 
+    def _draft_file(self, name: str) -> Path:
+        return self._flow_dir(name) / "flow.draft.json"
+
     def _node_file(self, flow: str, node_id: str) -> Path:
         return self._flow_dir(flow) / "nodes" / f"{node_id}.py"
 
+    def _draft_nodes_dir(self, flow: str) -> Path:
+        return self._flow_dir(flow) / "nodes.draft"
+
+    def _draft_node_file(self, flow: str, node_id: str) -> Path:
+        return self._draft_nodes_dir(flow) / f"{node_id}.py"
+
     # -------------------------------------------------------------------------
     # Flows
     # -------------------------------------------------------------------------
 
     def list_flows(self) -> list[str]:
-        return sorted(path.parent.name for path in self.root.glob("*/flow.json"))
+        """Every flow the editor knows, published or not."""
+        names = {path.parent.name for path in self.root.glob("*/flow.json")}
+        names |= {path.parent.name for path in self.root.glob("*/flow.draft.json")}
+        return sorted(names)
 
     def exists(self, name: str) -> bool:
+        return self._flow_file(name).exists() or self._draft_file(name).exists()
+
+    def is_published(self, name: str) -> bool:
         return self._flow_file(name).exists()
 
-    def read_flow(self, name: str) -> FlowDef:
+    def has_draft(self, name: str) -> bool:
+        """Are there unpublished changes — to the structure or to any source?"""
+        if self._draft_file(name).exists():
+            return True
+        drafts = self._draft_nodes_dir(name)
+        return drafts.exists() and any(drafts.glob("*.py"))
+
+    def read_flow(self, name: str, draft: bool = False) -> FlowDef:
+        """The published flow, or with ``draft`` the working copy."""
+        if draft:
+            path = self._draft_file(name)
+            if path.exists():
+                return FlowDef.model_validate_json(path.read_text())
         path = self._flow_file(name)
         if not path.exists():
             raise FlowNotFound(name)
         return FlowDef.model_validate_json(path.read_text())
 
     def read_all(self) -> list[FlowDef]:
+        """Every published flow — what the engine runs."""
         flows = []
-        for name in self.list_flows():
+        for path in sorted(self.root.glob("*/flow.json")):
+            name = path.parent.name
             try:
                 flows.append(self.read_flow(name))
             except Exception:
@@ -120,7 +178,7 @@ class FlowStore:
         return flows
 
     def write_flow(self, flow: FlowDef) -> bool:
-        """Save a flow. Returns False when nothing actually changed."""
+        """Publish a flow directly. Returns False when nothing actually changed."""
         path = self._flow_file(flow.name)
         content = flow.model_dump_json(indent=2) + "\n"
         if path.exists() and path.read_text() == content:
@@ -131,6 +189,68 @@ class FlowStore:
         self._commit(f"Update flow '{flow.name}'")
         return True
 
+    def write_draft(self, flow: FlowDef, base_version: int) -> FlowDef:
+        """Save unpublished changes, refusing to overwrite someone else's.
+
+        ``base_version`` is the version the editor last saw. It has to match
+        what is stored, otherwise another client saved in between and this
+        write would silently discard their work.
+        """
+        with self._write_lock:
+            current: FlowDef | None = None
+            if self.exists(flow.name):
+                current = self.read_flow(flow.name, draft=True)
+                if base_version != current.version:
+                    raise StaleVersion(flow.name, current.version)
+                if _same_content(current, flow):
+                    return current
+
+            stored = flow.model_copy(
+                update={"version": (current.version if current else 0) + 1}
+            )
+            path = self._draft_file(flow.name)
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(stored.model_dump_json(indent=2) + "\n")
+            self._commit(f"Update draft of flow '{flow.name}'")
+            return stored
+
+    def publish_flow(self, name: str, base_version: int) -> FlowDef:
+        """Promote the working copy to what the engine runs."""
+        with self._write_lock:
+            current = self.read_flow(name, draft=True)
+            if base_version != current.version:
+                raise StaleVersion(name, current.version)
+
+            draft = self._draft_file(name)
+            if draft.exists():
+                self._flow_file(name).write_text(
+                    current.model_dump_json(indent=2) + "\n"
+                )
+                draft.unlink()
+
+            drafts = self._draft_nodes_dir(name)
+            if drafts.exists():
+                target = self._flow_dir(name) / "nodes"
+                target.mkdir(parents=True, exist_ok=True)
+                for source in sorted(drafts.glob("*.py")):
+                    shutil.copy2(source, target / source.name)
+                shutil.rmtree(drafts)
+
+            self._commit(f"Publish flow '{name}'")
+            return current
+
+    def discard_draft(self, name: str) -> FlowDef:
+        """Throw the unpublished changes away and go back to what is running."""
+        with self._write_lock:
+            draft = self._draft_file(name)
+            if draft.exists():
+                draft.unlink()
+            drafts = self._draft_nodes_dir(name)
+            if drafts.exists():
+                shutil.rmtree(drafts)
+            self._commit(f"Discard draft of flow '{name}'")
+            return self.read_flow(name)
+
     def delete_flow(self, name: str) -> None:
         directory = self._flow_dir(name)
         if not directory.exists():
@@ -150,22 +270,30 @@ class FlowStore:
         if self.exists(new_name):
             raise FlowExists(new_name)
 
-        flow = self.read_flow(name)
         self._flow_dir(name).rename(self._flow_dir(new_name))
 
-        renamed = flow.model_copy(update={"name": new_name})
-        self._flow_file(new_name).write_text(renamed.model_dump_json(indent=2) + "\n")
-
-        for other in self.read_all():
-            if other.name == new_name:
+        # Published and draft alike carry the name, and both have to move.
+        for path in (self._flow_file(new_name), self._draft_file(new_name)):
+            if not path.exists():
                 continue
-            if self._retarget(other, f"{name}.", f"{new_name}."):
-                self._flow_file(other.name).write_text(
-                    other.model_dump_json(indent=2) + "\n"
-                )
+            flow = FlowDef.model_validate_json(path.read_text())
+            path.write_text(
+                flow.model_copy(update={"name": new_name}).model_dump_json(indent=2)
+                + "\n"
+            )
+
+        for other in self.list_flows():
+            if other == new_name:
+                continue
+            for path in (self._flow_file(other), self._draft_file(other)):
+                if not path.exists():
+                    continue
+                flow = FlowDef.model_validate_json(path.read_text())
+                if self._retarget(flow, f"{name}.", f"{new_name}."):
+                    path.write_text(flow.model_dump_json(indent=2) + "\n")
 
         self._commit(f"Rename flow '{name}' to '{new_name}'")
-        return renamed
+        return self.read_flow(new_name, draft=True)
 
     @staticmethod
     def _retarget(flow: FlowDef, old_prefix: str, new_prefix: str) -> bool:
@@ -186,18 +314,31 @@ class FlowStore:
     # Node source
     # -------------------------------------------------------------------------
 
-    def read_node_source(self, flow: str, node_id: str) -> str:
+    def read_node_source(self, flow: str, node_id: str, draft: bool = False) -> str:
+        if draft:
+            path = self._draft_node_file(flow, node_id)
+            if path.exists():
+                return path.read_text()
         path = self._node_file(flow, node_id)
         return path.read_text() if path.exists() else DEFAULT_SOURCE
 
-    def write_node_source(self, flow: str, node_id: str, code: str) -> bool:
-        path = self._node_file(flow, node_id)
-        if path.exists() and path.read_text() == code:
-            return False
+    def write_node_source(
+        self, flow: str, node_id: str, code: str, draft: bool = False
+    ) -> bool:
+        if draft:
+            # Compared against the working copy, so re-saving what is already
+            # published does not create a draft file that never gets promoted.
+            if self.read_node_source(flow, node_id, draft=True) == code:
+                return False
+            path = self._draft_node_file(flow, node_id)
+        else:
+            path = self._node_file(flow, node_id)
+            if path.exists() and path.read_text() == code:
+                return False
 
         path.parent.mkdir(parents=True, exist_ok=True)
         path.write_text(code)
-        self._commit(f"Update node '{flow}.{node_id}'")
+        self._commit(f"Update {'draft of ' if draft else ''}node '{flow}.{node_id}'")
         return True
 
     def delete_node_source(self, flow: str, node_id: str) -> None:
diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py
index 7a40922..b354d67 100644
--- a/backend/tests/api/routes/test_flows.py
+++ b/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:
diff --git a/backend/tests/flow/test_drafts.py b/backend/tests/flow/test_drafts.py
new file mode 100644
index 0000000..55d5d82
--- /dev/null
+++ b/backend/tests/flow/test_drafts.py
@@ -0,0 +1,133 @@
+"""Editing writes drafts; only publishing changes what the engine reads."""
+
+from pathlib import Path
+
+import pytest
+
+from app.flow.messages import MessageSpec
+from app.flow.schemas import FlowDef, NodeDef
+from app.flow.store import FlowStore, StaleVersion
+
+SOURCE = "def process(params):\n    return {}\n"
+EDITED = "def process(params):\n    return {'temp': 1}\n"
+
+
+@pytest.fixture
+def store(tmp_path: Path) -> FlowStore:
+    return FlowStore(tmp_path / "flows")
+
+
+def a_flow(title: str = "") -> FlowDef:
+    return FlowDef(
+        name="heating",
+        title=title,
+        nodes=[NodeDef(id="sensor", provides=[MessageSpec(name="temp")])],
+    )
+
+
+def test_a_draft_is_invisible_to_the_engine(store: FlowStore):
+    store.write_draft(a_flow(), 0)
+
+    assert store.has_draft("heating")
+    assert store.list_flows() == ["heating"]  # the editor sees it
+    assert store.read_all() == []  # the engine does not
+
+
+def test_the_version_counter_advances_per_save(store: FlowStore):
+    first = store.write_draft(a_flow(), 0)
+    assert first.version == 1
+
+    second = store.write_draft(a_flow(title="Warm"), first.version)
+    assert second.version == 2
+
+
+def test_resaving_identical_content_neither_bumps_nor_commits(store: FlowStore):
+    stored = store.write_draft(a_flow(), 0)
+
+    again = store.write_draft(a_flow(), stored.version)
+    assert again.version == stored.version
+
+
+def test_a_stale_write_is_refused(store: FlowStore):
+    store.write_draft(a_flow(), 0)
+    store.write_draft(a_flow(title="Mine"), 1)
+
+    # A second client still holding version 1 must not overwrite version 2.
+    with pytest.raises(StaleVersion) as excinfo:
+        store.write_draft(a_flow(title="Theirs"), 1)
+    assert excinfo.value.current == 2
+
+
+def test_publishing_promotes_the_draft_and_its_sources(store: FlowStore):
+    stored = store.write_draft(a_flow(title="Warm"), 0)
+    store.write_node_source("heating", "sensor", EDITED, draft=True)
+
+    published = store.publish_flow("heating", stored.version)
+
+    assert published.title == "Warm"
+    assert not store.has_draft("heating")
+    assert store.read_flow("heating").title == "Warm"
+    assert store.read_node_source("heating", "sensor") == EDITED
+    assert [flow.name for flow in store.read_all()] == ["heating"]
+
+
+def test_publishing_a_stale_draft_is_refused(store: FlowStore):
+    store.write_draft(a_flow(), 0)
+
+    with pytest.raises(StaleVersion):
+        store.publish_flow("heating", 99)
+
+
+def test_discarding_goes_back_to_what_is_running(store: FlowStore):
+    store.write_flow(a_flow(title="Published"))
+    store.write_node_source("heating", "sensor", SOURCE)
+    store.write_draft(a_flow(title="Edited"), a_flow().version)
+    store.write_node_source("heating", "sensor", EDITED, draft=True)
+
+    restored = store.discard_draft("heating")
+
+    assert restored.title == "Published"
+    assert not store.has_draft("heating")
+    assert store.read_node_source("heating", "sensor", draft=True) == SOURCE
+
+
+def test_an_edited_source_alone_counts_as_a_draft(store: FlowStore):
+    """Editing only code still has to be published before the engine runs it."""
+    store.write_flow(a_flow())
+    store.write_node_source("heating", "sensor", SOURCE)
+
+    store.write_node_source("heating", "sensor", EDITED, draft=True)
+
+    assert store.has_draft("heating")
+    assert store.read_node_source("heating", "sensor") == SOURCE
+    assert store.read_node_source("heating", "sensor", draft=True) == EDITED
+
+    store.publish_flow("heating", store.read_flow("heating").version)
+    assert store.read_node_source("heating", "sensor") == EDITED
+
+
+def test_resaving_the_published_source_creates_no_draft(store: FlowStore):
+    store.write_flow(a_flow())
+    store.write_node_source("heating", "sensor", SOURCE)
+
+    assert store.write_node_source("heating", "sensor", SOURCE, draft=True) is False
+    assert not store.has_draft("heating")
+
+
+def test_renaming_carries_the_draft_and_repoints_other_drafts(store: FlowStore):
+    store.write_draft(a_flow(), 0)
+    store.write_draft(
+        FlowDef(
+            name="display",
+            nodes=[NodeDef(id="gauge", requires=[MessageSpec(name="heating.temp")])],
+        ),
+        0,
+    )
+
+    renamed = store.rename_flow("heating", "warmth")
+
+    assert renamed.name == "warmth"
+    assert store.has_draft("warmth")
+    assert store.read_flow("display", draft=True).nodes[0].requires[0].name == (
+        "warmth.temp"
+    )
diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts
index 44b3f82..6e4cd57 100644
--- a/frontend/src/client/schemas.gen.ts
+++ b/frontend/src/client/schemas.gen.ts
@@ -161,12 +161,21 @@ export const FlowDetailSchema = {
             type: 'array',
             title: 'Issues',
             default: []
+        },
+        has_draft: {
+            type: 'boolean',
+            title: 'Has Draft',
+            default: false
         }
     },
     type: 'object',
     required: ['definition'],
     title: 'FlowDetail',
-    description: 'A flow plus how it is currently doing.'
+    description: `A flow plus how it is currently doing.
+
+\`\`definition\`\` is the working copy — the unpublished draft when there is
+one — because that is what the editor shows. \`\`nodes\`\` reports the
+published flow, which is what is actually running.`
 } as const;
 
 export const FlowInput_InputSchema = {
@@ -252,6 +261,11 @@ export const FlowSummarySchema = {
             type: 'integer',
             title: 'Error Count',
             default: 0
+        },
+        has_draft: {
+            type: 'boolean',
+            title: 'Has Draft',
+            default: false
         }
     },
     type: 'object',
@@ -631,6 +645,18 @@ export const PrivateUserCreateSchema = {
     title: 'PrivateUserCreate'
 } as const;
 
+export const PublishRequestSchema = {
+    properties: {
+        version: {
+            type: 'integer',
+            title: 'Version'
+        }
+    },
+    type: 'object',
+    required: ['version'],
+    title: 'PublishRequest'
+} as const;
+
 export const RenameRequestSchema = {
     properties: {
         new_name: {
diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts
index 4903d7d..49de469 100644
--- a/frontend/src/client/sdk.gen.ts
+++ b/frontend/src/client/sdk.gen.ts
@@ -3,7 +3,7 @@
 import type { CancelablePromise } from './core/CancelablePromise';
 import { OpenAPI } from './core/OpenAPI';
 import { request as __request } from './core/request';
-import type { FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, PrivateCreateUserData, PrivateCreateUserResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse } from './types.gen';
+import type { FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, PrivateCreateUserData, PrivateCreateUserResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse } from './types.gen';
 
 export class FlowsService {
     /**
@@ -55,7 +55,12 @@ export class FlowsService {
     
     /**
      * Save Flow
-     * Create or replace a flow. Saving the same content again changes nothing.
+     * Save unpublished changes to a flow.
+     *
+     * This writes a draft: the running pipeline keeps the published version until
+     * someone publishes. ``version`` is the one the editor last saw — a mismatch
+     * means another client saved in between and answers 409 rather than throwing
+     * their work away.
      * @param data The data for the request.
      * @param data.name
      * @param data.requestBody
@@ -98,6 +103,51 @@ export class FlowsService {
         });
     }
     
+    /**
+     * Publish Flow
+     * Deploy the unpublished changes: the engine picks them up from here.
+     * @param data The data for the request.
+     * @param data.name
+     * @param data.requestBody
+     * @returns FlowDetail Successful Response
+     * @throws ApiError
+     */
+    public static publishFlow(data: FlowsPublishFlowData): CancelablePromise<FlowsPublishFlowResponse> {
+        return __request(OpenAPI, {
+            method: 'POST',
+            url: '/api/v1/flows/{name}/publish',
+            path: {
+                name: data.name
+            },
+            body: data.requestBody,
+            mediaType: 'application/json',
+            errors: {
+                422: 'Validation Error'
+            }
+        });
+    }
+    
+    /**
+     * Discard Draft
+     * Throw the unpublished changes away and go back to what is running.
+     * @param data The data for the request.
+     * @param data.name
+     * @returns FlowDetail Successful Response
+     * @throws ApiError
+     */
+    public static discardDraft(data: FlowsDiscardDraftData): CancelablePromise<FlowsDiscardDraftResponse> {
+        return __request(OpenAPI, {
+            method: 'POST',
+            url: '/api/v1/flows/{name}/discard-draft',
+            path: {
+                name: data.name
+            },
+            errors: {
+                422: 'Validation Error'
+            }
+        });
+    }
+    
     /**
      * Rename Flow
      * Rename a flow, along with every reference to its messages.
@@ -124,7 +174,7 @@ export class FlowsService {
     
     /**
      * Read Node Source
-     * Read a node's Python source.
+     * Read a node's Python source, including unpublished edits.
      * @param data The data for the request.
      * @param data.name
      * @param data.nodeId
@@ -147,7 +197,11 @@ export class FlowsService {
     
     /**
      * Save Node Source
-     * Save a node's source and report whether it loads.
+     * Save a node's source as an unpublished edit and report whether it loads.
+     *
+     * The answer comes from compiling the code rather than from the running
+     * pipeline: a draft is not deployed, and compiling is both faster and more
+     * precise about what the author just typed.
      * @param data The data for the request.
      * @param data.name
      * @param data.nodeId
@@ -195,6 +249,9 @@ export class FlowsService {
     /**
      * Run Flow
      * Run every node of a flow once.
+     *
+     * With unpublished changes this runs the draft, so the button matches what is
+     * on the canvas. Nothing is deployed by running it.
      * @param data The data for the request.
      * @param data.name
      * @param data.requestBody
diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts
index c54ee5c..7b7d508 100644
--- a/frontend/src/client/types.gen.ts
+++ b/frontend/src/client/types.gen.ts
@@ -41,11 +41,16 @@ export type FlowDef_Output = {
 
 /**
  * A flow plus how it is currently doing.
+ *
+ * ``definition`` is the working copy — the unpublished draft when there is
+ * one — because that is what the editor shows. ``nodes`` reports the
+ * published flow, which is what is actually running.
  */
 export type FlowDetail = {
     definition: FlowDef_Output;
     nodes?: Array<NodeStatusPublic>;
     issues?: Array<ValidationIssue>;
+    has_draft?: boolean;
 };
 
 /**
@@ -81,6 +86,7 @@ export type FlowSummary = {
     title?: string;
     node_count?: number;
     error_count?: number;
+    has_draft?: boolean;
 };
 
 /**
@@ -213,6 +219,10 @@ export type PrivateUserCreate = {
     is_verified?: boolean;
 };
 
+export type PublishRequest = {
+    version: number;
+};
+
 export type RenameRequest = {
     new_name: string;
 };
@@ -337,6 +347,19 @@ export type FlowsDeleteFlowData = {
 
 export type FlowsDeleteFlowResponse = (Message);
 
+export type FlowsPublishFlowData = {
+    name: string;
+    requestBody: PublishRequest;
+};
+
+export type FlowsPublishFlowResponse = (FlowDetail);
+
+export type FlowsDiscardDraftData = {
+    name: string;
+};
+
+export type FlowsDiscardDraftResponse = (FlowDetail);
+
 export type FlowsRenameFlowData = {
     name: string;
     requestBody: RenameRequest;
diff --git a/frontend/src/components/Flow/FlowEditor.tsx b/frontend/src/components/Flow/FlowEditor.tsx
index b0b8097..885b683 100644
--- a/frontend/src/components/Flow/FlowEditor.tsx
+++ b/frontend/src/components/Flow/FlowEditor.tsx
@@ -24,6 +24,7 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"
 
 import {
   type FlowDef_Input,
+  type FlowDetail,
   FlowsService,
   type MessageSpec,
   type NodeDef_Input,
@@ -55,6 +56,8 @@ import {
   flowsQueryOptions,
   nodeTypesQueryOptions,
   useAutosave,
+  useDiscardDraft,
+  usePublish,
 } from "./queries"
 import { useFlowSocket } from "./useFlowSocket"
 
@@ -184,7 +187,13 @@ function uniqueNodeId(existing: NodeDef_Input[], type: string): string {
   }
 }
 
-function FlowEditorInner({ flowName }: { flowName: string }) {
+function FlowEditorInner({
+  flowName,
+  onReload,
+}: {
+  flowName: string
+  onReload: () => void
+}) {
   const navigate = useNavigate()
   const queryClient = useQueryClient()
   const { showErrorToast } = useCustomToast()
@@ -195,7 +204,15 @@ function FlowEditorInner({ flowName }: { flowName: string }) {
   const { data: detail } = useSuspenseQuery(flowQueryOptions(flowName))
   const { data: nodeTypeInfo } = useQuery(nodeTypesQueryOptions())
 
-  const { save, flush, mutation: saving } = useAutosave(flowName)
+  const {
+    save,
+    flush,
+    conflict,
+    resolveConflict,
+    mutation: saving,
+  } = useAutosave(flowName)
+  const publish = usePublish(flowName)
+  const discard = useDiscardDraft(flowName)
 
   const [definitions, setDefinitions] = useState<NodeDef_Input[]>(
     () => detail.definition.nodes ?? [],
@@ -672,6 +689,17 @@ function FlowEditorInner({ flowName }: { flowName: string }) {
             flows={flows.data}
             active={flowName}
             saving={saving.isPending}
+            hasDraft={detail.has_draft ?? false}
+            publishing={publish.isPending || saving.isPending}
+            onPublish={async () => {
+              // Publish what was actually stored: the version only advances
+              // once the queued save has landed.
+              await flush()
+              const current = queryClient.getQueryData<FlowDetail>(
+                flowKeys.detail(flowName),
+              )
+              publish.mutate(current?.definition.version ?? 1)
+            }}
             onEditFlow={() => {
               setSelectedId(null)
               setFlowPanelOpen(true)
@@ -685,8 +713,9 @@ function FlowEditorInner({ flowName }: { flowName: string }) {
             issues={issues}
             running={runMutation.isPending}
             onAddNode={() => setPaletteOpen(true)}
-            onRun={() => {
-              flush()
+            onRun={async () => {
+              // Running executes what is stored, so the queued edit goes first.
+              await flush()
               runMutation.mutate()
             }}
             onFocusNode={focusNode}
@@ -718,11 +747,23 @@ function FlowEditorInner({ flowName }: { flowName: string }) {
           flush()
           save({ ...next, nodes: definitions })
         }}
-        onRename={(newName) => {
-          flush()
+        onRename={async (newName) => {
+          await flush()
           renameMutation.mutate(newName)
         }}
         onDelete={() => deleteMutation.mutate()}
+        hasDraft={detail.has_draft ?? false}
+        discarding={discard.isPending}
+        onDiscardDraft={() => {
+          discard.mutate(undefined, {
+            // The published document replaces what is on the canvas, and the
+            // version counter goes back with it.
+            onSuccess: () => {
+              setFlowPanelOpen(false)
+              onReload()
+            },
+          })
+        }}
         onClose={() => setFlowPanelOpen(false)}
       />
 
@@ -833,6 +874,42 @@ function FlowEditorInner({ flowName }: { flowName: string }) {
           </DialogFooter>
         </DialogContent>
       </Dialog>
+
+      {/*
+       * Not dismissable: until one version wins, every further save fails, so
+       * there is nothing useful to go back to.
+       */}
+      <Dialog open={conflict}>
+        <DialogContent
+          data-testid="save-conflict"
+          showCloseButton={false}
+          onEscapeKeyDown={(event) => event.preventDefault()}
+          onPointerDownOutside={(event) => event.preventDefault()}
+        >
+          <DialogHeader>
+            <DialogTitle>Someone else changed this flow</DialogTitle>
+            <DialogDescription>
+              Another editor saved <span className="font-mono">{flowName}</span>{" "}
+              while you were working on it. Load their version, or keep yours
+              and write over theirs.
+            </DialogDescription>
+          </DialogHeader>
+          <DialogFooter className="sm:justify-between">
+            <Button
+              variant="ghost"
+              onClick={() => {
+                void resolveConflict("theirs")
+                onReload()
+              }}
+            >
+              Load theirs
+            </Button>
+            <Button onClick={() => void resolveConflict("mine")}>
+              Keep mine
+            </Button>
+          </DialogFooter>
+        </DialogContent>
+      </Dialog>
     </>
   )
 }
@@ -857,6 +934,10 @@ export function FlowEditor({ flowName }: { flowName: string }) {
   const onAuthFailure = useCallback(() => {
     navigate({ to: "/login" })
   }, [navigate])
+  // Bumped when the canvas has to take the server's document over its own:
+  // local edits live in state seeded on mount, so a remount is the reset.
+  const [epoch, setEpoch] = useState(0)
+  const reload = useCallback(() => setEpoch((n) => n + 1), [])
 
   useFlowSocket(onAuthFailure)
 
@@ -871,7 +952,11 @@ export function FlowEditor({ flowName }: { flowName: string }) {
        * it is what makes `fitView` run once per flow: xyflow queues the fit on
        * mount and resolves it as soon as the nodes have been measured.
        */}
-      <FlowEditorInner key={flowName} flowName={flowName} />
+      <FlowEditorInner
+        key={`${flowName}:${epoch}`}
+        flowName={flowName}
+        onReload={reload}
+      />
     </ReactFlowProvider>
   )
 }
diff --git a/frontend/src/components/Flow/FlowPanel.tsx b/frontend/src/components/Flow/FlowPanel.tsx
index da54506..8ac214e 100644
--- a/frontend/src/components/Flow/FlowPanel.tsx
+++ b/frontend/src/components/Flow/FlowPanel.tsx
@@ -29,6 +29,9 @@ export function FlowPanel({
   onChange,
   onRename,
   onDelete,
+  hasDraft,
+  discarding,
+  onDiscardDraft,
   onClose,
 }: {
   open: boolean
@@ -38,10 +41,14 @@ export function FlowPanel({
   onChange: (next: FlowDef_Input) => void
   onRename: (newName: string) => void
   onDelete: () => void
+  hasDraft: boolean
+  discarding: boolean
+  onDiscardDraft: () => void
   onClose: () => void
 }) {
   const [name, setName] = useState(definition.name)
   const [confirmOpen, setConfirmOpen] = useState(false)
+  const [discardOpen, setDiscardOpen] = useState(false)
 
   const valid = NAME_PATTERN.test(name)
   const changed = name !== definition.name
@@ -117,6 +124,26 @@ export function FlowPanel({
                 : `${nodeCount} node${nodeCount === 1 ? "" : "s"}.`}
             </p>
           </div>
+
+          {hasDraft ? (
+            <div className="grid gap-2">
+              <span className={PANEL_SECTION}>Unpublished changes</span>
+              <p className="text-sm text-muted-foreground">
+                The engine is still running the last published version of this
+                flow.
+              </p>
+              <Button
+                variant="outline"
+                size="sm"
+                className="h-8 justify-self-start"
+                disabled={discarding}
+                onClick={() => setDiscardOpen(true)}
+                data-testid="discard-draft"
+              >
+                {discarding ? "Discarding…" : "Discard changes"}
+              </Button>
+            </div>
+          ) : null}
         </div>
       </SidePanel>
 
@@ -149,6 +176,34 @@ export function FlowPanel({
           </DialogFooter>
         </DialogContent>
       </Dialog>
+
+      <Dialog open={discardOpen} onOpenChange={setDiscardOpen}>
+        <DialogContent>
+          <DialogHeader>
+            <DialogTitle>Discard the unpublished changes?</DialogTitle>
+            <DialogDescription>
+              The canvas goes back to the version the engine is running. What
+              you edited since is dropped, though the flow store's git history
+              keeps it.
+            </DialogDescription>
+          </DialogHeader>
+          <DialogFooter>
+            <Button variant="outline" onClick={() => setDiscardOpen(false)}>
+              Keep editing
+            </Button>
+            <Button
+              variant="destructive"
+              onClick={() => {
+                setDiscardOpen(false)
+                onDiscardDraft()
+              }}
+              data-testid="confirm-discard-draft"
+            >
+              Discard changes
+            </Button>
+          </DialogFooter>
+        </DialogContent>
+      </Dialog>
     </>
   )
 }
diff --git a/frontend/src/components/Flow/FlowTabs.tsx b/frontend/src/components/Flow/FlowTabs.tsx
index 050342d..c986e39 100644
--- a/frontend/src/components/Flow/FlowTabs.tsx
+++ b/frontend/src/components/Flow/FlowTabs.tsx
@@ -114,11 +114,17 @@ export function FlowTabs({
   flows,
   active,
   saving,
+  hasDraft,
+  publishing,
+  onPublish,
   onEditFlow,
 }: {
   flows: FlowSummary[]
   active: string
   saving: boolean
+  hasDraft: boolean
+  publishing: boolean
+  onPublish: () => void
   onEditFlow: () => void
 }) {
   const [dialogOpen, setDialogOpen] = useState(false)
@@ -143,13 +149,18 @@ export function FlowTabs({
               to="/flows/$flowName"
               params={{ flowName: flow.name }}
               className={cn(
-                "shrink-0 snap-start rounded-full px-3 py-1.5 text-sm transition-colors",
+                "flex shrink-0 snap-start items-center gap-1.5 rounded-full px-3 py-1.5 text-sm transition-colors",
                 flow.name === active
                   ? "bg-accent text-accent-foreground"
                   : "text-muted-foreground hover:bg-accent/50",
               )}
             >
               {flow.title || flow.name}
+              {flow.has_draft ? (
+                <span className="size-1.5 shrink-0 rounded-full bg-primary">
+                  <span className="sr-only">Unpublished changes</span>
+                </span>
+              ) : null}
             </Link>
           ))}
         </div>
@@ -202,9 +213,24 @@ export function FlowTabs({
               ? "Reconnecting to the engine"
               : saving
                 ? "Saving"
-                : "All changes saved"}
+                : hasDraft
+                  ? "Saved — publish to put it live"
+                  : "All changes saved"}
           </TooltipContent>
         </Tooltip>
+
+        {hasDraft ? (
+          <Button
+            variant="outline"
+            size="sm"
+            className="shrink-0 rounded-full"
+            onClick={onPublish}
+            disabled={publishing}
+            data-testid="publish-flow"
+          >
+            {publishing ? "Publishing…" : "Publish"}
+          </Button>
+        ) : null}
       </motion.div>
 
       <NewFlowDialog open={dialogOpen} onOpenChange={setDialogOpen} />
diff --git a/frontend/src/components/Flow/queries.ts b/frontend/src/components/Flow/queries.ts
index c20946c..08145ba 100644
--- a/frontend/src/components/Flow/queries.ts
+++ b/frontend/src/components/Flow/queries.ts
@@ -3,9 +3,9 @@ import {
   useMutation,
   useQueryClient,
 } from "@tanstack/react-query"
-import { useCallback, useEffect, useRef } from "react"
+import { useCallback, useEffect, useRef, useState } from "react"
 
-import { type FlowDef_Input, FlowsService } from "@/client"
+import { ApiError, type FlowDef_Input, FlowsService } from "@/client"
 
 export const flowKeys = {
   all: ["flows"] as const,
@@ -45,55 +45,139 @@ export const messageHistoryQueryOptions = (name: string, message: string) => ({
 })
 
 const AUTOSAVE_DELAY = 800
+/** How long to wait for a save in flight before sending the next one. */
+const RETRY_DELAY = 100
+
+/** Publish the unpublished changes, which is what puts them on the engine. */
+export function usePublish(name: string) {
+  const queryClient = useQueryClient()
+  return useMutation({
+    mutationFn: (version: number) =>
+      FlowsService.publishFlow({ name, requestBody: { version } }),
+    onSuccess: (detail) => {
+      queryClient.setQueryData(flowKeys.detail(name), detail)
+      queryClient.invalidateQueries({ queryKey: flowKeys.all, exact: true })
+    },
+  })
+}
+
+/** Throw the unpublished changes away and go back to what is running. */
+export function useDiscardDraft(name: string) {
+  const queryClient = useQueryClient()
+  return useMutation({
+    mutationFn: () => FlowsService.discardDraft({ name }),
+    onSuccess: (detail) => {
+      queryClient.setQueryData(flowKeys.detail(name), detail)
+      queryClient.invalidateQueries({ queryKey: flowKeys.all, exact: true })
+    },
+  })
+}
 
 /**
  * Saves the flow a moment after the last edit, and immediately when the editor
  * needs the server to be current (closing a panel, switching flow, running).
  *
- * Identical documents are skipped server-side, so a quiet canvas writes nothing.
+ * Saving writes a draft — the engine keeps running the published version until
+ * someone publishes. Identical documents are skipped server-side, so a quiet
+ * canvas writes nothing.
+ *
+ * Every save carries the version it is based on. If another client saved in
+ * between, the server refuses rather than discarding their work, and `conflict`
+ * turns true for the editor to ask which version wins.
  */
 export function useAutosave(name: string): {
   save: (definition: FlowDef_Input) => void
-  flush: () => void
+  /** Send what is queued and resolve once the server has it. */
+  flush: () => Promise<void>
+  conflict: boolean
+  resolveConflict: (mode: "theirs" | "mine") => Promise<void>
   mutation: UseMutationResult<unknown, unknown, FlowDef_Input, unknown>
 } {
   const queryClient = useQueryClient()
   const pending = useRef<FlowDef_Input | null>(null)
   const timer = useRef<ReturnType<typeof setTimeout> | null>(null)
+  // The freshest version this client has seen. Local edits keep the document
+  // they were made against, so the version has to be stamped on at send time.
+  const version = useRef<number | null>(null)
+  const sent = useRef<FlowDef_Input | null>(null)
+  const inFlight = useRef(false)
+  const [conflict, setConflict] = useState(false)
 
   const mutation = useMutation({
     mutationFn: (definition: FlowDef_Input) =>
-      FlowsService.saveFlow({ name, requestBody: definition }),
+      FlowsService.saveFlow({
+        name,
+        requestBody: { ...definition, version: version.current ?? undefined },
+      }),
+    onMutate: (definition) => {
+      inFlight.current = true
+      sent.current = definition
+    },
+    onSettled: () => {
+      inFlight.current = false
+    },
     onSuccess: (detail) => {
+      version.current = detail.definition.version ?? null
       // Write the server's answer straight into the cache: invalidating would
       // pull the document back out from under edits still in flight.
       queryClient.setQueryData(flowKeys.detail(name), detail)
       queryClient.invalidateQueries({ queryKey: flowKeys.all, exact: true })
     },
+    onError: (error) => {
+      if (error instanceof ApiError && error.status === 409) setConflict(true)
+    },
   })
 
   // react-query hands back a new mutation object on every render, so flushing
   // has to hang off `mutate`, which is stable. Depending on the whole mutation
   // re-ran the effect below on every render, and its cleanup cancelled the
   // pending save before it ever fired.
-  const { mutate } = mutation
-  const flush = useCallback(() => {
+  const { mutate, mutateAsync } = mutation
+  const flush = useCallback(async (): Promise<void> => {
     if (timer.current) {
       clearTimeout(timer.current)
       timer.current = null
     }
     const definition = pending.current
-    pending.current = null
-    if (definition) {
-      mutate(definition)
+    if (!definition) return
+    // Two saves in flight at once would race for the same version, and the
+    // loser would look like someone else's edit.
+    if (inFlight.current) {
+      await new Promise((resolve) => setTimeout(resolve, RETRY_DELAY))
+      return flush()
     }
-  }, [mutate])
+    pending.current = null
+    // A rejection here is a conflict, which the dialog handles; callers waiting
+    // on the flush only need to know the attempt is over.
+    await mutateAsync(definition).catch(() => undefined)
+  }, [mutateAsync])
+
+  const resolveConflict = useCallback(
+    async (mode: "theirs" | "mine") => {
+      setConflict(false)
+      if (mode === "theirs") {
+        pending.current = null
+        sent.current = null
+        version.current = null
+        await queryClient.invalidateQueries({
+          queryKey: flowKeys.detail(name),
+        })
+        return
+      }
+      const current = await FlowsService.readFlow({ name })
+      version.current = current.definition.version ?? null
+      const definition = pending.current ?? sent.current
+      pending.current = null
+      if (definition) mutate(definition)
+    },
+    [mutate, name, queryClient],
+  )
 
   const save = useCallback(
     (definition: FlowDef_Input) => {
       pending.current = definition
       if (timer.current) clearTimeout(timer.current)
-      timer.current = setTimeout(flush, AUTOSAVE_DELAY)
+      timer.current = setTimeout(() => void flush(), AUTOSAVE_DELAY)
     },
     [flush],
   )
@@ -101,15 +185,15 @@ export function useAutosave(name: string): {
   // Leaving the tab is the last chance to persist what is still queued.
   useEffect(() => {
     const onHidden = () => {
-      if (document.visibilityState === "hidden") flush()
+      if (document.visibilityState === "hidden") void flush()
     }
     document.addEventListener("visibilitychange", onHidden)
     return () => {
       document.removeEventListener("visibilitychange", onHidden)
       // Unmounting is a flow switch, not a reason to drop a queued edit.
-      flush()
+      void flush()
     }
   }, [flush])
 
-  return { save, flush, mutation }
+  return { save, flush, conflict, resolveConflict, mutation }
 }
diff --git a/frontend/src/components/Flow/useFlowSocket.ts b/frontend/src/components/Flow/useFlowSocket.ts
index ac37171..b2e11b2 100644
--- a/frontend/src/components/Flow/useFlowSocket.ts
+++ b/frontend/src/components/Flow/useFlowSocket.ts
@@ -1,7 +1,9 @@
+import { useQueryClient } from "@tanstack/react-query"
 import { useEffect, useRef } from "react"
 
 import { OpenAPI } from "@/client"
 import { liveStore } from "./liveStore"
+import { flowKeys } from "./queries"
 
 const RECONNECT_MIN = 1000
 const RECONNECT_MAX = 30000
@@ -42,6 +44,7 @@ export function useFlowSocket(onAuthFailure?: () => void): void {
   const retry = useRef(RECONNECT_MIN)
   const timer = useRef<ReturnType<typeof setTimeout> | null>(null)
   const closed = useRef(false)
+  const queryClient = useQueryClient()
 
   useEffect(() => {
     closed.current = false
@@ -87,6 +90,9 @@ export function useFlowSocket(onAuthFailure?: () => void): void {
             break
           case "pipeline_rebuilt":
             liveStore.setStatuses(message.nodes)
+            // Someone published, here or in another tab: the draft markers on
+            // the flow chips are stale until the list is fetched again.
+            queryClient.invalidateQueries({ queryKey: flowKeys.all })
             break
         }
       }
@@ -111,5 +117,5 @@ export function useFlowSocket(onAuthFailure?: () => void): void {
       socket.current?.close()
       liveStore.setConnected(false)
     }
-  }, [onAuthFailure])
+  }, [onAuthFailure, queryClient])
 }
diff --git a/frontend/tests/drafts.spec.ts b/frontend/tests/drafts.spec.ts
new file mode 100644
index 0000000..4a26dab
--- /dev/null
+++ b/frontend/tests/drafts.spec.ts
@@ -0,0 +1,94 @@
+import { expect, type Page, test } from "@playwright/test"
+
+/**
+ * Editing writes a draft; only publishing hands it to the engine. The two
+ * things worth proving here are that the engine ignores a draft until it is
+ * published, and that a second client cannot quietly overwrite the first.
+ */
+
+const flowName = `test_draft_${Date.now().toString(36)}`
+
+test.use({ storageState: "playwright/.auth/user.json" })
+
+test.describe.configure({ mode: "serial" })
+
+const apiUrl = process.env.VITE_API_URL || "http://api.localhost"
+
+async function api(page: Page, path: string, init: Record<string, unknown> = {}) {
+  const token = await page.evaluate(() => localStorage.getItem("access_token"))
+  return page.request.fetch(`${apiUrl}/api/v1${path}`, {
+    ...init,
+    headers: { Authorization: `Bearer ${token}` },
+  })
+}
+
+test("a draft stays off the engine until it is published", async ({ page }) => {
+  await page.goto("/flows")
+  await page.getByRole("button", { name: "New flow" }).click()
+  await page.getByTestId("flow-name-input").fill(flowName)
+  await page.getByRole("button", { name: "Create flow" }).click()
+  await page.waitForURL(`/flows/${flowName}`)
+
+  await page.getByTestId("add-node").click()
+  await page
+    .getByRole("option", { name: /function/i })
+    .first()
+    .click()
+  await expect(page.locator(".react-flow__node")).toHaveCount(1)
+  // Adding a node opens its panel, and the floating chrome steps aside for it.
+  await page.keyboard.press("Escape")
+
+  await expect(page.getByTestId("publish-flow")).toBeVisible()
+
+  // Nothing of this flow is loaded while it is only a draft.
+  await page.waitForTimeout(1500)
+  const before = await (await api(page, `/flows/${flowName}/state`)).json()
+  expect(before.nodes).toHaveLength(0)
+
+  await page.getByTestId("publish-flow").click()
+  await expect(page.getByTestId("publish-flow")).toBeHidden()
+
+  const after = await (await api(page, `/flows/${flowName}/state`)).json()
+  expect(after.nodes.length).toBeGreaterThan(0)
+})
+
+test("a save against a version someone else moved on from is refused", async ({
+  page,
+}) => {
+  await page.goto(`/flows/${flowName}`)
+  await page.waitForSelector(".react-flow__node")
+
+  const detail = await (await api(page, `/flows/${flowName}`)).json()
+
+  // Stand in for a second client that saved first.
+  const first = await api(page, `/flows/${flowName}`, {
+    method: "PUT",
+    data: { ...detail.definition, title: "Theirs" },
+  })
+  expect(first.ok()).toBeTruthy()
+
+  const stale = await api(page, `/flows/${flowName}`, {
+    method: "PUT",
+    data: { ...detail.definition, title: "Mine" },
+  })
+  expect(stale.status()).toBe(409)
+  expect((await stale.json()).detail.current_version).toBe(
+    detail.definition.version + 1,
+  )
+})
+
+test("discarding a draft goes back to the published flow", async ({ page }) => {
+  await page.goto(`/flows/${flowName}`)
+  await page.waitForSelector(".react-flow__node")
+
+  await page.getByTestId("edit-flow").click()
+  await page.getByTestId("discard-draft").click()
+  await page.getByTestId("confirm-discard-draft").click()
+
+  await expect(page.getByTestId("publish-flow")).toBeHidden()
+  const detail = await (await api(page, `/flows/${flowName}`)).json()
+  expect(detail.has_draft).toBe(false)
+  expect(detail.definition.title).not.toBe("Theirs")
+
+  await api(page, `/flows/${flowName}`, { method: "DELETE" })
+})