From 9c149f9ed4519671983f80af29eb625a25123e6b Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 21 Aug 2026 10:11:19 +0200 Subject: [PATCH] Let Postgres fold the rollups, and say when a run list was cut short MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /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) Claude-Session: https://claude.ai/code/session_01Uq8mtNb97A7praJLyeEYgs --- backend/app/api/routes/flows.py | 28 ++- backend/app/api/routes/observability.py | 170 +++++++++++------- backend/app/mcp/server.py | 8 +- backend/tests/api/routes/test_flows.py | 17 +- .../tests/api/routes/test_observability.py | 86 ++++++++- frontend/src/client/schemas.gen.ts | 25 ++- frontend/src/client/sdk.gen.ts | 4 +- frontend/src/client/types.gen.ts | 8 +- .../src/components/Health/HealthActivity.tsx | 31 +++- 9 files changed, 291 insertions(+), 86 deletions(-) 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 { 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; + count: number; +}; + export type RunRequest = { inputs?: { [key: string]: unknown; @@ -1399,7 +1403,7 @@ export type ObservabilityReadRunsData = { until?: (string | null); }; -export type ObservabilityReadRunsResponse = (Array); +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 (

{title}

@@ -147,6 +155,9 @@ function ListHeader({ title, moment }: { title: string; moment: Moment }) { {moment.pinned !== null ? "pinned to" : "showing"} {clock(moment.at)} ) : null} + {note ? ( + {note} + ) : null} {moment.pinned !== null ? : null}
) @@ -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 }) {
- +