Flows and dashboards are created by POST under an issued or superuser-pinned id; PUT only updates; deleted flows leave a tombstone with their key; the controller stores by id and runs by key, hooks live under /hooks/<flow id>; a key with state or queued work behind it is refused until cleared

backend/tests/flow/test_queue.py

@@ -680,3 +680,28 @@ def test_done_markers_do_not_grow_without_bound():
     # The newest are what redelivery would ask about.
     assert queue.was_done(f"entry-{queue_module.DONE_MARKERS + 499}", "f.node")
     assert not queue.was_done("entry-0", "f.node")
+
+
+def test_clearing_a_flow_drops_its_work_and_nothing_else():
+    """A new flow under the same key must never run what the old one queued."""
+    queue = MemoryWorkQueue()
+    queue.add(WorkItem(kind="cascade", node="gone.a", flow="gone"))
+    # Another flow's work carrying one of its messages, and a run, which is
+    # filed under a flow's id rather than its key: neither is the flow's.
+    queue.add(
+        WorkItem(kind="cascade", node="kept.a", flow="kept", outputs={"gone.x": 1})
+    )
+    queue.add(WorkItem(kind="run", node="", flow="gone", run_id="r1"))
+    queue.add_delayed(WorkItem(kind="flush", node="gone.b", flow="gone"), 9e9)
+    queue.add_delayed(WorkItem(kind="flush", node="kept.b", flow="kept"), 9e9)
+    queue.park("gone", WorkItem(kind="cascade", node="gone.c", flow="gone"))
+    assert len(queue.live_items()) == 6
+
+    queue.clear_flow("gone")
+
+    assert sorted((i.kind, i.node) for i in queue.live_items()) == [
+        ("cascade", "kept.a"),
+        ("flush", "kept.b"),
+        ("run", ""),
+    ]
+    assert queue.next_due() == 9e9