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.

backend/fluksio/api/routes/flows.py

@@ -16,7 +16,6 @@ from fastapi import (
 from fastapi.concurrency import run_in_threadpool
 from jwt.exceptions import InvalidTokenError
 from pydantic import BaseModel
-from sqlalchemy import delete
 from sqlmodel import Session, col, select
 
 from fluksio.api.deps import (
@@ -59,7 +58,7 @@ from fluksio.flow.store import (
     LibNotFound,
     StaleVersion,
 )
-from fluksio.models import Flavor, Message, Run, RunArtifact, RunMetric, RunNode
+from fluksio.models import Flavor, Message, Run
 
 router = APIRouter(
     prefix="/flows", tags=["flows"], dependencies=[Depends(get_current_user)]
@@ -242,25 +241,6 @@ def _audit(action: str, flow: str, user: CurrentUser) -> None:
     )
 
 
-def _forget_runs(session: Session, flow: str) -> None:
-    """A deleted flow's runs, and everything hanging off them.
-
-    Here rather than in ``FlowController.forget_flow`` because renaming a flow
-    calls that too, and a rename must keep its experiment history.
-
-    Only the run tables: ``flow_run``, ``metric_minute`` and ``engine_event``
-    are the observability rollups, deliberately kept as a record of what ran
-    and already pruned at OBS_RETENTION_DAYS.
-    """
-    # A subquery, not a materialised list of ids: a demo can hold thousands.
-    runs = select(col(Run.id)).where(col(Run.flow) == flow)
-    session.execute(delete(RunNode).where(col(RunNode.run_id).in_(runs)))
-    session.execute(delete(RunMetric).where(col(RunMetric.run_id).in_(runs)))
-    session.execute(delete(RunArtifact).where(col(RunArtifact.run_id).in_(runs)))
-    session.execute(delete(Run).where(col(Run.flow) == flow))
-    session.commit()
-
-
 def _source_ref(definition: FlowDef, node_id: str) -> str | None:
     """The library source this node runs, if it is a shared one."""
     node = next((n for n in definition.nodes if n.id == node_id), None)
@@ -483,7 +463,7 @@ async def discard_draft(name: str, controller: FlowControllerDep) -> Any:
 async def delete_flow(
     name: str, controller: FlowControllerDep, user: CurrentUser, session: SessionDep
 ) -> Any:
-    """Delete a flow and everything in it."""
+    """Delete a flow and everything in it, except the record of what it ran."""
     live = session.exec(
         select(Run.id).where(
             col(Run.flow) == name, col(Run.status).in_(("running", "queued"))
@@ -503,8 +483,10 @@ async def delete_flow(
         raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
     _audit("deleted", name, user)
     # Its files are gone; its values and queued work would otherwise linger.
+    # Its runs stay: a run record is a record of what ran, and surviving the
+    # flow it belonged to is the point of keeping one. `DELETE /runs?flow=`
+    # is what clears them.
     await run_in_threadpool(controller.forget_flow, name)
-    await run_in_threadpool(_forget_runs, session, name)
     await controller.reload_flow(name)
     return Message(message=f"Deleted flow '{name}'")
 

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,

backend/tests/api/routes/test_flows.py

@@ -299,10 +299,17 @@ def test_delete_flow(
         client.get(f"{PREFIX}/demo", headers=superuser_token_headers).status_code == 404
     )
 
-    # Deleting the flow takes its runs with it, so a reseeded demo starts clean.
+    # The run outlives the flow: a run record is a record of what ran, and
+    # `DELETE /runs?flow=` is the only thing that clears one.
     db.expire_all()
     for model in (Run, RunNode, RunMetric, RunArtifact):
-        assert db.exec(select(func.count()).select_from(model)).one() == 0
+        assert db.exec(select(func.count()).select_from(model)).one() == 1
+    assert (
+        client.get(
+            f"{settings.API_V1_STR}/runs/run-1", headers=superuser_token_headers
+        ).status_code
+        == 200
+    )
 
 
 def test_delete_flow_is_refused_while_a_run_is_live(

backend/tests/api/routes/test_runs.py

@@ -774,6 +774,49 @@ def test_a_running_run_is_refused_rather_than_raced(client, superuser_token_head
                 session.commit()
 
 
+def test_deleting_a_flows_runs_clears_that_flow_only(client, superuser_token_headers):
+    """The counterpart to the list's `flow` filter — what a reseed needs.
+
+    Deleting the flow itself leaves the history standing, so this is the tool
+    that clears one. It refuses while a run of that flow is still going, for
+    the same reason deleting a single run does.
+    """
+    made = datetime.now(UTC)
+    url = f"{settings.API_V1_STR}/runs"
+    with Session(db_engine) as session:
+        session.add(Run(id="bulk-1", flow="swept", status="ok", created_at=made))
+        session.add(RunNode(run_id="bulk-1", node="swept.a", status="ok"))
+        session.add(RunMetric(run_id="bulk-1", name="swept.loss", step=0, value=1.0))
+        session.add(Run(id="bulk-2", flow="swept", status="running", created_at=made))
+        session.add(Run(id="keep-1", flow="other", status="ok", created_at=made))
+        session.commit()
+
+    busy = client.delete(url, params={"flow": "swept"}, headers=superuser_token_headers)
+    assert busy.status_code == 409
+    assert "bulk-2" in busy.json()["detail"]
+
+    with Session(db_engine) as session:
+        session.get(Run, "bulk-2").status = "cancelled"
+        session.commit()
+
+    answer = client.delete(
+        url, params={"flow": "swept"}, headers=superuser_token_headers
+    )
+    assert answer.status_code == 200
+    assert "2" in answer.json()["message"]
+
+    with Session(db_engine) as session:
+        assert session.exec(select(Run).where(col(Run.flow) == "swept")).all() == []
+        for table in (RunNode, RunMetric):
+            assert (
+                session.exec(select(table).where(col(table.run_id) == "bulk-1")).all()
+                == []
+            )
+        assert session.get(Run, "keep-1") is not None
+        session.delete(session.get(Run, "keep-1"))
+        session.commit()
+
+
 # -----------------------------------------------------------------------------
 # A cached node's curve
 #
@@ -847,7 +890,7 @@ def test_a_cached_node_answers_with_the_curve_it_was_restored_from(
 def test_a_curve_whose_run_is_gone_is_empty_rather_than_an_error(
     client, superuser_token_headers
 ):
-    """Deleting a flow deletes its runs; what pointed at one is left holding it."""
+    """Delete the run a cache hit points at and the reusing run holds nothing."""
     with Session(db_engine) as session:
         session.delete(session.get(Run, "src-1"))
         session.commit()