diff --git a/Makefile b/Makefile index 827d1dc..30ab93f 100644 --- a/Makefile +++ b/Makefile @@ -3,7 +3,7 @@ # The workspace root delegates to these (see ../Makefile). .PHONY: dev-utils dev dev-local up down update install dev-backend dev-frontend \ - generate-client seed-example test test-backend test-frontend soak bench-startup lint lint-backend \ + generate-client seed-example seed-demo test test-backend test-frontend soak bench-startup lint lint-backend \ lint-frontend umami clean help COMPOSE_ROOT := $(CURDIR) @@ -94,6 +94,9 @@ generate-client: ## Regenerate the frontend SDK from the backend's OpenAPI sche seed-example: ## Seed the querying-chart example (needs a running stack + InfluxDB) cd backend && uv run python ../scripts/seed_example_chart.py +seed-demo: ## Seed the training-run example: a batch flow and its dashboard + cd backend && uv run python ../scripts/seed_demo_training.py + # ── Testing ─────────────────────────────────────────────────────── test: test-backend test-frontend ## Run all tests (backend + frontend) diff --git a/NOTEPAD.md b/NOTEPAD.md index 82fed35..072b1eb 100644 --- a/NOTEPAD.md +++ b/NOTEPAD.md @@ -13,6 +13,7 @@ should reopen it. ### To be sorted +- BUG/UI find a good way to represent the dictionary style of messages (truncated test, click to enlarge.. ). Show dtype instead of actual values -> actual values when clicking expand - BUG/UI loop-lag in home view can't have a value when there are no flows (push back if I'm wrong here) - BUG/UI brain view in the home view should be hidden when there are no flows/nodes - FEAT/UI in addition to the two issues above we should introduce a default flow and dashboard for every new user. These should be very minimal to quickly understand what's happening but still show off most of the features of the application. We could make this optional by offering a "Create Demo" button showing at the location of the brain view if there are no existing flows. This demo could then also serve as a demo.fluksio.com site later to play around without signup @@ -220,6 +221,9 @@ Open on purpose. Each names what should bring it back. - CHORE/INFRA: NATS JetStream as the work-queue backend — durable streams whose consumer semantics match the `WorkQueue` interface, in one small binary. Reopen with M5 remote workers, when the queue crosses hosts. NOTE: remote workers landed without it — a worker dials the engine's own socket and never touches Redis, so the queue still does not cross a host. Reopen if a second engine ever pulls from the same stream. - FEAT/RUNS: stage caching. `run_node.cache_key` is written on every run and the artifact store is content-addressed, so the pieces are in place; what is missing is computing the key from the node's source digest plus its input values and skipping a node whose key already has an `ok` row with its artifacts still present. The two research repos want this more than they want resume — neither persists checkpoints, and both re-run unchanged preprocessing every time. - FEAT/RUNS: per-label requirements overlays (`requirements-gpu.txt`) synced into a remote worker's venv, with drift surfaced against the engine's manifest. Today a worker's environment is whatever `--python` points at, which is fine for one hand-managed GPU box and not for several. `venv_digest` already arrives at attach and is shown on `/workers`, so the reporting half exists. +- FEAT/UI: a dashboard shows a run's curve only while it is running. Emissions reach the socket live, but a run's values live in its own state namespace, so reloading the panel afterwards leaves the chart empty — the durable series is on the run (`/runs/{id}/metrics`) and nothing binds a widget to it. A chart variant that reads a run's series, or the existing querying chart pointed at `/runs/series/compare`, is what would close it. This is also what a demo needs to show a finished experiment rather than only a live one. +- FEAT/UI: nothing submits a run with parameters from the UI. Pressing Run on a batch flow submits one with the declared defaults, which is enough to try it; choosing parameters, or launching a sweep, is API-only. A form generated from the flow's `inputs` is the obvious shape. +- FEAT/UI: `scripts/seed_demo_training.py` (`make seed-demo`) builds a batch flow and dashboard covering runs, streaming outputs, artifacts, device routing and sweeps. It is the content a "Create Demo" button would seed; what it needs to become one is a route that runs it server-side for the signed-in user. - FEAT/RUNS: a run detail screen. The API answers everything — params, per-node status with logs and tracebacks, artifacts, metrics, and `/runs/series/compare` in the chart widget's own `series` shape — but nothing in the dashboard reads it yet, so a run is inspected over HTTP. Comparing curves is a widget binding once someone builds the page around it. - FEAT/RUNS: a thin client CLI (`fluksio run/runs/sweep/worker`) over the same API. The engine being resident is what makes runs cheap; a CLI is ergonomics on top, and `curl` covers it until someone is running sweeps daily. - FEAT/RUNS: the step on a run's series is the count of emissions on that message, so a node yielding every tenth training step records steps 0, 1, 2 rather than 0, 10, 20 — a faithful x-axis of its own emissions, not of the loop inside it. If a real step number ever matters, a `record`-typed streaming port carrying its own `step` is the shape to read it from; the column is already there. diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py index 5335c8d..8a2ab5e 100644 --- a/backend/app/api/routes/flows.py +++ b/backend/app/api/routes/flows.py @@ -8,6 +8,7 @@ from fastapi import ( APIRouter, Depends, HTTPException, + Request, WebSocket, WebSocketDisconnect, ) @@ -27,6 +28,7 @@ from app.flow.dashboards import DashboardStore from app.flow.events import event_bus from app.flow.messages import qualify from app.flow.pipeline import ValidationIssue +from app.flow.runs import RunRejected from app.flow.schemas import ( NAME_PATTERN, BrainGraph, @@ -616,15 +618,38 @@ def validate_flow(name: str, controller: FlowControllerDep) -> Any: async def run_flow( name: str, body: RunRequest, + request: Request, controller: FlowControllerDep, + user: CurrentUser, ) -> Any: """Run every node of a flow once. With unpublished changes this runs the draft, so the button matches what is on the canvas. Nothing is deployed by running it. + + A *batch* flow is submitted as a run instead, because that is what running + one means: it is what keeps the parameters, the series and the result, and + a button that quietly did something else would be a trap. """ - _read_flow(controller, name) + definition = _read_flow(controller, name) _require_enabled(controller, name) + if definition.mode == "batch": + service = getattr(request.app.state, "run_service", None) + if service is None: + raise HTTPException(status_code=503, detail="Runs are not available") + try: + await run_in_threadpool( + service.submit, + name, + params=body.inputs, + cause="api", + actor=user.email, + draft=controller.store.has_draft(name), + ) + except RunRejected as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + return _flow_state(controller, name) + inputs = {qualify(name, key): value for key, value in body.inputs.items()} if controller.store.has_draft(name): await run_in_threadpool(controller.run_preview, name, inputs) diff --git a/backend/app/flow/runs.py b/backend/app/flow/runs.py index 8a363a2..d3cac98 100644 --- a/backend/app/flow/runs.py +++ b/backend/app/flow/runs.py @@ -124,15 +124,35 @@ def batch_issues(flow: FlowDef) -> list[str]: def required_labels(flow: FlowDef) -> list[str]: - """Worker labels this flow's nodes ask for.""" + """Worker labels this flow cannot run without. + + Only the nodes that *require* their device. One that merely prefers it runs + on the engine when no such worker is attached, so holding the whole run + back to wait for one would be waiting for something it does not need. + """ return sorted( - {node.device for node in flow.nodes if node.device and node.device.strip()} + { + node.device + for node in flow.nodes + if node.device and node.device.strip() and node.device_policy == "require" + } ) -def seed_values(flow: FlowDef, params: dict[str, Any]) -> dict[str, Any]: - """Turn a run's parameters into the messages the flow starts from.""" +def seed_values( + flow: FlowDef, params: dict[str, Any], seed: int | None = None +) -> dict[str, Any]: + """Turn a run's parameters into the messages the flow starts from. + + A run's ``seed`` fills an input of that name when the flow declares one, so + the field that distinguishes two runs of one configuration is the same + number the flow draws from — otherwise it would only look like it was, and + a sweep over seeds would run the same experiment N times. An explicit + parameter still wins, and a flow that declares no ``seed`` ignores it. + """ specs = {declared.spec.name: declared.spec for declared in flow.inputs} + if seed is not None and "seed" in specs and "seed" not in params: + params = {**params, "seed": seed} values: dict[str, Any] = {} for key, value in params.items(): spec = specs.get(key) @@ -335,7 +355,7 @@ class RunService: params = params or {} # Checked here rather than in the driver: a caller who mistyped a # parameter should be told now, not by a run that fails in a minute. - seed_values(flow, params) + seed_values(flow, params, seed) run = Run( id=new_run_id(), @@ -580,7 +600,7 @@ class RunService: if run_id in self._cancelled: pipeline.pause(flow.name) self._publish(run, "run_started") - pipeline.run(seed_values(flow, run.params)) + pipeline.run(seed_values(flow, run.params, run.seed)) result = collect_result(flow, state) with self._lock: cancelled = run_id in self._cancelled diff --git a/backend/tests/flow/test_runs.py b/backend/tests/flow/test_runs.py index 7a64ada..92dd664 100644 --- a/backend/tests/flow/test_runs.py +++ b/backend/tests/flow/test_runs.py @@ -308,3 +308,28 @@ def test_labels_come_from_the_nodes_that_ask_for_a_device(): assert required_labels(flow) == [] flow.nodes[0].device = "gpu" assert required_labels(flow) == ["gpu"] + + +def test_a_preferred_device_does_not_hold_a_run_back(): + flow = double_flow() + flow.nodes[0].device = "gpu" + flow.nodes[0].device_policy = "prefer" + # It runs on the engine when no such worker is attached, so waiting for + # one would be waiting for something the run does not need. + assert required_labels(flow) == [] + + +def test_a_runs_seed_fills_an_input_of_that_name(): + flow = double_flow() + flow.inputs.append(FlowInput(spec=spec("seed", DType.INT), initial=0)) + + # Otherwise the field that tells two runs of one configuration apart would + # only look like the number the flow draws from. + assert seed_values(flow, {}, seed=7)["study.seed"] == 7 + # An explicit parameter still wins. + assert seed_values(flow, {"seed": 3}, seed=7)["study.seed"] == 3 + + +def test_a_flow_without_a_seed_input_ignores_the_runs_seed(): + flow = double_flow() + assert seed_values(flow, {"lr": 1.0}, seed=7) == {"study.lr": 1.0} diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 56b7e92..13a368e 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -736,6 +736,10 @@ export class FlowsService { * * With unpublished changes this runs the draft, so the button matches what is * on the canvas. Nothing is deployed by running it. + * + * A *batch* flow is submitted as a run instead, because that is what running + * one means: it is what keeps the parameters, the series and the result, and + * a button that quietly did something else would be a trap. * @param data The data for the request. * @param data.name * @param data.requestBody