Report how long a run has been going, not just how long it took
Docs / docs (push) Successful in 20s
Playwright Tests / test-playwright (1, 2) (push) Failing after 2m39s
Playwright Tests / test-playwright (2, 2) (push) Failing after 1m39s
pre-commit / pre-commit (push) Failing after 2m51s
Playwright Tests / merge-reports (push) Canceled after 0s
Test Backend / test-backend (push) Canceled after 1m4s
Compose Smoke Test / test-compose (push) Canceled after 0s

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UytviPMJbXzD8P84nLvXcq
This commit is contained in:
2026-08-25 18:33:28 +02:00
co-authored by Claude Opus 5
parent 055f4d7d97
commit 3503512d05
4 changed files with 67 additions and 3 deletions
+10 -1
View File
@@ -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]
+16 -1
View File
@@ -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)
@@ -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:
+26 -1
View File
@@ -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(