Flow delete offers to delete its runs, history included

DELETE /runs?flow= removes every Run row of the flow but left its flow_run
rows, so a deleted flow kept appearing in the history. history=true now
drops those too, and the flow delete dialogs (editor and bulk) send it when
"Also delete its runs" is ticked.

backend/fluksio/api/routes/flows.py

@@ -556,7 +556,7 @@ async def delete_flow(
     # 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.
+    # is what clears them, and with `history=true` its cascades too.
     await run_in_threadpool(controller.forget_flow, name)
     await controller.reload_flow(name)
     return Message(message=f"Deleted flow '{name}'")

backend/fluksio/api/routes/runs.py

@@ -29,7 +29,7 @@ from fluksio.flow.events import event_bus
 from fluksio.flow.messages import requalify
 from fluksio.flow.runs import OUTPUT_CAP, RunRejected, RunService, new_run_id
 from fluksio.flow.store import FlowNotFound
-from fluksio.models import Message, Run, RunArtifact, RunMetric, RunNode
+from fluksio.models import FlowRun, Message, Run, RunArtifact, RunMetric, RunNode
 
 router = APIRouter(
     prefix="/runs", tags=["runs"], dependencies=[Depends(get_current_user)]
@@ -881,6 +881,7 @@ def delete_runs(
     session: SessionDep,
     user: CurrentUser,
     flow: Annotated[str, Query(min_length=1)],
+    history: bool = False,
 ) -> Any:
     """Every run of one flow at once.
 
@@ -888,6 +889,10 @@ def delete_runs(
     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.
+
+    ``history`` also drops the flow's rows in the run history (``flow_run``),
+    its cascades included. That is for a flow being deleted, which would
+    otherwise stay in the history until the retention window passes.
     """
     with writing(session):
         live = _live_run(session, flow)
@@ -906,6 +911,8 @@ def delete_runs(
             select(func.count()).select_from(Run).where(col(Run.flow) == flow)
         ).one()
         _drop_runs(session, chosen)
+        if history:
+            session.execute(delete(FlowRun).where(col(FlowRun.flow) == flow))
         session.commit()
     event_bus.publish(
         {

backend/tests/api/routes/test_runs.py

@@ -33,7 +33,7 @@ from fluksio.flow.runs import (
 )
 from fluksio.flow.schemas import FlowDef, FlowInput, NodeDef
 from fluksio.flow.store import FlowStore
-from fluksio.models import Run, RunArtifact, RunMetric, RunNode
+from fluksio.models import FlowRun, Run, RunArtifact, RunMetric, RunNode
 
 
 def test_a_run_cache_finds_what_an_earlier_run_recorded(tmp_path):
@@ -1041,6 +1041,33 @@ def test_deleting_a_flows_runs_clears_that_flow_only(client, superuser_token_hea
         session.commit()
 
 
+def test_history_also_clears_a_flows_cascades(client, superuser_token_headers):
+    """What deleting a flow together with its runs asks for."""
+    made = datetime.now(UTC)
+    url = f"{settings.API_V1_STR}/runs"
+    with Session(db_engine) as session:
+        session.add(Run(id="hist-1", flow="gone", status="ok", created_at=made))
+        session.add(FlowRun(id="hist-1", flow="gone", source="run", started_at=made))
+        session.add(FlowRun(id="hist-2", flow="gone", started_at=made))
+        session.add(FlowRun(id="hist-3", flow="other", started_at=made))
+        session.commit()
+
+    plain = client.delete(url, params={"flow": "gone"}, headers=superuser_token_headers)
+    assert plain.status_code == 200
+    with Session(db_engine) as session:
+        assert session.get(FlowRun, "hist-2") is not None
+
+    answer = client.delete(
+        url, params={"flow": "gone", "history": True}, headers=superuser_token_headers
+    )
+    assert answer.status_code == 200
+    with Session(db_engine) as session:
+        left = session.exec(select(FlowRun).where(col(FlowRun.flow) == "gone")).all()
+        assert left == []
+        session.delete(session.get(FlowRun, "hist-3"))
+        session.commit()
+
+
 # -----------------------------------------------------------------------------
 # A cached node's curve
 #