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/app/api/routes/flows.py

@@ -15,11 +15,13 @@ from fastapi import (
 from fastapi.concurrency import run_in_threadpool
 from jwt.exceptions import InvalidTokenError
 from pydantic import BaseModel
-from sqlmodel import Session
+from sqlalchemy import delete
+from sqlmodel import Session, col, select
 
 from app.api.deps import (
     CurrentUser,
     FlowControllerDep,
+    SessionDep,
     decode_token,
     get_current_user,
     user_from_token,
@@ -55,7 +57,7 @@ from app.flow.store import (
     LibNotFound,
     StaleVersion,
 )
-from app.models import Message
+from app.models import Message, Run, RunArtifact, RunMetric, RunNode
 
 router = APIRouter(
     prefix="/flows", tags=["flows"], dependencies=[Depends(get_current_user)]
@@ -209,6 +211,25 @@ 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)
@@ -400,7 +421,7 @@ async def discard_draft(name: str, controller: FlowControllerDep) -> Any:
 
 @router.delete("/{name}", response_model=Message)
 async def delete_flow(
-    name: str, controller: FlowControllerDep, user: CurrentUser
+    name: str, controller: FlowControllerDep, user: CurrentUser, session: SessionDep
 ) -> Any:
     """Delete a flow and everything in it."""
     try:
@@ -410,6 +431,7 @@ async def delete_flow(
     _audit("deleted", name, user)
     # Its files are gone; its values and queued work would otherwise linger.
     await run_in_threadpool(controller.forget_flow, name)
+    await run_in_threadpool(_forget_runs, session, name)
     await controller.reload()
     return Message(message=f"Deleted flow '{name}'")
 

backend/app/api/routes/observability.py

@@ -13,10 +13,12 @@ from typing import Any, Literal
 from fastapi import APIRouter, Depends, Request
 from fastapi.concurrency import run_in_threadpool
 from pydantic import BaseModel
-from sqlalchemy import func
+from sqlalchemy import ColumnElement, DateTime, Interval, cast, func, literal
+from sqlalchemy import select as sa_select
 from sqlmodel import col, select
 
 from app.api.deps import FlowControllerDep, SessionDep, get_current_user
+from app.core.config import settings
 from app.flow.controller import ADVISORY_ISSUES, NodeStatus
 from app.models import EngineEvent, FlowRun, MetricBucket
 
@@ -29,6 +31,10 @@ router = APIRouter(
 #: How many slices a per-flow sparkline is folded into.
 SPARK_SLICES = 60
 
+#: The origin fixed-stride slots are aligned to, which is the alignment the
+#: fold used to get from ``stamp - stamp % bucket_s``.
+EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)
+
 
 class HealthSummary(BaseModel):
     status: str
@@ -37,7 +43,6 @@ class HealthSummary(BaseModel):
     nodes: dict[str, int]
     queue: dict[str, Any]
     loop_lag: dict[str, float]
-    failures_24h: int
 
 
 class SeriesPoint(BaseModel):
@@ -74,6 +79,11 @@ class RunRow(BaseModel):
     deliveries: int
 
 
+class RunPage(BaseModel):
+    data: list[RunRow]
+    count: int
+
+
 class EventRow(BaseModel):
     id: int
     ts: datetime
@@ -97,15 +107,22 @@ def _since(hours: int) -> datetime:
     return datetime.now(timezone.utc) - timedelta(hours=hours)
 
 
+def _window_hours(hours: int) -> int:
+    """A window the rollups can answer for: an hour at least, retention at most.
+
+    Zero used to divide by nothing and answer 500, and nothing older than
+    retention exists, so a larger window is a scan that can only find less.
+    """
+    return max(1, min(hours, settings.OBS_RETENTION_DAYS * 24))
+
+
 def _aware(when: datetime) -> datetime:
     """A bound as the columns store it. A naive one is read as UTC."""
     return when if when.tzinfo else when.replace(tzinfo=timezone.utc)
 
 
 @router.get("/summary", response_model=HealthSummary)
-async def read_summary(
-    request: Request, controller: FlowControllerDep, session: SessionDep
-) -> Any:
+async def read_summary(request: Request, controller: FlowControllerDep) -> Any:
     """How the engine is doing right now. Always 200, degraded or not."""
     watchdog = getattr(request.app.state, "watchdog", None)
     problems: list[str] = []
@@ -144,14 +161,6 @@ async def read_summary(
     if invalid:
         problems.append(f"{len(invalid)} flow(s) cannot run: {', '.join(invalid)}")
 
-    statement = (
-        select(func.count())
-        .select_from(EngineEvent)
-        .where(col(EngineEvent.ts) >= _since(24), col(EngineEvent.type) != "audit")
-    )
-    # The health page polls this every ten seconds; the driver is synchronous.
-    failures = await run_in_threadpool(lambda: session.exec(statement).one())
-
     return HealthSummary(
         status="degraded" if problems else "ok",
         problems=problems,
@@ -181,7 +190,6 @@ async def read_summary(
             if watchdog is not None
             else {"ewma": 0.0, "max_60s": 0.0}
         ),
-        failures_24h=int(failures),
     )
 
 
@@ -194,63 +202,79 @@ def read_timeseries(
     bucket_s: int = 60,
 ) -> Any:
     """Executions, errors and timings over time, summed across nodes."""
-    # ponytail: the fold is in Python — a day is at most 1440 rows per node.
-    # date_bin() if the window ever grows past that.
-    statement = select(MetricBucket).where(col(MetricBucket.bucket) >= _since(hours))
+    hours = _window_hours(hours)
+    # Postgres does the fold: a week of minute rows per node used to cross the
+    # wire on every poll, and only the slices need to. The casts are load
+    # bearing — date_bin() is overloaded on timestamp and timestamptz, and an
+    # untyped bind parameter leaves the call ambiguous.
+    stride = timedelta(seconds=max(60, bucket_s))
+    slot = func.date_bin(
+        cast(literal(stride), Interval),
+        col(MetricBucket.bucket),
+        cast(literal(EPOCH), DateTime(timezone=True)),
+    ).label("slot")
+    statement = sa_select(
+        slot,
+        func.sum(col(MetricBucket.executions)).label("executions"),
+        func.sum(col(MetricBucket.errors)).label("errors"),
+        func.sum(col(MetricBucket.messages)).label("messages"),
+        func.sum(col(MetricBucket.duration_sum_ms)).label("duration_sum_ms"),
+        func.max(col(MetricBucket.duration_max_ms)).label("max_ms"),
+        func.sum(col(MetricBucket.lag_sum_ms)).label("lag_sum_ms"),
+        func.sum(col(MetricBucket.items)).label("items"),
+    ).where(col(MetricBucket.bucket) >= _since(hours))
     if flow:
         statement = statement.where(col(MetricBucket.flow) == flow)
     if node:
         statement = statement.where(col(MetricBucket.node) == node)
 
-    slices: dict[float, dict[str, float]] = {}
-    for row in session.exec(statement.order_by(col(MetricBucket.bucket))):
-        stamp = row.bucket.timestamp()
-        key = stamp - stamp % max(60, bucket_s)
-        point = slices.setdefault(
-            key,
-            {
-                "executions": 0.0,
-                "errors": 0.0,
-                "messages": 0.0,
-                "duration_sum_ms": 0.0,
-                "max_ms": 0.0,
-                "lag_sum_ms": 0.0,
-                "items": 0.0,
-            },
-        )
-        point["executions"] += row.executions
-        point["errors"] += row.errors
-        point["messages"] += row.messages
-        point["duration_sum_ms"] += row.duration_sum_ms
-        point["max_ms"] = max(point["max_ms"], row.duration_max_ms)
-        point["lag_sum_ms"] += row.lag_sum_ms
-        point["items"] += row.items
-
     return [
         SeriesPoint(
-            ts=ts,
-            executions=int(point["executions"]),
-            errors=int(point["errors"]),
-            messages=int(point["messages"]),
-            avg_ms=round(point["duration_sum_ms"] / (point["executions"] or 1), 2),
-            max_ms=round(point["max_ms"], 2),
-            avg_lag_ms=round(point["lag_sum_ms"] / (point["items"] or 1), 2),
+            ts=row.slot.timestamp(),
+            executions=int(row.executions),
+            errors=int(row.errors),
+            messages=int(row.messages),
+            avg_ms=round(row.duration_sum_ms / (row.executions or 1), 2),
+            max_ms=round(row.max_ms, 2),
+            avg_lag_ms=round(row.lag_sum_ms / (row.items or 1), 2),
         )
-        for ts, point in sorted(slices.items())
+        for row in session.execute(statement.group_by(slot).order_by(slot))
     ]
 
 
 @router.get("/flows", response_model=list[FlowRollup])
 def read_flow_rollups(session: SessionDep, hours: int = 24) -> Any:
     """One row per flow, with a coarse trend of how much it ran."""
+    hours = _window_hours(hours)
     since = _since(hours)
     window = hours * 3600
     start = since.timestamp()
 
+    # Binned to the sparkline slice rather than the minute, so a flow costs at
+    # most SPARK_SLICES rows however long the window is. The slice is the
+    # window over SPARK_SLICES, which for whole hours is whole minutes.
+    slot = func.date_bin(
+        cast(literal(timedelta(minutes=hours)), Interval),
+        col(MetricBucket.bucket),
+        cast(literal(since), DateTime(timezone=True)),
+    ).label("slot")
+    statement = (
+        sa_select(
+            col(MetricBucket.flow),
+            slot,
+            func.sum(col(MetricBucket.executions)).label("executions"),
+            func.sum(col(MetricBucket.errors)).label("errors"),
+            func.sum(col(MetricBucket.messages)).label("messages"),
+            func.sum(col(MetricBucket.duration_sum_ms)).label("duration_sum_ms"),
+            func.sum(col(MetricBucket.lag_sum_ms)).label("lag_sum_ms"),
+            func.sum(col(MetricBucket.items)).label("items"),
+        )
+        .where(col(MetricBucket.bucket) >= since)
+        .group_by(col(MetricBucket.flow), slot)
+    )
+
     rollups: dict[str, dict[str, Any]] = {}
-    for row in session.exec(
-        select(MetricBucket).where(col(MetricBucket.bucket) >= since)
-    ):
+    for row in session.execute(statement):
         entry = rollups.setdefault(
             row.flow,
             {
@@ -269,11 +293,14 @@ def read_flow_rollups(session: SessionDep, hours: int = 24) -> Any:
         entry["duration_sum_ms"] += row.duration_sum_ms
         entry["lag_sum_ms"] += row.lag_sum_ms
         entry["items"] += row.items
-        slot = min(
+        # A slot sits a whole number of slices from `since`, so this rounds
+        # rather than truncates: a float a hair short would lose a slice. The
+        # clamp holds the bucket landing exactly on the far edge in range.
+        index = min(
             SPARK_SLICES - 1,
-            max(0, int((row.bucket.timestamp() - start) / window * SPARK_SLICES)),
+            max(0, round((row.slot.timestamp() - start) / window * SPARK_SLICES)),
         )
-        entry["spark"][slot] += row.executions
+        entry["spark"][index] += row.executions
 
     # `.all()` first: a Result has `keys()`, so dict() would read it as a
     # mapping and subscript it.
@@ -302,7 +329,7 @@ def read_flow_rollups(session: SessionDep, hours: int = 24) -> Any:
     ]
 
 
-@router.get("/runs", response_model=list[RunRow])
+@router.get("/runs", response_model=RunPage)
 def read_runs(
     session: SessionDep,
     flow: str | None = None,
@@ -311,21 +338,38 @@ def read_runs(
     until: datetime | None = None,
     limit: int = 50,
 ) -> Any:
-    """Recent cascades, newest first.
+    """Recent cascades, newest first, and how many there were in total.
 
     ``since`` is inclusive and ``until`` exclusive, so a window of one minute
     holds exactly the runs of the minute bucket the charts are drawn from.
     """
-    statement = select(FlowRun).order_by(col(FlowRun.started_at).desc())
+    filters: list[ColumnElement[bool]] = []
     if flow:
-        statement = statement.where(col(FlowRun.flow) == flow)
+        filters.append(col(FlowRun.flow) == flow)
     if status:
-        statement = statement.where(col(FlowRun.status) == status)
+        filters.append(col(FlowRun.status) == status)
     if since:
-        statement = statement.where(col(FlowRun.started_at) >= _aware(since))
+        filters.append(col(FlowRun.started_at) >= _aware(since))
     if until:
-        statement = statement.where(col(FlowRun.started_at) < _aware(until))
-    return list(session.exec(statement.limit(min(limit, 200))))
+        filters.append(col(FlowRun.started_at) < _aware(until))
+
+    capped = min(limit, 200)
+    statement = select(FlowRun).where(*filters).order_by(col(FlowRun.started_at).desc())
+    rows = list(session.exec(statement.limit(capped)))
+    # A short page is its own total. The lists poll their whole range every
+    # thirty seconds, and counting on each of those would hand back what
+    # binning the metrics just saved — for a number that only ever says
+    # "there is more here than fits".
+    count = (
+        len(rows)
+        if len(rows) < capped
+        else int(
+            session.exec(
+                select(func.count()).select_from(FlowRun).where(*filters)
+            ).one()
+        )
+    )
+    return {"data": rows, "count": count}
 
 
 @router.get("/events", response_model=list[EventRow])

backend/app/mcp/server.py

@@ -284,7 +284,7 @@ async def apply_modules(requirements: str) -> Any:
 
 @mcp.tool()
 async def get_health() -> Any:
-    """How the engine is doing: flows, nodes, queue, loop lag and recent failures."""
+    """How the engine is doing right now: flows, nodes, queue and loop lag."""
     return await _call("GET", "/observability/summary")
 
 
@@ -312,7 +312,11 @@ async def list_failures(flow: str | None = None, limit: int = 50) -> Any:
 
 @mcp.tool()
 async def list_runs(flow: str | None = None, limit: int = 50) -> Any:
-    """Recent cascades: what triggered them, how long they took, how they ended."""
+    """Recent cascades: what triggered them, how long they took, how they ended.
+
+    A page of rows plus the total number matching, which says whether the limit
+    cut anything off.
+    """
     params: dict[str, Any] = {"limit": limit}
     if flow:
         params["flow"] = flow

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(