Act on a run, and read Home top-down

Runs: a run can now be deleted (DELETE /runs/{id}, cancelling a live one
first), exported as csv from the screen's own filters, and its flow label
opens the flow. Its "Parameters" panel became "Inputs" and lists every
input the flow declares, marking the ones that took the flow's own value
rather than the run's — the comparison table resolves the same defaults
instead of printing "unset".

Home reads brain, dashboards, health, flows: the mosaic is one full-width
scrolling strip, and the flows list and the flow-activity rollups merged
into a single left-joined table so a flow's state and its numbers sit on
one row.

Charts take a drag to narrow the x window and a double click or tap to
come back out. UplotChart holds the scale and passes resetScales:false
while a window is held, which is what the old comment said made this
impossible.

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

backend/fluksio/api/routes/runs.py

@@ -8,20 +8,22 @@ or to listen on the flow socket, which carries its start and finish.
 import csv
 import io
 import json
+import time
 from collections.abc import Iterator
 from datetime import UTC, datetime
 from itertools import groupby
 from typing import Any, Literal
 
-from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi import APIRouter, Depends, HTTPException, Request, Response
 from fastapi.concurrency import run_in_threadpool
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, Field, model_validator
-from sqlalchemy import func
+from sqlalchemy import delete, func
 from sqlalchemy import select as sa_select
 from sqlmodel import Session, col, select
 
 from fluksio.api.deps import CurrentUser, SessionDep, get_current_user
+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
@@ -613,6 +615,56 @@ async def cancel_run(run_id: str, request: Request, session: SessionDep) -> Any:
     return run
 
 
+@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.
+
+    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.
+
+    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``
+    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.
+    """
+    run = session.get(Run, run_id)
+    if run is None:
+        raise HTTPException(status_code=404, detail="No such run")
+    if run.status in ("running", "queued"):
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                f"Run {run_id} is {run.status}. Cancel it, or wait for it to "
+                "finish, before deleting it."
+            ),
+        )
+    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))
+    session.commit()
+    event_bus.publish(
+        {
+            "type": "audit",
+            "action": f"deleted run {run_id}",
+            "flow": flow,
+            "user": user.email,
+            "ts": time.time(),
+        }
+    )
+    return Response(status_code=204)
+
+
 def _series(session: Session, run_id: str, name: str = "") -> list[RunMetric]:
     """A run's numbers, including the ones a cached node points at.