Let Postgres fold the rollups, and say when a run list was cut short

/observability/timeseries and /flows read every metric_minute row in the window
and folded them in Python, so the 7d preset pulled a week of rows on each 30 s
poll. date_bin() does the binning now — the row count drops to the slices asked
for, and to flows × 60 for the sparklines. A window of zero hours used to divide
by nothing and answer 500; windows are clamped to an hour at the low end and to
the retention period at the high end, past which there is nothing to find.

/observability/runs returns {data, count} rather than a bare list, so a minute
busier than the 200-row cap says so instead of quietly showing its newest 200.
The count is only queried when the page comes back full, which keeps the poll
from handing back what the fold just saved.

failures_24h leaves the summary — the Home tile counts errors over the selected
window from the rollups, and nothing had read the field since.

Deleting a flow now takes its Run rows and their nodes, metrics and artifacts
with it. This lives in the route rather than in forget_flow because renaming a
flow calls that too, and a rename must keep its history. The observability
rollups stay: they are the record of what ran, and retention already prunes them.

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

backend/tests/api/routes/test_flows.py

@@ -1,6 +1,11 @@
+from datetime import UTC, datetime
+
 from fastapi.testclient import TestClient
+from sqlalchemy import func
+from sqlmodel import Session, select
 
 from app.core.config import settings
+from app.models import Run, RunArtifact, RunMetric, RunNode
 
 PREFIX = f"{settings.API_V1_STR}/flows"
 
@@ -273,9 +278,14 @@ def test_unconnected_input_is_surfaced(
 
 
 def test_delete_flow(
-    client: TestClient, superuser_token_headers: dict[str, str]
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
 ) -> None:
     client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
+    db.add(Run(id="run-1", flow="demo", created_at=datetime.now(UTC)))
+    db.add(RunNode(run_id="run-1", node="sensor"))
+    db.add(RunMetric(run_id="run-1", name="loss", step=-1))
+    db.add(RunArtifact(run_id="run-1", name="model.pt"))
+    db.commit()
 
     assert (
         client.delete(f"{PREFIX}/demo", headers=superuser_token_headers).status_code
@@ -285,6 +295,11 @@ 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.
+    db.expire_all()
+    for model in (Run, RunNode, RunMetric, RunArtifact):
+        assert db.exec(select(func.count()).select_from(model)).one() == 0
+
 
 def test_node_types_are_listed(
     client: TestClient, superuser_token_headers: dict[str, str]

backend/tests/api/routes/test_observability.py

@@ -76,6 +76,9 @@ def test_the_summary_answers_even_when_degraded(
         "invalid",
     }
     assert "error" in body["nodes"]
+    # A windowed count comes from /flows?hours= now, which is what the tile
+    # that used to read this actually sums.
+    assert "failures_24h" not in body
 
 
 def test_a_flow_that_cannot_run_makes_the_summary_degraded(
@@ -140,7 +143,7 @@ def test_the_history_reads_back(
     assert row["last_error_ts"] is not None
 
     runs = client.get(f"{PREFIX}/runs", headers=superuser_token_headers).json()
-    assert any(run["id"] == "9-0" and run["status"] == "ok" for run in runs)
+    assert any(run["id"] == "9-0" and run["status"] == "ok" for run in runs["data"])
 
     failures = client.get(f"{PREFIX}/events", headers=superuser_token_headers).json()
     assert any("Traceback" in event["detail"] for event in failures)
@@ -156,6 +159,85 @@ def test_the_history_reads_back(
     assert isinstance(dead.json(), list)
 
 
+def test_the_timeseries_folds_into_the_requested_bucket(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    """The fold moved into SQL, so the slices still have to line up as before.
+
+    Slots are aligned to the epoch, which is what the old modulo did, and a
+    slice only exists where a row does.
+    """
+    flow = "bucket-fold-test"
+    # The start of the current quarter hour, half an hour back so both slices
+    # sit inside a one hour window.
+    now = datetime.now(timezone.utc).timestamp()
+    first = datetime.fromtimestamp(now // 900 * 900 - 1800, timezone.utc)
+    for offset, executions in ((0, 1), (2, 2), (5, 4), (15, 8)):
+        db.add(
+            MetricBucket(
+                flow=flow,
+                node=f"{flow}.calc",
+                bucket=first + timedelta(minutes=offset),
+                executions=executions,
+            )
+        )
+    db.commit()
+
+    points = client.get(
+        f"{PREFIX}/timeseries",
+        headers=superuser_token_headers,
+        params={"flow": flow, "hours": 1, "bucket_s": 900},
+    ).json()
+
+    assert len(points) == 2
+    assert points[0]["executions"] == 1 + 2 + 4
+    assert points[1]["executions"] == 8
+    assert points[0]["ts"] % 900 == 0
+
+
+def test_a_zero_hour_window_is_still_an_hour(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    """`hours=0` divided by nothing and answered 500."""
+    for path in ("timeseries", "flows"):
+        response = client.get(
+            f"{PREFIX}/{path}", headers=superuser_token_headers, params={"hours": 0}
+        )
+        assert response.status_code == 200
+
+
+def test_the_runs_page_carries_its_total(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    """A full page says how much it left behind."""
+    minute = datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta(
+        hours=5
+    )
+    for index in range(3):
+        db.add(
+            FlowRun(
+                id=f"paged-{index}",
+                flow=FLOW,
+                started_at=minute + timedelta(seconds=index),
+                status="ok",
+            )
+        )
+    db.commit()
+
+    body = client.get(
+        f"{PREFIX}/runs",
+        headers=superuser_token_headers,
+        params={
+            "since": minute.isoformat(),
+            "until": (minute + timedelta(minutes=1)).isoformat(),
+            "limit": 2,
+        },
+    ).json()
+
+    assert len(body["data"]) == 2
+    assert body["count"] == 3
+
+
 def test_runs_narrow_to_one_minute(
     client: TestClient, superuser_token_headers: dict[str, str], db: Session
 ) -> None:
@@ -185,7 +267,7 @@ def test_runs_narrow_to_one_minute(
 
     # The upper bound is exclusive, so the run starting the next minute is not
     # in this one.
-    assert [run["id"] for run in runs] == ["minute-in"]
+    assert [run["id"] for run in runs["data"]] == ["minute-in"]
 
 
 def test_events_narrow_to_one_minute(