Start, stop and pause flows, and show what their nodes print

Flows can now be taken off the engine and put back. Stopped state lives in a
runtime.json beside the flow, not in the flow document: the canvas autosaves
that document, so a stopped flow would otherwise start itself again on the
next edit. A stopped flow gets no subscriptions, schedules or webhooks, its
nodes are skipped by the scheduler, and running it answers 409. Pausing holds
a flow's nodes while its values keep arriving, so the canvas still shows what
is coming in.

Node code is user code and print is how it says things, so stdout is teed
through a contextvar sink active only during a node execution — one event per
execution, capped, so a chatty node cannot outrun the stream. A node that
fails sends its traceback the same way, trimmed to the author's own frames.
The dock gains a logs panel and a pause control; the dashboard replaces its
placeholder with what is running, stopped or failing; the edge inspector can
send the last message again.

Single-stepping is deferred and noted: the scheduler keeps no progress between
calls, so a step button would re-run the same node rather than advance.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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),
             }
         )
 

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:

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

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("<node ")),
+        0,
+    )
+    return "".join(
+        ["Traceback (most recent call last):\n"]
+        + traceback.format_list(frames[start:])
+        + traceback.format_exception_only(exc_type, exc)
+    )
+
+
+class Collector:
+    """Gathers one execution's output, with a ceiling on how much it keeps."""
+
+    __slots__ = ("chunks", "truncated", "_limit", "_size", "_max_bytes")
+
+    def __init__(self, limit: int = 200, max_bytes: int = 8192) -> 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

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))
 

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():

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):

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
     # -------------------------------------------------------------------------

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(

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`.

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:

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',
+            "<node demo.broken>",
+            "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

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