diff --git a/backend/app/alembic/versions/a3f1c07b52d9_add_run_tables.py b/backend/app/alembic/versions/a3f1c07b52d9_add_run_tables.py
new file mode 100644
index 0000000..c8b8992
--- /dev/null
+++ b/backend/app/alembic/versions/a3f1c07b52d9_add_run_tables.py
@@ -0,0 +1,126 @@
+"""Add run, run_node, run_metric and run_artifact
+
+Runs are the batch shape of the engine: a flow taken from its inputs to its
+outputs once, with parameters that identify it and a result worth keeping.
+Deliberately its own tables rather than columns on flow_run — a cascade row is
+rolled up and pruned on a retention window, and an experiment must not be.
+
+Revision ID: a3f1c07b52d9
+Revises: 087c44e16304
+Create Date: 2026-08-18 09:12:44.108312
+
+"""
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = 'a3f1c07b52d9'
+down_revision = '087c44e16304'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        'run',
+        sa.Column('id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
+        sa.Column('flow', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column('flow_version', sa.Integer(), nullable=False),
+        sa.Column('commit', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
+        sa.Column('params', sa.JSON(), nullable=True),
+        sa.Column(
+            'params_digest',
+            sqlmodel.sql.sqltypes.AutoString(length=64),
+            nullable=False,
+        ),
+        sa.Column('seed', sa.Integer(), nullable=True),
+        sa.Column('group_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True),
+        sa.Column('parent_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True),
+        sa.Column('cause', sqlmodel.sql.sqltypes.AutoString(length=32), nullable=False),
+        sa.Column('status', sqlmodel.sql.sqltypes.AutoString(length=16), nullable=False),
+        sa.Column(
+            'status_reason',
+            sqlmodel.sql.sqltypes.AutoString(length=1024),
+            nullable=False,
+        ),
+        sa.Column('labels', sa.JSON(), nullable=True),
+        sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+        sa.Column('started_at', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('finished_at', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('duration_ms', sa.Float(), nullable=False),
+        sa.Column('result', sa.JSON(), nullable=True),
+        sa.Column('engine', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
+        sa.Column('lease_at', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('actor', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.PrimaryKeyConstraint('id'),
+    )
+    op.create_index(op.f('ix_run_flow'), 'run', ['flow'], unique=False)
+    op.create_index(op.f('ix_run_status'), 'run', ['status'], unique=False)
+    op.create_index(op.f('ix_run_group_id'), 'run', ['group_id'], unique=False)
+    op.create_index(op.f('ix_run_created_at'), 'run', ['created_at'], unique=False)
+    op.create_index(
+        op.f('ix_run_params_digest'), 'run', ['params_digest'], unique=False
+    )
+
+    op.create_table(
+        'run_node',
+        sa.Column('run_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
+        sa.Column('node', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column('status', sqlmodel.sql.sqltypes.AutoString(length=16), nullable=False),
+        sa.Column('attempt', sa.Integer(), nullable=False),
+        sa.Column('started_at', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('duration_ms', sa.Float(), nullable=False),
+        sa.Column('worker', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
+        sa.Column('error', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column('logs', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column(
+            'cache_key', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False
+        ),
+        sa.PrimaryKeyConstraint('run_id', 'node'),
+    )
+    op.create_index(
+        op.f('ix_run_node_cache_key'), 'run_node', ['cache_key'], unique=False
+    )
+
+    op.create_table(
+        'run_metric',
+        sa.Column('run_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
+        sa.Column('name', sqlmodel.sql.sqltypes.AutoString(length=128), nullable=False),
+        sa.Column('step', sa.Integer(), nullable=False),
+        sa.Column('node', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column('ts', sa.Float(), nullable=False),
+        sa.Column('value', sa.Float(), nullable=False),
+        sa.PrimaryKeyConstraint('run_id', 'name', 'step'),
+    )
+
+    op.create_table(
+        'run_artifact',
+        sa.Column('run_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
+        sa.Column('name', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column('node', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+        sa.Column('digest', sqlmodel.sql.sqltypes.AutoString(length=71), nullable=False),
+        sa.Column('size', sa.BigInteger(), nullable=False),
+        sa.Column(
+            'media_type', sqlmodel.sql.sqltypes.AutoString(length=128), nullable=False
+        ),
+        sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint('run_id', 'name'),
+    )
+    op.create_index(
+        op.f('ix_run_artifact_digest'), 'run_artifact', ['digest'], unique=False
+    )
+
+
+def downgrade():
+    op.drop_index(op.f('ix_run_artifact_digest'), table_name='run_artifact')
+    op.drop_table('run_artifact')
+    op.drop_table('run_metric')
+    op.drop_index(op.f('ix_run_node_cache_key'), table_name='run_node')
+    op.drop_table('run_node')
+    op.drop_index(op.f('ix_run_params_digest'), table_name='run')
+    op.drop_index(op.f('ix_run_created_at'), table_name='run')
+    op.drop_index(op.f('ix_run_group_id'), table_name='run')
+    op.drop_index(op.f('ix_run_status'), table_name='run')
+    op.drop_index(op.f('ix_run_flow'), table_name='run')
+    op.drop_table('run')
diff --git a/backend/app/api/main.py b/backend/app/api/main.py
index af9d48d..6c18c4b 100644
--- a/backend/app/api/main.py
+++ b/backend/app/api/main.py
@@ -10,6 +10,7 @@ from app.api.routes import (
     oauth,
     observability,
     private,
+    runs,
     secrets,
     users,
     utils,
@@ -27,6 +28,7 @@ api_router.include_router(dashboards.router)
 api_router.include_router(messages.router)
 api_router.include_router(modules.router)
 api_router.include_router(observability.router)
+api_router.include_router(runs.router)
 # Always mounted so the generated SDK stays the same shape; the endpoints
 # themselves refuse to work unless MCP is switched on.
 api_router.include_router(oauth.router)
diff --git a/backend/app/api/routes/runs.py b/backend/app/api/routes/runs.py
new file mode 100644
index 0000000..6bf2f22
--- /dev/null
+++ b/backend/app/api/routes/runs.py
@@ -0,0 +1,282 @@
+"""Runs over the API: submit one, watch it, read what it made.
+
+Submitting returns immediately with a queued run — a training run is measured
+in hours, so nothing here waits for one. The way to follow a run is to poll it
+or to listen on the flow socket, which carries its start and finish.
+"""
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi.concurrency import run_in_threadpool
+from pydantic import BaseModel, Field
+from sqlmodel import col, select
+
+from app.api.deps import CurrentUser, SessionDep, get_current_user
+from app.flow.runs import RunRejected, RunService, new_run_id
+from app.flow.store import FlowNotFound
+from app.models import Run, RunArtifact, RunMetric, RunNode
+
+router = APIRouter(
+    prefix="/runs", tags=["runs"], dependencies=[Depends(get_current_user)]
+)
+
+#: A sweep bigger than this is almost always a mistake in a loop.
+MAX_SWEEP = 1000
+
+
+class RunCreate(BaseModel):
+    params: dict[str, Any] = Field(default_factory=dict)
+    seed: int | None = None
+    #: Run the unpublished draft instead of what is published.
+    draft: bool = False
+
+
+class SweepEntry(BaseModel):
+    params: dict[str, Any] = Field(default_factory=dict)
+    seed: int | None = None
+
+
+class SweepCreate(BaseModel):
+    runs: list[SweepEntry] = Field(default_factory=list)
+    draft: bool = False
+
+
+class RunNodeRow(BaseModel):
+    node: str
+    status: str
+    attempt: int
+    duration_ms: float
+    worker: str
+    error: str
+    logs: str
+
+
+class ArtifactRow(BaseModel):
+    name: str
+    node: str
+    digest: str
+    size: int
+    media_type: str
+
+
+class RunRow(BaseModel):
+    """A run without its result, which is the part that can be large."""
+
+    id: str
+    flow: str
+    status: str
+    status_reason: str
+    cause: str
+    params: dict[str, Any]
+    params_digest: str
+    seed: int | None
+    group_id: str | None
+    labels: list[str]
+    created_at: Any
+    started_at: Any = None
+    finished_at: Any = None
+    duration_ms: float
+    actor: str
+
+
+class RunDetail(RunRow):
+    result: dict[str, Any] = Field(default_factory=dict)
+    commit: str = ""
+    flow_version: int = 1
+    nodes: list[RunNodeRow] = Field(default_factory=list)
+    artifacts: list[ArtifactRow] = Field(default_factory=list)
+
+
+class MetricPoint(BaseModel):
+    step: int
+    ts: float
+    value: float
+
+
+class MetricSeries(BaseModel):
+    """The shape a chart widget already draws, so comparing runs is a binding."""
+
+    label: str
+    points: list[list[float]] = Field(default_factory=list)
+
+
+class SeriesAnswer(BaseModel):
+    metric: str
+    lines: list[MetricSeries] = Field(default_factory=list)
+
+
+def _service(request: Request) -> RunService:
+    service: RunService | None = getattr(request.app.state, "run_service", None)
+    if service is None:
+        raise HTTPException(status_code=503, detail="Runs are not available")
+    return service
+
+
+@router.post("/flows/{name}", response_model=RunRow, status_code=202)
+async def create_run(
+    name: str, body: RunCreate, request: Request, user: CurrentUser
+) -> Any:
+    """Queue one run of a flow."""
+    service = _service(request)
+    try:
+        return await run_in_threadpool(
+            service.submit,
+            name,
+            params=body.params,
+            seed=body.seed,
+            cause="api",
+            actor=user.email,
+            draft=body.draft,
+        )
+    except FlowNotFound as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    except RunRejected as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
+
+@router.post("/flows/{name}/sweep", response_model=list[RunRow], status_code=202)
+async def create_sweep(
+    name: str, body: SweepCreate, request: Request, user: CurrentUser
+) -> Any:
+    """Queue many runs of one flow under a shared group.
+
+    An ensemble is this with the same parameters and different seeds; a grid
+    search is this with the parameters spread out. Either way the caller
+    builds the list — the engine does not own a sweep grammar.
+    """
+    if not body.runs:
+        raise HTTPException(status_code=422, detail="A sweep needs at least one run")
+    if len(body.runs) > MAX_SWEEP:
+        raise HTTPException(
+            status_code=422, detail=f"A sweep is capped at {MAX_SWEEP} runs"
+        )
+    service = _service(request)
+    group = new_run_id()
+
+    def submit_all() -> list[Run]:
+        return [
+            service.submit(
+                name,
+                params=entry.params,
+                seed=entry.seed,
+                group_id=group,
+                cause="sweep",
+                actor=user.email,
+                draft=body.draft,
+            )
+            for entry in body.runs
+        ]
+
+    try:
+        return await run_in_threadpool(submit_all)
+    except FlowNotFound as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    except RunRejected as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
+
+@router.get("", response_model=list[RunRow])
+def read_runs(
+    session: SessionDep,
+    flow: str | None = None,
+    status: str | None = None,
+    group: str | None = None,
+    digest: str | None = None,
+    limit: int = 50,
+) -> Any:
+    """Runs, newest first. The queryable table an experiment log needs."""
+    statement = select(Run).order_by(col(Run.created_at).desc())
+    if flow:
+        statement = statement.where(col(Run.flow) == flow)
+    if status:
+        statement = statement.where(col(Run.status) == status)
+    if group:
+        statement = statement.where(col(Run.group_id) == group)
+    if digest:
+        statement = statement.where(col(Run.params_digest) == digest)
+    return list(session.exec(statement.limit(min(limit, 500))))
+
+
+@router.get("/{run_id}", response_model=RunDetail)
+def read_run(run_id: str, session: SessionDep) -> Any:
+    """One run in full: what it was asked, what each node did, what it made."""
+    run = session.get(Run, run_id)
+    if run is None:
+        raise HTTPException(status_code=404, detail="No such run")
+    nodes = session.exec(select(RunNode).where(col(RunNode.run_id) == run_id)).all()
+    artifacts = session.exec(
+        select(RunArtifact).where(col(RunArtifact.run_id) == run_id)
+    ).all()
+    detail = RunDetail.model_validate(run, from_attributes=True)
+    detail.nodes = [RunNodeRow.model_validate(n, from_attributes=True) for n in nodes]
+    detail.artifacts = [
+        ArtifactRow.model_validate(a, from_attributes=True) for a in artifacts
+    ]
+    return detail
+
+
+@router.post("/{run_id}/cancel", response_model=RunRow)
+async def cancel_run(run_id: str, request: Request, session: SessionDep) -> Any:
+    """Stop a run. One already past its last node is left as it finished."""
+    run = session.get(Run, run_id)
+    if run is None:
+        raise HTTPException(status_code=404, detail="No such run")
+    service = _service(request)
+    await run_in_threadpool(service.cancel, run_id)
+    session.refresh(run)
+    return run
+
+
+@router.get("/{run_id}/metrics", response_model=list[MetricPoint])
+def read_metrics(run_id: str, session: SessionDep, name: str, stride: int = 1) -> Any:
+    """One metric's series, in step order.
+
+    ``stride`` thins a long curve down: 3000 steps drawn on a 400-pixel chart
+    is 3000 points nobody can see.
+    """
+    statement = (
+        select(RunMetric)
+        .where(col(RunMetric.run_id) == run_id, col(RunMetric.name) == name)
+        .order_by(col(RunMetric.step))
+    )
+    rows = list(session.exec(statement))
+    if stride > 1:
+        rows = rows[:: max(1, stride)]
+    return rows
+
+
+@router.get("/series/compare", response_model=SeriesAnswer)
+def compare_metric(session: SessionDep, ids: str, metric: str) -> Any:
+    """One metric across several runs, as the chart widget's series shape.
+
+    This is the comparison view: it answers in the same shape a flow answers a
+    chart's query with, so putting three training curves beside each other is
+    a widget binding rather than a screen of its own.
+    """
+    run_ids = [part for part in ids.split(",") if part]
+    if not run_ids:
+        raise HTTPException(status_code=422, detail="Name at least one run")
+    runs = {
+        run.id: run
+        for run in session.exec(select(Run).where(col(Run.id).in_(run_ids))).all()
+    }
+    lines: list[MetricSeries] = []
+    for run_id in run_ids:
+        run = runs.get(run_id)
+        if run is None:
+            continue
+        rows = session.exec(
+            select(RunMetric)
+            .where(col(RunMetric.run_id) == run_id, col(RunMetric.name) == metric)
+            .order_by(col(RunMetric.step))
+        ).all()
+        label = run_id
+        if run.seed is not None:
+            label = f"{run_id} (seed {run.seed})"
+        lines.append(
+            MetricSeries(
+                label=label, points=[[float(row.step), row.value] for row in rows]
+            )
+        )
+    return SeriesAnswer(metric=metric, lines=lines)
diff --git a/backend/app/flow/controller.py b/backend/app/flow/controller.py
index 205fe2a..f0aa47b 100644
--- a/backend/app/flow/controller.py
+++ b/backend/app/flow/controller.py
@@ -11,6 +11,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import traceback
+from collections.abc import Callable
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any, cast
@@ -40,7 +41,7 @@ from app.flow.nodes import (
     SwitchNode,
     TriggerNode,
 )
-from app.flow.pipeline import Pipeline, ValidationIssue, ValueSource
+from app.flow.pipeline import NodeOutcome, Pipeline, ValidationIssue, ValueSource
 from app.flow.schemas import (
     BrainEdge,
     BrainGraph,
@@ -262,6 +263,8 @@ class FlowController:
         self.loaded: dict[str, LoadedNode] = {}
         self.issues: list[ValidationIssue] = []
         self.disabled: set[str] = set()
+        #: Flows that only run when a run asks them to.
+        self.batch: set[str] = set()
         self.supervisor = Supervisor(events)
         self.history_limits: dict[str, int] = {}
         self._lock = asyncio.Lock()
@@ -306,6 +309,10 @@ class FlowController:
                 for flow in published
                 if not self.store.read_enabled(flow.name)
             }
+            # A batch flow is built — the canvas draws it and validation covers
+            # it — but never activated: it runs when a run asks it to, and a
+            # subscription of its own would be a second way in.
+            self.batch = {flow.name for flow in published if flow.mode == "batch"}
             # Off the loop: building a python node asks its worker to compile,
             # which waits for a free slot — and a busy node holds one for as
             # long as its timeout. On the loop that stalls every request, the
@@ -373,8 +380,9 @@ class FlowController:
             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:
+            # that is what stopping it means. Nor does a batch flow, which has
+            # no outside to listen to.
+            if entry.flow in self.disabled or entry.flow in self.batch:
                 continue
             node.supervisor = self.supervisor
             try:
@@ -834,6 +842,33 @@ class FlowController:
             return
         self.pipeline.run(inputs or {}, nodes=self.pipeline.flow_nodes(flow))
 
+    def build_run_pipeline(
+        self,
+        flow: FlowDef,
+        state: StateBackend,
+        draft: bool = False,
+        observer: Callable[[NodeOutcome], None] | None = None,
+    ) -> Pipeline:
+        """Build one flow as a pipeline of its own, for a single run.
+
+        Nothing here is activated: a run executes the graph from its inputs
+        rather than waiting to be told something, so subscriptions, schedules
+        and webhooks would only be a second copy of what the live pipeline
+        already holds. The state is the run's, which is what keeps two runs of
+        one flow from overwriting each other's messages.
+        """
+        nodes, _loaded, initial_values, _inputs = self._build_flows([(flow, draft)])
+        pipeline = Pipeline(
+            nodes=nodes,
+            state=state,
+            events=self.events,
+            max_workers=self.max_workers,
+            initial_values=initial_values,
+            observer=observer,
+        )
+        pipeline.history_limits = self.history_limits
+        return pipeline
+
     def run_preview(self, flow_name: str, inputs: dict[str, Any] | None = None) -> None:
         """Run a flow's unpublished draft once, as the editor shows it.
 
diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py
index d3c53dd..c452d72 100644
--- a/backend/app/flow/pipeline.py
+++ b/backend/app/flow/pipeline.py
@@ -13,7 +13,7 @@ import threading
 import time
 import uuid
 from collections import deque
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from concurrent.futures import Future, ThreadPoolExecutor, wait
 from contextlib import contextmanager
 from typing import Any, Literal
@@ -76,6 +76,22 @@ def node_source(node: Node) -> ValueSource:
     return ValueSource(kind="node", id=node.id, label=node.local_id)
 
 
+class NodeOutcome(BaseModel):
+    """How one node execution went.
+
+    Handed to whoever is watching a particular pipeline rather than published:
+    a run has to record every node it ran, and the event bus drops what it
+    cannot keep up with.
+    """
+
+    node: str
+    ok: bool
+    duration_ms: float = 0.0
+    outputs: int = 0
+    error: str = ""
+    logs: str = ""
+
+
 class Pipeline:
     """Directed graph of nodes with automatic dependency resolution."""
 
@@ -96,6 +112,7 @@ class Pipeline:
         "_queue",
         "_node_pool",
         "history_limits",
+        "observer",
     )
 
     def __init__(
@@ -108,6 +125,7 @@ class Pipeline:
         disabled_flows: set[str] | None = None,
         work_queue: WorkQueue | None = None,
         node_pool: ThreadPoolExecutor | None = None,
+        observer: Callable[[NodeOutcome], None] | None = None,
     ) -> None:
         self._nodes = nodes or []
         # Stopped flows are stored and survive a restart; paused ones are a
@@ -127,6 +145,8 @@ class Pipeline:
         self._queue = work_queue
         # A pool owned by the execution service, so a wave does not build one.
         self._node_pool = node_pool
+        # Set by a run, which needs every node it executed written down.
+        self.observer = observer
         # How deep to keep each message's series; a chart asking for more
         # than the default puts its message in here. Swapped, never mutated.
         self.history_limits: dict[str, int] = {}
@@ -665,6 +685,7 @@ class Pipeline:
                         }
                     )
 
+            duration_ms = round((time.perf_counter() - started) * 1000, 2)
             self._publish(
                 {
                     "type": "node_executed",
@@ -673,17 +694,44 @@ class Pipeline:
                     # A node that returns nothing ran but published nothing,
                     # which is a different thing to show than one that emitted.
                     "outputs": len(result or {}),
-                    "duration_ms": round((time.perf_counter() - started) * 1000, 2),
+                    "duration_ms": duration_ms,
                     "run": entry_id,
                     "ts": time.time(),
                 }
             )
+            self._observe(
+                NodeOutcome(
+                    node=node.id,
+                    ok=True,
+                    duration_ms=duration_ms,
+                    outputs=len(result or {}),
+                    logs=collected.text,
+                )
+            )
             return result
         except Exception as exc:
             # One failing node must not take the rest of the graph down.
-            self.publish_error(node, exc, collected, entry_id)
+            error = self.publish_error(node, exc, collected, entry_id)
+            self._observe(
+                NodeOutcome(
+                    node=node.id,
+                    ok=False,
+                    duration_ms=round((time.perf_counter() - started) * 1000, 2),
+                    error=error,
+                    logs=collected.text,
+                )
+            )
             return None
 
+    def _observe(self, outcome: NodeOutcome) -> None:
+        """Tell the run watching this pipeline, if there is one."""
+        if self.observer is None:
+            return
+        try:
+            self.observer(outcome)
+        except Exception:
+            logger.exception("Run observer failed for '%s'", outcome.node)
+
     # -------------------------------------------------------------------------
     # Running, stopped, paused
     # -------------------------------------------------------------------------
diff --git a/backend/app/flow/queue.py b/backend/app/flow/queue.py
index ad3c468..35f70a3 100644
--- a/backend/app/flow/queue.py
+++ b/backend/app/flow/queue.py
@@ -40,7 +40,8 @@ class WorkItem:
     """One unit of journaled work.
 
     :param kind: ``cascade`` replays a node's outputs and runs what is
-        downstream; ``flush`` lets out what a node's rate limits held back.
+        downstream; ``flush`` lets out what a node's rate limits held back;
+        ``run`` is a whole batch run, and carries only its id.
     :param node: The node the item is about — the source for a cascade, the
         target for a node item.
     :param flow: The flow that node belongs to, so gating needs no lookup.
@@ -68,6 +69,9 @@ class WorkItem:
     entry_id: str = ""
     deliveries: int = 1
     enqueued_at: float = field(default_factory=time.time)
+    #: The run this item is, on a ``run`` item. Everything else about a run is
+    #: in its database row, so the journal only has to name it.
+    run_id: str = ""
 
     def to_fields(self) -> dict[str, str]:
         return {
@@ -80,6 +84,7 @@ class WorkItem:
             "guard_key": self.guard_key,
             "guard_value": self.guard_value,
             "enqueued_at": str(self.enqueued_at),
+            "run_id": self.run_id,
         }
 
     @classmethod
@@ -98,6 +103,7 @@ class WorkItem:
             entry_id=entry_id,
             deliveries=deliveries,
             enqueued_at=float(fields.get("enqueued_at") or time.time()),
+            run_id=fields.get("run_id", ""),
         )
 
 
diff --git a/backend/app/flow/runs.py b/backend/app/flow/runs.py
new file mode 100644
index 0000000..d0ef1c7
--- /dev/null
+++ b/backend/app/flow/runs.py
@@ -0,0 +1,516 @@
+"""Runs: a flow taken from its inputs to its outputs, once.
+
+A cascade is what an always-on flow does when a value arrives — it has no
+beginning and no end worth recording. A *run* is the other shape the same
+engine can take: parameters go in, the graph executes until it drains, and
+what it produced is kept. That is what an ML experiment is, and what a
+CI-style job is, so both are this one entity.
+
+Three things make a run different from a cascade, and each is deliberate:
+
+* **Its own state.** Message names are global keys, so two runs of one flow
+  sharing the engine's state would overwrite each other's values. A run gets a
+  state backend namespaced to itself, which is a constructor argument rather
+  than a change to the pipeline — every key the engine keeps, versions and
+  node memory included, already goes through that backend.
+* **Its own record.** The event bus drops what it cannot keep up with, which
+  is right for a live canvas and wrong for a result. The driver writes the
+  run's rows itself, from the thread that is running it.
+* **Its own durability.** The queue wakes an engine up; from the moment a run
+  starts, its database row is the truth. Redelivering hours of training
+  because an acknowledgement was late is not recovery, so a run is
+  acknowledged as soon as it is claimed and a stale lease — not an unacked
+  stream entry — is what marks a run its engine died in the middle of.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import socket
+import threading
+import time
+import uuid
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timedelta, timezone
+from typing import TYPE_CHECKING, Any
+
+from sqlalchemy import update
+from sqlmodel import Session, col, select
+
+from app.core.db import engine as db_engine
+from app.flow.messages import qualify
+from app.flow.pipeline import NodeOutcome, Pipeline
+from app.flow.queue import WorkItem, WorkQueue
+from app.flow.schemas import FlowDef
+from app.flow.state import MemoryState, StateBackend
+from app.models import Run, RunNode
+
+if TYPE_CHECKING:
+    from app.flow.controller import FlowController
+
+logger = logging.getLogger(__name__)
+
+#: How long a finished run's state is kept before Redis drops it. Long enough
+#: to look at what a failed run left behind, short enough not to accumulate.
+RUN_STATE_TTL = 24 * 3600
+#: How often a running run says it is still alive.
+LEASE_INTERVAL_S = 20.0
+#: A lease older than this belongs to an engine that is not coming back.
+LEASE_STALE_S = 90.0
+#: How often stale leases are looked for.
+SWEEP_INTERVAL_S = 30.0
+#: Runs driven at once. Node bodies are bounded by the worker pool anyway;
+#: this only bounds how many graphs are in flight.
+MAX_PARALLEL = 4
+CLAIM_COUNT = 4
+CLAIM_BLOCK_MS = 1000
+ERROR_CAP = 2000
+LOG_CAP = 8000
+
+#: Where a run's state lives, so it can never collide with the engine's own.
+RUN_NAMESPACE = "run"
+
+
+class RunRejected(ValueError):
+    """The run cannot be made: bad parameters, or a flow that cannot batch."""
+
+
+def new_run_id() -> str:
+    """Time-ordered, so the newest runs sort last without reading a column."""
+    return f"{int(time.time() * 1000):013d}-{uuid.uuid4().hex[:8]}"
+
+
+def digest_of(params: dict[str, Any], seed: int | None) -> str:
+    """What identifies a run's inputs: same parameters, same digest."""
+    canonical = json.dumps(
+        {"params": params, "seed": seed}, sort_keys=True, separators=(",", ":")
+    )
+    return hashlib.sha256(canonical.encode()).hexdigest()
+
+
+def batch_issues(flow: FlowDef) -> list[str]:
+    """Why this flow cannot be run as a batch, if it cannot.
+
+    Only one thing genuinely breaks: a port with a discretization interval
+    holds values back for a timer to release, and a run has no timer — the
+    engine would drop them instead. A delay node is fine; without a queue to
+    defer into it simply sleeps, which in a run is what was asked for.
+    """
+    issues: list[str] = []
+    for node in flow.nodes:
+        for spec in list(node.requires) + list(node.provides):
+            if spec.interval > 0:
+                issues.append(
+                    f"Node '{node.id}' rate-limits '{spec.port or spec.name}'. "
+                    "A run has no timer to release what that holds back, so "
+                    "the value would be dropped. Remove the interval to run "
+                    "this flow as a batch."
+                )
+    return issues
+
+
+def required_labels(flow: FlowDef) -> list[str]:
+    """Worker labels this flow's nodes ask for."""
+    return sorted(
+        {node.device for node in flow.nodes if node.device and node.device.strip()}
+    )
+
+
+def seed_values(flow: FlowDef, params: dict[str, Any]) -> dict[str, Any]:
+    """Turn a run's parameters into the messages the flow starts from."""
+    specs = {declared.spec.name: declared.spec for declared in flow.inputs}
+    values: dict[str, Any] = {}
+    for key, value in params.items():
+        spec = specs.get(key)
+        if spec is None:
+            known = ", ".join(sorted(specs)) or "none"
+            raise RunRejected(
+                f"'{key}' is not an input of flow '{flow.name}' (it declares: {known})"
+            )
+        try:
+            spec.check(value)
+        except TypeError as exc:
+            raise RunRejected(f"Parameter '{key}': {exc}") from exc
+        values[qualify(flow.name, spec.name)] = value
+    return values
+
+
+def collect_result(flow: FlowDef, state: StateBackend) -> dict[str, Any]:
+    """What the run produced, keyed by message name without the flow prefix."""
+    prefix = f"{flow.name}."
+    if flow.outputs:
+        names = [qualify(flow.name, name) for name in flow.outputs]
+    else:
+        # Everything the flow ended up holding. The engine's own bookkeeping is
+        # keyed by `__thing__:message`, so it never starts with the flow name.
+        names = sorted(key for key in state.keys() if key.startswith(prefix))
+    result: dict[str, Any] = {}
+    for name in names:
+        if name in state:
+            result[name[len(prefix) :] if name.startswith(prefix) else name] = state[
+                name
+            ]
+    return result
+
+
+class RunService:
+    """Accepts runs, drives them, and writes down what they did."""
+
+    def __init__(
+        self,
+        controller: FlowController,
+        queue: WorkQueue,
+        state_factory: Callable[[str], StateBackend] | None = None,
+        parallel: int = MAX_PARALLEL,
+    ) -> None:
+        self.controller = controller
+        self.queue = queue
+        # Without one, a run gets a private in-memory state — which is exactly
+        # the isolation it wants, minus surviving the process.
+        self._state_factory = state_factory or (lambda _ns: MemoryState())
+        self.engine_name = f"{socket.gethostname()}-{os.getpid()}"[:64]
+        self._pool = ThreadPoolExecutor(max_workers=parallel, thread_name_prefix="run")
+        self._stop = threading.Event()
+        self._consumer: threading.Thread | None = None
+        self._keeper: threading.Thread | None = None
+        # Runs this process is driving, and the pipeline each is running, so a
+        # cancel has something to hold on to.
+        self._active: dict[str, Pipeline] = {}
+        self._cancelled: set[str] = set()
+        self._lock = threading.Lock()
+
+    # -------------------------------------------------------------------------
+    # Lifecycle
+    # -------------------------------------------------------------------------
+
+    def start(self) -> None:
+        if self._consumer is not None:
+            return
+        self._consumer = threading.Thread(
+            target=self._consume, name="run-consumer", daemon=True
+        )
+        self._consumer.start()
+        self._keeper = threading.Thread(
+            target=self._keep_leases, name="run-leases", daemon=True
+        )
+        self._keeper.start()
+
+    def stop(self) -> None:
+        self._stop.set()
+        for thread in (self._consumer, self._keeper):
+            if thread is not None:
+                thread.join(timeout=5)
+        self._consumer = None
+        self._keeper = None
+        self._pool.shutdown(wait=False)
+        self.queue.close()
+
+    def alive(self) -> bool:
+        return self._consumer is not None and self._consumer.is_alive()
+
+    # -------------------------------------------------------------------------
+    # Accepting work
+    # -------------------------------------------------------------------------
+
+    def submit(
+        self,
+        flow_name: str,
+        params: dict[str, Any] | None = None,
+        seed: int | None = None,
+        group_id: str | None = None,
+        cause: str = "api",
+        actor: str = "",
+        draft: bool = False,
+    ) -> Run:
+        """Journal a run and wake an engine up for it. Never blocks on it."""
+        flow = self.controller.store.read_flow(flow_name, draft=draft)
+        issues = batch_issues(flow)
+        if issues:
+            raise RunRejected(" ".join(issues))
+        params = params or {}
+        # Checked here rather than in the driver: a caller who mistyped a
+        # parameter should be told now, not by a run that fails in a minute.
+        seed_values(flow, params)
+
+        run = Run(
+            id=new_run_id(),
+            flow=flow.name,
+            flow_version=flow.version,
+            commit=self.controller.store.head(),
+            params=params,
+            params_digest=digest_of(params, seed),
+            seed=seed,
+            group_id=group_id,
+            cause=cause,
+            status="queued",
+            labels=required_labels(flow),
+            created_at=datetime.now(timezone.utc),
+            actor=actor,
+        )
+        with Session(db_engine) as session:
+            session.add(run)
+            session.commit()
+            session.refresh(run)
+
+        self.queue.add(WorkItem(kind="run", node="", flow=flow.name, run_id=run.id))
+        return run
+
+    def cancel(self, run_id: str) -> bool:
+        """Stop a run: nothing further is scheduled once its nodes return.
+
+        A node already executing is left to finish. Killing one needs the
+        worker pool to know which run it belongs to, which is what the metric
+        stream adds next; until then a cancel that arrives mid-node is a stop
+        rather than an interruption.
+        """
+        with self._lock:
+            pipeline = self._active.get(run_id)
+            if pipeline is None:
+                # Not running here — if it is still queued, refusing to start
+                # is all the cancelling it needs.
+                cancelled = self._finish_queued(run_id)
+                if cancelled:
+                    self._cancelled.add(run_id)
+                return cancelled
+            self._cancelled.add(run_id)
+        # Reusing the pause gate: a gated node is never submitted, so the graph
+        # drains instead of going further.
+        pipeline.pause(self._flow_of(run_id) or "")
+        return True
+
+    def _flow_of(self, run_id: str) -> str | None:
+        with Session(db_engine) as session:
+            run = session.get(Run, run_id)
+            return run.flow if run else None
+
+    def _finish_queued(self, run_id: str) -> bool:
+        with Session(db_engine) as session:
+            result = session.exec(
+                update(Run)
+                .where(col(Run.id) == run_id, col(Run.status) == "queued")
+                .values(
+                    status="cancelled",
+                    finished_at=datetime.now(timezone.utc),
+                    status_reason="Cancelled before it started",
+                )
+            )
+            session.commit()
+            return bool(result.rowcount)
+
+    # -------------------------------------------------------------------------
+    # Threads
+    # -------------------------------------------------------------------------
+
+    def _consume(self) -> None:
+        failures = 0
+        while not self._stop.is_set():
+            try:
+                items = self.queue.claim(CLAIM_COUNT, CLAIM_BLOCK_MS)
+                failures = 0
+            except Exception as exc:
+                failures += 1
+                logger.error("Could not claim runs: %s", exc)
+                self._stop.wait(min(30.0, 2.0**failures))
+                continue
+            for item in items:
+                # Acknowledged before it runs: from here on the row is the
+                # record, and a lease that stops moving is what says otherwise.
+                self.queue.ack(item)
+                if not item.run_id:
+                    continue
+                try:
+                    self._pool.submit(self._drive, item.run_id)
+                except RuntimeError:
+                    logger.warning("Run %s not started: shutting down", item.run_id)
+
+    def _keep_leases(self) -> None:
+        """Say the local runs are alive, and clean up after engines that died."""
+        last_sweep = 0.0
+        while not self._stop.is_set():
+            self._stop.wait(LEASE_INTERVAL_S)
+            if self._stop.is_set():
+                break
+            with self._lock:
+                mine = list(self._active)
+            now = datetime.now(timezone.utc)
+            try:
+                if mine:
+                    with Session(db_engine) as session:
+                        session.exec(
+                            update(Run)
+                            .where(col(Run.id).in_(mine))
+                            .values(lease_at=now)
+                        )
+                        session.commit()
+                if time.monotonic() - last_sweep >= SWEEP_INTERVAL_S:
+                    last_sweep = time.monotonic()
+                    self._sweep_abandoned(now)
+            except Exception:
+                logger.exception("Could not refresh run leases")
+
+    def _sweep_abandoned(self, now: datetime) -> None:
+        cutoff = now - timedelta(seconds=LEASE_STALE_S)
+        with Session(db_engine) as session:
+            result = session.exec(
+                update(Run)
+                .where(
+                    col(Run.status) == "running",
+                    col(Run.lease_at) < cutoff,
+                )
+                .values(
+                    status="abandoned",
+                    finished_at=now,
+                    status_reason="The engine running it stopped reporting",
+                )
+            )
+            session.commit()
+            if result.rowcount:
+                logger.warning("Marked %d run(s) abandoned", result.rowcount)
+
+    # -------------------------------------------------------------------------
+    # Driving one run
+    # -------------------------------------------------------------------------
+
+    def _claim(self, run_id: str) -> Run | None:
+        """Take the run, or leave it: whoever moves it out of `queued` owns it.
+
+        The compare-and-swap is what makes a redelivered item harmless — the
+        second engine to arrive updates nothing and walks away.
+        """
+        now = datetime.now(timezone.utc)
+        with Session(db_engine) as session:
+            result = session.exec(
+                update(Run)
+                .where(col(Run.id) == run_id, col(Run.status) == "queued")
+                .values(
+                    status="running",
+                    started_at=now,
+                    lease_at=now,
+                    engine=self.engine_name,
+                )
+            )
+            session.commit()
+            if not result.rowcount:
+                return None
+            return session.exec(select(Run).where(col(Run.id) == run_id)).first()
+
+    def _drive(self, run_id: str) -> None:
+        run = self._claim(run_id)
+        if run is None:
+            return
+        started = time.perf_counter()
+        status = "ok"
+        reason = ""
+        result: dict[str, Any] = {}
+        state: StateBackend | None = None
+        errors = 0
+
+        def observe(outcome: NodeOutcome) -> None:
+            nonlocal errors
+            if not outcome.ok:
+                errors += 1
+            self._record_node(run_id, outcome)
+
+        try:
+            flow = self.controller.store.read_flow(run.flow)
+            state = self._state_factory(f"{RUN_NAMESPACE}:{run_id}")
+            pipeline = self.controller.build_run_pipeline(
+                flow, state=state, observer=observe
+            )
+            with self._lock:
+                self._active[run_id] = pipeline
+                if run_id in self._cancelled:
+                    pipeline.pause(flow.name)
+            self._publish(run, "run_started")
+            pipeline.run(seed_values(flow, run.params))
+            result = collect_result(flow, state)
+            with self._lock:
+                cancelled = run_id in self._cancelled
+            if cancelled:
+                status = "cancelled"
+                reason = "Cancelled while running"
+            elif errors:
+                status = "error"
+                reason = f"{errors} node(s) failed"
+        except Exception as exc:
+            logger.exception("Run %s failed", run_id)
+            status = "error"
+            reason = f"{type(exc).__name__}: {exc}"[:1024]
+        finally:
+            with self._lock:
+                self._active.pop(run_id, None)
+                self._cancelled.discard(run_id)
+            duration = round((time.perf_counter() - started) * 1000, 2)
+            self._finish(run_id, status, reason, result, duration)
+            run.status = status
+            self._publish(run, "run_finished")
+            # Its values were only ever this run's; nothing reads them once it
+            # has a result. On Redis the namespace would expire anyway.
+            if state is not None and status != "error":
+                try:
+                    state.clear()
+                except Exception:
+                    logger.warning("Could not clear state of run %s", run_id)
+
+    def _record_node(self, run_id: str, outcome: NodeOutcome) -> None:
+        row = RunNode(
+            run_id=run_id,
+            node=outcome.node[:255],
+            status="ok" if outcome.ok else "error",
+            started_at=datetime.now(timezone.utc),
+            duration_ms=outcome.duration_ms,
+            error=outcome.error[:ERROR_CAP],
+            logs=outcome.logs[:LOG_CAP],
+        )
+        try:
+            with Session(db_engine) as session:
+                session.merge(row)
+                session.commit()
+        except Exception:
+            logger.exception(
+                "Could not record node '%s' of run %s", outcome.node, run_id
+            )
+
+    def _finish(
+        self,
+        run_id: str,
+        status: str,
+        reason: str,
+        result: dict[str, Any],
+        duration_ms: float,
+    ) -> None:
+        try:
+            with Session(db_engine) as session:
+                session.exec(
+                    update(Run)
+                    .where(col(Run.id) == run_id)
+                    .values(
+                        status=status,
+                        status_reason=reason,
+                        result=result,
+                        duration_ms=duration_ms,
+                        finished_at=datetime.now(timezone.utc),
+                    )
+                )
+                session.commit()
+        except Exception:
+            logger.exception("Could not close run %s", run_id)
+
+    def _publish(self, run: Run, kind: str) -> None:
+        events = self.controller.events
+        if events is None:
+            return
+        events.publish(
+            {
+                "type": kind,
+                "flow": run.flow,
+                "run": run.id,
+                "status": run.status,
+                "group": run.group_id or "",
+                "ts": time.time(),
+            }
+        )
diff --git a/backend/app/flow/schemas.py b/backend/app/flow/schemas.py
index b598726..dd34a2d 100644
--- a/backend/app/flow/schemas.py
+++ b/backend/app/flow/schemas.py
@@ -48,7 +48,23 @@ class NodeDef(BaseModel):
             "Seconds this node's code may run before it is stopped. This "
             "covers the first call's imports, which can be much slower than "
             "the body. Above 60 the engine may deliver its work again while "
-            "it is still running."
+            "it is still running — in a batch run, which never redelivers, "
+            "it is an idle timeout instead: silence this long is a kill."
+        ),
+    )
+    device: str | None = Field(
+        default=None,
+        description=(
+            "Label of the worker this node's code must run on, such as 'gpu'. "
+            "Empty means the engine's own workers. A run needing a label no "
+            "attached worker carries waits rather than failing."
+        ),
+    )
+    device_policy: Literal["require", "prefer"] = Field(
+        default="require",
+        description=(
+            "What to do when no worker carries `device`: wait for one, or run "
+            "locally anyway."
         ),
     )
 
@@ -73,6 +89,22 @@ class FlowDef(BaseModel):
     nodes: list[NodeDef] = Field(default_factory=list)
     inputs: list[FlowInput] = Field(default_factory=list)
     version: int = 1
+    mode: Literal["live", "batch"] = Field(
+        default="live",
+        description=(
+            "A live flow reacts to what arrives: its subscriptions, schedules "
+            "and webhooks run until it is stopped. A batch flow only runs when "
+            "a run asks it to, from its inputs to its outputs, and is never "
+            "activated."
+        ),
+    )
+    outputs: list[str] = Field(
+        default_factory=list,
+        description=(
+            "Messages a batch run reports as its result, unqualified. Empty "
+            "means every message the flow ends up holding."
+        ),
+    )
 
     @field_validator("name")
     @classmethod
diff --git a/backend/app/flow/store.py b/backend/app/flow/store.py
index 9739abc..24c60da 100644
--- a/backend/app/flow/store.py
+++ b/backend/app/flow/store.py
@@ -160,6 +160,11 @@ class FlowStore:
     def _lib_file(self, name: str) -> Path:
         return self.root / LIB_DIR / f"{name}.py"
 
+    def head(self) -> str:
+        """The commit the store is at, so a result can name the code it ran."""
+        result = self._git("rev-parse", "HEAD")
+        return result.stdout.strip() if result.returncode == 0 else ""
+
     # -------------------------------------------------------------------------
     # Module requirements
     #
diff --git a/backend/app/main.py b/backend/app/main.py
index 85fde6d..70fac28 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -25,6 +25,7 @@ from app.flow.nodes.http import close_shared_client
 from app.flow.pipeline import ValueSource
 from app.flow.plugins import load_plugins
 from app.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkQueue
+from app.flow.runs import RUN_STATE_TTL, RunService
 from app.flow.secrets import init_secrets
 from app.flow.state import MemoryState, RedisState, StateBackend
 from app.flow.store import FlowStore
@@ -46,13 +47,31 @@ def _state_backend() -> StateBackend:
     return MemoryState()
 
 
-def _work_queue() -> WorkQueue:
+def _work_queue(namespace: str = "pipeline") -> WorkQueue:
     """Redis makes queued work survive the process; memory does not pretend to."""
     if settings.REDIS_HOST:
-        return RedisWorkQueue(host=settings.REDIS_HOST, port=settings.REDIS_PORT)
+        return RedisWorkQueue(
+            host=settings.REDIS_HOST, port=settings.REDIS_PORT, namespace=namespace
+        )
     return MemoryWorkQueue()
 
 
+def _run_state(namespace: str) -> StateBackend:
+    """A state backend of a run's own, which is what isolates it.
+
+    It expires: a finished run's messages are read out into its result, and
+    what is left is only worth keeping while someone might look at it.
+    """
+    if settings.REDIS_HOST:
+        return RedisState(
+            host=settings.REDIS_HOST,
+            port=settings.REDIS_PORT,
+            namespace=namespace,
+            ttl=RUN_STATE_TTL,
+        )
+    return MemoryState()
+
+
 def _mcp_sessions() -> AbstractAsyncContextManager[None]:
     """The MCP session manager's run scope, or nothing when MCP is off."""
     if not settings.MCP_ENABLED:
@@ -110,6 +129,15 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     # Charts need a deeper series than the default; tell the engine
     # before it starts recording.
     controller.set_history_limits(dashboards.history_requirements())
+    # Runs read from a stream of their own: a burst of sweep runs must not
+    # stand between the automations and their work, and a run that takes an
+    # hour must not be judged by the cascade reaper's timings.
+    run_service = RunService(
+        controller=controller,
+        queue=_work_queue("run"),
+        state_factory=_run_state,
+    )
+    app.state.run_service = run_service
     watchdog = LoopWatchdog(event_bus)
     app.state.watchdog = watchdog
     watchdog_task = asyncio.create_task(watchdog.run(), name="loop-watchdog")
@@ -118,6 +146,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         MetricsCollector(event_bus).run(), name="metrics-collector"
     )
     await controller.start()
+    run_service.start()
     try:
         # A mounted sub-app gets no lifespan of its own, so the MCP session
         # manager is entered here; without it every /mcp request fails.
@@ -127,6 +156,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         watchdog_task.cancel()
         alerts_task.cancel()
         metrics_task.cancel()
+        await run_in_threadpool(run_service.stop)
         await controller.stop()
         pool.stop()
         close_shared_client()
diff --git a/backend/app/models.py b/backend/app/models.py
index 7653873..3adb9f6 100644
--- a/backend/app/models.py
+++ b/backend/app/models.py
@@ -1,5 +1,6 @@
 import uuid
 from datetime import datetime, timezone
+from typing import Any
 
 from pydantic import EmailStr
 from sqlalchemy import JSON, Column, DateTime
@@ -279,3 +280,131 @@ class FlowRun(SQLModel, table=True):
     errors: int = 0
     duration_ms: float = 0.0
     deliveries: int = 1
+
+
+# -----------------------------------------------------------------------------
+# Runs
+#
+# A cascade above is what an always-on flow does when a value arrives. A run is
+# the other shape: a batch flow taken from its inputs to its outputs once, with
+# parameters that identify it and a result worth keeping. An experiment is a
+# run, and so is a CI-style job — same entity, different caller.
+#
+# Kept apart from the observability tables on purpose: those are rolled up and
+# pruned on a retention window, and an experiment nobody wants deleted must not
+# share that fate.
+# -----------------------------------------------------------------------------
+
+
+class Run(SQLModel, table=True):
+    """One finite execution of a flow, with what it was asked and what it made."""
+
+    __tablename__ = "run"
+
+    id: str = Field(primary_key=True, max_length=64)
+    flow: str = Field(index=True, max_length=255)
+    #: The flow document this ran, and the commit it was read at, so a result
+    #: can be traced back to the code that produced it.
+    flow_version: int = 1
+    commit: str = Field(default="", max_length=64)
+    params: dict[str, Any] = Field(sa_column=Column(JSON), default_factory=dict)
+    #: sha256 over params and seed. Two runs of the same thing share it, which
+    #: is what makes "have I already run this?" a lookup.
+    params_digest: str = Field(default="", index=True, max_length=64)
+    seed: int | None = None
+    #: Runs submitted together — a sweep, an ensemble.
+    group_id: str | None = Field(default=None, index=True, max_length=64)
+    #: The run this one was made from, on a retry.
+    parent_id: str | None = Field(default=None, max_length=64)
+    #: api, hook, sweep or cli.
+    cause: str = Field(default="api", max_length=32)
+    #: queued, running, ok, error, cancelled or abandoned.
+    status: str = Field(default="queued", index=True, max_length=16)
+    #: Why it is where it is: what it waits for, or what went wrong.
+    status_reason: str = Field(default="", max_length=1024)
+    #: Worker labels its nodes need, so a run with nowhere to go can say so.
+    labels: list[str] = Field(sa_column=Column(JSON), default_factory=list)
+    created_at: datetime = Field(
+        index=True,
+        sa_type=DateTime(timezone=True),  # type: ignore
+    )
+    started_at: datetime | None = Field(
+        default=None,
+        sa_type=DateTime(timezone=True),  # type: ignore
+    )
+    finished_at: datetime | None = Field(
+        default=None,
+        sa_type=DateTime(timezone=True),  # type: ignore
+    )
+    duration_ms: float = 0.0
+    #: The flow's declared outputs once it finished.
+    result: dict[str, Any] = Field(sa_column=Column(JSON), default_factory=dict)
+    #: Which engine holds it, and when it last said so. A run whose lease has
+    #: gone stale is one whose engine died mid-run.
+    engine: str = Field(default="", max_length=64)
+    lease_at: datetime | None = Field(
+        default=None,
+        sa_type=DateTime(timezone=True),  # type: ignore
+    )
+    actor: str = Field(default="", max_length=255)
+
+
+class RunNode(SQLModel, table=True):
+    """What one node did inside a run: the per-stage view of it."""
+
+    __tablename__ = "run_node"
+
+    run_id: str = Field(primary_key=True, max_length=64)
+    node: str = Field(primary_key=True, max_length=255)
+    #: ok, error, skipped, cached or cancelled.
+    status: str = Field(default="ok", max_length=16)
+    attempt: int = 1
+    started_at: datetime | None = Field(
+        default=None,
+        sa_type=DateTime(timezone=True),  # type: ignore
+    )
+    duration_ms: float = 0.0
+    #: Which worker ran it: "local", or a remote worker's name.
+    worker: str = Field(default="local", max_length=64)
+    error: str = ""
+    logs: str = ""
+    #: Everything this node's output depends on, hashed. Recorded from the
+    #: start so that skipping a stage whose inputs have not changed is later a
+    #: lookup rather than a migration.
+    cache_key: str = Field(default="", index=True, max_length=64)
+
+
+class RunMetric(SQLModel, table=True):
+    """One number a run reported, at one step.
+
+    Written by the run's own driver rather than folded off the event bus: the
+    bus drops what it cannot keep up with, which is the right trade for a live
+    canvas and the wrong one for a training curve.
+    """
+
+    __tablename__ = "run_metric"
+
+    run_id: str = Field(primary_key=True, max_length=64)
+    name: str = Field(primary_key=True, max_length=128)
+    #: -1 for a value with no step of its own — a final score.
+    step: int = Field(primary_key=True)
+    node: str = Field(default="", max_length=255)
+    ts: float = 0.0
+    value: float = 0.0
+
+
+class RunArtifact(SQLModel, table=True):
+    """A file a run produced, addressed by the hash of its content."""
+
+    __tablename__ = "run_artifact"
+
+    run_id: str = Field(primary_key=True, max_length=64)
+    name: str = Field(primary_key=True, max_length=255)
+    node: str = Field(default="", max_length=255)
+    digest: str = Field(default="", index=True, max_length=71)
+    size: int = 0
+    media_type: str = Field(default="application/octet-stream", max_length=128)
+    created_at: datetime = Field(
+        default_factory=lambda: datetime.now(timezone.utc),
+        sa_type=DateTime(timezone=True),  # type: ignore
+    )
diff --git a/backend/tests/flow/test_runs.py b/backend/tests/flow/test_runs.py
new file mode 100644
index 0000000..2e052c0
--- /dev/null
+++ b/backend/tests/flow/test_runs.py
@@ -0,0 +1,170 @@
+"""What a run is made of: isolation, the per-node record, and what it reports.
+
+The service itself needs a database, so what is checked here is the part that
+decides whether a run is correct — that two runs of one flow cannot see each
+other's messages, that every node executed is reported once, and that the
+parameters a caller sends are refused before anything runs if they are wrong.
+"""
+
+import pytest
+
+from app.flow.messages import DType, MessageSpec
+from app.flow.nodes import Node
+from app.flow.pipeline import Pipeline
+from app.flow.runs import (
+    RunRejected,
+    batch_issues,
+    collect_result,
+    digest_of,
+    required_labels,
+    seed_values,
+)
+from app.flow.schemas import FlowDef, FlowInput, NodeDef
+from app.flow.state import MemoryState
+
+
+def spec(name: str, dtype: DType = DType.FLOAT, **kwargs) -> MessageSpec:
+    return MessageSpec(name=name, dtype=dtype, **kwargs)
+
+
+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 double_flow() -> FlowDef:
+    """A flow with one input, one node and one declared output."""
+    return FlowDef(
+        name="study",
+        mode="batch",
+        inputs=[FlowInput(spec=spec("lr"), initial=0.1)],
+        outputs=["loss"],
+        nodes=[
+            NodeDef(
+                id="train",
+                requires=[spec("lr")],
+                provides=[spec("loss")],
+            )
+        ],
+    )
+
+
+def build(flow: FlowDef, state: MemoryState, observer=None) -> Pipeline:
+    """The pipeline a run drives, without the controller that normally builds it."""
+    node = make_node(
+        "train",
+        flow.name,
+        lambda lr, params: {"loss": lr * 2},
+        requires=[spec("lr")],
+        provides=[spec("loss")],
+    )
+    return Pipeline(nodes=[node], state=state, observer=observer)
+
+
+# -----------------------------------------------------------------------------
+# Isolation — the reason a run has a state backend of its own
+# -----------------------------------------------------------------------------
+
+
+def test_two_runs_of_one_flow_do_not_see_each_other():
+    flow = double_flow()
+    first, second = MemoryState(), MemoryState()
+
+    build(flow, first).run(seed_values(flow, {"lr": 0.5}))
+    build(flow, second).run(seed_values(flow, {"lr": 4.0}))
+
+    assert collect_result(flow, first) == {"loss": 1.0}
+    assert collect_result(flow, second) == {"loss": 8.0}
+
+
+def test_result_falls_back_to_everything_the_flow_holds():
+    flow = double_flow()
+    flow.outputs = []
+    state = MemoryState()
+
+    build(flow, state).run(seed_values(flow, {"lr": 1.0}))
+
+    # The input is part of what the flow ended up holding; the engine's own
+    # bookkeeping keys are not.
+    assert collect_result(flow, state) == {"lr": 1.0, "loss": 2.0}
+
+
+# -----------------------------------------------------------------------------
+# The record — every node a run executed, reported once
+# -----------------------------------------------------------------------------
+
+
+def test_observer_sees_every_node_once():
+    flow = double_flow()
+    seen = []
+
+    build(flow, MemoryState(), observer=seen.append).run(seed_values(flow, {"lr": 1.0}))
+
+    assert [(o.node, o.ok) for o in seen] == [("study.train", True)]
+    assert seen[0].outputs == 1
+
+
+def test_observer_reports_a_failing_node_with_its_error():
+    seen = []
+
+    def boom(params):
+        raise ValueError("no convergence")
+
+    node = make_node("train", "study", boom, provides=[spec("loss")])
+    Pipeline(nodes=[node], state=MemoryState(), observer=seen.append).run()
+
+    assert len(seen) == 1
+    assert not seen[0].ok
+    assert "no convergence" in seen[0].error
+
+
+def test_a_failing_observer_does_not_take_the_node_down():
+    def refuse(_outcome):
+        raise RuntimeError("the database is gone")
+
+    node = make_node(
+        "train", "study", lambda params: {"loss": 1.0}, provides=[spec("loss")]
+    )
+    pipeline = Pipeline(nodes=[node], state=MemoryState(), observer=refuse)
+    pipeline.run()
+
+    assert pipeline.state["study.loss"] == 1.0
+
+
+# -----------------------------------------------------------------------------
+# What a caller may ask for
+# -----------------------------------------------------------------------------
+
+
+def test_parameters_must_be_declared_inputs():
+    flow = double_flow()
+    with pytest.raises(RunRejected, match="not an input"):
+        seed_values(flow, {"learning_rate": 0.1})
+
+
+def test_parameters_are_type_checked_before_anything_runs():
+    flow = double_flow()
+    with pytest.raises(RunRejected, match="lr"):
+        seed_values(flow, {"lr": "fast"})
+
+
+def test_a_rate_limited_port_cannot_be_run_as_a_batch():
+    flow = double_flow()
+    flow.nodes[0].provides = [spec("loss", interval=30)]
+    # Without a queue there is no timer to release what an interval holds, so
+    # the value would be dropped rather than delayed.
+    assert batch_issues(flow)
+    assert not batch_issues(double_flow())
+
+
+def test_the_digest_identifies_the_inputs_not_their_order():
+    assert digest_of({"a": 1, "b": 2}, 3) == digest_of({"b": 2, "a": 1}, 3)
+    assert digest_of({"a": 1}, 3) != digest_of({"a": 1}, 4)
+
+
+def test_labels_come_from_the_nodes_that_ask_for_a_device():
+    flow = double_flow()
+    assert required_labels(flow) == []
+    flow.nodes[0].device = "gpu"
+    assert required_labels(flow) == ["gpu"]