Report how long a run has been going, not just how long it took

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]

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)

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:

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(