app/backend/fluksio/api/routes/runs.py

"""Runs over the API: submit one, watch it, read what it made.

Submitting returns immediately with a queued run — a training run is measured
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.
"""

import csv
import io
import json
import time
from collections.abc import Iterator
from datetime import UTC, datetime
from itertools import groupby
from typing import Annotated, Any, Literal

from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field, model_validator
from sqlalchemy import delete, func
from sqlalchemy import select as sa_select
from sqlmodel import Session, col, select

from fluksio.api.deps import CurrentUser, SessionDep, get_current_user
from fluksio.core.db import writing
from fluksio.flow.events import event_bus
from fluksio.flow.messages import requalify
from fluksio.flow.runs import RunRejected, RunService, new_run_id
from fluksio.flow.store import FlowNotFound
from fluksio.models import Run, RunArtifact, RunMetric, RunNode

router = APIRouter(
    prefix="/runs", tags=["runs"], dependencies=[Depends(get_current_user)]
)

#: A sweep bigger than this is almost always a mistake in a loop.
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)


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=UTC)


#: Where a caller may say a run came from. "sweep" is not here because the
#: sweep route writes it itself, and neither is a value a client made up: the
#: column is only worth a table row if it means the same thing every time.
RunCause = Literal["api", "cli", "sdk"]


class RunCreate(BaseModel):
    params: dict[str, Any] = Field(default_factory=dict)
    seed: int | None = None
    #: Run the unpublished draft instead of what is published.
    draft: bool = False
    #: Execute every node, whatever an earlier run already worked out.
    no_cache: bool = False
    #: Who is asking. The dashboard leaves it, and is the "api" default.
    cause: RunCause = "api"
    #: A key the caller minted for this submission. Sending it again returns
    #: the run it already made, so a retry after a timeout cannot double-submit.
    idempotency_key: str | None = Field(default=None, max_length=64)


class SweepEntry(BaseModel):
    params: dict[str, Any] = Field(default_factory=dict)
    seed: int | None = None
    #: One per entry, so retrying a half-created sweep recreates only the runs
    #: whose rows never landed.
    idempotency_key: str | None = Field(default=None, max_length=64)


class SweepCreate(BaseModel):
    runs: list[SweepEntry] = Field(default_factory=list)
    draft: bool = False
    no_cache: bool = False


class RunNodeRow(BaseModel):
    node: str
    status: str
    attempt: int
    duration_ms: float
    worker: str
    error: str
    logs: str
    #: What this node's result was looked up by. Empty when it may not be
    #: reused; `status` is "cached" when it was.
    cache_key: str = ""
    #: Which run it was restored from, when it was. That run is also where this
    #: node's series was recorded.
    cached_from: str = ""


class ArtifactRow(BaseModel):
    name: str
    node: str
    digest: str
    size: int
    media_type: str
    #: What it was called where it was written, so something downloading it can
    #: give it that name back rather than the message's. Absent when the node
    #: never said one.
    filename: str | None = None


class RunRow(BaseModel):
    """A run without its result, which is the part that can be large."""

    id: str
    flow: str
    status: str
    status_reason: str
    cause: str
    params: dict[str, Any]
    params_digest: str
    #: The user repository's commit, for a flow declared in code with the
    #: decorators. Empty for one drawn on the canvas, where `commit` is the
    #: whole answer to what produced the number.
    origin_commit: str = ""
    #: What that repository's python files hashed to when the run started.
    #: Two runs of one dirty tree share a commit and differ here, which is the
    #: only way to tell apart what they actually executed.
    code_digest: str = ""
    #: The flow store's own commit. Short, unlike `result`, so the list
    #: carries it: "what code produced this" is a question asked of a table.
    commit: str = ""
    seed: int | None
    group_id: str | None
    labels: list[str]
    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)
    flow_version: int = 1
    nodes: list[RunNodeRow] = Field(default_factory=list)
    artifacts: list[ArtifactRow] = Field(default_factory=list)


class FlowRunsRow(BaseModel):
    """How much a flow has been run, for the screen's list of flows."""

    flow: str
    runs: int
    running: int
    queued: int
    last_created_at: Any = None


class MetricPoint(BaseModel):
    step: int
    ts: float
    value: float
    name: str = ""


class MetricSeries(BaseModel):
    """The shape a chart widget already draws, so comparing runs is a binding."""

    label: str
    points: list[list[float]] = Field(default_factory=list)


class SeriesAnswer(BaseModel):
    metric: str
    #: What the x values are: "step", "time" (seconds since this run's first
    #: reading), or the name of another metric this one was plotted against.
    x: str = "step"
    lines: list[MetricSeries] = Field(default_factory=list)


def _service(request: Request) -> RunService:
    service: RunService | None = getattr(request.app.state, "run_service", None)
    if service is None:
        raise HTTPException(status_code=503, detail="Runs are not available")
    return service


@router.post("/flows/{name}", response_model=RunRow, status_code=202)
async def create_run(
    name: str, body: RunCreate, request: Request, user: CurrentUser
) -> Any:
    """Queue one run of a flow."""
    service = _service(request)
    try:
        return await run_in_threadpool(
            service.submit,
            name,
            params=body.params,
            seed=body.seed,
            cause=body.cause,
            actor=user.email,
            draft=body.draft,
            no_cache=body.no_cache,
            idempotency_key=body.idempotency_key,
        )
    except FlowNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except RunRejected as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc


@router.post("/flows/{name}/sweep", response_model=list[RunRow], status_code=202)
async def create_sweep(
    name: str, body: SweepCreate, request: Request, user: CurrentUser
) -> Any:
    """Queue many runs of one flow under a shared group.

    An ensemble is this with the same parameters and different seeds; a grid
    search is this with the parameters spread out. Either way the caller
    builds the list — the engine does not own a sweep grammar.
    """
    if not body.runs:
        raise HTTPException(status_code=422, detail="A sweep needs at least one run")
    if len(body.runs) > MAX_SWEEP:
        raise HTTPException(
            status_code=422, detail=f"A sweep is capped at {MAX_SWEEP} runs"
        )
    service = _service(request)
    group = new_run_id()

    def submit_all() -> list[Run]:
        return [
            service.submit(
                name,
                params=entry.params,
                seed=entry.seed,
                group_id=group,
                cause="sweep",
                actor=user.email,
                draft=body.draft,
                no_cache=body.no_cache,
                idempotency_key=entry.idempotency_key,
            )
            for entry in body.runs
        ]

    try:
        return await run_in_threadpool(submit_all)
    except FlowNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except RunRejected as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc


@router.get("", response_model=list[RunRow])
def read_runs(
    session: SessionDep,
    flow: str | None = None,
    status: str | None = None,
    group: str | None = None,
    digest: str | None = None,
    since: datetime | None = None,
    before: datetime | None = None,
    limit: Annotated[int, Query(ge=1, le=500)] = 50,
    offset: Annotated[int, Query(ge=0)] = 0,
) -> Any:
    """Runs, newest first. The queryable table an experiment log needs.

    ``before`` is the cursor a long history is paged by: rows are newest
    first, so handing back the last row's ``created_at`` reads the next page
    whatever landed meanwhile — which ``offset`` cannot, since a run submitted
    between two pages shifts every row down one. ``since`` bounds the other
    end and is inclusive.
    """
    statement = select(Run).order_by(col(Run.created_at).desc())
    if flow:
        statement = statement.where(col(Run.flow) == flow)
    if status:
        statement = statement.where(col(Run.status) == status)
    if group:
        statement = statement.where(col(Run.group_id) == group)
    if digest:
        statement = statement.where(col(Run.params_digest) == digest)
    if since:
        statement = statement.where(col(Run.created_at) >= _aware(since))
    if before:
        statement = statement.where(col(Run.created_at) < _aware(before))
    statement = statement.offset(max(0, offset)).limit(min(limit, 500))
    return list(session.exec(statement))


@router.get("/overview", response_model=list[FlowRunsRow])
def read_overview(session: SessionDep) -> Any:
    """One row per flow that has ever run, busiest-recent first.

    The list caps at 500 newest runs, so counting flows on the client goes
    wrong the moment a history outgrows one page. The database counts instead.
    """
    # Deliberately unwindowed. It is a group-by over the whole `run` table,
    # which is never pruned — but a run kept for years is the point of the
    # table, and a window here would drop a finished experiment off the rail
    # rather than making it cheaper to find. A rollup is the answer if the
    # scan ever measures.
    statement = sa_select(
        col(Run.flow),
        col(Run.status),
        func.count(col(Run.id)),
        func.max(col(Run.created_at)),
    ).group_by(col(Run.flow), col(Run.status))

    rows: dict[str, FlowRunsRow] = {}
    for flow, status, count, latest in session.execute(statement):
        row = rows.setdefault(flow, FlowRunsRow(flow=flow, runs=0, running=0, queued=0))
        row.runs += count
        if status == "running":
            row.running += count
        elif status == "queued":
            row.queued += count
        if row.last_created_at is None or latest > row.last_created_at:
            row.last_created_at = latest
    return sorted(rows.values(), key=lambda row: row.last_created_at, reverse=True)


# ---------------------------------------------------------------------------
# Export
#
# Both routes are declared above `/{run_id}`, or "export" is read as the id of
# a run nobody has. What they are for: an analysis wants a dataframe, and the
# alternatives are a call per run or somebody reading our schema out of
# `fluksio.db`. The two shapes below are what an analysis actually asks for.
# ---------------------------------------------------------------------------

#: What an export is written as. Parquet is a conversion the client does over
#: jsonl, because keeping dtypes is worth a dependency only to whoever wants it.
ExportFormat = Literal["csv", "jsonl"]

#: The columns of the long table, in order. The run is on every row: it is the
#: join back to the run page and to what the run made, and it is what makes an
#: exported file auditable rather than loose.
METRIC_COLUMNS = ("run", "name", "step", "ts", "value")

#: A run's own columns in the wide table. Its inputs and its final numbers
#: follow, prefixed, so an input named "status" cannot collide with the run's.
#: The most runs one export covers. High enough that a real sweep fits and low
#: enough that an unfiltered request cannot read the whole table into memory —
#: which it did, before a byte was streamed, while holding one of the fifteen
#: pooled database connections for the length of the download.
EXPORT_CAP = 10_000

RUN_COLUMNS = (
    "id",
    "flow",
    "status",
    "created_at",
    "started_at",
    "finished_at",
    "duration_ms",
    "seed",
    "group_id",
    "code_digest",
    "origin_commit",
)


def _selected(
    session: Session,
    flow: str | None,
    status: str | None,
    group: str | None,
    ids: str,
    since: datetime | None,
    until: datetime | None,
) -> list[Run]:
    """The runs an export covers, newest first — the filters the list takes.

    Capped: the filters bound a *sensible* request and nothing bounded an
    unfiltered one, so asking for everything read every row of the table into
    memory before a byte was sent. A truncated export says so in a header.
    """
    statement = select(Run).order_by(col(Run.created_at).desc())
    if flow:
        statement = statement.where(col(Run.flow) == flow)
    if status:
        statement = statement.where(col(Run.status) == status)
    if group:
        statement = statement.where(col(Run.group_id) == group)
    named = [part for part in ids.split(",") if part]
    if named:
        statement = statement.where(col(Run.id).in_(named))
    if since:
        statement = statement.where(col(Run.created_at) >= _aware(since))
    if until:
        statement = statement.where(col(Run.created_at) < _aware(until))
    return list(session.exec(statement.limit(EXPORT_CAP + 1)))


def _cell(value: Any) -> Any:
    """A value as a table holds it: a scalar, or JSON when it is not one."""
    if hasattr(value, "isoformat"):
        return value.isoformat()
    if value is None or isinstance(value, (str, int, float, bool)):
        return value
    return json.dumps(value)


def _leaves(value: Any, prefix: str = "") -> Iterator[tuple[str, Any]]:
    """Everything a record holds, by its dotted path.

    A node returning a record rather than a scalar is the ordinary shape —
    the numbers arrive inside `final_metrics` — and a record in one cell is
    not a column anybody can compare. Lists are left whole: a curve belongs in
    the long table, not in a cell of this one.
    """
    if isinstance(value, dict):
        for key, inner in value.items():
            yield from _leaves(inner, f"{prefix}.{key}" if prefix else str(key))
    else:
        yield prefix, value


def _dig(record: dict[str, Any], path: str) -> Any:
    """A dotted path into a record: `final_metrics.train_loss`.

    A key with a dot in its own name is not reachable this way, which is the
    price of the spelling.
    """
    value: Any = record
    for part in path.split("."):
        if not isinstance(value, dict) or part not in value:
            return None
        value = value[part]
    return value


def _scored(runs: list[Run]) -> list[str]:
    """A run's final numbers: every number its declared outputs carry, however
    deep it sits. A flag is not a number, and neither is a label."""
    return sorted(
        {
            path
            for run in runs
            for path, value in _leaves(run.result)
            if isinstance(value, (int, float)) and not isinstance(value, bool)
        }
    )


def _drain(buffer: io.StringIO) -> str:
    text = buffer.getvalue()
    buffer.seek(0)
    buffer.truncate(0)
    return text


def _csv(columns: list[str], chunks: Iterator[list[dict[str, Any]]]) -> Iterator[str]:
    """Header first, then a chunk at a time through one reused buffer."""
    buffer = io.StringIO()
    writer = csv.DictWriter(buffer, fieldnames=columns)
    writer.writeheader()
    yield _drain(buffer)
    for chunk in chunks:
        writer.writerows(chunk)
        yield _drain(buffer)


def _stream(
    fmt: str,
    name: str,
    columns: list[str],
    chunks: Iterator[list[dict[str, Any]]],
    truncated: bool = False,
) -> StreamingResponse:
    """The rows out, a chunk at a time rather than a list.

    Streaming because the point of an export is that it is bigger than what a
    screen reads: a sweep's curves are millions of rows.
    """
    if fmt == "jsonl":
        body: Iterator[str] = (
            "".join(json.dumps(row) + "\n" for row in chunk) for chunk in chunks
        )
        media = "application/x-ndjson"
    else:
        body = _csv(columns, chunks)
        media = "text/csv; charset=utf-8"
    headers = {"Content-Disposition": f'attachment; filename="{name}.{fmt}"'}
    if truncated:
        # The body is otherwise indistinguishable from a complete one, and an
        # export quietly missing its oldest runs is the wrong thing to hand
        # somebody who is about to plot it.
        headers["X-Truncated"] = str(EXPORT_CAP)
    return StreamingResponse(body, media_type=media, headers=headers)


@router.get("/export/metrics")
def export_metrics(
    session: SessionDep,
    flow: str | None = None,
    status: str | None = None,
    group: str | None = None,
    ids: str = "",
    since: datetime | None = None,
    until: datetime | None = None,
    name: str = "",
    stride: Annotated[int, Query(ge=1, le=100_000)] = 1,
    format: ExportFormat = "csv",
) -> Any:
    """Every selected run's series as one long table: run, name, step, ts, value.

    The tidy shape a plotting library takes without reshaping. ``name`` keeps
    the metrics it lists; ``stride`` thins each curve — per series, so asking
    for every tenth point of two metrics gives every tenth point of both.
    """
    runs = _selected(session, flow, status, group, ids, since, until)
    truncated = len(runs) > EXPORT_CAP
    runs = runs[:EXPORT_CAP]
    wanted = [part for part in name.split(",") if part]
    step = max(1, stride)

    def chunks() -> Iterator[list[dict[str, Any]]]:
        for run in runs:
            rows: list[dict[str, Any]] = []
            for series, points in groupby(
                _series(session, run.id), key=lambda row: row.name
            ):
                if wanted and series not in wanted:
                    continue
                rows.extend(
                    {
                        "run": run.id,
                        "name": row.name,
                        "step": row.step,
                        "ts": row.ts,
                        "value": row.value,
                    }
                    for row in list(points)[::step]
                )
            yield rows

    return _stream(format, "metrics", list(METRIC_COLUMNS), chunks(), truncated)


@router.get("/export/runs")
def export_runs(
    session: SessionDep,
    flow: str | None = None,
    status: str | None = None,
    group: str | None = None,
    ids: str = "",
    since: datetime | None = None,
    until: datetime | None = None,
    params: str = "",
    metrics: str = "",
    format: ExportFormat = "csv",
) -> Any:
    """One row per run: what it was given, what it scored, what code it ran.

    The arm-comparison table. Inputs are columns rather than one JSON blob —
    every input the selection recorded, so the schema is the same whichever
    runs are asked for; ``params`` narrows it. ``metrics`` narrows the final
    numbers to a few of a run's declared outputs.

    Both take dotted paths into a record a node returned:
    ``metrics=final_metrics.train_loss,test_metrics.known.perfect`` selects
    three fields rather than two blobs, and the defaults reach the same
    depth.
    """
    runs = _selected(session, flow, status, group, ids, since, until)
    truncated = len(runs) > EXPORT_CAP
    runs = runs[:EXPORT_CAP]
    inputs = [part for part in params.split(",") if part] or sorted(
        {path for run in runs for path, _ in _leaves(run.params)}
    )
    scores = [part for part in metrics.split(",") if part] or _scored(runs)
    columns = [
        *RUN_COLUMNS,
        *(f"param.{key}" for key in inputs),
        *(f"metric.{key}" for key in scores),
    ]

    def chunks() -> Iterator[list[dict[str, Any]]]:
        for run in runs:
            row = {column: _cell(getattr(run, column)) for column in RUN_COLUMNS}
            row.update((f"param.{k}", _cell(_dig(run.params, k))) for k in inputs)
            row.update((f"metric.{k}", _cell(_dig(run.result, k))) for k in scores)
            yield [row]

    return _stream(format, "runs", columns, chunks(), truncated)


@router.get("/{run_id}", response_model=RunDetail)
def read_run(run_id: str, session: SessionDep) -> Any:
    """One run in full: what it was asked, what each node did, what it made."""
    run = session.get(Run, run_id)
    if run is None:
        raise HTTPException(status_code=404, detail="No such run")
    nodes = session.exec(select(RunNode).where(col(RunNode.run_id) == run_id)).all()
    artifacts = session.exec(
        select(RunArtifact).where(col(RunArtifact.run_id) == run_id)
    ).all()
    detail = RunDetail.model_validate(run, from_attributes=True)
    detail.nodes = [RunNodeRow.model_validate(n, from_attributes=True) for n in nodes]
    detail.artifacts = [
        ArtifactRow.model_validate(a, from_attributes=True) for a in artifacts
    ]
    return detail


@router.post("/{run_id}/cancel", response_model=RunRow)
async def cancel_run(run_id: str, request: Request, session: SessionDep) -> Any:
    """Stop a run. One already past its last node is left as it finished."""
    run = session.get(Run, run_id)
    if run is None:
        raise HTTPException(status_code=404, detail="No such run")
    service = _service(request)
    await run_in_threadpool(service.cancel, run_id)
    session.refresh(run)
    return run


@router.delete("/{run_id}", status_code=204)
def delete_run(run_id: str, session: SessionDep, user: CurrentUser) -> Response:
    """Forget a run and everything hanging off it.

    The same four statements ``_forget_runs`` uses when a flow goes: the run
    tables carry a plain string ``run_id`` and no foreign key, so nothing
    cascades on its own. ``flow_run``, ``metric_minute`` and ``engine_event``
    stay — they are the observability record and are pruned on their own window.

    A live run is refused rather than raced: the driver writes its nodes back
    when it finishes, and those rows would arrive for a run that no longer
    exists. Cancel it first.

    Two things this costs, both deliberate. ``RunNode.outputs`` *is* the stage
    cache, so a later run loses hits this one would have served. And a node
    restored from this run points here through ``cached_from`` — ``_series``
    already reads a missing source as an empty curve, which is what ``NO_CURVE``
    explains on the screen. The artifact bytes need no help: ``sweep_artifacts``
    keeps whatever a ``run_artifact`` row or a live message still names, so
    dropping the rows is enough and the hourly sweep reclaims the blobs.
    """
    # Reads the run, then decides whether to delete it — so it takes the write
    # lock up front rather than upgrading and losing to whichever flush
    # committed in between.
    with writing(session):
        run = session.get(Run, run_id)
        if run is None:
            raise HTTPException(status_code=404, detail="No such run")
        if run.status in ("running", "queued"):
            raise HTTPException(
                status_code=409,
                detail=(
                    f"Run {run_id} is {run.status}. Cancel it, or wait for it to "
                    "finish, before deleting it."
                ),
            )
        flow = run.flow
        session.execute(delete(RunNode).where(col(RunNode.run_id) == run_id))
        session.execute(delete(RunMetric).where(col(RunMetric.run_id) == run_id))
        session.execute(delete(RunArtifact).where(col(RunArtifact.run_id) == run_id))
        session.execute(delete(Run).where(col(Run.id) == run_id))
        session.commit()
    event_bus.publish(
        {
            "type": "audit",
            "action": f"deleted run {run_id}",
            "flow": flow,
            "user": user.email,
            "ts": time.time(),
        }
    )
    return Response(status_code=204)


def _series(session: Session, run_id: str, name: str = "") -> list[RunMetric]:
    """A run's numbers, including the ones a cached node points at.

    A cache hit replays no emissions, so a node restored from an earlier run has
    no rows of its own — it carries that run's id instead, and its series is read
    from there. Names are re-qualified on the way out, because the same node
    reached through two flows publishes under two names and the caller asked for
    this run's.
    """
    statement = select(RunMetric).where(col(RunMetric.run_id) == run_id)
    if name:
        statement = statement.where(col(RunMetric.name) == name)
    rows = list(session.exec(statement))

    restored = session.exec(
        select(RunNode).where(
            col(RunNode.run_id) == run_id, col(RunNode.cached_from) != ""
        )
    ).all()
    if restored:
        run = session.get(Run, run_id)
        flow = run.flow if run is not None else ""
        # Two queries for the lot, rather than a `Run` lookup and a `RunMetric`
        # query per restored node. A comparison of twenty runs calls this
        # twenty times, so the per-node round trips multiplied.
        sources = {
            row.id: row.flow
            for row in session.exec(
                select(Run).where(col(Run.id).in_({n.cached_from for n in restored}))
            )
        }
        wanted: dict[tuple[str, str], list[RunNode]] = {}
        for node_row in restored:
            source_flow = sources.get(node_row.cached_from)
            if source_flow is None:
                # The run it came from is gone — deleted with its flow. The
                # outputs are still on this run; the curve is not recoverable.
                continue
            key = (
                node_row.cached_from,
                requalify(node_row.node, flow, source_flow),
            )
            wanted.setdefault(key, []).append(node_row)
        if wanted:
            cached = session.exec(
                select(RunMetric).where(
                    col(RunMetric.run_id).in_({run_id for run_id, _ in wanted}),
                    col(RunMetric.node).in_({node for _, node in wanted}),
                )
            )
            for row in cached:
                for node_row in wanted.get((row.run_id, row.node), ()):
                    source_flow = sources[node_row.cached_from]
                    renamed = requalify(row.name, source_flow, flow)
                    if name and renamed != name:
                        continue
                    rows.append(
                        RunMetric(
                            run_id=run_id,
                            name=renamed,
                            step=row.step,
                            node=node_row.node,
                            ts=row.ts,
                            value=row.value,
                        )
                    )

    rows.sort(key=lambda row: (row.name, row.step))
    return rows


@router.get("/{run_id}/metrics", response_model=list[MetricPoint])
def read_metrics(
    run_id: str,
    session: SessionDep,
    name: str = "",
    stride: Annotated[int, Query(ge=1, le=100_000)] = 1,
) -> Any:
    """One metric's series, in step order — or every one of them, unnamed.

    ``stride`` thins a long curve down: 3000 steps drawn on a 400-pixel chart
    is 3000 points nobody can see.
    """
    rows = _series(session, run_id, name)
    if stride > 1:
        rows = rows[:: max(1, stride)]
    return rows


def _points(session: Session, run_id: str, metric: str, x: str) -> list[list[float]]:
    """One run's readings of ``metric``, against whichever x was asked for.

    The step is the default because it is what every run has. Time answers
    "which one got there sooner", and is measured from this run's own first
    reading so that runs started hours apart still lie on top of each other.
    Another metric answers "against what the loop was actually counting" — an
    epoch, or samples seen — and is joined on the step the two share, which is
    the only thing they have in common.
    """
    rows = _series(session, run_id, metric)
    if x == "time":
        if not rows:
            return []
        start = min(row.ts for row in rows)
        return [[row.ts - start, row.value] for row in rows]
    if x and x != "step":
        against = {row.step: row.value for row in _series(session, run_id, x)}
        return [[against[row.step], row.value] for row in rows if row.step in against]
    return [[float(row.step), row.value] for row in rows]


@router.get("/series/compare", response_model=SeriesAnswer)
def compare_metric(session: SessionDep, ids: str, metric: str, x: str = "") -> Any:
    """One metric across several runs, as the chart widget's series shape.

    This is the comparison view: it answers in the same shape a flow answers a
    chart's query with, so putting three training curves beside each other is
    a widget binding rather than a screen of its own.

    ``x`` names what to plot against — nothing or "step", "time", or another
    metric of the same runs.
    """
    run_ids = [part for part in ids.split(",") if part]
    if not run_ids:
        raise HTTPException(status_code=422, detail="Name at least one run")
    runs = {
        run.id: run
        for run in session.exec(select(Run).where(col(Run.id).in_(run_ids))).all()
    }
    lines: list[MetricSeries] = []
    for run_id in run_ids:
        run = runs.get(run_id)
        if run is None:
            continue
        label = run_id
        if run.seed is not None:
            label = f"{run_id} (seed {run.seed})"
        lines.append(
            MetricSeries(label=label, points=_points(session, run_id, metric, x))
        )
    return SeriesAnswer(metric=metric, x=x or "step", lines=lines)