Keep a flow's runs when the flow goes

A run record is a record of what ran, so deleting a flow no longer sweeps its
history: `_forget_runs` is gone from the flow-delete path, and `DELETE /runs/{id}`
is the only thing that removes a run one at a time. The in-flight guard stays —
that is about work, not history.

`DELETE /runs?flow=` is the counterpart to the list's flow filter and what a
reseed needs, sharing the four statements with the single-run delete and
refusing the same way while a run of that flow is still going.

A run whose flow is gone reads as one: `useFlowInputs` reports the 404 rather
than an empty declaration set, so the run page says "flow deleted", explains it,
and disables Retry, which the route would refuse anyway.

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

backend/fluksio/api/routes/runs.py

@@ -28,7 +28,7 @@ from fluksio.flow.events import event_bus
 from fluksio.flow.messages import requalify
 from fluksio.flow.runs import RunRejected, RunService, new_run_id
 from fluksio.flow.store import FlowNotFound
-from fluksio.models import Run, RunArtifact, RunMetric, RunNode
+from fluksio.models import Message, Run, RunArtifact, RunMetric, RunNode
 
 router = APIRouter(
     prefix="/runs", tags=["runs"], dependencies=[Depends(get_current_user)]
@@ -703,27 +703,88 @@ async def retry_run(run_id: str, request: Request, user: CurrentUser) -> Any:
         raise HTTPException(status_code=status, detail=detail) from exc
 
 
-@router.delete("/{run_id}", status_code=204)
-def delete_run(run_id: str, session: SessionDep, user: CurrentUser) -> Response:
-    """Forget a run and everything hanging off it.
+def _drop_runs(session: Session, chosen: Any) -> None:
+    """A selection of runs and everything hanging off them.
 
-    The same four statements ``_forget_runs`` uses when a flow goes: the run
-    tables carry a plain string ``run_id`` and no foreign key, so nothing
-    cascades on its own. ``flow_run``, ``metric_minute`` and ``engine_event``
-    stay — they are the observability record and are pruned on their own window.
-
-    A live run is refused rather than raced: the driver writes its nodes back
-    when it finishes, and those rows would arrive for a run that no longer
-    exists. Cancel it first.
+    The run tables carry a plain string ``run_id`` and no foreign key, so
+    nothing cascades on its own and these four statements are the whole of it.
+    ``flow_run``, ``metric_minute`` and ``engine_event`` stay — they are the
+    observability record and are pruned on their own window.
 
     Two things this costs, both deliberate. ``RunNode.outputs`` *is* the stage
-    cache, so a later run loses hits this one would have served. And a node
-    restored from this run points here through ``cached_from`` — ``_series``
+    cache, so a later run loses hits these would have served. And a node
+    restored from one of them points here through ``cached_from`` — ``_series``
     already reads a missing source as an empty curve, which is what ``NO_CURVE``
     explains on the screen. The artifact bytes need no help: ``sweep_artifacts``
     keeps whatever a ``run_artifact`` row or a live message still names, so
     dropping the rows is enough and the hourly sweep reclaims the blobs.
     """
+    session.execute(delete(RunNode).where(col(RunNode.run_id).in_(chosen)))
+    session.execute(delete(RunMetric).where(col(RunMetric.run_id).in_(chosen)))
+    session.execute(delete(RunArtifact).where(col(RunArtifact.run_id).in_(chosen)))
+    session.execute(delete(Run).where(col(Run.id).in_(chosen)))
+
+
+def _live_run(session: Session, flow: str) -> str | None:
+    """A run of this flow the driver is still writing to, if there is one."""
+    return session.exec(
+        select(Run.id).where(
+            col(Run.flow) == flow, col(Run.status).in_(("running", "queued"))
+        )
+    ).first()
+
+
+@router.delete("", response_model=Message)
+def delete_runs(
+    session: SessionDep,
+    user: CurrentUser,
+    flow: Annotated[str, Query(min_length=1)],
+) -> Any:
+    """Every run of one flow at once.
+
+    The counterpart to the list's ``flow`` filter, and what a reseed needs:
+    deleting a flow leaves its runs standing on purpose, so this is the only
+    way to clear a history without a call per run. ``flow`` is required —
+    there is no "delete every run of everything" here, by design.
+    """
+    with writing(session):
+        live = _live_run(session, flow)
+        if live is not None:
+            raise HTTPException(
+                status_code=409,
+                detail=(
+                    f"Flow '{flow}' has a run in progress ({live}). Cancel it, or "
+                    "wait for it to finish, before deleting its runs."
+                ),
+            )
+        # Counted before the delete, and a subquery rather than a materialised
+        # list of ids on the way in: a demo can hold thousands.
+        chosen = select(col(Run.id)).where(col(Run.flow) == flow)
+        gone = session.exec(
+            select(func.count()).select_from(Run).where(col(Run.flow) == flow)
+        ).one()
+        _drop_runs(session, chosen)
+        session.commit()
+    event_bus.publish(
+        {
+            "type": "audit",
+            "action": f"deleted {gone} run(s)",
+            "flow": flow,
+            "user": user.email,
+            "ts": time.time(),
+        }
+    )
+    return Message(message=f"Deleted {gone} run(s) of '{flow}'")
+
+
+@router.delete("/{run_id}", status_code=204)
+def delete_run(run_id: str, session: SessionDep, user: CurrentUser) -> Response:
+    """Forget a run and everything hanging off it.
+
+    A live run is refused rather than raced: the driver writes its nodes back
+    when it finishes, and those rows would arrive for a run that no longer
+    exists. Cancel it first.
+    """
     # Reads the run, then decides whether to delete it — so it takes the write
     # lock up front rather than upgrading and losing to whichever flush
     # committed in between.
@@ -740,10 +801,7 @@ def delete_run(run_id: str, session: SessionDep, user: CurrentUser) -> Response:
                 ),
             )
         flow = run.flow
-        session.execute(delete(RunNode).where(col(RunNode.run_id) == run_id))
-        session.execute(delete(RunMetric).where(col(RunMetric.run_id) == run_id))
-        session.execute(delete(RunArtifact).where(col(RunArtifact.run_id) == run_id))
-        session.execute(delete(Run).where(col(Run.id) == run_id))
+        _drop_runs(session, [run_id])
         session.commit()
     event_bus.publish(
         {
@@ -792,8 +850,8 @@ def _series(session: Session, run_id: str, name: str = "") -> list[RunMetric]:
         for node_row in restored:
             source_flow = sources.get(node_row.cached_from)
             if source_flow is None:
-                # The run it came from is gone — deleted with its flow. The
-                # outputs are still on this run; the curve is not recoverable.
+                # The run it came from was deleted. The outputs are still on
+                # this run; the curve is not recoverable.
                 continue
             key = (
                 node_row.cached_from,