diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py
index cae0d94..6ea4fa7 100644
--- a/backend/app/api/routes/flows.py
+++ b/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}'")
 
diff --git a/backend/app/api/routes/observability.py b/backend/app/api/routes/observability.py
index 2134035..5e3b6f8 100644
--- a/backend/app/api/routes/observability.py
+++ b/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])
diff --git a/backend/app/mcp/server.py b/backend/app/mcp/server.py
index 866e251..e17f074 100644
--- a/backend/app/mcp/server.py
+++ b/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
diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py
index 1b68a0e..eb77632 100644
--- a/backend/tests/api/routes/test_flows.py
+++ b/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]
diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py
index 24ba915..ec03183 100644
--- a/backend/tests/api/routes/test_observability.py
+++ b/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(
diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts
index 03eb6d2..ba22498 100644
--- a/frontend/src/client/schemas.gen.ts
+++ b/frontend/src/client/schemas.gen.ts
@@ -1081,14 +1081,10 @@ export const HealthSummarySchema = {
             },
             type: 'object',
             title: 'Loop Lag'
-        },
-        failures_24h: {
-            type: 'integer',
-            title: 'Failures 24H'
         }
     },
     type: 'object',
-    required: ['status', 'problems', 'flows', 'nodes', 'queue', 'loop_lag', 'failures_24h'],
+    required: ['status', 'problems', 'flows', 'nodes', 'queue', 'loop_lag'],
     title: 'HealthSummary'
 } as const;
 
@@ -2421,6 +2417,25 @@ export const RunNodeRowSchema = {
     title: 'RunNodeRow'
 } as const;
 
+export const RunPageSchema = {
+    properties: {
+        data: {
+            items: {
+                '$ref': '#/components/schemas/app__api__routes__observability__RunRow'
+            },
+            type: 'array',
+            title: 'Data'
+        },
+        count: {
+            type: 'integer',
+            title: 'Count'
+        }
+    },
+    type: 'object',
+    required: ['data', 'count'],
+    title: 'RunPage'
+} as const;
+
 export const RunRequestSchema = {
     properties: {
         inputs: {
diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts
index 0fe6389..aa4c463 100644
--- a/frontend/src/client/sdk.gen.ts
+++ b/frontend/src/client/sdk.gen.ts
@@ -1308,7 +1308,7 @@ export class ObservabilityService {
     
     /**
      * Read Runs
-     * 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.
@@ -1318,7 +1318,7 @@ export class ObservabilityService {
      * @param data.since
      * @param data.until
      * @param data.limit
-     * @returns app__api__routes__observability__RunRow Successful Response
+     * @returns RunPage Successful Response
      * @throws ApiError
      */
     public static readRuns(data: ObservabilityReadRunsData = {}): CancelablePromise<ObservabilityReadRunsResponse> {
diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts
index cc4f532..d71d107 100644
--- a/frontend/src/client/types.gen.ts
+++ b/frontend/src/client/types.gen.ts
@@ -408,7 +408,6 @@ export type HealthSummary = {
     loop_lag: {
         [key: string]: (number);
     };
-    failures_24h: number;
 };
 
 /**
@@ -856,6 +855,11 @@ export type RunNodeRow = {
     logs: string;
 };
 
+export type RunPage = {
+    data: Array<app__api__routes__observability__RunRow>;
+    count: number;
+};
+
 export type RunRequest = {
     inputs?: {
         [key: string]: unknown;
@@ -1399,7 +1403,7 @@ export type ObservabilityReadRunsData = {
     until?: (string | null);
 };
 
-export type ObservabilityReadRunsResponse = (Array<app__api__routes__observability__RunRow>);
+export type ObservabilityReadRunsResponse = (RunPage);
 
 export type ObservabilityReadEventsData = {
     flow?: (string | null);
diff --git a/frontend/src/components/Health/HealthActivity.tsx b/frontend/src/components/Health/HealthActivity.tsx
index 911b42e..b6a9fce 100644
--- a/frontend/src/components/Health/HealthActivity.tsx
+++ b/frontend/src/components/Health/HealthActivity.tsx
@@ -8,7 +8,7 @@ import { UplotChart } from "@/components/Common/UplotChart"
 import { useEngineEvents } from "@/components/Flow/liveStore"
 import { PANEL_SECTION } from "@/components/Flow/SidePanel"
 import { Button } from "@/components/ui/button"
-import { cn, dur } from "@/lib/utils"
+import { cn, dur, si } from "@/lib/utils"
 import {
   ago,
   auditQueryOptions,
@@ -138,7 +138,15 @@ function Chart({
 }
 
 /** What a list is showing, and the way back to all of it. */
-function ListHeader({ title, moment }: { title: string; moment: Moment }) {
+function ListHeader({
+  title,
+  moment,
+  note,
+}: {
+  title: string
+  moment: Moment
+  note?: string
+}) {
   return (
     <div className="flex min-h-6 flex-wrap items-center gap-2">
       <h2 className={PANEL_SECTION}>{title}</h2>
@@ -147,6 +155,9 @@ function ListHeader({ title, moment }: { title: string; moment: Moment }) {
           {moment.pinned !== null ? "pinned to" : "showing"} {clock(moment.at)}
         </span>
       ) : null}
+      {note ? (
+        <span className="text-xs text-muted-foreground">{note}</span>
+      ) : null}
       {moment.pinned !== null ? <ClearPin moment={moment} /> : null}
     </div>
   )
@@ -249,10 +260,18 @@ export function HealthActivity({ range }: { range: Range }) {
   // minute the pointer rests on.
   const shownRuns =
     runsAt.pinned !== null
-      ? (pinnedRuns ?? [])
+      ? (pinnedRuns?.data ?? [])
       : runsAt.at === null
-        ? (runs ?? []).slice(0, RUNS_SHOWN)
-        : (runs ?? []).filter((run) => minuteOf(run.started_at) === runsAt.at)
+        ? (runs?.data ?? []).slice(0, RUNS_SHOWN)
+        : (runs?.data ?? []).filter(
+            (run) => minuteOf(run.started_at) === runsAt.at,
+          )
+  // A busy minute writes more runs than one page carries. Without the total
+  // behind it, a truncated list is indistinguishable from a quiet minute.
+  const runsNote =
+    runsAt.pinned !== null && (pinnedRuns?.count ?? 0) > shownRuns.length
+      ? `showing ${shownRuns.length} of ${si(pinnedRuns?.count ?? 0)}`
+      : undefined
   const shownFailures =
     failuresAt.pinned !== null
       ? (pinnedFailures ?? [])
@@ -290,7 +309,7 @@ export function HealthActivity({ range }: { range: Range }) {
 
       <section className="grid content-start gap-4 lg:grid-cols-2">
         <div className="grid content-start gap-3">
-          <ListHeader title="Recent runs" moment={runsAt} />
+          <ListHeader title="Recent runs" moment={runsAt} note={runsNote} />
           <div
             className={cn(
               CARD,