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.
"""

from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel, Field
from sqlmodel import col, select

from fluksio.api.deps import CurrentUser, SessionDep, get_current_user
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


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


class SweepEntry(BaseModel):
    params: dict[str, Any] = Field(default_factory=dict)
    seed: int | None = None


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


class RunNodeRow(BaseModel):
    node: str
    status: str
    attempt: int
    duration_ms: float
    worker: str
    error: str
    logs: str


class ArtifactRow(BaseModel):
    name: str
    node: str
    digest: str
    size: int
    media_type: str


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 = ""
    seed: int | None
    group_id: str | None
    labels: list[str]
    created_at: Any
    started_at: Any = None
    finished_at: Any = None
    duration_ms: float
    actor: str


class RunDetail(RunRow):
    result: dict[str, Any] = Field(default_factory=dict)
    commit: str = ""
    flow_version: int = 1
    nodes: list[RunNodeRow] = Field(default_factory=list)
    artifacts: list[ArtifactRow] = Field(default_factory=list)


class MetricPoint(BaseModel):
    step: int
    ts: float
    value: float


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
    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="api",
            actor=user.email,
            draft=body.draft,
        )
    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,
            )
            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,
    limit: int = 50,
) -> Any:
    """Runs, newest first. The queryable table an experiment log needs."""
    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)
    return list(session.exec(statement.limit(min(limit, 500))))


@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.get("/{run_id}/metrics", response_model=list[MetricPoint])
def read_metrics(run_id: str, session: SessionDep, name: str, stride: int = 1) -> Any:
    """One metric's series, in step order.

    ``stride`` thins a long curve down: 3000 steps drawn on a 400-pixel chart
    is 3000 points nobody can see.
    """
    statement = (
        select(RunMetric)
        .where(col(RunMetric.run_id) == run_id, col(RunMetric.name) == name)
        .order_by(col(RunMetric.step))
    )
    rows = list(session.exec(statement))
    if stride > 1:
        rows = rows[:: max(1, stride)]
    return rows


@router.get("/series/compare", response_model=SeriesAnswer)
def compare_metric(session: SessionDep, ids: str, metric: 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.
    """
    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
        rows = session.exec(
            select(RunMetric)
            .where(col(RunMetric.run_id) == run_id, col(RunMetric.name) == metric)
            .order_by(col(RunMetric.step))
        ).all()
        label = run_id
        if run.seed is not None:
            label = f"{run_id} (seed {run.seed})"
        lines.append(
            MetricSeries(
                label=label, points=[[float(row.step), row.value] for row in rows]
            )
        )
    return SeriesAnswer(metric=metric, lines=lines)