From 5b122341d5fb2888c92b7e74ebc3dbb7fb1eee0f Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 2 Sep 2026 15:10:24 +0200 Subject: [PATCH] New run: start a run from the app, on the working copy The site promises simulated inputs and mocked sensor values, and nothing in the app was that. A run already is: the values are the caller's, the state is the run's own namespace, and nothing it computes reaches the live flow. What was missing was a screen to do it from, and the draft flag being honoured. `/runs/new` is a flow, a field per declared input, a seed and Run; `/runs` stays the log. A comma-separated list in a number field expands into the grid `fluksio sweep --param` builds and goes to the sweep route, so launching one no longer needs a terminal. Only numbers split: a comma in a string is content, and one in JSON is syntax. `RunCreate.draft` was validated at submit and dropped before the run executed, so "try the working copy" ran the published one. `Run.draft` is a column now, the driver reads the same copy the submit checked, and a retry carries it. `FlowSummary.mode` came with it so the rail can say which flows are batch before one is picked. Also here: a Retry button on a finished run, which the route has always had and the UI never did, and parameter cells truncated to their column with the full value on hover. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_013TTfoK82awm8wvxXhHz3XF --- .../versions/a8f6f79e40be_run_draft.py | 33 ++ backend/fluksio/api/routes/flows.py | 1 + backend/fluksio/api/routes/runs.py | 2 + backend/fluksio/flow/runs.py | 13 +- backend/fluksio/flow/schemas.py | 2 + backend/fluksio/models.py | 3 + backend/tests/api/routes/test_runs.py | 56 ++ docs/concepts/runs.md | 25 +- docs/interface/index.md | 2 + frontend/src/client/schemas.gen.ts | 16 + frontend/src/client/sdk.gen.ts | 8 +- frontend/src/client/types.gen.ts | 4 + frontend/src/components/Flow/RunDialog.tsx | 165 +++--- frontend/src/components/Runs/NewRun.tsx | 486 ++++++++++++++++++ frontend/src/components/Runs/RunDetail.tsx | 25 +- frontend/src/components/Runs/RunsScreen.tsx | 84 ++- frontend/src/components/Runs/queries.ts | 20 + frontend/src/routeTree.gen.ts | 21 + frontend/src/routes/_layout/runs/new.tsx | 26 + frontend/tests/runs.spec.ts | 25 + 20 files changed, 926 insertions(+), 91 deletions(-) create mode 100644 backend/fluksio/alembic/versions/a8f6f79e40be_run_draft.py create mode 100644 frontend/src/components/Runs/NewRun.tsx create mode 100644 frontend/src/routes/_layout/runs/new.tsx diff --git a/backend/fluksio/alembic/versions/a8f6f79e40be_run_draft.py b/backend/fluksio/alembic/versions/a8f6f79e40be_run_draft.py new file mode 100644 index 0000000..1cb49d6 --- /dev/null +++ b/backend/fluksio/alembic/versions/a8f6f79e40be_run_draft.py @@ -0,0 +1,33 @@ +"""run.draft + +A run submitted against the working copy validated the draft and then executed +what was published. The row now records which copy was asked for, so the +driver reads the same one the submit checked. + +Revision ID: a8f6f79e40be +Revises: e7d3b1a9c624 +Create Date: 2026-09-02 + +""" + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision = "a8f6f79e40be" +down_revision = "e7d3b1a9c624" +branch_labels = None +depends_on = None + + +def upgrade(): + # Not null with a default, so every run recorded before this reads as what + # it was: a run of the published flow. + op.add_column( + "run", + sa.Column("draft", sa.Boolean(), nullable=False, server_default=sa.text("0")), + ) + + +def downgrade(): + op.drop_column("run", "draft") diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 7c6992f..5519c6d 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -303,6 +303,7 @@ def read_flows(controller: FlowControllerDep) -> Any: FlowSummary( name=definition.name, title=definition.title, + mode=definition.mode, node_count=len(definition.nodes), error_count=sum(1 for s in statuses if s.status == "error"), has_draft=controller.store.has_draft(name), diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py index 2b86006..169bf2b 100644 --- a/backend/fluksio/api/routes/runs.py +++ b/backend/fluksio/api/routes/runs.py @@ -134,6 +134,8 @@ class RunRow(BaseModel): commit: str = "" seed: int | None group_id: str | None + #: Whether it ran the working copy rather than what is published. + draft: bool = False labels: list[str] created_at: Any started_at: Any = None diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 867bc83..b143333 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -906,6 +906,7 @@ class RunService: group_id=group_id, cause=cause, no_cache=no_cache, + draft=draft, status="queued", labels=required_labels(flow), needs=required_resources(flow), @@ -973,6 +974,10 @@ class RunService: inputs, the same group, so a sweep is completed rather than repeated. The stage cache is what makes it cheap — the nodes that finished are restored rather than run again. + + A run of the working copy retries as one, which means whatever the + draft is now rather than what it was — there is no older draft to go + back to, and the point of retrying a draft run is the code on disk. """ with Session(db_engine) as session: run = session.get(Run, run_id) @@ -989,6 +994,7 @@ class RunService: cause="retry", actor=actor, no_cache=source.no_cache, + draft=source.draft, parent_id=source.id, ) @@ -1203,17 +1209,20 @@ class RunService: sink = MetricSink(run_id) try: - flow = self.controller.store.read_flow(run.flow) + flow = self.controller.store.read_flow(run.flow, draft=run.draft) # Read again, now that it is this run's turn: a sweep queues every # run at once, and the code on disk is free to move in the hours # before the last of them starts. What the record must name is the - # state that ran, not the state that was submitted. + # state that ran, not the state that was submitted. A draft that + # was published meanwhile reads as the published copy, which is + # the same document under a different name. digests = node_digests(flow) run.code_digest = self._restamp(run, run_digest(flow)) state = self._state_factory(f"{RUN_NAMESPACE}:{run_id}") pipeline = self.controller.build_run_pipeline( flow, state=state, + draft=run.draft, observer=observe, emission_observer=sink.handle, run=RunContext( diff --git a/backend/fluksio/flow/schemas.py b/backend/fluksio/flow/schemas.py index 71acdcb..a0e47af 100644 --- a/backend/fluksio/flow/schemas.py +++ b/backend/fluksio/flow/schemas.py @@ -355,6 +355,8 @@ class MessageHistory(BaseModel): class FlowSummary(BaseModel): name: str title: str = "" + #: Whether running it means one finite execution or leaving it running. + mode: Literal["live", "batch"] = "live" node_count: int = 0 error_count: int = 0 has_draft: bool = False diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 726d721..e36649c 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -374,6 +374,9 @@ class Run(SQLModel, table=True): cause: str = Field(default="api", max_length=32) #: Re-execute every node, whatever the stage cache holds for it. no_cache: bool = False + #: Run the unpublished working copy rather than what is published. Read + #: again when the run starts, so a draft published in between is what runs. + draft: bool = False #: queued, running, ok, error, cancelled or abandoned. status: str = Field(default="queued", index=True, max_length=16) #: Why it is where it is: what it waits for, or what went wrong. diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index f4d20b4..8df5a0f 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -14,6 +14,7 @@ from sqlmodel import Session, col, select from fluksio.core.config import settings from fluksio.core.db import engine as db_engine from fluksio.flow.artifacts import ArtifactStore +from fluksio.flow.controller import FlowController from fluksio.flow.messages import DType, MessageSpec from fluksio.flow.pipeline import NodeOutcome from fluksio.flow.runs import ( @@ -27,6 +28,7 @@ from fluksio.flow.runs import ( seed_values, ) from fluksio.flow.schemas import FlowDef, FlowInput, NodeDef +from fluksio.flow.store import FlowStore from fluksio.models import Run, RunArtifact, RunMetric, RunNode @@ -492,6 +494,56 @@ def test_the_seed_is_recorded_the_same_way_however_it_arrived(): session.commit() +DRAFT_SOURCE = "def process():\n return {'answer': 'draft'}\n" +PUBLISHED_SOURCE = "def process():\n return {'answer': 'published'}\n" + + +def test_a_draft_run_executes_the_draft(tmp_path): + """The flag was checked at submit and forgotten by the time it ran. + + Which made trying an edit before publishing it impossible from anywhere: + the submit validated the working copy and the driver then executed what + was published, and the two only agree when there is no draft. + """ + flow = FlowDef( + name="study", + mode="batch", + outputs=["answer"], + nodes=[ + NodeDef( + id="answer", + provides=[MessageSpec(name="answer", dtype=DType.STR)], + ) + ], + ) + store = FlowStore(tmp_path / "flows") + store.write_flow(flow) + store.write_node_source("study", "answer", PUBLISHED_SOURCE) + store.write_node_source("study", "answer", DRAFT_SOURCE, draft=True) + + service = RunService(controller=FlowController(store), queue=_Collect()) + made = [] + try: + for draft, expected in ((True, "draft"), (False, "published")): + run = service.submit("study", {}, draft=draft) + made.append(run.id) + assert run.draft is draft + service._drive(run.id) + with Session(db_engine) as session: + stored = session.get(Run, run.id) + assert stored.status == "ok", stored.status_reason + assert stored.result == {"answer": expected} + finally: + with Session(db_engine) as session: + for run in session.exec(select(Run).where(col(Run.id).in_(made))).all(): + session.delete(run) + for node in session.exec( + select(RunNode).where(col(RunNode.run_id).in_(made)) + ).all(): + session.delete(node) + session.commit() + + def test_a_retry_is_a_new_run_that_names_the_one_it_repeats(): """The way back from a run an engine restart interrupted. @@ -516,6 +568,7 @@ def test_a_retry_is_a_new_run_that_names_the_one_it_repeats(): status="abandoned", params={"lr": 0.3}, group_id="sweep-9", + draft=True, created_at=datetime.now(UTC), ) ) @@ -537,6 +590,9 @@ def test_a_retry_is_a_new_run_that_names_the_one_it_repeats(): "sweep-9", "retry", ) + # What it was a run of comes with it: retrying a run of the working + # copy that silently ran the published one would say nothing at all. + assert again.draft is True assert queue.items[-1].run_id == again.id # A run that has not finished is cancelled, not retried. diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index 99866f3..cf5a4ca 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -41,6 +41,13 @@ value would be dropped rather than delayed, so submitting is refused instead. ## Submitting +**Runs → New run** in the app is the same call with a form in front of it: pick +a flow, fill in its declared inputs, press Run. A comma-separated list in a +number field runs every combination of them as a sweep. **Use draft** runs the +working copy rather than what is published, which is how a change is tried +before it is deployed, and each run keeps its own state, so nothing it computes +reaches the live flow. + ```bash curl -X POST $FLUKSIO/runs/flows/train_polymer_gnn \ -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \ @@ -54,6 +61,11 @@ result, per-node record and artifacts. Wrong parameters are refused before anything executes: an undeclared name, or a value of the wrong type, comes back as a 422 naming the problem. +`"draft": true` on the body runs the unpublished working copy — the flow +document and the node sources the editor is showing. The copy is read again +when the run starts, so a draft published while it waited runs as the +published one. + ### Sweeps An ensemble is the same parameters at different seeds; a grid search is the @@ -313,10 +325,11 @@ that for the runs of a sweep that did not finish. ## Looking at what ran -The **Runs** screen is the experiment log: every run newest-first, filtered by -flow, by status, or down to one sweep. A sweep is worth filtering to, since the -table then draws a column per parameter that actually varied, which is what -makes fifty runs of one flow readable. +**New run** is where one is started, and the **Runs** screen is the experiment +log: every run newest-first, filtered by flow, by status, or down to one +sweep. A sweep is worth filtering to, since the table then draws a column per +parameter that actually varied, which is what makes fifty runs of one flow +readable. Tick two or more and their curves go side by side. Shift-click to take a range, or the header box to take everything on screen. That comparison is the @@ -329,7 +342,9 @@ still lie on top of each other; or against another metric of the same runs (an epoch, or samples seen) joined on the step the two share. One run in full is params, the per-node record with its logs and traceback, -the artifacts it made, its metrics and its result. +the artifacts it made, its metrics and its result. **Retry** submits it again +as a run of its own, keeping the flow, the inputs, the seed and the sweep it +belonged to. ### Taking it into a dataframe diff --git a/docs/interface/index.md b/docs/interface/index.md index 7207b03..e2680c1 100644 --- a/docs/interface/index.md +++ b/docs/interface/index.md @@ -18,6 +18,8 @@ phone the sidebar collapses to a sheet. | **Home** | the brain graph, health, and everything that recently happened | | **Flows** | the list of flows, and the canvas for each | | **Dashboards** | the widget canvases, and the panels that display them | +| **Runs** | starting a run with values you choose, and the log of what ran | +| **Workers** | the machines that nodes can be placed on | | **Secrets** | credentials your nodes reference without holding | | **Modules** | the Python packages your node code may import | | **Alerts** | where failures get sent | diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 69e9df2..ec8bf58 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -1263,6 +1263,12 @@ export const FlowSummarySchema = { title: 'Title', default: '' }, + mode: { + type: 'string', + enum: ['live', 'batch'], + title: 'Mode', + default: 'live' + }, node_count: { type: 'integer', title: 'Node Count', @@ -2849,6 +2855,11 @@ export const RunDetailSchema = { ], title: 'Group Id' }, + draft: { + type: 'boolean', + title: 'Draft', + default: false + }, labels: { items: { type: 'string' @@ -4123,6 +4134,11 @@ export const fluksio__api__routes__runs__RunRowSchema = { ], title: 'Group Id' }, + draft: { + type: 'boolean', + title: 'Draft', + default: false + }, labels: { items: { type: 'string' diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 3ef869d..c51d483 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -123,9 +123,14 @@ export class ArtifactsService { * legitimate a body here as a checkpoint, and neither should have to fit in * memory twice. Capped, because nothing else here was: any account, and any * worker credential, could otherwise fill the data volume. + * + * ``volatile`` puts it in the ring instead of the store — a frame a screen is + * watching now, which the oldest of falls out of memory rather than being + * kept. A worker on another host publishing a camera sends this. * @param data The data for the request. * @param data.name * @param data.mediaType + * @param data.volatile * @returns ArtifactRef Successful Response * @throws ApiError */ @@ -135,7 +140,8 @@ export class ArtifactsService { url: '/api/v1/artifacts', query: { name: data.name, - media_type: data.mediaType + media_type: data.mediaType, + volatile: data.volatile }, errors: { 422: 'Validation Error' diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 428c47d..9b55317 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -400,6 +400,7 @@ export type FlowStatePublic = { export type FlowSummary = { name: string; title?: string; + mode?: 'live' | 'batch'; node_count?: number; error_count?: number; has_draft?: boolean; @@ -470,6 +471,7 @@ export type fluksio__api__routes__runs__RunRow = { commit?: string; seed: (number | null); group_id: (string | null); + draft?: boolean; labels: Array<(string)>; created_at: unknown; started_at?: unknown; @@ -1018,6 +1020,7 @@ export type RunDetail = { commit?: string; seed: (number | null); group_id: (string | null); + draft?: boolean; labels: Array<(string)>; created_at: unknown; started_at?: unknown; @@ -1356,6 +1359,7 @@ export type AlertsTestChannelResponse = (Message); export type ArtifactsPutArtifactData = { mediaType?: string; name?: string; + volatile?: boolean; }; export type ArtifactsPutArtifactResponse = (ArtifactRef); diff --git a/frontend/src/components/Flow/RunDialog.tsx b/frontend/src/components/Flow/RunDialog.tsx index 329b015..eb80859 100644 --- a/frontend/src/components/Flow/RunDialog.tsx +++ b/frontend/src/components/Flow/RunDialog.tsx @@ -1,6 +1,6 @@ import { useEffect, useState } from "react" -import type { FlowDef_Input } from "@/client" +import type { DType, FlowDef_Input, FlowInput_Input } from "@/client" import { Button } from "@/components/ui/button" import { Dialog, @@ -22,6 +22,83 @@ import { import { parseByDtype } from "./FlowBoundary" import { asText } from "./NodePanel" +/** + * A field per declared input, the way its type is entered. + * + * Held as text rather than as values: half-typed input is normal ("-", "1.", + * a JSON object with one brace), and a list of them is a sweep — both of + * which a parsed value would have thrown away. The caller parses on submit. + */ +export function ParamFields({ + inputs, + values, + placeholderFor, + onChange, +}: { + inputs: FlowInput_Input[] + values: Record + /** What an empty field says it takes, when the default is not enough. */ + placeholderFor?: (dtype: DType | undefined) => string + onChange: (values: Record) => void +}) { + return ( +
+ {inputs.map((declared) => { + const name = declared.spec?.name ?? "" + const dtype = declared.spec?.dtype + return ( +
+ + {dtype === "bool" ? ( + + ) : ( + . or sha256:…" + : `${dtype ?? "float"} or @run:.`) + } + className="text-sm" + onChange={(event) => + onChange({ ...values, [name]: event.target.value }) + } + /> + )} +
+ ) + })} +
+ ) +} + +/** What the fields start from: the declared value of each, as text. */ +export function initialText(inputs: FlowInput_Input[]): Record { + return Object.fromEntries( + inputs.map((one) => [one.spec?.name ?? "", asText(one.initial)]), + ) +} + +/** The declared inputs worth drawing a field for. */ +export function declaredInputs(definition: FlowDef_Input): FlowInput_Input[] { + return (definition.inputs ?? []).filter((one) => one.spec?.name) +} + /** * The parameters of one run, taken from the flow's inputs. * @@ -42,8 +119,8 @@ export function RunDialog({ onOpenChange: (open: boolean) => void onRun: (params: Record) => void }) { - const inputs = (definition.inputs ?? []).filter((one) => one.spec?.name) - const [values, setValues] = useState>({}) + const inputs = declaredInputs(definition) + const [values, setValues] = useState>({}) // Opening is what fills the form: an edit to the flow between two runs // should show up, and the last run's values should not linger. The inputs @@ -51,11 +128,7 @@ export function RunDialog({ // biome-ignore lint/correctness/useExhaustiveDependencies: opening is the dependency. useEffect(() => { if (!open) return - setValues( - Object.fromEntries( - inputs.map((one) => [one.spec?.name ?? "", one.initial ?? null]), - ), - ) + setValues(initialText(inputs)) }, [open]) return ( @@ -70,52 +143,7 @@ export function RunDialog({ -
- {inputs.map((declared) => { - const name = declared.spec?.name ?? "" - const dtype = declared.spec?.dtype - return ( -
- - {dtype === "bool" ? ( - - ) : ( - . or sha256:…" - : `${dtype ?? "float"} or @run:.` - } - className="text-sm" - onChange={(event) => - setValues({ - ...values, - [name]: parseByDtype(dtype, event.target.value), - }) - } - /> - )} -
- ) - })} -
+ @@ -142,3 +162,22 @@ export function RunDialog({ ) } + +/** + * The fields as the engine takes them. + * + * A field nobody filled in is left out, which is what keeps its declared + * value — an empty parameter is not the same as a zero. + */ +export function parseParams( + inputs: FlowInput_Input[], + values: Record, +): Record { + const params: Record = {} + for (const declared of inputs) { + const name = declared.spec?.name ?? "" + const parsed = parseByDtype(declared.spec?.dtype, values[name] ?? "") + if (parsed !== null) params[name] = parsed + } + return params +} diff --git a/frontend/src/components/Runs/NewRun.tsx b/frontend/src/components/Runs/NewRun.tsx new file mode 100644 index 0000000..2167198 --- /dev/null +++ b/frontend/src/components/Runs/NewRun.tsx @@ -0,0 +1,486 @@ +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query" +import { Link, useNavigate } from "@tanstack/react-router" +import { FlaskConical, SquarePen } from "lucide-react" +import { useEffect, useState } from "react" + +import { + type DType, + type FlowInput_Input, + type fluksio__api__routes__runs__RunRow as RunRow, + RunsService, +} from "@/client" +import { flowQueryOptions, flowsQueryOptions } from "@/components/Flow/queries" +import { + declaredInputs, + initialText, + ParamFields, + parseParams, +} from "@/components/Flow/RunDialog" +import { FieldLabel } from "@/components/Flow/SidePanel" +import { Button } from "@/components/ui/button" +import { Input } from "@/components/ui/input" +import { Skeleton } from "@/components/ui/skeleton" +import { Switch } from "@/components/ui/switch" +import { + Table, + TableBody, + TableCell, + TableHead, + TableHeader, + TableRow, +} from "@/components/ui/table" +import useCustomToast from "@/hooks/useCustomToast" +import { cn, dur } from "@/lib/utils" +import { handleError } from "@/utils" +import { + CARD, + isLive, + paramsSummary, + runKeys, + runOverviewQueryOptions, + runQueryOptions, + runsListQueryOptions, + shortId, +} from "./queries" +import { Entry } from "./RunDetail" +import { RunStatusBadge, statusReason } from "./RunStatus" +import { FlowRail } from "./RunsScreen" + +/** How many recent runs of the picked flow the page keeps in front of you. */ +const RECENT = 8 + +/** + * Starting a run: a flow, values for its inputs, and what it did. + * + * A run is how this instance is tried without touching it — the values are + * yours rather than a sensor's, the state is the run's own, and nothing it + * computes reaches the live flow. The working copy is what it runs by + * default, so an edit is answered before it is published. + */ +export function NewRun({ + flow, + onPick, +}: { + flow?: string + onPick: (flow: string | undefined) => void +}) { + const { data: flows } = useQuery(flowsQueryOptions()) + const { data: overview } = useQuery(runOverviewQueryOptions()) + + const counts = new Map((overview ?? []).map((row) => [row.flow, row])) + const rows = (flows?.data ?? []).map((one) => { + const seen = counts.get(one.name) + const notes = [ + one.mode === "batch" ? "batch" : "", + one.has_draft ? "draft" : "", + ].filter(Boolean) + return { + flow: one.name, + runs: seen?.runs ?? 0, + running: seen?.running ?? 0, + queued: seen?.queued ?? 0, + note: notes.join(" · ") || undefined, + } + }) + + return ( +
+ No flows yet. Draw one first, and it can be run from here.

+ } + onPick={onPick} + /> + +
+
+
+

New run

+

+ Runs a flow with the values you give it. Each run keeps its own + state, so nothing it computes reaches the live flow. +

+
+ {flow && ( + + )} +
+ + {flow ? ( + + ) : ( +
+ + Pick a flow to run. +
+ )} +
+
+ ) +} + +function RunForm({ flow }: { flow: string }) { + const client = useQueryClient() + const navigate = useNavigate() + const { showErrorToast } = useCustomToast() + const { data: detail, isPending } = useQuery(flowQueryOptions(flow)) + + const [values, setValues] = useState>({}) + const [seed, setSeed] = useState("") + const [draft, setDraft] = useState(true) + const [noCache, setNoCache] = useState(false) + // The run this page is watching: the one just started, or the newest of the + // flow when the page is arrived at with runs already behind it. + const [latest, setLatest] = useState() + + const definition = detail?.definition + const inputs: FlowInput_Input[] = definition ? declaredInputs(definition) : [] + const hasDraft = Boolean(detail?.has_draft) + + // Filling the form is what arriving at a flow does, and switching flows + // remounts this, so the declared values are read once per flow. + // biome-ignore lint/correctness/useExhaustiveDependencies: the definition arriving is the dependency. + useEffect(() => { + if (definition) setValues(initialText(inputs)) + }, [definition]) + + const { data: recent } = useQuery({ + ...runsListQueryOptions({ flow, limit: RECENT }), + refetchInterval: (query: { state: { data?: RunRow[] } }) => + (query.state.data ?? []).some((run) => isLive(run.status)) + ? 3_000 + : (false as const), + }) + + const grid = definition ? expand(inputs, values) : [] + const seeded = seed === "" ? null : Number(seed) + + // Which copy and which cache, shared by both shapes of submit. The seed is + // not: the sweep route takes one per entry, so that a grid can vary it. + const flags = { draft: draft && hasDraft, no_cache: noCache } + const withSeed = + seeded === null || Number.isNaN(seeded) ? {} : { seed: seeded } + + const submit = useMutation({ + mutationFn: async () => { + if (grid.length > 1) { + return RunsService.createSweep({ + name: flow, + requestBody: { + ...flags, + runs: grid.map((params) => ({ + params, + ...withSeed, + idempotency_key: submissionKey(), + })), + }, + }) + } + return RunsService.createRun({ + name: flow, + requestBody: { + ...flags, + ...withSeed, + params: grid[0] ?? {}, + idempotency_key: submissionKey(), + }, + }) + }, + onSuccess: (made) => { + client.invalidateQueries({ queryKey: runKeys.all }) + if (Array.isArray(made)) { + // A sweep is a table rather than a result: it goes where fifty runs + // are readable, filtered to the group it just made. + navigate({ + to: "/runs", + search: { flow, group: made[0]?.group_id ?? undefined }, + }) + return + } + setLatest(made.id) + }, + // The engine's own words: a refusal names the port or the node it is + // about, and a rewrite here would lose that. + onError: handleError.bind(showErrorToast), + }) + + if (isPending || !definition) { + return + } + + const watching = latest ?? recent?.[0]?.id + + return ( + <> +
+ {inputs.length === 0 ? ( +

+ This flow declares no inputs, so there is nothing to choose. Add one + in the flow panel to run it with a value. +

+ ) : ( + + )} + +
+
+ + Seed + + setSeed(event.target.value)} + /> +
+ + {hasDraft && ( +
+ + Use draft + + +
+ )} + +
+ + Skip cache + + +
+ + +
+ +

+ A comma-separated list in a number field runs every combination of + them as a sweep. +

+
+ + {watching && } + + {(recent?.length ?? 0) > 0 && ( +
+

Recent

+ + + + Run + Status + Parameters + Took + + + + + {(recent ?? []).map((run) => { + const summary = paramsSummary(run.params) + return ( + + + + {shortId(run.id)} + + + + + + + + {summary || "—"} + + + + {run.duration_ms ? dur(run.duration_ms) : "—"} + + + + + + ) + })} + +
+
+ )} + + ) +} + +/** What the run being watched did, without leaving the form to read it. */ +function Result({ id }: { id: string }) { + const { data: run } = useQuery(runQueryOptions(id)) + if (!run) return null + + const reason = statusReason(run) + const failed = (run.nodes ?? []).find((node) => node.error) + const result = Object.entries(run.result ?? {}) + + return ( +
+
+

+ {shortId(run.id)} +

+ + +
+ + {reason &&

{reason}

} + + {failed?.error && ( + 
+          {failed.node}: {failed.error}
+
+ )} + + {result.length > 0 ? ( +
+ {result.map(([name, value]) => ( + + ))} +
+ ) : ( + !isLive(run.status) && ( +

+ This run declares no result. Its numbers are on the run itself. +

+ ) + )} +
+ ) +} + +/** + * A key for one submission, so a retry after a timeout cannot double-submit. + * + * `crypto.randomUUID` is secure-context only, and this app is reached over + * plain HTTP on a LAN, where it is not defined at all — so the fallback is + * not theoretical. + */ +function submissionKey(): string { + if (typeof crypto?.randomUUID === "function") return crypto.randomUUID() + return `${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 10)}` +} + +/** A stored parameter back in the field it was typed into. */ +function asParamText(value: unknown): string { + if (value === null || value === undefined) return "" + if (typeof value === "object") return JSON.stringify(value) + return String(value) +} + +function placeholderFor(dtype: DType | undefined): string { + if (dtype === "artifact") return "@run:. or sha256:…" + if (dtype === "int" || dtype === "float") { + return `${dtype}, or a list for a sweep` + } + return `${dtype ?? "float"} or @run:.` +} + +/** + * The runs these fields ask for. + * + * A list in a number field is the sweep grammar, and the product of the lists + * is the grid — the same one `fluksio sweep --param` builds. Only numbers are + * split: a comma in a string is content, and one in JSON is syntax. + * + * ponytail: no per-field "this is a list" toggle. Add one if a numeric input + * ever legitimately takes a comma. + */ +export function expand( + inputs: FlowInput_Input[], + values: Record, +): Record[] { + let grid: Record[] = [{}] + for (const declared of inputs) { + const name = declared.spec?.name ?? "" + const dtype = declared.spec?.dtype + const raw = values[name] ?? "" + const listed = + (dtype === "int" || dtype === "float") && raw.includes(",") + ? raw.split(",").map((one) => one.trim()) + : [raw] + grid = grid.flatMap((row) => + listed.map((one) => { + const parsed = parseParams([declared], { [name]: one }) + return { ...row, ...parsed } + }), + ) + } + return grid +} diff --git a/frontend/src/components/Runs/RunDetail.tsx b/frontend/src/components/Runs/RunDetail.tsx index c70c086..5d60f14 100644 --- a/frontend/src/components/Runs/RunDetail.tsx +++ b/frontend/src/components/Runs/RunDetail.tsx @@ -1,6 +1,6 @@ import { useQuery } from "@tanstack/react-query" import { Link } from "@tanstack/react-router" -import { ChevronDown, ChevronRight, Download } from "lucide-react" +import { ChevronDown, ChevronRight, Download, RotateCcw } from "lucide-react" import { useState } from "react" import type { ArtifactRow, RunNodeRow } from "@/client" @@ -31,6 +31,7 @@ import { shortId, useCancelRun, useFlowInputs, + useRetryRun, } from "./queries" import { NodeStatusBadge, RunStatusBadge, statusReason } from "./RunStatus" @@ -39,6 +40,7 @@ const LABEL = "text-muted-foreground text-xs" export function RunDetail({ id }: { id: string }) { const { data: run, isPending } = useQuery(runQueryOptions(id)) const cancel = useCancelRun() + const retry = useRetryRun() const names = useMetricNames(id) const declared = useFlowInputs(run?.flow) const [metric, setMetric] = useState("") @@ -89,10 +91,15 @@ export function RunDetail({ id }: { id: string }) { in a sweep )} + {run.draft && ( + + ran the working copy + + )}
- {isLive(run.status) && ( + {isLive(run.status) ? ( + ) : ( + )}
@@ -193,7 +212,7 @@ export function RunDetail({ id }: { id: string }) { * record like any other, and serialising it onto one truncated line answers * nothing. */ -function Entry({ +export function Entry({ name, value, note, diff --git a/frontend/src/components/Runs/RunsScreen.tsx b/frontend/src/components/Runs/RunsScreen.tsx index 006bab5..0f357eb 100644 --- a/frontend/src/components/Runs/RunsScreen.tsx +++ b/frontend/src/components/Runs/RunsScreen.tsx @@ -4,7 +4,8 @@ import { useQueryClient, } from "@tanstack/react-query" import { Link } from "@tanstack/react-router" -import { Download, FlaskConical, Trash2, X } from "lucide-react" +import { Download, FlaskConical, Plus, Trash2, X } from "lucide-react" +import type { ReactNode } from "react" import { useRef, useState } from "react" import type { fluksio__api__routes__runs__RunRow as RunRow } from "@/client" @@ -153,6 +154,7 @@ export function RunsScreen({ update({ flow, group: undefined, compare: undefined }) } @@ -164,6 +166,17 @@ export function RunsScreen({ {search.flow ?? "Runs"} + +