diff --git a/NOTEPAD.md b/NOTEPAD.md index 906a0bd..432f865 100644 --- a/NOTEPAD.md +++ b/NOTEPAD.md @@ -11,7 +11,12 @@ Deferring because out of scope is fine, but don't mention deferring than. - BUG/UI: enlarge the icon in the sidebar slightly - BUG/UI: clicking outside the panel does not discard the flow edit panel - BUG/UI: the graph showed in the node edit panel should also be shown for a specific edge inside the pop-up panel when clicking the edge -- BUG/UI: the "run" button in the toolbar should be stateful; i.e. when a flow is running (i.e. with nodes that automatically emit data such as triggers), it should be possible to pause the flow and resume it. Also for flows without a trigger, the button should change state to allow interrupting a running node +- FEAT/FLOW: single-stepping a paused flow. Pause and resume are in; a step button needs the + scheduler to keep its per-run progress between calls, which the one-shot executor does not — + without that it re-runs the first ready node instead of advancing. Needs a persistent + per-flow work queue that a step pops from and resume drains. +- FEAT/UI: interrupting a node that is already running. Pause holds nodes that have not been + submitted yet; one already executing runs to completion. - BUG/UI: the enlarged panel (for code editing) should still maintain its floating style - BUG/UI: `SidePanel`'s mobile branch does not set `data-testid` on the `SheetContent`, so `[data-testid=node-panel]` does not exist on a phone. Mobile specs cannot address the panel. @@ -57,9 +62,9 @@ Deferring because out of scope is fine, but don't mention deferring than. ## Blocked -- FEAT/UI: a "Bug" icon on the node error bubble opening the console at the full error. The - bubble now shows one line and the traceback only reaches the server log; there is no - console in the roadmap yet for it to open. +- FEAT/UI: a "Bug" icon on the node error bubble opening the logs panel at that node's + traceback. The panel now has the traceback; wiring the bubble to open and filter it needs + the panel's open state lifted into `FlowEditor`. - FEAT/INFRA: MQTT broker and InfluxDB compose services for local development. The node types exist; a local broker would make them testable without external hardware. - CHORE/INFRA: `bun install` inside the frontend Docker build intermittently fails with diff --git a/ROADMAP.md b/ROADMAP.md index 7600f6a..d981963 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -52,6 +52,11 @@ external interfaces. See `docs/architecture/structure.canvas` → *Backend – M - [x] REST + WebSocket API over the engine: create/read/update flows, edit node source, run, and stream values, node status and execution events - [x] Dependency-loop detection and graph validation surfaced as API errors +- [x] Per-flow start/stop, stored in a `runtime.json` beside the flow so it + survives a restart and stays out of the autosaved document; pause/resume + holds a flow's nodes while its values keep arriving +- [x] Node log streaming: what a node prints, and the traceback of one that + fails, reach the editor as `node_log` events - [ ] MQTT broker / InfluxDB compose services for local development - [x] Git-based versioning of the flow store (one commit per saved change) - [x] Draft/publish split: edits autosave to `flow.draft.json` / `nodes.draft/`, @@ -94,6 +99,10 @@ React + Vite, primarily desktop but usable on mobile. See `docs/architecture/str - [x] Validation shown on the node it belongs to, and summarised in the dock - [x] Publish control and draft markers in the flow bar, discard in the flow panel, and a conflict dialog when another client got there first +- [x] Dashboard showing which flows run, which are stopped and which have + errors, with a switch per flow +- [x] Logs panel in the canvas dock, pause/resume beside Run, and replaying an + edge's last message from the inspector - [ ] Device assignment per node, selectable from compatible devices - [ ] Test-node affordance on the canvas - [ ] User management screens diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py index 8904830..664c81b 100644 --- a/backend/app/api/routes/flows.py +++ b/backend/app/api/routes/flows.py @@ -57,6 +57,8 @@ class FlowDetail(BaseModel): nodes: list[NodeStatusPublic] = [] issues: list[ValidationIssue] = [] has_draft: bool = False + enabled: bool = True + paused: bool = False class ValidationResult(BaseModel): @@ -81,6 +83,10 @@ class TriggerRequest(BaseModel): def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail: name = definition.name + running = { + "enabled": controller.is_enabled(name), + "paused": controller.is_paused(name), + } if controller.store.has_draft(name): # Report the draft the editor is showing, not the version running # underneath it — otherwise a node the author just broke looks fine. @@ -90,11 +96,13 @@ def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail: nodes=preview.nodes, issues=preview.issues, has_draft=True, + **running, ) return FlowDetail( definition=definition, nodes=controller.node_statuses(name), issues=controller.flow_issues(name), + **running, ) @@ -106,6 +114,14 @@ def _read_flow(controller: FlowController, name: str) -> FlowDef: raise HTTPException(status_code=404, detail=f"No flow named '{name}'") +def _require_enabled(controller: FlowController, name: str) -> None: + if not controller.is_enabled(name): + raise HTTPException( + status_code=409, + detail=f"Flow '{name}' is stopped — start it before running it", + ) + + def _flow_state(controller: FlowController, name: str) -> FlowStatePublic: return FlowStatePublic( values={ @@ -137,6 +153,8 @@ def read_flows(controller: FlowControllerDep) -> Any: node_count=len(definition.nodes), error_count=sum(1 for s in statuses if s.status == "error"), has_draft=controller.store.has_draft(name), + enabled=controller.is_enabled(name), + paused=controller.is_paused(name), ) ) return FlowsPublic(data=summaries, count=len(summaries)) @@ -316,6 +334,43 @@ async def save_node_source( ) +# ----------------------------------------------------------------------------- +# Running, stopped, paused +# ----------------------------------------------------------------------------- + + +@router.post("/{name}/start", response_model=FlowDetail) +async def start_flow(name: str, controller: FlowControllerDep) -> Any: + """Let the engine run this flow again.""" + _read_flow(controller, name) + await controller.set_enabled(name, True) + return _detail(controller, _read_flow(controller, name)) + + +@router.post("/{name}/stop", response_model=FlowDetail) +async def stop_flow(name: str, controller: FlowControllerDep) -> Any: + """Take this flow off the engine: no subscriptions, schedules or webhooks.""" + _read_flow(controller, name) + await controller.set_enabled(name, False) + return _detail(controller, _read_flow(controller, name)) + + +@router.post("/{name}/pause", response_model=Message) +def pause_flow(name: str, controller: FlowControllerDep) -> Any: + """Hold the flow's nodes so its messages can be stepped through.""" + _read_flow(controller, name) + controller.pause_flow(name) + return Message(message=f"Paused flow '{name}'") + + +@router.post("/{name}/resume", response_model=Message) +async def resume_flow(name: str, controller: FlowControllerDep) -> Any: + """Let the flow carry on, running whatever was held back.""" + _read_flow(controller, name) + await run_in_threadpool(controller.resume_flow, name) + return Message(message=f"Resumed flow '{name}'") + + # ----------------------------------------------------------------------------- # Validation and execution # ----------------------------------------------------------------------------- @@ -342,6 +397,7 @@ async def run_flow( on the canvas. Nothing is deployed by running it. """ _read_flow(controller, name) + _require_enabled(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) @@ -358,6 +414,7 @@ async def trigger_node( controller: FlowControllerDep, ) -> Any: """Feed values into a single node.""" + _require_enabled(controller, name) try: await run_in_threadpool( controller.trigger_node, f"{name}.{node_id}", body.values @@ -429,6 +486,8 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None: "values": controller.values(), "nodes": [s.model_dump() for s in controller.node_statuses()], "issues": [i.model_dump() for i in controller.issues], + "paused": controller.paused_flows(), + "logs": list(event_bus.recent_logs), } ) diff --git a/backend/app/flow/controller.py b/backend/app/flow/controller.py index 8102314..6c930f6 100644 --- a/backend/app/flow/controller.py +++ b/backend/app/flow/controller.py @@ -20,6 +20,7 @@ from types import ModuleType from typing import Any, cast from fastapi import FastAPI +from fastapi.concurrency import run_in_threadpool from app.flow.events import EventBus from app.flow.messages import MessageSpec, qualify @@ -188,6 +189,7 @@ class FlowController: self.pipeline: Pipeline | None = None self.loaded: dict[str, LoadedNode] = {} self.issues: list[ValidationIssue] = [] + self.disabled: set[str] = set() self._lock = asyncio.Lock() # ------------------------------------------------------------------------- @@ -200,13 +202,24 @@ class FlowController: async def stop(self) -> None: await self._teardown() + async def set_enabled(self, flow: str, enabled: bool) -> None: + """Stop or start one flow. Rebuilding is what applies it.""" + await run_in_threadpool(self.store.write_enabled, flow, enabled) + await self.reload() + async def reload(self) -> None: """Rebuild the whole pipeline from what is currently stored.""" async with self._lock: await self._teardown() + published = self.store.read_all() + self.disabled = { + flow.name + for flow in published + if not self.store.read_enabled(flow.name) + } nodes, loaded, initial_values, flow_inputs = self._build_flows( - [(flow, False) for flow in self.store.read_all()] + [(flow, False) for flow in published] ) self.loaded = loaded @@ -216,6 +229,7 @@ class FlowController: events=self.events, max_workers=self.max_workers, initial_values=initial_values, + disabled_flows=self.disabled, ) self.issues = _collect_issues(loaded, self.pipeline, flow_inputs) @@ -226,6 +240,8 @@ class FlowController: "type": "pipeline_rebuilt", "issues": [issue.model_dump() for issue in self.issues], "nodes": [status.model_dump() for status in self.node_statuses()], + # A rebuild is a fresh pipeline, so nothing is paused any more. + "paused": self.paused_flows(), } ) @@ -255,6 +271,10 @@ class FlowController: node = entry.node if node is None: continue + # A stopped flow gets no subscriptions, schedules or webhooks — + # that is what stopping it means. + if entry.flow in self.disabled: + continue try: if isinstance(node, MqttNode) and node.mode == MqttNode.Mode.SUBSCRIBER: await node.start_subscription() @@ -392,9 +412,7 @@ class FlowController: for entry in loaded.values() if entry.flow == name ], - issues=[ - issue for issue in issues if not issue.flow or issue.flow == name - ], + issues=[issue for issue in issues if not issue.flow or issue.flow == name], ) def compile_check(self, flow: str, node_id: str, code: str) -> str | None: @@ -416,6 +434,24 @@ class FlowController: # Execution # ------------------------------------------------------------------------- + def is_enabled(self, flow: str) -> bool: + return flow not in self.disabled + + def is_paused(self, flow: str) -> bool: + return self.pipeline is not None and flow in self.pipeline.paused_flows() + + def paused_flows(self) -> list[str]: + return self.pipeline.paused_flows() if self.pipeline else [] + + def pause_flow(self, flow: str) -> None: + if self.pipeline is not None: + self.pipeline.pause(flow) + + def resume_flow(self, flow: str) -> None: + """Blocking — call from a worker thread: held-back nodes run on resume.""" + if self.pipeline is not None: + self.pipeline.resume(flow) + def run_flow(self, flow: str, inputs: dict[str, Any] | None = None) -> None: """Run every node of one flow. Blocking — call from a worker thread.""" if self.pipeline is None: diff --git a/backend/app/flow/events.py b/backend/app/flow/events.py index a7487ee..87d7f6a 100644 --- a/backend/app/flow/events.py +++ b/backend/app/flow/events.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio import logging +from collections import deque from collections.abc import AsyncIterator from contextlib import asynccontextmanager from typing import Any @@ -16,6 +17,7 @@ from typing import Any logger = logging.getLogger(__name__) QUEUE_SIZE = 256 +LOG_HISTORY = 400 class EventBus: @@ -24,6 +26,9 @@ class EventBus: def __init__(self) -> None: self._loop: asyncio.AbstractEventLoop | None = None self._subscribers: set[asyncio.Queue[dict[str, Any]]] = set() + # Kept whether or not anyone is listening, so opening the log panel + # shows what just happened rather than an empty box. + self.recent_logs: deque[dict[str, Any]] = deque(maxlen=LOG_HISTORY) def bind(self, loop: asyncio.AbstractEventLoop) -> None: """Attach the bus to the running event loop (called once at startup).""" @@ -31,6 +36,9 @@ class EventBus: def publish(self, event: dict[str, Any]) -> None: """Publish an event from any thread.""" + if event.get("type") == "node_log": + # deque.append is atomic, so worker threads need no lock here. + self.recent_logs.append(event) loop = self._loop if loop is None or not self._subscribers: return diff --git a/backend/app/flow/logs.py b/backend/app/flow/logs.py new file mode 100644 index 0000000..ec2d739 --- /dev/null +++ b/backend/app/flow/logs.py @@ -0,0 +1,108 @@ +"""Capture what a node prints, so the editor can show it. + +Node code is written by the user and ``print`` is the obvious way to look at a +value, but a node runs on a pool thread and its output would otherwise land +unattributed in the server log. ``sys.stdout`` is replaced once by a tee that +also hands what it is given to whichever capture is active on *this* thread — +so only node executions are captured, and everything else passes through +untouched. +""" + +from __future__ import annotations + +import io +import sys +import traceback +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from contextvars import ContextVar +from typing import TextIO + +# Set for the duration of one node execution, on the thread running it. +_sink: ContextVar[Callable[[str], None] | None] = ContextVar( + "fluksio_node_log_sink", default=None +) + + +class _Tee(io.TextIOBase): + """Writes through to the real stream, and to the active capture.""" + + def __init__(self, real: TextIO) -> None: + self._real = real + + def write(self, text: str) -> int: + sink = _sink.get() + if sink is not None and text: + sink(text) + return self._real.write(text) + + def flush(self) -> None: + self._real.flush() + + def isatty(self) -> bool: + return self._real.isatty() + + +def install() -> None: + """Put the tee in place. Safe to call more than once.""" + if not isinstance(sys.stdout, _Tee): + sys.stdout = _Tee(sys.stdout) + if not isinstance(sys.stderr, _Tee): + sys.stderr = _Tee(sys.stderr) + + +@contextmanager +def capture(sink: Callable[[str], None]) -> Iterator[None]: + """Send everything printed on this thread to ``sink`` for the duration.""" + token = _sink.set(sink) + try: + yield + finally: + _sink.reset(token) + + +def node_traceback() -> str: + """The exception being handled, from the node's own code onward. + + The frames above it are the engine calling the node, which is noise to the + person who wrote it — the same trimming ``_short_error`` does for the one + line shown on the node itself. + """ + exc_type, exc, tb = sys.exc_info() + if exc is None: + return "" + frames = traceback.extract_tb(tb) + start = next( + (i for i, frame in enumerate(frames) if frame.filename.startswith(" None: + self.chunks: list[str] = [] + self.truncated = False + self._limit = limit + self._max_bytes = max_bytes + self._size = 0 + + def __call__(self, text: str) -> None: + if len(self.chunks) >= self._limit or self._size >= self._max_bytes: + self.truncated = True + return + self.chunks.append(text) + self._size += len(text) + + @property + def text(self) -> str: + out = "".join(self.chunks) + return out[: self._max_bytes] if len(out) > self._max_bytes else out diff --git a/backend/app/flow/nodes.py b/backend/app/flow/nodes.py index 5d91d0e..1140ee4 100644 --- a/backend/app/flow/nodes.py +++ b/backend/app/flow/nodes.py @@ -13,6 +13,7 @@ import httpx import numpy as np from pydantic import BaseModel, ConfigDict, Field +from app.flow import logs from app.flow.messages import MessageSpec, qualify if TYPE_CHECKING: @@ -236,7 +237,10 @@ class Node: # A source node asked to inject nothing produces its own data. if not outputs and not self.requires: - outputs = self.f(params=self.params) or {} + collected = logs.Collector() + with logs.capture(collected): + outputs = self.f(params=self.params) or {} + self._pipeline.publish_log(self, collected, "") return self._pipeline.trigger(self, self._to_messages(outputs)) diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py index c8bcefb..3c286b3 100644 --- a/backend/app/flow/pipeline.py +++ b/backend/app/flow/pipeline.py @@ -9,6 +9,7 @@ allowed: each publication triggers the consumers, and the latest value wins. from __future__ import annotations import logging +import threading import time from collections import deque from concurrent.futures import Future, ThreadPoolExecutor, wait @@ -16,6 +17,7 @@ from typing import Any, Literal from pydantic import BaseModel +from app.flow import logs from app.flow.events import EventBus from app.flow.messages import flow_of from app.flow.nodes import Node @@ -55,6 +57,9 @@ class Pipeline: "_edges", "_execution_order", "_downstream_cache", + "_disabled", + "_paused", + "_gate_lock", ) def __init__( @@ -64,8 +69,14 @@ class Pipeline: events: EventBus | None = None, max_workers: int | None = None, initial_values: dict[str, Any] | None = None, + disabled_flows: set[str] | None = None, ) -> None: self._nodes = nodes or [] + # Stopped flows are stored and survive a restart; paused ones are a + # debugging state that a rebuild is meant to clear. + self._disabled = frozenset(disabled_flows or ()) + self._paused: set[str] = set() + self._gate_lock = threading.Lock() # An empty state backend is falsy, so this cannot be ``state or ...``: # that would quietly hand the pipeline a second, private state and # leave everyone reading the shared one seeing nothing. @@ -309,14 +320,34 @@ class Pipeline: if self._events is not None: self._events.publish(event) + def publish_log(self, node: Node, collected: logs.Collector, error: str) -> None: + """One event per execution, so a chatty node cannot outrun the stream.""" + text = collected.text + error + if not text: + return + self._publish( + { + "type": "node_log", + "flow": node.flow, + "node": node.id, + "text": text, + "level": "error" if error else "info", + "truncated": collected.truncated, + "ts": time.time(), + } + ) + def _execute_node(self, node: Node, state: StateBackend) -> dict[str, Any] | None: """Run one node and record its outputs. Never raises.""" started = time.perf_counter() + collected = logs.Collector() try: with state.lock(): inputs = {k: state[k] for k in node.requires if k in state} - result = node.execute(inputs) + with logs.capture(collected): + result = node.execute(inputs) + self.publish_log(node, collected, "") if result: ts = time.time() @@ -355,6 +386,9 @@ class Pipeline: except Exception as exc: # One failing node must not take the rest of the graph down. logger.exception("Node '%s' failed", node.id) + # The one-line error goes on the node; the traceback goes to the + # log panel, which is where there is room to read it. + self.publish_log(node, collected, logs.node_traceback()) self._publish( { "type": "node_error", @@ -366,6 +400,37 @@ class Pipeline: ) return None + # ------------------------------------------------------------------------- + # Running, stopped, paused + # ------------------------------------------------------------------------- + + def is_disabled(self, flow: str) -> bool: + return flow in self._disabled + + def paused_flows(self) -> list[str]: + with self._gate_lock: + return sorted(self._paused) + + def pause(self, flow: str) -> None: + """Hold this flow's nodes. Values still arrive; nothing acts on them.""" + with self._gate_lock: + self._paused.add(flow) + self._publish({"type": "flow_paused", "flow": flow, "paused": True}) + + def resume(self, flow: str) -> None: + with self._gate_lock: + self._paused.discard(flow) + self._publish({"type": "flow_paused", "flow": flow, "paused": False}) + # Whatever was held back is free to run now. + self._execute_parallel(self.flow_nodes(flow), self._state, check_ready=True) + + def _gate_blocks(self, node: Node) -> bool: + """Is this node's flow held back from executing?""" + if node.flow in self._disabled: + return True + with self._gate_lock: + return node.flow in self._paused + def _execute_parallel( self, nodes_subset: set[Node] | None, @@ -395,10 +460,16 @@ class Pipeline: def submit_ready(executor: ThreadPoolExecutor) -> None: for n in target_nodes: - if n not in submitted and n not in skipped and is_ready(n): + if n in submitted or n in skipped: + continue + # Gate before the readiness check, so a held-back node does not + # spend the synchronous claim it would need once it may run. + if self._gate_blocks(n): + continue + if is_ready(n): submitted.add(n) node_futures[n] = executor.submit(self._execute_node, n, state) - elif n not in submitted and n.synchronous and in_degree[n] == 0: + elif n.synchronous and in_degree[n] == 0: # Not ready now; a later trigger may make it ready. skipped.add(n) @@ -433,9 +504,19 @@ class Pipeline: return self._execute_parallel(nodes, self._state, check_ready=False) def trigger(self, node: Node, outputs: dict[str, Any] | None) -> StateBackend: - """Publish a node's outputs and run everything downstream of it.""" + """Publish a node's outputs and run everything downstream of it. + + A stopped flow drops the event: its subscriptions and schedules are torn + down anyway, and anything still arriving from another thread would be + work the flow was explicitly told not to do. A *paused* flow still + publishes, so the incoming value is visible on the canvas, and holds + the nodes downstream of it for stepping. + """ state = self._state + if node.flow in self._disabled: + return state + if outputs: ts = time.time() with state.lock(): diff --git a/backend/app/flow/schemas.py b/backend/app/flow/schemas.py index d7fcbd2..fd24c83 100644 --- a/backend/app/flow/schemas.py +++ b/backend/app/flow/schemas.py @@ -116,6 +116,8 @@ class FlowSummary(BaseModel): node_count: int = 0 error_count: int = 0 has_draft: bool = False + enabled: bool = True + paused: bool = False class FlowsPublic(BaseModel): diff --git a/backend/app/flow/store.py b/backend/app/flow/store.py index 8081971..57a319f 100644 --- a/backend/app/flow/store.py +++ b/backend/app/flow/store.py @@ -14,6 +14,7 @@ unpublished — which is what every flow written before this existed looks like. from __future__ import annotations +import json import logging import shutil import subprocess @@ -132,6 +133,36 @@ class FlowStore: def _draft_node_file(self, flow: str, node_id: str) -> Path: return self._draft_nodes_dir(flow) / f"{node_id}.py" + def _runtime_file(self, name: str) -> Path: + return self._flow_dir(name) / "runtime.json" + + # ------------------------------------------------------------------------- + # Runtime state + # ------------------------------------------------------------------------- + + def read_enabled(self, name: str) -> bool: + """Whether the engine should run this flow. Missing means yes.""" + path = self._runtime_file(name) + if not path.exists(): + return True + try: + return bool(json.loads(path.read_text()).get("enabled", True)) + except (ValueError, OSError): + logger.warning("Unreadable runtime state for flow '%s'", name) + return True + + def write_enabled(self, name: str, enabled: bool) -> None: + """Stop or start a flow, in a file the editor never writes. + + This is deliberately not a field on the flow document: that one is + autosaved from the canvas, so a stopped flow would start itself again + on the next edit. + """ + path = self._runtime_file(name) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"enabled": enabled}, indent=2) + "\n") + self._commit(f"{'Start' if enabled else 'Stop'} flow '{name}'") + # ------------------------------------------------------------------------- # Flows # ------------------------------------------------------------------------- diff --git a/backend/app/main.py b/backend/app/main.py index 5b20493..52fd0ef 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -9,6 +9,7 @@ from starlette.middleware.cors import CORSMiddleware from app.api.main import api_router from app.core.config import settings +from app.flow import logs from app.flow.controller import FlowController from app.flow.events import event_bus from app.flow.secrets import init_secrets @@ -34,6 +35,8 @@ def _state_backend() -> StateBackend: async def lifespan(app: FastAPI) -> AsyncIterator[None]: """Start the flow engine alongside the API.""" event_bus.bind(asyncio.get_running_loop()) + # Node code is user code, and `print` is how it says things. + logs.install() init_secrets(settings.SECRETS_FILE, settings.SECRET_KEY) controller = FlowController( diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 994a1ba..6376324 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -79,6 +79,9 @@ ignore = [ # contract the engine calls them with. "app/flow/nodes.py" = ["ARG001", "ARG002"] "tests/flow/*" = ["ARG001"] +# Printing is what this one is about: node code is user code, and `print` is +# how it says things. +"tests/flow/test_logs.py" = ["ARG001", "T201"] [tool.ruff.lint.pyupgrade] # Preserve types, even if a file imports `from __future__ import annotations`. diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py index b354d67..afeb2f3 100644 --- a/backend/tests/api/routes/test_flows.py +++ b/backend/tests/api/routes/test_flows.py @@ -193,6 +193,71 @@ def test_discarding_a_draft_restores_what_is_running( client.delete(f"{PREFIX}/reverted", headers=superuser_token_headers) +def test_stopping_a_flow_takes_it_off_the_engine( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + saved = client.put( + f"{PREFIX}/halted", headers=superuser_token_headers, json=a_flow("halted") + ).json() + client.post( + f"{PREFIX}/halted/publish", + headers=superuser_token_headers, + json={"version": saved["definition"]["version"]}, + ) + + stopped = client.post(f"{PREFIX}/halted/stop", headers=superuser_token_headers) + assert stopped.status_code == 200 + assert stopped.json()["enabled"] is False + # Nothing of it is loaded, so there is nothing to run. + assert ( + client.post( + f"{PREFIX}/halted/run", headers=superuser_token_headers, json={"inputs": {}} + ).status_code + == 409 + ) + + started = client.post(f"{PREFIX}/halted/start", headers=superuser_token_headers) + assert started.json()["enabled"] is True + assert ( + client.post( + f"{PREFIX}/halted/run", headers=superuser_token_headers, json={"inputs": {}} + ).status_code + == 200 + ) + + client.delete(f"{PREFIX}/halted", headers=superuser_token_headers) + + +def test_pausing_is_reported_back( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + saved = client.put( + f"{PREFIX}/held", headers=superuser_token_headers, json=a_flow("held") + ).json() + client.post( + f"{PREFIX}/held/publish", + headers=superuser_token_headers, + json={"version": saved["definition"]["version"]}, + ) + + assert ( + client.post(f"{PREFIX}/held/pause", headers=superuser_token_headers).status_code + == 200 + ) + assert ( + client.get(f"{PREFIX}/held", headers=superuser_token_headers).json()["paused"] + is True + ) + + client.post(f"{PREFIX}/held/resume", headers=superuser_token_headers) + assert ( + client.get(f"{PREFIX}/held", headers=superuser_token_headers).json()["paused"] + is False + ) + + client.delete(f"{PREFIX}/held", headers=superuser_token_headers) + + def test_unconnected_input_is_surfaced( client: TestClient, superuser_token_headers: dict[str, str] ) -> None: diff --git a/backend/tests/flow/test_logs.py b/backend/tests/flow/test_logs.py new file mode 100644 index 0000000..d0b0043 --- /dev/null +++ b/backend/tests/flow/test_logs.py @@ -0,0 +1,124 @@ +"""What a node prints reaches the editor, attributed to that node.""" + +import sys +from typing import Any + +from app.flow import logs +from app.flow.messages import DType, MessageSpec +from app.flow.nodes import Node +from app.flow.pipeline import Pipeline + + +def run_with_capture(nodes: list[Node], bus: "RecordingBus") -> None: + """Run a graph the way the app does, tee first. + + The app installs the tee once at startup; pytest replaces ``sys.stdout`` + around each test, so it is installed here rather than in a fixture. + """ + logs.install() + Pipeline(nodes=nodes, events=bus).run({}) + + +class RecordingBus: + """Stands in for the event bus without an event loop behind it.""" + + def __init__(self) -> None: + self.events: list[dict[str, Any]] = [] + + def publish(self, event: dict[str, Any]) -> None: + self.events.append(event) + + +def make_node(node_id: str, f, provides=()) -> Node: + node = Node(f=f, provides=list(provides), name=node_id) + node.assign_flow("demo", node_id) + return node + + +def logs_of(bus: RecordingBus) -> list[dict[str, Any]]: + return [event for event in bus.events if event["type"] == "node_log"] + + +def test_what_a_node_prints_is_reported_against_it(): + def talkative(params): + print("value looks fine") + return {"temp": 20.0} + + bus = RecordingBus() + node = make_node("chatty", talkative, provides=[MessageSpec(name="temp")]) + run_with_capture([node], bus) + + captured = logs_of(bus) + assert len(captured) == 1 + assert captured[0]["node"] == "demo.chatty" + assert captured[0]["level"] == "info" + assert "value looks fine" in captured[0]["text"] + + +def test_a_quiet_node_produces_no_log_event(): + bus = RecordingBus() + node = make_node( + "quiet", + lambda params: {"temp": 20.0}, + provides=[MessageSpec(name="temp", dtype=DType.FLOAT)], + ) + run_with_capture([node], bus) + + assert logs_of(bus) == [] + + +def test_a_failing_node_reports_its_traceback(): + # Compiled the way the controller compiles node source, because the frame + # trimming keys on that filename. + namespace: dict[str, Any] = {} + exec( + compile( + 'def process(params):\n print("about to fail")\n' + ' raise RuntimeError("boom")\n', + "", + "exec", + ), + namespace, + ) + + bus = RecordingBus() + run_with_capture([make_node("broken", namespace["process"])], bus) + + captured = logs_of(bus) + assert len(captured) == 1 + assert captured[0]["level"] == "error" + # Both what it printed and where it broke, which the node bubble has no + # room for. + assert "about to fail" in captured[0]["text"] + assert "RuntimeError: boom" in captured[0]["text"] + # The frames above the node belong to the engine that called it. + assert "pipeline.py" not in captured[0]["text"] + + +def test_a_flood_is_truncated_rather_than_streamed(): + def noisy(params): + for index in range(1000): + print(f"line {index}") + return None + + bus = RecordingBus() + run_with_capture([make_node("noisy", noisy)], bus) + + captured = logs_of(bus) + # One event per execution, whatever the node prints. + assert len(captured) == 1 + assert captured[0]["truncated"] is True + assert len(captured[0]["text"]) <= 8192 + + +def test_printing_outside_a_node_still_reaches_the_real_stream(capsys): + logs.install() + print("server talking") + assert "server talking" in capsys.readouterr().out + + +def test_installing_twice_does_not_stack_tees(): + logs.install() + once = sys.stdout + logs.install() + assert sys.stdout is once diff --git a/backend/tests/flow/test_runtime_control.py b/backend/tests/flow/test_runtime_control.py new file mode 100644 index 0000000..4bb40e0 --- /dev/null +++ b/backend/tests/flow/test_runtime_control.py @@ -0,0 +1,103 @@ +"""Stopping a flow takes it off the engine; pausing holds its nodes.""" + +from pathlib import Path + +from app.flow.messages import DType, MessageSpec +from app.flow.nodes import Node +from app.flow.pipeline import Pipeline +from app.flow.store import FlowStore + + +def spec(name: str) -> MessageSpec: + return MessageSpec(name=name, dtype=DType.FLOAT) + + +def make_node(node_id: str, flow: str, f, requires=(), provides=()) -> Node: + node = Node(f=f, requires=list(requires), provides=list(provides), name=node_id) + node.assign_flow(flow, node_id) + return node + + +def a_chain(flow: str, ran: list[str]) -> list[Node]: + """source → middle: two nodes, so stepping has somewhere to stop.""" + + def source(params): + ran.append(f"{flow}.source") + return {"temp": 20.0} + + def middle(temp, params): + ran.append(f"{flow}.middle") + return {"warm": temp > 10} + + return [ + make_node("source", flow, source, provides=[spec("temp")]), + make_node( + "middle", + flow, + middle, + requires=[spec("temp")], + provides=[MessageSpec(name="warm", dtype=DType.BOOL)], + ), + ] + + +def test_a_stopped_flow_runs_nothing_and_its_neighbours_carry_on(): + ran: list[str] = [] + pipeline = Pipeline( + nodes=a_chain("stopped", ran) + a_chain("running", ran), + disabled_flows={"stopped"}, + ) + + pipeline.run({}) + + assert not any(name.startswith("stopped.") for name in ran) + assert {"running.source", "running.middle"} <= set(ran) + + +def test_a_stopped_flow_ignores_a_trigger_from_its_own_nodes(): + ran: list[str] = [] + nodes = a_chain("stopped", ran) + pipeline = Pipeline(nodes=nodes, disabled_flows={"stopped"}) + + # What an MQTT subscription or a webhook would do. + nodes[0].inject({"temp": 20.0}) + + assert ran == [] + assert "stopped.temp" not in pipeline.state + + +def test_a_paused_flow_still_takes_values_but_acts_on_none_of_them(): + ran: list[str] = [] + nodes = a_chain("demo", ran) + pipeline = Pipeline(nodes=nodes) + + pipeline.pause("demo") + nodes[0].inject({"temp": 20.0}) + + # The value is there to look at; nothing downstream of it ran. + assert pipeline.state["demo.temp"] == 20.0 + assert ran == [] + + +def test_resuming_runs_what_was_held_back(): + ran: list[str] = [] + pipeline = Pipeline(nodes=a_chain("demo", ran)) + + pipeline.pause("demo") + pipeline.run({}) + assert ran == [] + + pipeline.resume("demo") + assert ran == ["demo.source", "demo.middle"] + assert pipeline.paused_flows() == [] + + +def test_stopped_survives_a_restart(tmp_path: Path): + store = FlowStore(tmp_path / "flows") + assert store.read_enabled("heating") is True + + store.write_enabled("heating", False) + assert store.read_enabled("heating") is False + + # A second store over the same directory is what a restart looks like. + assert FlowStore(tmp_path / "flows").read_enabled("heating") is False diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 6e4cd57..9aa50cc 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -166,6 +166,16 @@ export const FlowDetailSchema = { type: 'boolean', title: 'Has Draft', default: false + }, + enabled: { + type: 'boolean', + title: 'Enabled', + default: true + }, + paused: { + type: 'boolean', + title: 'Paused', + default: false } }, type: 'object', @@ -266,6 +276,16 @@ export const FlowSummarySchema = { type: 'boolean', title: 'Has Draft', default: false + }, + enabled: { + type: 'boolean', + title: 'Enabled', + default: true + }, + paused: { + type: 'boolean', + title: 'Paused', + default: false } }, type: 'object', diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 49de469..25bd75d 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, PrivateCreateUserData, PrivateCreateUserResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse } from './types.gen'; +import type { FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, PrivateCreateUserData, PrivateCreateUserResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse } from './types.gen'; export class FlowsService { /** @@ -225,6 +225,90 @@ export class FlowsService { }); } + /** + * Start Flow + * Let the engine run this flow again. + * @param data The data for the request. + * @param data.name + * @returns FlowDetail Successful Response + * @throws ApiError + */ + public static startFlow(data: FlowsStartFlowData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/flows/{name}/start', + path: { + name: data.name + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Stop Flow + * Take this flow off the engine: no subscriptions, schedules or webhooks. + * @param data The data for the request. + * @param data.name + * @returns FlowDetail Successful Response + * @throws ApiError + */ + public static stopFlow(data: FlowsStopFlowData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/flows/{name}/stop', + path: { + name: data.name + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Pause Flow + * Hold the flow's nodes so its messages can be stepped through. + * @param data The data for the request. + * @param data.name + * @returns Message Successful Response + * @throws ApiError + */ + public static pauseFlow(data: FlowsPauseFlowData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/flows/{name}/pause', + path: { + name: data.name + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Resume Flow + * Let the flow carry on, running whatever was held back. + * @param data The data for the request. + * @param data.name + * @returns Message Successful Response + * @throws ApiError + */ + public static resumeFlow(data: FlowsResumeFlowData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/flows/{name}/resume', + path: { + name: data.name + }, + errors: { + 422: 'Validation Error' + } + }); + } + /** * Validate Flow * Report what would keep this flow from running. diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 7b7d508..0178eec 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -51,6 +51,8 @@ export type FlowDetail = { nodes?: Array; issues?: Array; has_draft?: boolean; + enabled?: boolean; + paused?: boolean; }; /** @@ -87,6 +89,8 @@ export type FlowSummary = { node_count?: number; error_count?: number; has_draft?: boolean; + enabled?: boolean; + paused?: boolean; }; /** @@ -382,6 +386,30 @@ export type FlowsSaveNodeSourceData = { export type FlowsSaveNodeSourceResponse = (NodeStatusPublic); +export type FlowsStartFlowData = { + name: string; +}; + +export type FlowsStartFlowResponse = (FlowDetail); + +export type FlowsStopFlowData = { + name: string; +}; + +export type FlowsStopFlowResponse = (FlowDetail); + +export type FlowsPauseFlowData = { + name: string; +}; + +export type FlowsPauseFlowResponse = (Message); + +export type FlowsResumeFlowData = { + name: string; +}; + +export type FlowsResumeFlowResponse = (Message); + export type FlowsValidateFlowData = { name: string; }; diff --git a/frontend/src/components/Flow/EdgeInspector.tsx b/frontend/src/components/Flow/EdgeInspector.tsx index d45f118..51d4a10 100644 --- a/frontend/src/components/Flow/EdgeInspector.tsx +++ b/frontend/src/components/Flow/EdgeInspector.tsx @@ -1,10 +1,13 @@ -import { ArrowRight, Trash2 } from "lucide-react" +import { useMutation } from "@tanstack/react-query" +import { ArrowRight, RotateCcw, Trash2 } from "lucide-react" import { motion } from "motion/react" import { useEffect, useRef, useState } from "react" +import { FlowsService } from "@/client" import { Button } from "@/components/ui/button" import { Popover, PopoverAnchor, PopoverContent } from "@/components/ui/popover" import { ScrollArea } from "@/components/ui/scroll-area" +import useCustomToast from "@/hooks/useCustomToast" import { cn } from "@/lib/utils" import { displayName } from "./deriveEdges" import { useLiveValue } from "./liveStore" @@ -77,6 +80,8 @@ export type InspectedEdge = { /** Node titles, so the popover names the two ends in the user's own words. */ from: string to: string + /** The producing node's id, which is what replaying the message goes through. */ + sourceId: string x: number y: number } @@ -96,6 +101,19 @@ export function EdgeInspector({ onUnbind: (message: string) => void }) { const live = useLiveValue(edge?.message) + const { showErrorToast } = useCustomToast() + // Publishing the value again from the node that produced it runs everything + // downstream exactly as the original did. + const replay = useMutation({ + mutationFn: (value: unknown) => + FlowsService.triggerNode({ + name: flow, + nodeId: edge?.sourceId ?? "", + requestBody: { values: { [edge?.message ?? ""]: value } }, + }), + onError: () => showErrorToast("The message could not be sent again."), + }) + if (!edge) return null const scalar = live === undefined ? null : formatScalar(live.value) @@ -115,6 +133,20 @@ export function EdgeInspector({ + {live === undefined ? null : ( + + )} + + + + + + + + {!enabled + ? "This flow is stopped" + : paused + ? "Let the flow carry on" + : "Hold the nodes; values still arrive"} + + + + + + + + + + + {enabled ? "Run every node once" : "Start the flow to run it"} + + ) } diff --git a/frontend/src/components/Flow/FlowEditor.tsx b/frontend/src/components/Flow/FlowEditor.tsx index 885b683..c180b0d 100644 --- a/frontend/src/components/Flow/FlowEditor.tsx +++ b/frontend/src/components/Flow/FlowEditor.tsx @@ -49,7 +49,7 @@ import { FlowTabs } from "./FlowTabs" import { LiveEdge } from "./LiveEdge" import { NodePanel } from "./NodePanel" import "./flow.css" -import { liveStore } from "./liveStore" +import { liveStore, useFlowPaused } from "./liveStore" import { flowKeys, flowQueryOptions, @@ -230,6 +230,7 @@ function FlowEditorInner({ const [editorExpanded, setEditorExpanded] = useState(false) const issues = detail.issues ?? [] + const paused = useFlowPaused(flowName) // Keep the latest document in a ref so autosave never captures a stale copy. const latest = useRef(detail.definition) @@ -389,6 +390,27 @@ function FlowEditorInner({ showErrorToast("The flow could not run. Check the node errors."), }) + const enableMutation = useMutation({ + mutationFn: (next: boolean) => + next + ? FlowsService.startFlow({ name: flowName }) + : FlowsService.stopFlow({ name: flowName }), + onSuccess: (detail) => { + queryClient.setQueryData(flowKeys.detail(flowName), detail) + queryClient.invalidateQueries({ queryKey: flowKeys.all, exact: true }) + }, + onError: () => showErrorToast("The flow could not be started or stopped."), + }) + + const pauseMutation = useMutation({ + mutationFn: (next: boolean) => + next + ? FlowsService.pauseFlow({ name: flowName }) + : FlowsService.resumeFlow({ name: flowName }), + // The engine answers with a flow_paused event, which is what the dock reads. + onError: () => showErrorToast("The flow could not be paused."), + }) + const renameMutation = useMutation({ mutationFn: (newName: string) => FlowsService.renameFlow({ @@ -659,6 +681,7 @@ function FlowEditorInner({ message: (edge.data as { message: string }).message, from: label(edge.source), to: label(edge.target), + sourceId: edge.source, x: event.clientX, y: event.clientY, }) @@ -710,14 +733,18 @@ function FlowEditorInner({ {panelOpen ? null : ( setPaletteOpen(true)} onRun={async () => { // Running executes what is stored, so the queued edit goes first. await flush() runMutation.mutate() }} + onTogglePause={() => pauseMutation.mutate(!paused)} onFocusNode={focusNode} /> )} @@ -752,6 +779,9 @@ function FlowEditorInner({ renameMutation.mutate(newName) }} onDelete={() => deleteMutation.mutate()} + enabled={detail.enabled ?? true} + toggling={enableMutation.isPending} + onToggleEnabled={(next) => enableMutation.mutate(next)} hasDraft={detail.has_draft ?? false} discarding={discard.isPending} onDiscardDraft={() => { diff --git a/frontend/src/components/Flow/FlowPanel.tsx b/frontend/src/components/Flow/FlowPanel.tsx index 8ac214e..3823ad2 100644 --- a/frontend/src/components/Flow/FlowPanel.tsx +++ b/frontend/src/components/Flow/FlowPanel.tsx @@ -11,6 +11,7 @@ import { DialogTitle, } from "@/components/ui/dialog" import { Input } from "@/components/ui/input" +import { Switch } from "@/components/ui/switch" import { PANEL_SECTION, SidePanel } from "./SidePanel" const NAME_PATTERN = /^[a-z][a-z0-9_]*$/ @@ -32,6 +33,9 @@ export function FlowPanel({ hasDraft, discarding, onDiscardDraft, + enabled, + toggling, + onToggleEnabled, onClose, }: { open: boolean @@ -44,6 +48,9 @@ export function FlowPanel({ hasDraft: boolean discarding: boolean onDiscardDraft: () => void + enabled: boolean + toggling: boolean + onToggleEnabled: (next: boolean) => void onClose: () => void }) { const [name, setName] = useState(definition.name) @@ -85,6 +92,24 @@ export function FlowPanel({ } >
+
+ Running +
+

+ {enabled + ? "The engine runs this flow: subscriptions, schedules and webhooks are live." + : "Stopped. Nothing of this flow is subscribed, scheduled or reachable."} +

+ +
+
+
Name
diff --git a/frontend/src/components/Flow/LogsPanel.tsx b/frontend/src/components/Flow/LogsPanel.tsx new file mode 100644 index 0000000..457bdef --- /dev/null +++ b/frontend/src/components/Flow/LogsPanel.tsx @@ -0,0 +1,120 @@ +import { Terminal } from "lucide-react" +import { useEffect, useRef } from "react" + +import { Button } from "@/components/ui/button" +import { + Popover, + PopoverContent, + PopoverTrigger, +} from "@/components/ui/popover" +import { ScrollArea } from "@/components/ui/scroll-area" +import { + Tooltip, + TooltipContent, + TooltipTrigger, +} from "@/components/ui/tooltip" +import { cn } from "@/lib/utils" +import { liveStore, useLiveLogs } from "./liveStore" + +function shortTime(ts: number): string { + return new Date(ts * 1000).toLocaleTimeString(undefined, { + hour12: false, + hour: "2-digit", + minute: "2-digit", + second: "2-digit", + }) +} + +/** Drop the flow prefix: every line in here belongs to the open flow. */ +function nodeLabel(nodeId: string, flow: string): string { + return nodeId.startsWith(`${flow}.`) ? nodeId.slice(flow.length + 1) : nodeId +} + +/** + * What the nodes of this flow printed, and the tracebacks of the ones that + * failed — the detail the one-line error bubble on a node has no room for. + */ +export function LogsPanel({ flow }: { flow: string }) { + const lines = useLiveLogs().filter((line) => line.flow === flow) + const bottom = useRef(null) + + // Follow the tail, which is where a running flow puts what just happened. + // biome-ignore lint/correctness/useExhaustiveDependencies: a new line is what scrolls. + useEffect(() => { + bottom.current?.scrollIntoView({ block: "end" }) + }, [lines.length]) + + return ( + + + + + + + + What this flow printed + + + +
+

+ Logs +

+ +
+ + {lines.length === 0 ? ( +

+ Nothing yet. Anything a node prints shows up here. +

+ ) : ( + +
    + {lines.map((line, index) => ( +
  • + + {shortTime(line.ts)}{" "} + + {nodeLabel(line.node, flow)} + + + + {line.text.replace(/\n+$/, "")} + {line.truncated ? "\n… truncated" : ""} + +
    • + ))} +
  • +
+ + )} + + + ) +} diff --git a/frontend/src/components/Flow/liveStore.ts b/frontend/src/components/Flow/liveStore.ts index 03da5a7..1b078ac 100644 --- a/frontend/src/components/Flow/liveStore.ts +++ b/frontend/src/components/Flow/liveStore.ts @@ -13,14 +13,28 @@ export type LiveStatus = { status: "active" | "error" | "running" | "success" error?: string | null } +/** One node execution's output, as the log panel shows it. */ +export type LogLine = { + flow: string + node: string + text: string + level: "info" | "error" + truncated?: boolean + ts: number +} type Listener = () => void +/** Enough to see what a flow has been doing, not a log store. */ +const LOG_LIMIT = 500 + const values = new Map() const statuses = new Map() // How many times a node has emitted. The number itself means nothing; a change // is what restarts the pulse. const emits = new Map() +let logLines: LogLine[] = [] +const paused = new Set() const listeners = new Map>() let connected = false @@ -79,6 +93,33 @@ export const liveStore = { emits.set(nodeId, (emits.get(nodeId) ?? 0) + 1) notify(`emit:${nodeId}`) }, + appendLog(line: LogLine) { + // A new array each time, so the hook's snapshot comparison sees the change. + logLines = [...logLines, line].slice(-LOG_LIMIT) + notify("logs") + }, + setLogs(lines: LogLine[]) { + logLines = lines.slice(-LOG_LIMIT) + notify("logs") + }, + clearLogs() { + logLines = [] + notify("logs") + }, + setPaused(flow: string, isPaused: boolean) { + if (isPaused) paused.add(flow) + else paused.delete(flow) + notify(`paused:${flow}`) + }, + setPausedFlows(flows: string[]) { + const next = new Set(flows) + for (const flow of new Set([...paused, ...next])) { + if (paused.has(flow) === next.has(flow)) continue + if (next.has(flow)) paused.add(flow) + else paused.delete(flow) + notify(`paused:${flow}`) + } + }, setConnected(next: boolean) { if (connected === next) return connected = next @@ -94,6 +135,10 @@ export const liveStore = { statuses.clear() for (const key of emits.keys()) notify(`emit:${key}`) emits.clear() + logLines = [] + notify("logs") + for (const flow of paused) notify(`paused:${flow}`) + paused.clear() }, } @@ -119,6 +164,21 @@ export function useNodeEmits(nodeId: string): number { ) } +/** Every captured line, newest last. Filtered by the panel that shows it. */ +export function useLiveLogs(): LogLine[] { + return useSyncExternalStore( + (listener) => subscribeKey("logs", listener), + () => logLines, + ) +} + +export function useFlowPaused(flow: string): boolean { + return useSyncExternalStore( + (listener) => subscribeKey(`paused:${flow}`, listener), + () => paused.has(flow), + ) +} + export function useLiveConnection(): boolean { return useSyncExternalStore( (listener) => { diff --git a/frontend/src/components/Flow/useFlowSocket.ts b/frontend/src/components/Flow/useFlowSocket.ts index b2e11b2..c279ef2 100644 --- a/frontend/src/components/Flow/useFlowSocket.ts +++ b/frontend/src/components/Flow/useFlowSocket.ts @@ -2,7 +2,7 @@ import { useQueryClient } from "@tanstack/react-query" import { useEffect, useRef } from "react" import { OpenAPI } from "@/client" -import { liveStore } from "./liveStore" +import { type LogLine, liveStore } from "./liveStore" import { flowKeys } from "./queries" const RECONNECT_MIN = 1000 @@ -13,14 +13,19 @@ type FlowEvent = type: "snapshot" values: Record nodes: { id: string; status: string; error?: string | null }[] + paused?: string[] + logs?: LogLine[] } | { type: "message_value"; name: string; value: unknown; ts: number } | { type: "node_executed"; node: string; outputs: number } | { type: "node_error"; node: string; error: string } | { type: "node_status"; node: string; status: string; error?: string | null } + | ({ type: "node_log" } & LogLine) + | { type: "flow_paused"; flow: string; paused: boolean } | { type: "pipeline_rebuilt" nodes: { id: string; status: string; error?: string | null }[] + paused?: string[] } function socketUrl(): string { @@ -65,6 +70,8 @@ export function useFlowSocket(onAuthFailure?: () => void): void { case "snapshot": liveStore.setValues(message.values) liveStore.setStatuses(message.nodes) + liveStore.setPausedFlows(message.paused ?? []) + liveStore.setLogs(message.logs ?? []) break case "message_value": liveStore.setValue(message.name, { @@ -88,10 +95,17 @@ export function useFlowSocket(onAuthFailure?: () => void): void { error: message.error, }) break + case "node_log": + liveStore.appendLog(message) + break + case "flow_paused": + liveStore.setPaused(message.flow, message.paused) + break case "pipeline_rebuilt": liveStore.setStatuses(message.nodes) - // Someone published, here or in another tab: the draft markers on - // the flow chips are stale until the list is fetched again. + liveStore.setPausedFlows(message.paused ?? []) + // Someone published or started a flow, here or in another tab: the + // markers on the flow chips are stale until the list is refetched. queryClient.invalidateQueries({ queryKey: flowKeys.all }) break } diff --git a/frontend/src/routes/_layout/index.tsx b/frontend/src/routes/_layout/index.tsx index fb06673..cc87acd 100644 --- a/frontend/src/routes/_layout/index.tsx +++ b/frontend/src/routes/_layout/index.tsx @@ -1,6 +1,15 @@ -import { createFileRoute } from "@tanstack/react-router" +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query" +import { createFileRoute, Link } from "@tanstack/react-router" +import { AlertCircle, Workflow } from "lucide-react" +import { type FlowSummary, FlowsService } from "@/client" +import { flowKeys, flowsQueryOptions } from "@/components/Flow/queries" +import { Badge } from "@/components/ui/badge" +import { Card } from "@/components/ui/card" +import { Skeleton } from "@/components/ui/skeleton" +import { Switch } from "@/components/ui/switch" import useAuth from "@/hooks/useAuth" +import useCustomToast from "@/hooks/useCustomToast" export const Route = createFileRoute("/_layout/")({ component: Dashboard, @@ -13,19 +22,137 @@ export const Route = createFileRoute("/_layout/")({ }), }) -function Dashboard() { - const { user: currentUser } = useAuth() +/** Another tab can stop a flow, and the engine can fail one on its own. */ +const REFRESH_INTERVAL = 10_000 + +function Tile({ label, value }: { label: string; value: number }) { + return ( + +
+

{value}

+

{label}

+
+ + ) +} + +function FlowRow({ flow }: { flow: FlowSummary }) { + const queryClient = useQueryClient() + const { showErrorToast } = useCustomToast() + const enabled = flow.enabled ?? true + + const toggle = useMutation({ + mutationFn: (next: boolean) => + next + ? FlowsService.startFlow({ name: flow.name }) + : FlowsService.stopFlow({ name: flow.name }), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: flowKeys.all }) + queryClient.invalidateQueries({ queryKey: flowKeys.detail(flow.name) }) + }, + onError: () => showErrorToast("The flow could not be started or stopped."), + }) return ( -
-
-

- Hi, {currentUser?.full_name || currentUser?.email} 👋 -

-

- Welcome back, nice to see you again!!! +

+ +

{flow.title || flow.name}

+

+ {flow.node_count === 1 ? "1 node" : `${flow.node_count} nodes`} + {flow.has_draft ? " · unpublished changes" : ""}

+ + + {(flow.error_count ?? 0) > 0 ? ( + + + {flow.error_count} + + ) : null} + + + {enabled ? (flow.paused ? "Paused" : "Running") : "Stopped"} + + + toggle.mutate(next)} + aria-label={`Run ${flow.title || flow.name}`} + data-testid="flow-enabled-switch" + /> +
+ ) +} + +function Dashboard() { + const { user: currentUser } = useAuth() + const { data, isPending } = useQuery({ + ...flowsQueryOptions(), + refetchInterval: REFRESH_INTERVAL, + }) + + const flows = data?.data ?? [] + const running = flows.filter((flow) => flow.enabled ?? true).length + const failing = flows.filter((flow) => (flow.error_count ?? 0) > 0).length + + return ( +
+
+

+ Hi, {currentUser?.full_name || currentUser?.email} 👋 +

+

+ {flows.length === 0 + ? "No flows yet." + : `${running} of ${flows.length} flows are running.`} +

+ +
+ + + +
+ + + {isPending ? ( +
+ + +
+ ) : flows.length === 0 ? ( +
+ + + +

+ Flows you build show up here, with what they are doing. +

+ + Go to flows + +
+ ) : ( + flows.map((flow) => ) + )} +
) } diff --git a/frontend/tests/drafts.spec.ts b/frontend/tests/drafts.spec.ts index 4a26dab..8c64565 100644 --- a/frontend/tests/drafts.spec.ts +++ b/frontend/tests/drafts.spec.ts @@ -14,7 +14,11 @@ test.describe.configure({ mode: "serial" }) const apiUrl = process.env.VITE_API_URL || "http://api.localhost" -async function api(page: Page, path: string, init: Record = {}) { +async function api( + page: Page, + path: string, + init: Record = {}, +) { const token = await page.evaluate(() => localStorage.getItem("access_token")) return page.request.fetch(`${apiUrl}/api/v1${path}`, { ...init, diff --git a/frontend/tests/runtime.spec.ts b/frontend/tests/runtime.spec.ts new file mode 100644 index 0000000..ae12b5c --- /dev/null +++ b/frontend/tests/runtime.spec.ts @@ -0,0 +1,128 @@ +import { expect, type Page, test } from "@playwright/test" + +/** + * Flows can be taken off the engine and put back, and what a node prints — or + * the traceback of one that fails — is readable without leaving the canvas. + */ + +const flowName = `test_runtime_${Date.now().toString(36)}` + +test.use({ storageState: "playwright/.auth/user.json" }) + +test.describe.configure({ mode: "serial" }) + +const apiUrl = process.env.VITE_API_URL || "http://api.localhost" + +const PRINTING_NODE = `def process(params): + print("sensor read 21.5 degrees") + return {"reading": 21.5} +` +const BROKEN_NODE = `def process(reading, params): + raise RuntimeError("downstream blew up") +` + +async function api( + page: Page, + path: string, + init: Record = {}, +) { + const token = await page.evaluate(() => localStorage.getItem("access_token")) + return page.request.fetch(`${apiUrl}/api/v1${path}`, { + ...init, + headers: { Authorization: `Bearer ${token}` }, + }) +} + +test.beforeAll(async ({ browser }) => { + const page = await browser.newPage({ + storageState: "playwright/.auth/user.json", + }) + await page.goto("/") + await api(page, `/flows/${flowName}`, { + method: "PUT", + data: { + name: flowName, + title: "Runtime", + nodes: [ + { + id: "sensor", + type: "python", + position: { x: 0, y: 0 }, + provides: [{ name: "reading", dtype: "float" }], + }, + { + id: "logger", + type: "python", + position: { x: 260, y: 0 }, + requires: [{ name: "reading", dtype: "float" }], + }, + ], + }, + }) + await api(page, `/flows/${flowName}/nodes/sensor/source`, { + method: "PUT", + data: { code: PRINTING_NODE }, + }) + await api(page, `/flows/${flowName}/nodes/logger/source`, { + method: "PUT", + data: { code: BROKEN_NODE }, + }) + const detail = await (await api(page, `/flows/${flowName}`)).json() + await api(page, `/flows/${flowName}/publish`, { + method: "POST", + data: { version: detail.definition.version }, + }) + await page.close() +}) + +test("the dashboard lists flows and can stop one", async ({ page }) => { + await page.goto("/") + const row = page + .getByTestId("dashboard-flow-row") + .filter({ hasText: "Runtime" }) + await expect(row).toBeVisible() + await expect(row).toContainText("Running") + + await row.getByTestId("flow-enabled-switch").click() + await expect(row).toContainText("Stopped") + + // A stopped flow is not something the engine will run. + const refused = await api(page, `/flows/${flowName}/run`, { + method: "POST", + data: { inputs: {} }, + }) + expect(refused.status()).toBe(409) + + await row.getByTestId("flow-enabled-switch").click() + await expect(row).toContainText("Running") +}) + +test("the logs panel shows what a node printed and why one failed", async ({ + page, +}) => { + await page.goto(`/flows/${flowName}`) + await page.waitForSelector(".react-flow__node") + + await page.getByTestId("run-flow").click() + await page.getByTestId("flow-logs").click() + + const panel = page.locator('[data-slot="popover-content"]') + await expect(panel).toContainText("sensor read 21.5 degrees") + await expect(panel).toContainText("RuntimeError") +}) + +test("a flow can be paused and let go again", async ({ page }) => { + await page.goto(`/flows/${flowName}`) + await page.waitForSelector(".react-flow__node") + + await page.getByTestId("pause-flow").click() + await expect(page.getByTestId("resume-flow")).toBeVisible() + expect((await (await api(page, `/flows/${flowName}`)).json()).paused).toBe( + true, + ) + + await page.getByTestId("resume-flow").click() + await expect(page.getByTestId("pause-flow")).toBeVisible() + + await api(page, `/flows/${flowName}`, { method: "DELETE" }) +}) diff --git a/frontend/tests/utils/user.ts b/frontend/tests/utils/user.ts index 65cb4f1..7ad04f6 100644 --- a/frontend/tests/utils/user.ts +++ b/frontend/tests/utils/user.ts @@ -7,7 +7,7 @@ export async function logInUser(page: Page, email: string, password: string) { await page.getByTestId("password-input").fill(password) await page.getByRole("button", { name: "Log In" }).click() await page.waitForURL("/") - await expect( - page.getByText("Welcome back, nice to see you again!"), - ).toBeVisible() + // The greeting is the one part of the dashboard that is there whether or not + // any flows are. + await expect(page.getByRole("heading", { name: /^Hi, / })).toBeVisible() }