diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py index 4f091d2..da738df 100644 --- a/backend/fluksio/cli.py +++ b/backend/fluksio/cli.py @@ -356,6 +356,14 @@ def _free_port(host: str, start: int) -> int: def cmd_serve(args: argparse.Namespace) -> int: + # At a terminal this is a dashboard with the engine as a child of it. The + # import is here rather than at the top because it is only ever needed on + # that path, and `serve` in a container must not pay for it. + if not args.plain and sys.stdout.isatty() and sys.stdin.isatty(): + from fluksio.tui import run_tui + + return run_tui(args) + data_dir = _data_dir(args.data_dir, args.shared) for flag, name in CONCURRENCY_FLAGS.items(): value = getattr(args, flag, None) @@ -566,6 +574,11 @@ def _parser() -> argparse.ArgumentParser: metavar="N", help="python worker processes (default 4, FLOW_MAX_WORKERS)", ) + serve.add_argument( + "--plain", + action="store_true", + help="the log stream rather than the dashboard (the default with no terminal)", + ) serve.add_argument( "--gpus", type=_at_least(0), diff --git a/backend/fluksio/tui.py b/backend/fluksio/tui.py new file mode 100644 index 0000000..a02223a --- /dev/null +++ b/backend/fluksio/tui.py @@ -0,0 +1,342 @@ +"""The dashboard `fluksio serve` opens at a terminal. + +A supervisor, not an engine: it starts `serve --plain` as a child and talks to +it over the same HTTP API every other client uses. So the engine is a process +that can be stopped and started under the screen watching it, an engine +somebody else started can be adopted rather than duplicated, and quitting the +dashboard is not the same as stopping the engine. + +Nothing of the engine is imported here — that lives in the child. +""" + +from __future__ import annotations + +import argparse +import json +import os +import signal +import subprocess +import sys +from pathlib import Path +from typing import Any + +from textual import work +from textual.app import App, ComposeResult +from textual.containers import Horizontal, Vertical +from textual.screen import ModalScreen +from textual.widgets import Button, DataTable, Footer, Header, Input, Label, RichLog + +from fluksio.cli import ( + DEFAULT_PORT, + DEFAULT_PORTAL, + _data_dir, + _token_for, + probe_engine, + read_pidfile, +) +from fluksio.sdk.cli import WATCH_INTERVAL_S, _status_screen +from fluksio.sdk.client import Client, config_path + +#: How long to keep asking the child for the credential it writes on the way +#: up, before giving up and letting the panels say so. +STARTUP_TRIES = 60 + + +def child_argv(argv: list[str]) -> list[str]: + """The same command, told to serve without a dashboard of its own. + + Every flag is passed through: whatever `serve` was asked for is what the + engine under this screen is running with. + """ + passed = [flag for flag in argv if flag != "--plain"] + return [sys.executable, "-m", "fluksio.cli", *passed, "--plain"] + + +class Enroll(ModalScreen[tuple[str, str] | None]): + """The claim code a portal minted, and which portal minted it.""" + + BINDINGS = [("escape", "dismiss(None)", "cancel")] + + def compose(self) -> ComposeResult: + with Vertical(id="enroll"): + yield Label("Pair this installation with a portal") + yield Input(placeholder="claim code", id="code") + yield Input(value=DEFAULT_PORTAL, id="portal") + with Horizontal(): + yield Button("Enroll", variant="primary", id="go") + yield Button("Cancel", id="cancel") + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "cancel": + self.dismiss(None) + return + code = self.query_one("#code", Input).value.strip() + portal = self.query_one("#portal", Input).value.strip() + self.dismiss((code, portal or DEFAULT_PORTAL) if code else None) + + +class ServeApp(App[int]): + """One screen: how the engine is, what has run, and what it is saying.""" + + CSS = """ + #status { height: auto; padding: 0 1; } + #runs { height: 2fr; } + #log { height: 1fr; border-top: solid $panel; } + #enroll { width: 60; height: auto; padding: 1 2; background: $surface; } + """ + + BINDINGS = [ + ("q", "quit", "quit (engine keeps running)"), + ("s", "stop_start", "stop/start"), + ("r", "restart", "restart"), + ("c", "cancel_run", "cancel run"), + ("e", "enroll", "enroll"), + ] + + def __init__(self, args: argparse.Namespace) -> None: + super().__init__() + self.args = args + self.data_dir: Path = _data_dir(args.data_dir, args.shared) + self.child: subprocess.Popen[str] | None = None + #: The pid of an engine this dashboard did not start. Only ever one + #: whose data directory is this one — the probe is what proves it. + self.adopted: int | None = None + self.client: Client | None = None + self.url = "" + + # -- layout --------------------------------------------------------------- + + def compose(self) -> ComposeResult: + yield Header() + yield RichLog(id="status", markup=True, wrap=True) + table: DataTable[str] = DataTable(id="runs", cursor_type="row") + yield table + yield RichLog(id="log", markup=False, highlight=False, max_lines=2000) + yield Footer() + + def on_mount(self) -> None: + self.title = f"fluksio — {self.data_dir}" + table = self.query_one("#runs", DataTable) + table.add_columns("run", "status", "flow", "took", "params") + self.start_engine(first=True) + self.set_interval(WATCH_INTERVAL_S, self.refresh_panels) + + # -- the engine under the screen ------------------------------------------ + + def note(self, message: str) -> None: + self.query_one("#log", RichLog).write(message) + + def start_engine(self, first: bool = False) -> None: + """Adopt whatever is already serving this directory, or start one.""" + host = self.args.host + reachable = "127.0.0.1" if host in ("0.0.0.0", "::", "") else host + wanted = self.args.port or DEFAULT_PORT + url = f"http://{reachable}:{wanted}" + who = probe_engine(url, _token_for(self.data_dir)) if first else "other" + + if who == "ours": + running = read_pidfile(self.data_dir) + self.adopted = running["pid"] if running else None + self.url = url + named = f" (pid {self.adopted})" if self.adopted else "" + self.note(f"Adopted the engine already serving this directory{named}.") + if self.adopted is None: + self.note("It wrote no pidfile, so this screen cannot stop it.") + self.connect() + return + if who == "foreign": + self.note(f"Port {wanted} holds another installation's Fluksio.") + + self.child = subprocess.Popen( # noqa: S603 + child_argv(sys.argv[1:]), + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + env={**os.environ, "PYTHONUNBUFFERED": "1"}, + ) + self.tail_child(self.child) + self.await_engine() + + @work(thread=True, exclusive=False) + def tail_child(self, child: subprocess.Popen[str]) -> None: + """The engine's own output, which is why a second terminal was needed.""" + if child.stdout is None: + return + for line in child.stdout: + self.call_from_thread(self.note, line.rstrip()) + self.call_from_thread(self.note, f"The engine stopped ({child.wait()}).") + + @work(thread=True, exclusive=True, group="startup") + def await_engine(self) -> None: + """Wait for the child to say where it landed, then talk to it. + + The port is the child's to choose — it moves off a taken one — and + `client.json` is where it says which it took. + """ + import time + + for _ in range(STARTUP_TRIES): + if self.child is not None and self.child.poll() is not None: + return + try: + stored = json.loads(config_path(self.data_dir).read_text()) + except (OSError, ValueError): + stored = {} + if stored.get("url") and stored.get("token"): + self.url = str(stored["url"]) + self.call_from_thread(self.connect) + return + time.sleep(0.5) + self.call_from_thread(self.note, "The engine did not come up.") + + def connect(self) -> None: + stored = config_path(self.data_dir) + try: + config = json.loads(stored.read_text()) + self.client = Client( + url=str(config["url"]), token=str(config["token"]), retries=0 + ) + except (OSError, ValueError, KeyError) as exc: + self.note(f"No credential to talk to the engine with: {exc}") + return + self.url = self.client.url + self.refresh_panels() + + def engine_pid(self) -> int | None: + return self.child.pid if self.child is not None else self.adopted + + def stop_engine(self) -> None: + if self.child is not None: + self.note(f"Stopping the engine (pid {self.child.pid}).") + self.child.terminate() + self.child = None + elif self.adopted is not None: + self.note(f"Stopping the adopted engine (pid {self.adopted}).") + try: + os.kill(self.adopted, signal.SIGTERM) + except OSError as exc: + self.note(f"Could not stop it: {exc}") + self.adopted = None + self.client = None + + # -- what the panels show ------------------------------------------------- + + def refresh_panels(self) -> None: + if self.client is not None: + self.read_engine() + + @work(thread=True, exclusive=True, group="poll") + def read_engine(self) -> None: + client = self.client + if client is None: + return + try: + screen = _status_screen(client) + rows = client.runs(limit=20) + except Exception as exc: + self.call_from_thread(self.show_offline, exc) + return + self.call_from_thread(self.show, screen, rows) + + def show(self, screen: Any, rows: list[dict[str, Any]]) -> None: + status = self.query_one("#status", RichLog) + status.clear() + status.write(screen) + table = self.query_one("#runs", DataTable) + table.clear() + for row in rows: + table.add_row( + str(row.get("id", ""))[-8:], + str(row.get("status", "")), + str(row.get("flow", "")), + f"{(row.get('duration_ms') or 0) / 1000:.1f}s", + json.dumps(row.get("params") or {})[:60], + key=str(row.get("id", "")), + ) + + def show_offline(self, exc: Exception) -> None: + status = self.query_one("#status", RichLog) + status.clear() + which = "starting" if self.engine_pid() is not None else "not running" + status.write(f"[yellow]The engine is {which}.[/] ({type(exc).__name__})") + + # -- keys ----------------------------------------------------------------- + + def action_stop_start(self) -> None: + if self.engine_pid() is not None: + self.stop_engine() + else: + self.start_engine() + + def action_restart(self) -> None: + self.stop_engine() + self.start_engine() + + def action_cancel_run(self) -> None: + table = self.query_one("#runs", DataTable) + if self.client is None or not table.row_count: + return + # The cell holds the tail of the id, which is what fits; the row's key + # is the whole of it, which is what the engine is asked about. + key = table.coordinate_to_cell_key(table.cursor_coordinate).row_key + if key.value: + self.cancel_run(str(key.value)) + + @work(thread=True) + def cancel_run(self, run_id: str) -> None: + client = self.client + if client is None: + return + try: + client.cancel(run_id) + except Exception as exc: + self.call_from_thread(self.note, f"Could not cancel {run_id}: {exc}") + return + self.call_from_thread(self.note, f"Cancelled {run_id}.") + self.call_from_thread(self.refresh_panels) + + def action_enroll(self) -> None: + self.push_screen(Enroll(), self.enrolled) + + def enrolled(self, answer: tuple[str, str] | None) -> None: + if answer is None: + return + code, portal = answer + self.run_enroll(code, portal) + + @work(thread=True) + def run_enroll(self, code: str, portal: str) -> None: + """`fluksio enroll`, as its own process for the same reason serve is. + + A running engine picks the configuration up on its own; this screen + only has to report what the command said. + """ + done = subprocess.run( # noqa: S603 + [ + sys.executable, + "-m", + "fluksio.cli", + "enroll", + code, + "--portal", + portal, + "--data-dir", + str(self.data_dir), + ], + capture_output=True, + text=True, + ) + for line in (done.stdout + done.stderr).splitlines(): + self.call_from_thread(self.note, line) + + +def run_tui(args: argparse.Namespace) -> int: + """Open the dashboard, and say what is still running when it closes.""" + app = ServeApp(args) + app.run() + pid, url = app.engine_pid(), app.url + if pid is not None: + print(f"The engine is still running: pid {pid} at {url or 'its port'}.") + print(f" fluksio serve reattaches to it; kill {pid} stops it.") + return 0 diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 2c8b82a..7158390 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -50,6 +50,9 @@ dependencies = [ # `fluksio status` draws with it. Already here underneath fastapi's CLI, # named because a command that depends on it should say so. "rich>=13", + # `fluksio serve` opens a dashboard with it at a terminal. Pure python and + # mostly rich underneath, which is already here. + "textual>=1.0", ] [project.optional-dependencies] @@ -152,6 +155,8 @@ ignore = [ # It talks to whoever ran it; that is what a command line is. "fluksio/cli.py" = ["T201"] "fluksio/sdk/cli.py" = ["T201"] +# What it prints is the line left in the terminal after the dashboard closes. +"fluksio/tui.py" = ["T201"] # Node functions take `params` whether or not they use it — that is the # contract the engine calls them with. "fluksio/flow/nodes.py" = ["ARG001", "ARG002"] diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index afca09e..030303c 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -567,11 +567,51 @@ def test_run_and_sweep_take_what_to_sync() -> None: assert parser.parse_args(["sweep", "train"]).sync == [] +def test_the_dashboard_runs_the_engine_as_a_child_of_itself(monkeypatch) -> None: + """At a terminal `serve` is a dashboard; the engine is a plain serve. + + Every flag is passed through, so what the child runs with is what serve + was asked for — and `--plain` is what stops it opening a second one. + """ + import sys + + from fluksio import cli + from fluksio.tui import child_argv + + argv = child_argv(["serve", "--port", "8123", "--gpus", "1"]) + assert argv[:3] == [sys.executable, "-m", "fluksio.cli"] + assert argv[3:] == ["serve", "--port", "8123", "--gpus", "1", "--plain"] + # Already plain: told once, not twice. + assert child_argv(["serve", "--plain"])[3:] == ["serve", "--plain"] + + opened: list[str] = [] + monkeypatch.setattr( + "fluksio.tui.run_tui", lambda args: opened.append("tui") or 0, raising=False + ) + monkeypatch.setattr(sys.stdout, "isatty", lambda: True, raising=False) + monkeypatch.setattr(sys.stdin, "isatty", lambda: True, raising=False) + + parser = cli._parser() + assert cli.cmd_serve(parser.parse_args(["serve"])) == 0 + assert opened == ["tui"] + + # `--plain` goes past it, which is what the child and every container does. + # Nothing else of serve runs here, so it fails on the data directory it is + # given rather than opening a dashboard. + opened.clear() + monkeypatch.setattr( + cli, "_data_dir", lambda *a, **k: (_ for _ in ()).throw(SystemExit(3)) + ) + with __import__("pytest").raises(SystemExit): + cli.cmd_serve(parser.parse_args(["serve", "--plain"])) + assert opened == [] + + def test_a_serving_engine_records_itself_until_it_stops(tmp_path) -> None: """A pid nobody is running is the same as no pidfile at all.""" import os - from fluksio.cli import PIDFILE, read_pidfile, write_pidfile + from fluksio.cli import read_pidfile, write_pidfile assert read_pidfile(tmp_path) is None @@ -586,7 +626,7 @@ def test_a_serving_engine_records_itself_until_it_stops(tmp_path) -> None: assert read_pidfile(tmp_path) is None -def test_who_holds_the_port_is_told_apart_by_the_token(tmp_path) -> None: +def test_who_holds_the_port_is_told_apart_by_the_token() -> None: """Only this directory's own engine may be reported as already up. The token is signed with this directory's secret key, so an engine that diff --git a/uv.lock b/uv.lock index 849c897..1519009 100644 --- a/uv.lock +++ b/uv.lock @@ -891,6 +891,7 @@ dependencies = [ { name = "redis" }, { name = "rich" }, { name = "sqlmodel" }, + { name = "textual" }, { name = "uv" }, ] @@ -940,6 +941,7 @@ requires-dist = [ { name = "rich", specifier = ">=13" }, { name = "sentry-sdk", extras = ["fastapi"], marker = "extra == 'server'", specifier = ">=2.20.0" }, { name = "sqlmodel", specifier = ">=0.0.21,<1.0.0" }, + { name = "textual", specifier = ">=1.0" }, { name = "uv", specifier = ">=0.5" }, ] provides-extras = ["parquet", "server"] @@ -1293,6 +1295,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fc/85/69f92b2a7b3c0f88ffe107c86b952b397004b5b8ea5a81da3d9c04c04422/librt-0.7.8-cp314-cp314t-win_arm64.whl", hash = "sha256:8766ece9de08527deabcd7cb1b4f1a967a385d26e33e536d6d8913db6ef74f06", size = 40550, upload-time = "2026-01-14T12:56:01.542Z" }, ] +[[package]] +name = "linkify-it-py" +version = "2.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/45/98/7a1a5f31fd5c7ba93e963b168e244b8e3dd705b3d2a718e3c3307583bf57/linkify_it_py-2.2.0.tar.gz", hash = "sha256:907acd2d17ac1fbb9ddb62c8957ccbd6158cac602231a15c3b0cd1e215f03cee", size = 32939, upload-time = "2026-08-29T07:07:08.305Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/13/d4/1152d1c7ab42d8b908be64fd200ddc870dc9d4925e951198702084aa1a7d/linkify_it_py-2.2.0-py3-none-any.whl", hash = "sha256:3adc40eb5af300b2605fcfdb968c24e1d780a90f1f2221af7c15e5111e94d443", size = 21971, upload-time = "2026-08-29T07:07:07.164Z" }, +] + [[package]] name = "lxml" version = "6.0.2" @@ -1397,6 +1408,11 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/94/54/e7d793b573f298e1c9013b8c4dade17d481164aa517d1d7148619c2cedbf/markdown_it_py-4.0.0-py3-none-any.whl", hash = "sha256:87327c59b172c5011896038353a81343b6754500a08cd7a4973bb48c6d578147", size = 87321, upload-time = "2025-08-11T12:57:51.923Z" }, ] +[package.optional-dependencies] +linkify = [ + { name = "linkify-it-py" }, +] + [[package]] name = "markupsafe" version = "3.0.3" @@ -1486,6 +1502,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/01/c8/248b201f6d753d69fd5d6506011abbb35a946d9142b2ae311a948fd0be3d/mcp-1.29.0-py3-none-any.whl", hash = "sha256:f5a075bb611f23d6f4d080c6a1699fa62772eebc562ba9e66b306ddde1c755f7", size = 223436, upload-time = "2026-07-28T13:41:40.337Z" }, ] +[[package]] +name = "mdit-py-plugins" +version = "0.6.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/59/fc/f8d0863f8862f25602c0404d75568e89fb6b4109804645e5cdfb1be5cf56/mdit_py_plugins-0.6.1.tar.gz", hash = "sha256:a2bca0f039f39dbd35fb74ae1b5f998608c437463371f0ff7f49a19a17a114d0", size = 56114, upload-time = "2026-05-13T09:03:38.91Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a5/69/6da5581c6a7fede7dc261bf4e67d6adca4196f176b43288b55b3db395b6e/mdit_py_plugins-0.6.1-py3-none-any.whl", hash = "sha256:214c82fb2ac524472ab6a5bcab1de80f73b50443e187f401bfd77efbc7c6481d", size = 66663, upload-time = "2026-05-13T09:03:37.76Z" }, +] + [[package]] name = "mdurl" version = "0.1.2" @@ -1785,6 +1813,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/32/2b/121e912bd60eebd623f873fd090de0e84f322972ab25a7f9044c056804ed/pathspec-1.0.3-py3-none-any.whl", hash = "sha256:e80767021c1cc524aa3fb14bedda9c34406591343cc42797b386ce7b9354fb6c", size = 55021, upload-time = "2026-01-09T15:46:44.652Z" }, ] +[[package]] +name = "platformdirs" +version = "4.11.5" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/ea/06/cf1564dcc2e2261c8c8c6c05628dc8b418943bdae2a4e58640ceb2f770fa/platformdirs-4.11.5.tar.gz", hash = "sha256:e8b31f4f8bcbbedef91a6b57a706255e4f148d2a4e01648382a0a47342539173", size = 34823, upload-time = "2026-08-27T21:36:37.46Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c7/12/6f3fcd5067a9cbf4f8664b32957973498da8b083455203c8d9cab83a725c/platformdirs-4.11.5-py3-none-any.whl", hash = "sha256:89f8d42695853b89c7170bd49bc3dc593f98a71e695ede88e06a3b247bc4563b", size = 23900, upload-time = "2026-08-27T21:36:36.227Z" }, +] + [[package]] name = "pluggy" version = "1.6.0" @@ -2673,6 +2710,23 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c8/cb/6a6a47d5b464bd08695d254f3da6e7986cc70c9fa5d778eda57538edfe56/starlette-1.6.0-py3-none-any.whl", hash = "sha256:a86dd39d14bb45f85a3d18525215a9ef0cfd1f192ac793220e72598c90335f0c", size = 75969, upload-time = "2026-08-08T18:27:56.196Z" }, ] +[[package]] +name = "textual" +version = "8.2.8" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py", extra = ["linkify"] }, + { name = "mdit-py-plugins" }, + { name = "platformdirs" }, + { name = "pygments" }, + { name = "rich" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/00/21/39a76b01bd5eea82a04baaca7580e105d8c59450df03998345bb2cfb307b/textual-8.2.8.tar.gz", hash = "sha256:3f106a9fbc73e39dd266c9712432087de78a6d644084c7c241d6a25c3169115b", size = 1860502, upload-time = "2026-06-30T06:51:24.495Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fb/be/35261223d9416a0751cdff1c7b4a6f881387218a12d439fe22fefebc8c04/textual-8.2.8-py3-none-any.whl", hash = "sha256:267375fd402dc8d981457212efa71f0e3365fd17bba144ba9bb3ed7563cb374a", size = 731418, upload-time = "2026-06-30T06:51:26.364Z" }, +] + [[package]] name = "typer" version = "0.21.1"