From 3503512d05157731661a36dfea8cd9d29f49c3bb Mon Sep 17 00:00:00 2001 From: stroblme Date: Tue, 25 Aug 2026 18:33:21 +0200 Subject: [PATCH] Report how long a run has been going, not just how long it took Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01UytviPMJbXzD8P84nLvXcq --- backend/fluksio/api/routes/observability.py | 11 +++++++- backend/fluksio/api/routes/runs.py | 17 +++++++++++- .../tests/api/routes/test_observability.py | 15 +++++++++++ backend/tests/api/routes/test_runs.py | 27 ++++++++++++++++++- 4 files changed, 67 insertions(+), 3 deletions(-) diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index ea75662..409f9c4 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -12,12 +12,13 @@ from typing import Any, Literal from fastapi import APIRouter, Depends, Request from fastapi.concurrency import run_in_threadpool -from pydantic import BaseModel +from pydantic import BaseModel, model_validator from sqlalchemy import ColumnElement, Integer, cast, func from sqlalchemy import select as sa_select from sqlmodel import col, select from fluksio.api.deps import FlowControllerDep, SessionDep, get_current_user +from fluksio.api.routes.runs import elapsed_ms from fluksio.core.config import settings from fluksio.flow.controller import ADVISORY_ISSUES, NodeStatus from fluksio.models import EngineEvent, FlowRun, MetricBucket @@ -81,9 +82,17 @@ class RunRow(BaseModel): finished_at: datetime | None = None nodes: int errors: int + #: How long it took, or — while it is still going — how long it has been + #: going: the collector only writes a duration once a cascade finishes. duration_ms: float deliveries: int + @model_validator(mode="after") + def _running_duration(self) -> "RunRow": + if self.status == "running" and not self.duration_ms: + self.duration_ms = elapsed_ms(self.started_at) + return self + class RunPage(BaseModel): data: list[RunRow] diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py index d3f90fa..7018d19 100644 --- a/backend/fluksio/api/routes/runs.py +++ b/backend/fluksio/api/routes/runs.py @@ -5,11 +5,12 @@ in hours, so nothing here waits for one. The way to follow a run is to poll it or to listen on the flow socket, which carries its start and finish. """ +from datetime import UTC, datetime from typing import Any from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.concurrency import run_in_threadpool -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator from sqlalchemy import func from sqlalchemy import select as sa_select from sqlmodel import Session, col, select @@ -28,6 +29,12 @@ router = APIRouter( MAX_SWEEP = 1000 +def elapsed_ms(since: datetime) -> float: + """Milliseconds since an instant the columns stored as UTC.""" + start = since if since.tzinfo else since.replace(tzinfo=UTC) + return round((datetime.now(UTC) - start).total_seconds() * 1000, 2) + + class RunCreate(BaseModel): params: dict[str, Any] = Field(default_factory=dict) seed: int | None = None @@ -95,9 +102,17 @@ class RunRow(BaseModel): created_at: Any started_at: Any = None finished_at: Any = None + #: How long it took, or — while it is still going — how long it has been + #: going: a duration of its own is only written once a run finishes. duration_ms: float actor: str + @model_validator(mode="after") + def _running_duration(self) -> "RunRow": + if self.status == "running" and not self.duration_ms and self.started_at: + self.duration_ms = elapsed_ms(self.started_at) + return self + class RunDetail(RunRow): result: dict[str, Any] = Field(default_factory=dict) diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py index 9c966d1..f67e8e2 100644 --- a/backend/tests/api/routes/test_observability.py +++ b/backend/tests/api/routes/test_observability.py @@ -266,6 +266,21 @@ def test_runs_narrow_to_one_minute( assert [run["id"] for run in runs["data"]] == ["minute-in"] +def test_a_running_cascade_reports_how_long_it_has_been_going( + client: TestClient, superuser_token_headers: dict[str, str], db: Session +) -> None: + """The collector writes a duration at the end; until then, time since.""" + started = datetime.now(UTC) - timedelta(seconds=30) + db.add(FlowRun(id="in-flight", flow=FLOW, started_at=started, status="running")) + db.commit() + + runs = client.get( + f"{PREFIX}/runs", headers=superuser_token_headers, params={"status": "running"} + ).json() + + assert runs["data"][0]["duration_ms"] >= 30_000 + + def test_events_narrow_to_one_minute( client: TestClient, superuser_token_headers: dict[str, str], db: Session ) -> None: diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 633d432..ed2a93c 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -5,7 +5,7 @@ The pipeline half — what a hit restores and what a key is made of — is in """ import json -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta import pytest from sqlmodel import Session, col, select @@ -286,6 +286,31 @@ def test_the_overview_counts_a_flow_the_list_page_would_not_reach( assert (row["runs"], row["running"], row["queued"]) == (3, 1, 0) +def test_a_running_run_reports_how_long_it_has_been_going( + client, superuser_token_headers +): + """A duration is only written at the end; until then, time since it began.""" + with Session(db_engine) as session: + session.add( + Run( + id="in-flight", + flow="timed", + status="running", + created_at=datetime.now(UTC), + started_at=datetime.now(UTC) - timedelta(seconds=30), + ) + ) + session.commit() + + rows = client.get( + f"{settings.API_V1_STR}/runs", + headers=superuser_token_headers, + params={"flow": "timed"}, + ).json() + + assert rows[0]["duration_ms"] >= 30_000 + + def test_overview_is_not_read_as_a_run_id(client, superuser_token_headers): """`/overview` is declared before `/{run_id}`, which would swallow it.""" answer = client.get(