diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py index cc2d2d3..ce4c54c 100644 --- a/backend/fluksio/flow/executor.py +++ b/backend/fluksio/flow/executor.py @@ -171,7 +171,6 @@ class ExecutionService: except Exception as exc: logger.error("Could not touch claimed work: %s", exc) - if now - last_reclaim < RECLAIM_INTERVAL_S: continue last_reclaim = now diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index b4d2bc9..d37c734 100644 --- a/backend/fluksio/flow/queue.py +++ b/backend/fluksio/flow/queue.py @@ -126,7 +126,7 @@ class WorkQueue(ABC): def ack(self, item: WorkItem) -> None: """Mark an item done, so it is never redelivered.""" - def touch(self, entry_ids: list[str]) -> None: + def touch(self, entry_ids: list[str]) -> None: # noqa: B027 """Say these items are still being worked on, not abandoned. A node with no timeout may run far longer than the reclaim window, and diff --git a/backend/fluksio/sdk/__init__.py b/backend/fluksio/sdk/__init__.py index d89fe69..5929cca 100644 --- a/backend/fluksio/sdk/__init__.py +++ b/backend/fluksio/sdk/__init__.py @@ -275,7 +275,9 @@ def node( if device_policy not in ("require", "prefer"): raise SyncError("device_policy is 'require' or 'prefer'") if timeout is not None and timeout < 0: - raise SyncError(f"node timeout must be 0 or more (0 disables it), got {timeout}") + raise SyncError( + f"node timeout must be 0 or more (0 disables it), got {timeout}" + ) def decorate(fn: F) -> F: required, _ = _ports(requires) diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 0579e96..cb2a249 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -1,4 +1,4 @@ -"""The `fluksio sync`, `run`, `runs`, `sweep` and `login` commands. +"""The `fluksio sync`, `run`, `runs`, `sweep`, `status` and `login` commands. Kept beside the SDK rather than in `fluksio.cli`: these are the client half of the tool, and none of them needs the engine to be importable — `--local`, which @@ -308,6 +308,33 @@ def _input_types(definition: dict[str, Any]) -> dict[str, str]: } +def _ask_params(definition: dict[str, Any]) -> dict[str, Any]: + """The run dialog, in a terminal: one line per declared input. + + An empty answer leaves the input out, which is what keeps its declared + value — the same thing the browser's dialog does with a field nobody + filled in. So pressing Enter through the lot runs the defaults. + """ + params: dict[str, Any] = {} + for entry in definition.get("inputs") or []: + spec = entry.get("spec") or {} + name = str(spec.get("name", "")) + if not name: + continue + dtype = str(spec.get("dtype", "float")) + initial = entry.get("initial") + shown = "" if initial is None else json.dumps(initial) + prompt = f"{name} ({dtype})" + (f" [{shown}]" if shown else "") + ": " + answer = input(prompt).strip() + if not answer: + continue + try: + params[name] = _coerce(answer, dtype) + except ValueError as exc: + raise SyncError(f"'{name}' takes {dtype}: {exc}") from exc + return params + + def _params(definition: dict[str, Any], rest: list[str]) -> dict[str, Any]: """Turn `--lr 0.05` into a typed parameter, using the flow's own inputs.""" types = _input_types(definition) @@ -420,7 +447,18 @@ def cmd_run(args: argparse.Namespace, rest: list[str]) -> int: stored = client.get_flow(args.flow) if stored is None: return _fail(f"no flow '{args.flow}' on that engine") - params = _params(stored.get("definition") or {}, rest) + definition = stored.get("definition") or {} + params = _params(definition, rest) + # Nothing on the command line and somebody watching: ask, the way + # pressing Run in the browser asks. A scripted run — piped, or + # carrying parameters already — is left exactly as it was. + if ( + not params + and not args.defaults + and (definition.get("inputs") or []) + and sys.stdin.isatty() + ): + params = _ask_params(definition) handle = client.submit( args.flow, params, seed=args.seed, no_cache=args.no_cache ) @@ -443,6 +481,147 @@ def cmd_run(args: argparse.Namespace, rest: list[str]) -> int: return _fail(str(exc)) +# --------------------------------------------------------------------------- +# Status +# +# The home screen's top half, in a terminal: how the engine is, what each flow +# is doing, and what failed recently. Drawn with rich, which is already here +# under fastapi's own CLI. +# --------------------------------------------------------------------------- + + +#: How often `--watch` asks again. The dashboard polls its summary every ten +#: seconds; nothing here moves faster than that. +WATCH_INTERVAL_S = 5.0 + + +def _flow_state(flow: dict[str, Any]) -> str: + if flow.get("quarantined"): + return "quarantined" + if not flow.get("enabled", True): + return "stopped" + if flow.get("paused"): + return "paused" + return "running" + + +def _status_screen(client: Client) -> Any: + """One frame: health, the flows, and the failures under them.""" + from rich.console import Group + from rich.table import Table + from rich.text import Text + + summary = client.summary() + flows = client.flows() + # Both, because an installation is usually one or the other: a live flow + # fails as an engine event, while a batch run fails on its own row. + failures = client.events(kind="failure", limit=5) + runs = client.runs(limit=5) + + healthy = summary.get("status") == "ok" + head = Text() + head.append( + "healthy" if healthy else str(summary.get("status", "unknown")), + style="green" if healthy else "yellow", + ) + problems = summary.get("problems") or [] + if problems: + head.append(" " + " · ".join(str(p) for p in problems), style="yellow") + + counts = summary.get("flows") or {} + nodes = summary.get("nodes") or {} + queue = summary.get("queue") or {} + lag = summary.get("loop_lag") or {} + facts = Text( + f"{counts.get('running', 0)}/{counts.get('total', 0)} flows running " + f"{nodes.get('total', 0)} nodes" + + (f", {nodes['error']} failed" if nodes.get("error") else "") + + f" queue {queue.get('pending', 0)} pending" + + (f", {queue['parked']} parked" if queue.get("parked") else "") + + f" loop lag {lag.get('ewma', 0.0) * 1000:.0f}ms", + style="dim", + ) + + table = Table(box=None, pad_edge=False, header_style="dim") + table.add_column("flow") + table.add_column("state") + table.add_column("nodes", justify="right") + table.add_column("") + for flow in sorted(flows, key=lambda one: str(one.get("name", ""))): + state = _flow_state(flow) + notes = [] + if flow.get("error_count"): + notes.append(f"{flow['error_count']} in error") + if flow.get("has_draft"): + notes.append("unpublished changes") + table.add_row( + str(flow.get("title") or flow.get("name", "")), + Text( + state, + style={"running": "green", "paused": "yellow"}.get(state, "red"), + ), + str(flow.get("node_count", 0)), + Text(" · ".join(notes), style="red" if flow.get("error_count") else "dim"), + ) + + parts: list[Any] = [head, facts, ""] + parts.append( + table if flows else Text("No flows yet. `fluksio sync` uploads yours.", "dim") + ) + if runs: + parts += ["", Text("recent runs", style="dim")] + for row in runs: + state = str(row.get("status", "")) + parts.append( + Text(f" {str(row.get('id', ''))[-8:]} ") + + Text( + f"{state:<9}", + style={"ok": "green", "cached": "cyan", "running": "cyan"}.get( + state, "red" if state == "error" else "yellow" + ), + ) + + Text( + f"{str(row.get('flow', '')):<16}" + f"{(row.get('duration_ms') or 0) / 1000:7.1f}s", + style="dim", + ) + ) + if failures: + parts += ["", Text("recent failures", style="dim")] + for event in failures: + where = " ".join( + str(event.get(key, "")) for key in ("flow", "node") if event.get(key) + ) + parts.append( + Text(f" {where} ", style="red") + + Text(str(event.get("detail", ""))[:100], style="dim") + ) + return Group(*parts) + + +def cmd_status(args: argparse.Namespace) -> int: + """How the engine is doing, once or until Ctrl-C.""" + from rich.console import Console + from rich.live import Live + + console = Console() + try: + with _client_for(args) as client: + if not args.watch: + console.print(_status_screen(client)) + return 0 + if not sys.stdout.isatty(): + return _fail("--watch needs a terminal; without one, drop it") + with Live(_status_screen(client), console=console) as live: + while True: + time.sleep(WATCH_INTERVAL_S) + live.update(_status_screen(client)) + except KeyboardInterrupt: + return 130 + except (SyncError, ApiError) as exc: + return _fail(str(exc)) + + def cmd_runs(args: argparse.Namespace) -> int: try: with _client_for(args) as client: @@ -593,9 +772,25 @@ def add_parsers(subparsers: Any) -> None: action="store_true", help="execute every node, even one an earlier run already answered", ) + parser.add_argument( + "--defaults", + action="store_true", + help="take every input's declared value instead of asking for it", + ) with_engine(parser, local=True) parser.set_defaults(func=cmd_run) + parser = subparsers.add_parser( + "status", help="how the engine is doing, and what each flow is up to" + ) + parser.add_argument( + "--watch", + action="store_true", + help="keep it on screen, refreshed until Ctrl-C", + ) + with_engine(parser) + parser.set_defaults(func=cmd_status) + parser = subparsers.add_parser("runs", help="the runs an engine has recorded") parser.add_argument("--flow", default="") parser.add_argument("--limit", type=int, default=20) diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 539cc5b..db3a956 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -194,6 +194,25 @@ class Client: """Retire the engine's workers, so the next run imports the code as it is.""" self._call("POST", "/modules/refresh") + def flows(self) -> list[dict[str, Any]]: + """Every flow with its node and error counts, as the home screen lists.""" + result = self._call("GET", "/flows/") + return list(result.get("data") or []) + + # -- how the engine is doing ------------------------------------------- + + def summary(self) -> dict[str, Any]: + """Health as the dashboard reads it: flows, nodes, queue, loop lag.""" + result: dict[str, Any] = self._call("GET", "/observability/summary") + return result + + def events(self, kind: str = "failure", limit: int = 10) -> list[dict[str, Any]]: + """What went wrong, or who changed what. Newest first.""" + result = self._call( + "GET", "/observability/events", params={"kind": kind, "limit": limit} + ) + return list(result or []) + # -- runs -------------------------------------------------------------- def submit( diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 2671917..c5dd036 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -48,6 +48,9 @@ dependencies = [ # the user's own. Present in the image; a pip install would otherwise have # to find one on PATH, and quietly fall back to the engine's interpreter. "uv>=0.5", + # `fluksio status` draws with it. Already here underneath fastapi's CLI, + # named because a command that depends on it should say so. + "rich>=13", ] [project.urls] diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index a7e155a..cbb9902 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -149,7 +149,9 @@ def made_artifact(): def test_a_run_reference_resolves_to_what_that_run_produced(made_artifact): run_id, reference = made_artifact - resolved = resolve_references(artifact_flow(), {"dataset": f"@run:{run_id}.dataset"}) + resolved = resolve_references( + artifact_flow(), {"dataset": f"@run:{run_id}.dataset"} + ) # The producer's own reference, file name and all — not one rebuilt from # the row, which carries the message name instead. diff --git a/backend/tests/flow/test_queue.py b/backend/tests/flow/test_queue.py index 5410be9..3f69ff0 100644 --- a/backend/tests/flow/test_queue.py +++ b/backend/tests/flow/test_queue.py @@ -314,9 +314,7 @@ def test_work_in_flight_is_touched_until_it_finishes(monkeypatch): consumer.assign_flow("f", "consumer") queue = RecordingQueue() - pipeline = Pipeline( - nodes=[source, consumer], state=MemoryState(), work_queue=queue - ) + pipeline = Pipeline(nodes=[source, consumer], state=MemoryState(), work_queue=queue) service = ExecutionService(queue) service.bind(pipeline) service.start() diff --git a/backend/tests/flow/test_runs.py b/backend/tests/flow/test_runs.py index 744abf1..7ed5fc9 100644 --- a/backend/tests/flow/test_runs.py +++ b/backend/tests/flow/test_runs.py @@ -443,4 +443,3 @@ def test_a_node_with_no_fingerprint_is_never_looked_up(): assert calls == [1] assert cache.asked == [] assert seen[0].cache_key == "" - diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index 770b79e..e2e7898 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -255,3 +255,53 @@ def test_ctrl_c_while_waiting_cancels_the_run(monkeypatch) -> None: args = _parser().parse_args(["run", "train", "--local", "--no-sync"]) assert cli.cmd_run(args, []) == 130 assert cancelled == ["run-1"] + + +def test_an_artifact_input_may_be_named_rather_than_pasted() -> None: + """The engine resolves either spelling; the CLI just stops mangling them.""" + import json + + from fluksio.sdk.cli import _coerce + + assert _coerce("@run:123-abc.dataset", "artifact") == "@run:123-abc.dataset" + digest = "sha256:" + "a1" * 32 + assert _coerce(digest, "artifact") == digest + # A reference a script already holds still arrives as the object it is. + reference = {"digest": digest, "size": 3} + assert _coerce(json.dumps(reference), "artifact") == reference + + +def test_the_run_prompt_keeps_declared_values_for_anything_left_blank() -> None: + """Enter through the lot is what running with the defaults looks like.""" + from unittest.mock import patch + + from fluksio.sdk.cli import _ask_params + + definition = { + "inputs": [ + {"spec": {"name": "lr", "dtype": "float"}, "initial": 0.05}, + {"spec": {"name": "epochs", "dtype": "int"}, "initial": 10}, + ] + } + + with patch("builtins.input", side_effect=["", ""]): + # Nothing sent: an input the caller leaves out keeps what it declares, + # which is the same contract the browser's dialog has. + assert _ask_params(definition) == {} + + with patch("builtins.input", side_effect=["0.01", "50"]): + assert _ask_params(definition) == {"lr": 0.01, "epochs": 50} + + +def test_the_run_prompt_names_an_answer_of_the_wrong_type() -> None: + from unittest.mock import patch + + import pytest + + from fluksio.sdk import SyncError + from fluksio.sdk.cli import _ask_params + + definition = {"inputs": [{"spec": {"name": "lr", "dtype": "float"}}]} + with patch("builtins.input", side_effect=["fast"]): + with pytest.raises(SyncError, match="lr"): + _ask_params(definition) diff --git a/docs/code/cli.md b/docs/code/cli.md index 59ad161..012ef8f 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -12,8 +12,8 @@ should only *run nodes* for an engine elsewhere. It has none of the engine in it. See [Remote workers](workers.md). The command is two things at once: `serve`, `enroll` and `worker` *are* an -installation, while `login`, `sync`, `run`, `runs` and `sweep` talk to one that -may be anywhere. +installation, while `login`, `sync`, `run`, `runs`, `sweep` and `status` talk to +one that may be anywhere. ## Where an installation lives @@ -196,6 +196,19 @@ whole reference as JSON still works and is what a script that already holds one does — which is the same thing `flow.submit(dataset=run.result["dataset"])` does from Python. +Run a flow with no parameters at a terminal and it asks for them, one line per +declared input, with the declared value in brackets: + +```text +lr (float) [0.05]: 0.01 +epochs (int) [10]: +dataset (artifact): @run:1758042000123-9f2ab41c.dataset +``` + +Enter keeps what is in brackets, so pressing it through the lot runs the +defaults. Nothing changes for a scripted run: passing any parameter, or piping +the command, skips the questions, and `--defaults` skips them explicitly. + `--no-sync` runs what is already on the engine. Worth it in a tight loop where you know nothing changed, since syncing retires the workers and the next call pays its imports again. A directory that declares no flows syncs nothing and @@ -214,6 +227,20 @@ worker pool and module reconcile, against the ~15 ms of submitting to an engine that is already up: `--local` is for "I just want to run it", not for a loop you are iterating in. +### `fluksio status` + +```sh +fluksio status [--watch] +``` + +The home screen's top half in a terminal: whether the engine is healthy and +what is wrong if not, then every flow with its state, its node count and +whether it has unpublished changes, and the last few failures under them. + +`--watch` keeps it on screen and refreshes every five seconds until Ctrl-C — +the cadence the dashboard polls at, since nothing here moves faster. It needs a +terminal; without one, run it without `--watch` and the output pipes cleanly. + ### `fluksio runs` ```sh diff --git a/uv.lock b/uv.lock index 2d03361..8788b56 100644 --- a/uv.lock +++ b/uv.lock @@ -5,8 +5,8 @@ resolution-markers = [ "python_full_version >= '3.14' and sys_platform == 'win32'", "python_full_version >= '3.14' and sys_platform != 'win32'", "python_full_version == '3.13.*' and sys_platform == 'win32'", - "python_full_version < '3.13' and sys_platform == 'win32'", "python_full_version == '3.13.*' and sys_platform != 'win32'", + "python_full_version < '3.13' and sys_platform == 'win32'", "python_full_version < '3.13' and sys_platform != 'win32'", ] @@ -891,6 +891,7 @@ dependencies = [ { name = "pyjwt" }, { name = "python-multipart" }, { name = "redis" }, + { name = "rich" }, { name = "sentry-sdk", extra = ["fastapi"] }, { name = "sqlmodel" }, { name = "tenacity" }, @@ -927,6 +928,7 @@ requires-dist = [ { name = "pyjwt", specifier = ">=2.8.0,<3.0.0" }, { name = "python-multipart", specifier = ">=0.0.7,<1.0.0" }, { name = "redis", specifier = ">=7.1.0" }, + { name = "rich", specifier = ">=13" }, { name = "sentry-sdk", extras = ["fastapi"], specifier = ">=2.20.0" }, { name = "sqlmodel", specifier = ">=0.0.21,<1.0.0" }, { name = "tenacity", specifier = ">=8.2.3,<9.0.0" }, @@ -2546,12 +2548,12 @@ version = "0.46.2" source = { registry = "https://pypi.org/simple" } resolution-markers = [ "python_full_version == '3.13.*' and sys_platform == 'win32'", - "python_full_version < '3.13' and sys_platform == 'win32'", "python_full_version == '3.13.*' and sys_platform != 'win32'", + "python_full_version < '3.13' and sys_platform == 'win32'", "python_full_version < '3.13' and sys_platform != 'win32'", ] dependencies = [ - { name = "anyio", marker = "python_full_version < '3.14'" }, + { name = "anyio" }, ] sdist = { url = "https://files.pythonhosted.org/packages/ce/20/08dfcd9c983f6a6f4a1000d934b9e6d626cff8d2eeb77a89a68eef20a2b7/starlette-0.46.2.tar.gz", hash = "sha256:7f7361f34eed179294600af672f565727419830b54b7b084efe44bb82d2fccd5", size = 2580846, upload-time = "2025-04-13T13:56:17.942Z" } wheels = [ @@ -2567,7 +2569,7 @@ resolution-markers = [ "python_full_version >= '3.14' and sys_platform != 'win32'", ] dependencies = [ - { name = "anyio", marker = "python_full_version >= '3.14'" }, + { name = "anyio" }, ] sdist = { url = "https://files.pythonhosted.org/packages/b5/b4/205b0d5241d934e8add0c38aa924c4f9fb7330834ff11e5444db964ec3f9/starlette-1.6.0.tar.gz", hash = "sha256:d4e3ac5e546444960c710297a3c9fc3f7ebae1b7e963f3d36173b49da535be9b", size = 2716969, upload-time = "2026-08-08T18:27:57.512Z" } wheels = [