diff --git a/backend/fluksio/alembic/env.py b/backend/fluksio/alembic/env.py
index d28a18b..d4f4fa4 100755
--- a/backend/fluksio/alembic/env.py
+++ b/backend/fluksio/alembic/env.py
@@ -20,7 +20,7 @@ if config.config_file_name is not None:
 # target_metadata = None
 
 from fluksio.models import SQLModel  # noqa
-from fluksio.core.config import settings # noqa
+from fluksio.core.config import settings  # noqa
 from fluksio.core.types import UTCDateTime  # noqa
 
 target_metadata = SQLModel.metadata
@@ -38,6 +38,7 @@ def render_item(type_, obj, autogen_context):
         return "sa.DateTime(timezone=True)"
     return False
 
+
 # other values from the config, defined by the needs of env.py,
 # can be acquired:
 # my_important_option = config.get_main_option("my_important_option")
diff --git a/backend/fluksio/alembic/versions/e7d3b1a9c624_list_indexes.py b/backend/fluksio/alembic/versions/e7d3b1a9c624_list_indexes.py
new file mode 100644
index 0000000..0d899ce
--- /dev/null
+++ b/backend/fluksio/alembic/versions/e7d3b1a9c624_list_indexes.py
@@ -0,0 +1,37 @@
+"""Composite indexes for the three list screens.
+
+Each of these filters on one column and orders by another, and every index on
+the table was single-column — so SQLite used one of them and sorted the rest by
+hand. `engine_event` is the worst of the three: its filter is `type != 'audit'`,
+which no index on `type` can serve at all.
+
+Revision ID: e7d3b1a9c624
+Revises: d4a71e9c2b58
+Create Date: 2026-08-29
+
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "e7d3b1a9c624"
+down_revision = "d4a71e9c2b58"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # `GET /runs` filters flow and/or status, newest first.
+    op.create_index("ix_run_flow_created", "run", ["flow", "created_at"])
+    op.create_index("ix_run_status_created", "run", ["status", "created_at"])
+    # `GET /observability/runs` does the same over the execution history.
+    op.create_index("ix_flow_run_flow_started", "flow_run", ["flow", "started_at"])
+    # `GET /observability/events` excludes audit rows, newest first.
+    op.create_index("ix_engine_event_type_ts", "engine_event", ["type", "ts"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_engine_event_type_ts", table_name="engine_event")
+    op.drop_index("ix_flow_run_flow_started", table_name="flow_run")
+    op.drop_index("ix_run_status_created", table_name="run")
+    op.drop_index("ix_run_flow_created", table_name="run")
diff --git a/backend/fluksio/api/routes/artifacts.py b/backend/fluksio/api/routes/artifacts.py
index 605887e..471737e 100644
--- a/backend/fluksio/api/routes/artifacts.py
+++ b/backend/fluksio/api/routes/artifacts.py
@@ -19,6 +19,7 @@ from starlette.concurrency import run_in_threadpool
 
 from fluksio.api.deps import user_from_token
 from fluksio.core import security
+from fluksio.core.config import settings
 from fluksio.core.db import engine
 from fluksio.flow.artifacts import ArtifactStore
 
@@ -84,14 +85,32 @@ async def put_artifact(
 
     Spooled to disk as it arrives rather than buffered: a video segment is as
     legitimate a body here as a checkpoint, and neither should have to fit in
-    memory twice.
+    memory twice. Capped, because nothing else here was: any account, and any
+    worker credential, could otherwise fill the data volume.
     """
     store = _store(request)
+    cap = settings.MAX_ARTIFACT_BYTES
+    declared = request.headers.get("content-length")
+    if cap and declared and declared.isdigit() and int(declared) > cap:
+        raise HTTPException(
+            status_code=413, detail=f"An artifact may be at most {cap} bytes"
+        )
     handle = tempfile.NamedTemporaryFile(dir=store.root, delete=False)
+    written = 0
     try:
         with handle:
             async for chunk in request.stream():
-                handle.write(chunk)
+                written += len(chunk)
+                # A chunked body declares no length, so the stream is what
+                # actually holds the limit.
+                if cap and written > cap:
+                    raise HTTPException(
+                        status_code=413,
+                        detail=f"An artifact may be at most {cap} bytes",
+                    )
+                # Off the event loop: this is a write syscall per chunk, for
+                # as long as the upload lasts.
+                await run_in_threadpool(handle.write, chunk)
         return await run_in_threadpool(
             store.put_file, Path(handle.name), media_type, name
         )
diff --git a/backend/fluksio/api/routes/login.py b/backend/fluksio/api/routes/login.py
index 2fec5b2..37ccae4 100644
--- a/backend/fluksio/api/routes/login.py
+++ b/backend/fluksio/api/routes/login.py
@@ -1,18 +1,21 @@
 from datetime import timedelta
+from secrets import compare_digest
 from typing import Annotated, Any
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
 from fastapi.responses import HTMLResponse
 from fastapi.security import OAuth2PasswordRequestForm
 
 from fluksio import crud
 from fluksio.api.deps import CurrentUser, SessionDep, get_current_active_superuser
+from fluksio.api.routes.oauth import too_many
 from fluksio.core import security
 from fluksio.core.config import settings
 from fluksio.models import Message, NewPassword, Token, UserPublic, UserUpdate
 from fluksio.utils import (
     generate_password_reset_token,
     generate_reset_password_email,
+    password_fingerprint,
     send_email,
     verify_password_reset_token,
 )
@@ -20,17 +23,35 @@ from fluksio.utils import (
 router = APIRouter(tags=["login"])
 
 
+#: Attempts per address per window, and the window. Argon2 is deliberately
+#: expensive — tens of megabytes and several passes per verification — and
+#: this route is unauthenticated and runs in the shared threadpool, so a
+#: burst of guesses is a memory and a concurrency problem before it is a
+#: credential one. A person who mistypes twice is nowhere near it.
+LOGIN_ATTEMPTS = 10
+LOGIN_WINDOW_S = 300.0
+
+
 @router.post("/login/access-token")
 def login_access_token(
-    session: SessionDep, form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
+    request: Request,
+    session: SessionDep,
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
 ) -> Token:
     """
     OAuth2 compatible token login, get an access token for future requests
     """
+    if too_many(request, "login", LOGIN_ATTEMPTS, LOGIN_WINDOW_S, record=False):
+        raise HTTPException(
+            status_code=429, detail="Too many sign-in attempts. Try again shortly."
+        )
     user = crud.authenticate(
         session=session, email=form_data.username, password=form_data.password
     )
     if not user:
+        # Only the failures are counted, so somebody signing in all day is
+        # never near the limit and somebody guessing is.
+        too_many(request, "login", LOGIN_ATTEMPTS, LOGIN_WINDOW_S)
         raise HTTPException(status_code=400, detail="Incorrect email or password")
     elif not user.is_active:
         raise HTTPException(status_code=400, detail="Inactive user")
@@ -51,7 +72,9 @@ def test_token(current_user: CurrentUser) -> Any:
 
 
 @router.post("/password-recovery/{email}")
-def recover_password(email: str, session: SessionDep) -> Message:
+def recover_password(
+    email: str, session: SessionDep, background: BackgroundTasks
+) -> Message:
     """
     Password Recovery
     """
@@ -60,11 +83,18 @@ def recover_password(email: str, session: SessionDep) -> Message:
     # Always return the same response to prevent email enumeration attacks
     # Only send email if user actually exists
     if user:
-        password_reset_token = generate_password_reset_token(email=email)
+        password_reset_token = generate_password_reset_token(
+            email=email, hashed_password=user.hashed_password
+        )
         email_data = generate_reset_password_email(
             email_to=user.email, email=email, token=password_reset_token
         )
-        send_email(
+        # After the response, not before it: sending is a synchronous SMTP
+        # conversation, and doing it inline made the reply slow exactly when
+        # the address exists — which is the enumeration the identical wording
+        # below is there to prevent.
+        background.add_task(
+            send_email,
             email_to=user.email,
             subject=email_data.subject,
             html_content=email_data.html_content,
@@ -79,15 +109,21 @@ def reset_password(session: SessionDep, body: NewPassword) -> Message:
     """
     Reset password
     """
-    email = verify_password_reset_token(token=body.token)
-    if not email:
+    claims = verify_password_reset_token(token=body.token)
+    if not claims:
         raise HTTPException(status_code=400, detail="Invalid token")
+    email, fingerprint = claims
     user = crud.get_user_by_email(session=session, email=email)
     if not user:
         # Don't reveal that the user doesn't exist - use same error as invalid token
         raise HTTPException(status_code=400, detail="Invalid token")
     elif not user.is_active:
         raise HTTPException(status_code=400, detail="Inactive user")
+    if not compare_digest(fingerprint, password_fingerprint(user.hashed_password)):
+        # The link already set a password, or the password moved since it was
+        # sent. Either way this one is spent — a reset link used to work over
+        # and over for the whole of its 48 hours.
+        raise HTTPException(status_code=400, detail="Invalid token")
     user_in_update = UserUpdate(password=body.new_password)
     crud.update_user(
         session=session,
@@ -113,7 +149,9 @@ def recover_password_html_content(email: str, session: SessionDep) -> Any:
             status_code=404,
             detail="The user with this username does not exist in the system.",
         )
-    password_reset_token = generate_password_reset_token(email=email)
+    password_reset_token = generate_password_reset_token(
+        email=email, hashed_password=user.hashed_password
+    )
     email_data = generate_reset_password_email(
         email_to=user.email, email=email, token=password_reset_token
     )
diff --git a/backend/fluksio/api/routes/oauth.py b/backend/fluksio/api/routes/oauth.py
index 07e2a4b..ce77e21 100644
--- a/backend/fluksio/api/routes/oauth.py
+++ b/backend/fluksio/api/routes/oauth.py
@@ -26,7 +26,7 @@ import re
 import secrets
 import time
 import uuid
-from collections import defaultdict, deque
+from collections import OrderedDict, defaultdict, deque
 from datetime import UTC, datetime, timedelta
 from typing import Any
 from urllib.parse import urlencode, urlparse
@@ -81,24 +81,56 @@ def _error(code: str, description: str, status: int = 400) -> JSONResponse:
 # enough: the engine is one process, and this is a speed bump, not a boundary.
 # -----------------------------------------------------------------------------
 
-_hits: dict[str, deque[float]] = defaultdict(deque)
+#: How many distinct buckets are remembered at once. Entries were pruned
+#: *within* a deque and never removed, so one key per source address survived
+#: for the life of the process.
+_MAX_BUCKETS = 4096
+_hits: OrderedDict[str, deque[float]] = OrderedDict()
 
 
-def _too_many(bucket: str, limit: int, window: float) -> bool:
+def _too_many(bucket: str, limit: int, window: float, record: bool = True) -> bool:
     now = time.monotonic()
-    seen = _hits[bucket]
+    seen = _hits.get(bucket)
+    if seen is None:
+        seen = _hits[bucket] = deque()
+        while len(_hits) > _MAX_BUCKETS:
+            _hits.popitem(last=False)
+    else:
+        _hits.move_to_end(bucket)
     while seen and now - seen[0] > window:
         seen.popleft()
     if len(seen) >= limit:
         return True
-    seen.append(now)
+    if record:
+        seen.append(now)
     return False
 
 
 def _client_ip(request: Request) -> str:
+    """The caller's address, or the proxy's if it did not forward one.
+
+    Behind Traefik `request.client.host` is the proxy for every request, which
+    made every per-address limit here one global bucket — so one caller could
+    lock out everyone. The leftmost forwarded entry is the client, and is what
+    the panel pairing screen already reads.
+    """
+    forwarded = request.headers.get("x-forwarded-for", "").split(",")[0].strip()
+    if forwarded:
+        return forwarded
     return request.client.host if request.client else "unknown"
 
 
+def too_many(
+    request: Request, action: str, limit: int, window: float, record: bool = True
+) -> bool:
+    """Per-address rate limit, for anything unauthenticated on any router.
+
+    ``record=False`` asks the question without spending an attempt, for a
+    caller that only wants to count the ones that failed.
+    """
+    return _too_many(f"{action}:{_client_ip(request)}", limit, window, record)
+
+
 # -----------------------------------------------------------------------------
 # Helpers
 # -----------------------------------------------------------------------------
diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py
index 54b6176..75bcd86 100644
--- a/backend/fluksio/api/routes/observability.py
+++ b/backend/fluksio/api/routes/observability.py
@@ -8,9 +8,9 @@ cannot render while the engine is degraded is the wrong way round.
 """
 
 from datetime import UTC, datetime, timedelta
-from typing import Any, Literal
+from typing import Annotated, Any, Literal
 
-from fastapi import APIRouter, Depends, Request
+from fastapi import APIRouter, Depends, Query, Request
 from fastapi.concurrency import run_in_threadpool
 from pydantic import BaseModel, model_validator
 from sqlalchemy import ColumnElement, Integer, cast, func
@@ -141,6 +141,11 @@ def _closed_before(stride: int) -> datetime:
     return datetime.fromtimestamp(now // stride * stride, UTC)
 
 
+#: The most slices one timeseries answers with. `hours=720&bucket_s=60` is
+#: 43 200 points into a single JSON array; a chart draws a few hundred.
+MAX_POINTS = 2000
+
+
 def _window_hours(hours: int) -> int:
     """A window the rollups can answer for: an hour at least, retention at most.
 
@@ -245,14 +250,28 @@ def read_timeseries(
     session: SessionDep,
     flow: str | None = None,
     node: str | None = None,
-    hours: int = 24,
-    bucket_s: int = 60,
+    hours: Annotated[int, Query(ge=0, le=8760)] = 24,
+    bucket_s: Annotated[int, Query(ge=1, le=86400)] = 60,
+    since: datetime | None = None,
+    until: datetime | None = None,
 ) -> Any:
-    """Executions, errors and timings over time, summed across nodes."""
+    """Executions, errors and timings over time, summed across nodes.
+
+    ``hours`` measures back from now, which is what a range picker asks for.
+    ``since`` and ``until`` name a window instead — inclusive and exclusive,
+    the same pair ``/runs`` and ``/events`` take — so a chart dragged to a
+    span can be re-fetched at that span's own resolution rather than
+    magnifying the buckets it already has.
+    """
     hours = _window_hours(hours)
     # The database does the fold: a week of minute rows per node used to cross
     # the wire on every poll, and only the slices need to.
     stride = max(60, bucket_s)
+    if since is not None and until is not None:
+        # Enough slices to draw with, and never so many that the response is
+        # the problem the fold was there to solve.
+        span = (_aware(until) - _aware(since)).total_seconds()
+        stride = max(stride, int(span // MAX_POINTS) + 1)
     slot = (_epoch(col(MetricBucket.bucket)) // stride * stride).label("slot")
     statement = sa_select(
         slot,
@@ -264,8 +283,8 @@ def read_timeseries(
         func.sum(col(MetricBucket.lag_sum_ms)).label("lag_sum_ms"),
         func.sum(col(MetricBucket.items)).label("items"),
     ).where(
-        col(MetricBucket.bucket) >= _since(hours),
-        col(MetricBucket.bucket) < _closed_before(stride),
+        col(MetricBucket.bucket) >= (_aware(since) if since else _since(hours)),
+        col(MetricBucket.bucket) < (_aware(until) if until else _closed_before(stride)),
     )
     if flow:
         statement = statement.where(col(MetricBucket.flow) == flow)
@@ -287,7 +306,9 @@ def read_timeseries(
 
 
 @router.get("/flows", response_model=list[FlowRollup])
-def read_flow_rollups(session: SessionDep, hours: int = 24) -> Any:
+def read_flow_rollups(
+    session: SessionDep, hours: Annotated[int, Query(ge=0, le=8760)] = 24
+) -> Any:
     """One row per flow, with a coarse trend of how much it ran."""
     hours = _window_hours(hours)
     window = hours * 3600
@@ -388,7 +409,7 @@ def read_runs(
     status: str | None = None,
     since: datetime | None = None,
     until: datetime | None = None,
-    limit: int = 50,
+    limit: Annotated[int, Query(ge=1, le=200)] = 50,
 ) -> Any:
     """Recent cascades, newest first, and how many there were in total.
 
@@ -432,7 +453,7 @@ def read_events(
     run: str | None = None,
     since: datetime | None = None,
     until: datetime | None = None,
-    limit: int = 100,
+    limit: Annotated[int, Query(ge=1, le=500)] = 100,
 ) -> Any:
     """What went wrong, or who changed what. Newest first.
 
@@ -459,7 +480,9 @@ def read_events(
 
 
 @router.get("/dead-letter", response_model=list[DeadLetter])
-async def read_dead_letters(controller: FlowControllerDep, limit: int = 50) -> Any:
+async def read_dead_letters(
+    controller: FlowControllerDep, limit: Annotated[int, Query(ge=1, le=200)] = 50
+) -> Any:
     """Work the engine gave up on, which nothing else surfaces."""
     if controller.execution is None:
         return []
diff --git a/backend/fluksio/api/routes/panels.py b/backend/fluksio/api/routes/panels.py
index 33adec9..85e21e7 100644
--- a/backend/fluksio/api/routes/panels.py
+++ b/backend/fluksio/api/routes/panels.py
@@ -35,7 +35,15 @@ from fluksio.cloud import config as cloud_config
 from fluksio.core import security
 from fluksio.core.config import settings
 from fluksio.flow.events import event_bus
-from fluksio.flow.panels import PanelDef, PanelsConfig, find, read_config, write_config
+from fluksio.flow.panels import (
+    PanelDef,
+    PanelsConfig,
+    edit_lock,
+    find,
+    read_config,
+    write_config,
+)
+from fluksio.flow.state import CONNECT_TIMEOUT_S, SOCKET_TIMEOUT_S
 from fluksio.models import Message
 
 #: Gated per route rather than on the router: the two pairing endpoints are the
@@ -104,6 +112,10 @@ class _PendingStore:
                 host=settings.REDIS_HOST,
                 port=settings.REDIS_PORT,
                 decode_responses=True,
+                # A silent Redis must fail this request, not hold it open.
+                socket_timeout=SOCKET_TIMEOUT_S,
+                socket_connect_timeout=CONNECT_TIMEOUT_S,
+                health_check_interval=30,
             )
             if settings.REDIS_HOST
             else None
@@ -288,7 +300,6 @@ async def save_panels(body: PanelsConfig) -> Any:
     how a device is unpaired along with its assignment. Re-pairing one screen
     and keeping the assignment is ``/{panel_id}/unpair`` below.
     """
-    stored = {p.id: p.nonce for p in (await run_in_threadpool(read_config)).panels}
     seen = set()
     for panel in body.panels:
         if panel.id in seen:
@@ -296,12 +307,20 @@ async def save_panels(body: PanelsConfig) -> Any:
                 status_code=422, detail=f"Two panels named {panel.id!r}"
             )
         seen.add(panel.id)
-        # The nonce belongs to this installation, not to whoever is writing the
-        # panels back: a client holding an older copy must not be able to undo
-        # a revocation by saving an arrangement.
-        panel.nonce = stored.get(panel.id, 0)
 
-    await run_in_threadpool(write_config, body)
+    def _save() -> None:
+        # Read and write under one lock. The nonce belongs to this
+        # installation, not to whoever is writing the panels back: a client
+        # holding an older copy must not be able to undo a revocation by
+        # saving an arrangement — and reading it in a separate step from
+        # writing it is exactly how an unpair in between was undone.
+        with edit_lock:
+            stored = {p.id: p.nonce for p in read_config().panels}
+            for panel in body.panels:
+                panel.nonce = stored.get(panel.id, 0)
+            write_config(body)
+
+    await run_in_threadpool(_save)
     # Which dashboards hang on which panel just changed. An empty name says
     # that much and no more: every screen listening rescopes and refetches
     # what it shows, rather than waiting for whenever it next reads.
@@ -436,12 +455,19 @@ async def unpair_panel(panel_id: str) -> Any:
     A credential the portal minted for a remote screen carries no nonce, so
     this does not reach it; that one is revoked at the hub.
     """
-    config = await run_in_threadpool(read_config)
-    panel = next((p for p in config.panels if p.id == panel_id), None)
-    if panel is None:
+
+    def _unpair() -> bool:
+        with edit_lock:
+            config = read_config()
+            panel = next((p for p in config.panels if p.id == panel_id), None)
+            if panel is None:
+                return False
+            panel.nonce += 1
+            write_config(config)
+            return True
+
+    if not await run_in_threadpool(_unpair):
         raise HTTPException(status_code=404, detail=f"No panel named {panel_id!r}")
-    panel.nonce += 1
-    await run_in_threadpool(write_config, config)
     # The screen is still holding a socket. One event and it refetches, which
     # is where it meets the 401 that sends it back to the pairing code.
     event_bus.publish({"type": "dashboard_changed", "dashboard": "", "ts": time.time()})
diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py
index bea6b67..e71f2d1 100644
--- a/backend/fluksio/api/routes/runs.py
+++ b/backend/fluksio/api/routes/runs.py
@@ -12,9 +12,9 @@ import time
 from collections.abc import Iterator
 from datetime import UTC, datetime
 from itertools import groupby
-from typing import Any, Literal
+from typing import Annotated, Any, Literal
 
-from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
 from fastapi.concurrency import run_in_threadpool
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, Field, model_validator
@@ -23,6 +23,7 @@ from sqlalchemy import select as sa_select
 from sqlmodel import Session, col, select
 
 from fluksio.api.deps import CurrentUser, SessionDep, get_current_user
+from fluksio.core.db import writing
 from fluksio.flow.events import event_bus
 from fluksio.flow.messages import requalify
 from fluksio.flow.runs import RunRejected, RunService, new_run_id
@@ -271,8 +272,8 @@ def read_runs(
     digest: str | None = None,
     since: datetime | None = None,
     before: datetime | None = None,
-    limit: int = 50,
-    offset: int = 0,
+    limit: Annotated[int, Query(ge=1, le=500)] = 50,
+    offset: Annotated[int, Query(ge=0)] = 0,
 ) -> Any:
     """Runs, newest first. The queryable table an experiment log needs.
 
@@ -306,6 +307,11 @@ def read_overview(session: SessionDep) -> Any:
     The list caps at 500 newest runs, so counting flows on the client goes
     wrong the moment a history outgrows one page. The database counts instead.
     """
+    # Deliberately unwindowed. It is a group-by over the whole `run` table,
+    # which is never pruned — but a run kept for years is the point of the
+    # table, and a window here would drop a finished experiment off the rail
+    # rather than making it cheaper to find. A rollup is the answer if the
+    # scan ever measures.
     statement = sa_select(
         col(Run.flow),
         col(Run.status),
@@ -346,6 +352,12 @@ METRIC_COLUMNS = ("run", "name", "step", "ts", "value")
 
 #: A run's own columns in the wide table. Its inputs and its final numbers
 #: follow, prefixed, so an input named "status" cannot collide with the run's.
+#: The most runs one export covers. High enough that a real sweep fits and low
+#: enough that an unfiltered request cannot read the whole table into memory —
+#: which it did, before a byte was streamed, while holding one of the fifteen
+#: pooled database connections for the length of the download.
+EXPORT_CAP = 10_000
+
 RUN_COLUMNS = (
     "id",
     "flow",
@@ -370,7 +382,12 @@ def _selected(
     since: datetime | None,
     until: datetime | None,
 ) -> list[Run]:
-    """The runs an export covers, newest first — the filters the list takes."""
+    """The runs an export covers, newest first — the filters the list takes.
+
+    Capped: the filters bound a *sensible* request and nothing bounded an
+    unfiltered one, so asking for everything read every row of the table into
+    memory before a byte was sent. A truncated export says so in a header.
+    """
     statement = select(Run).order_by(col(Run.created_at).desc())
     if flow:
         statement = statement.where(col(Run.flow) == flow)
@@ -385,7 +402,7 @@ def _selected(
         statement = statement.where(col(Run.created_at) >= _aware(since))
     if until:
         statement = statement.where(col(Run.created_at) < _aware(until))
-    return list(session.exec(statement))
+    return list(session.exec(statement.limit(EXPORT_CAP + 1)))
 
 
 def _cell(value: Any) -> Any:
@@ -458,7 +475,11 @@ def _csv(columns: list[str], chunks: Iterator[list[dict[str, Any]]]) -> Iterator
 
 
 def _stream(
-    fmt: str, name: str, columns: list[str], chunks: Iterator[list[dict[str, Any]]]
+    fmt: str,
+    name: str,
+    columns: list[str],
+    chunks: Iterator[list[dict[str, Any]]],
+    truncated: bool = False,
 ) -> StreamingResponse:
     """The rows out, a chunk at a time rather than a list.
 
@@ -473,11 +494,13 @@ def _stream(
     else:
         body = _csv(columns, chunks)
         media = "text/csv; charset=utf-8"
-    return StreamingResponse(
-        body,
-        media_type=media,
-        headers={"Content-Disposition": f'attachment; filename="{name}.{fmt}"'},
-    )
+    headers = {"Content-Disposition": f'attachment; filename="{name}.{fmt}"'}
+    if truncated:
+        # The body is otherwise indistinguishable from a complete one, and an
+        # export quietly missing its oldest runs is the wrong thing to hand
+        # somebody who is about to plot it.
+        headers["X-Truncated"] = str(EXPORT_CAP)
+    return StreamingResponse(body, media_type=media, headers=headers)
 
 
 @router.get("/export/metrics")
@@ -490,7 +513,7 @@ def export_metrics(
     since: datetime | None = None,
     until: datetime | None = None,
     name: str = "",
-    stride: int = 1,
+    stride: Annotated[int, Query(ge=1, le=100_000)] = 1,
     format: ExportFormat = "csv",
 ) -> Any:
     """Every selected run's series as one long table: run, name, step, ts, value.
@@ -500,6 +523,8 @@ def export_metrics(
     for every tenth point of two metrics gives every tenth point of both.
     """
     runs = _selected(session, flow, status, group, ids, since, until)
+    truncated = len(runs) > EXPORT_CAP
+    runs = runs[:EXPORT_CAP]
     wanted = [part for part in name.split(",") if part]
     step = max(1, stride)
 
@@ -523,7 +548,7 @@ def export_metrics(
                 )
             yield rows
 
-    return _stream(format, "metrics", list(METRIC_COLUMNS), chunks())
+    return _stream(format, "metrics", list(METRIC_COLUMNS), chunks(), truncated)
 
 
 @router.get("/export/runs")
@@ -552,6 +577,8 @@ def export_runs(
     depth.
     """
     runs = _selected(session, flow, status, group, ids, since, until)
+    truncated = len(runs) > EXPORT_CAP
+    runs = runs[:EXPORT_CAP]
     inputs = [part for part in params.split(",") if part] or sorted(
         {path for run in runs for path, _ in _leaves(run.params)}
     )
@@ -569,7 +596,7 @@ def export_runs(
             row.update((f"metric.{k}", _cell(_dig(run.result, k))) for k in scores)
             yield [row]
 
-    return _stream(format, "runs", columns, chunks())
+    return _stream(format, "runs", columns, chunks(), truncated)
 
 
 @router.get("/{run_id}", response_model=RunDetail)
@@ -623,23 +650,27 @@ def delete_run(run_id: str, session: SessionDep, user: CurrentUser) -> Response:
     keeps whatever a ``run_artifact`` row or a live message still names, so
     dropping the rows is enough and the hourly sweep reclaims the blobs.
     """
-    run = session.get(Run, run_id)
-    if run is None:
-        raise HTTPException(status_code=404, detail="No such run")
-    if run.status in ("running", "queued"):
-        raise HTTPException(
-            status_code=409,
-            detail=(
-                f"Run {run_id} is {run.status}. Cancel it, or wait for it to "
-                "finish, before deleting it."
-            ),
-        )
-    flow = run.flow
-    session.execute(delete(RunNode).where(col(RunNode.run_id) == run_id))
-    session.execute(delete(RunMetric).where(col(RunMetric.run_id) == run_id))
-    session.execute(delete(RunArtifact).where(col(RunArtifact.run_id) == run_id))
-    session.execute(delete(Run).where(col(Run.id) == run_id))
-    session.commit()
+    # Reads the run, then decides whether to delete it — so it takes the write
+    # lock up front rather than upgrading and losing to whichever flush
+    # committed in between.
+    with writing(session):
+        run = session.get(Run, run_id)
+        if run is None:
+            raise HTTPException(status_code=404, detail="No such run")
+        if run.status in ("running", "queued"):
+            raise HTTPException(
+                status_code=409,
+                detail=(
+                    f"Run {run_id} is {run.status}. Cancel it, or wait for it to "
+                    "finish, before deleting it."
+                ),
+            )
+        flow = run.flow
+        session.execute(delete(RunNode).where(col(RunNode.run_id) == run_id))
+        session.execute(delete(RunMetric).where(col(RunMetric.run_id) == run_id))
+        session.execute(delete(RunArtifact).where(col(RunArtifact.run_id) == run_id))
+        session.execute(delete(Run).where(col(Run.id) == run_id))
+        session.commit()
     event_bus.publish(
         {
             "type": "audit",
@@ -674,32 +705,50 @@ def _series(session: Session, run_id: str, name: str = "") -> list[RunMetric]:
     if restored:
         run = session.get(Run, run_id)
         flow = run.flow if run is not None else ""
+        # Two queries for the lot, rather than a `Run` lookup and a `RunMetric`
+        # query per restored node. A comparison of twenty runs calls this
+        # twenty times, so the per-node round trips multiplied.
+        sources = {
+            row.id: row.flow
+            for row in session.exec(
+                select(Run).where(col(Run.id).in_({n.cached_from for n in restored}))
+            )
+        }
+        wanted: dict[tuple[str, str], list[RunNode]] = {}
         for node_row in restored:
-            source = session.get(Run, node_row.cached_from)
-            if source is None:
+            source_flow = sources.get(node_row.cached_from)
+            if source_flow is None:
                 # The run it came from is gone — deleted with its flow. The
                 # outputs are still on this run; the curve is not recoverable.
                 continue
-            source_node = requalify(node_row.node, flow, source.flow)
-            for row in session.exec(
+            key = (
+                node_row.cached_from,
+                requalify(node_row.node, flow, source_flow),
+            )
+            wanted.setdefault(key, []).append(node_row)
+        if wanted:
+            cached = session.exec(
                 select(RunMetric).where(
-                    col(RunMetric.run_id) == node_row.cached_from,
-                    col(RunMetric.node) == source_node,
+                    col(RunMetric.run_id).in_({run_id for run_id, _ in wanted}),
+                    col(RunMetric.node).in_({node for _, node in wanted}),
                 )
-            ):
-                renamed = requalify(row.name, source.flow, flow)
-                if name and renamed != name:
-                    continue
-                rows.append(
-                    RunMetric(
-                        run_id=run_id,
-                        name=renamed,
-                        step=row.step,
-                        node=node_row.node,
-                        ts=row.ts,
-                        value=row.value,
+            )
+            for row in cached:
+                for node_row in wanted.get((row.run_id, row.node), ()):
+                    source_flow = sources[node_row.cached_from]
+                    renamed = requalify(row.name, source_flow, flow)
+                    if name and renamed != name:
+                        continue
+                    rows.append(
+                        RunMetric(
+                            run_id=run_id,
+                            name=renamed,
+                            step=row.step,
+                            node=node_row.node,
+                            ts=row.ts,
+                            value=row.value,
+                        )
                     )
-                )
 
     rows.sort(key=lambda row: (row.name, row.step))
     return rows
@@ -707,7 +756,10 @@ def _series(session: Session, run_id: str, name: str = "") -> list[RunMetric]:
 
 @router.get("/{run_id}/metrics", response_model=list[MetricPoint])
 def read_metrics(
-    run_id: str, session: SessionDep, name: str = "", stride: int = 1
+    run_id: str,
+    session: SessionDep,
+    name: str = "",
+    stride: Annotated[int, Query(ge=1, le=100_000)] = 1,
 ) -> Any:
     """One metric's series, in step order — or every one of them, unnamed.
 
diff --git a/backend/fluksio/api/routes/secrets.py b/backend/fluksio/api/routes/secrets.py
index 7584d12..635398f 100644
--- a/backend/fluksio/api/routes/secrets.py
+++ b/backend/fluksio/api/routes/secrets.py
@@ -9,12 +9,18 @@ from fastapi import APIRouter, Depends, HTTPException
 from fastapi.concurrency import run_in_threadpool
 from pydantic import BaseModel
 
-from fluksio.api.deps import get_current_user
+from fluksio.api.deps import get_current_active_superuser
 from fluksio.flow.secrets import SecretNotFound, get_secrets
 from fluksio.models import Message
 
+# Superuser, not merely signed in. The names alone say what this installation
+# talks to, and `PUT /{name}` takes any name — so an ordinary account could
+# overwrite the credential a flow authenticates with. `/search` already gates
+# secret names this way and said so; this router was the half that did not.
 router = APIRouter(
-    prefix="/secrets", tags=["secrets"], dependencies=[Depends(get_current_user)]
+    prefix="/secrets",
+    tags=["secrets"],
+    dependencies=[Depends(get_current_active_superuser)],
 )
 
 
diff --git a/backend/fluksio/api/routes/users.py b/backend/fluksio/api/routes/users.py
index 4e5f743..103e6c5 100644
--- a/backend/fluksio/api/routes/users.py
+++ b/backend/fluksio/api/routes/users.py
@@ -1,7 +1,7 @@
 import uuid
-from typing import Any
+from typing import Annotated, Any
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlmodel import col, func, select
 
 from fluksio import crud
@@ -34,7 +34,11 @@ router = APIRouter(prefix="/users", tags=["users"])
     dependencies=[Depends(get_current_active_superuser)],
     response_model=UsersPublic,
 )
-def read_users(session: SessionDep, skip: int = 0, limit: int = 100) -> Any:
+def read_users(
+    session: SessionDep,
+    skip: Annotated[int, Query(ge=0)] = 0,
+    limit: Annotated[int, Query(ge=1, le=500)] = 100,
+) -> Any:
     """
     Retrieve users.
     """
diff --git a/backend/fluksio/cloud/connector.py b/backend/fluksio/cloud/connector.py
index d022284..3cbfdb7 100644
--- a/backend/fluksio/cloud/connector.py
+++ b/backend/fluksio/cloud/connector.py
@@ -50,6 +50,10 @@ MAX_WS_MESSAGE = 1024 * 1024
 #: Only the versioned API is served over the tunnel. The MCP mount and the
 #: OAuth endpoints live outside it and stay local-only.
 ALLOWED_PREFIX = "/api/v1/"
+#: Proxied calls, and bridged streams, this installation will run at once.
+#: Neither dict was bounded, and an id the hub reused silently dropped the
+#: reference to a task that was still running.
+MAX_IN_FLIGHT = 256
 #: How often to look for a config that appeared while the engine was up.
 #: Enrolment is something a person does and then waits on, so this is the
 #: delay they sit through — short enough not to be worth a restart.
@@ -380,6 +384,16 @@ class CloudConnector:
         call_id = str(frame.get("id") or "")
         if not call_id:
             return
+        if len(self._calls) >= MAX_IN_FLIGHT:
+            logger.warning(
+                "Refusing proxied call %s: %d already in flight",
+                call_id,
+                len(self._calls),
+            )
+            return
+        # An id the hub reuses would otherwise drop the reference to a task
+        # still running, leaving it with nothing able to cancel it.
+        self._cancel(call_id)
         task = asyncio.create_task(self._serve(socket, frame))
         self._calls[call_id] = task
         task.add_done_callback(lambda _t: self._calls.pop(call_id, None))
@@ -465,6 +479,14 @@ class CloudConnector:
         stream_id = str(frame.get("id") or "")
         if not stream_id:
             return
+        if len(self._streams) >= MAX_IN_FLIGHT:
+            logger.warning(
+                "Refusing stream %s: %d already open", stream_id, len(self._streams)
+            )
+            return
+        existing = self._streams.pop(stream_id, None)
+        if existing is not None:
+            existing.cancel()
         task = asyncio.create_task(self._stream(socket, frame))
         self._streams[stream_id] = task
         task.add_done_callback(lambda _t: self._streams.pop(stream_id, None))
diff --git a/backend/fluksio/cloud/enroll.py b/backend/fluksio/cloud/enroll.py
index 64faa8a..4a06d90 100644
--- a/backend/fluksio/cloud/enroll.py
+++ b/backend/fluksio/cloud/enroll.py
@@ -11,6 +11,7 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 from typing import Any
+from urllib.parse import urlsplit
 
 import httpx
 from sqlmodel import Session, select
@@ -34,11 +35,25 @@ class AlreadyEnrolled(EnrollError):
         super().__init__(409, "This installation is already connected to a portal")
 
 
+def _is_local(url: str) -> bool:
+    """Whether the address is this machine or a compose-internal service."""
+    host = urlsplit(url).hostname or ""
+    return host in {"localhost", "127.0.0.1", "::1"} or host.endswith(".local")
+
+
 def redeem_claim(
     portal_url: str, claim_code: str, *, timeout: float = 15.0
 ) -> dict[str, Any]:
     """Trade a claim code for this installation's credential and the portal's keys."""
     base = portal_url.rstrip("/")
+    # The claim code and, from here on, this installation's credential go to
+    # this address. Over plain http both are readable by anything on the path,
+    # so refuse rather than enrol insecurely — bar a loopback portal, which is
+    # how the stack is developed against itself.
+    if not base.startswith("https://") and not _is_local(base):
+        raise EnrollError(
+            422, "The portal address must be https:// (or a local address)"
+        )
     try:
         response = httpx.post(
             f"{base}/api/v1/enroll/",
diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py
index 91497fc..22b7c83 100644
--- a/backend/fluksio/core/config.py
+++ b/backend/fluksio/core/config.py
@@ -57,8 +57,12 @@ class Settings(BaseSettings):
     #: secrets, artifacts and the user venv. The paths below derive from it
     #: unless they are set explicitly.
     DATA_DIR: Path = Path("flow-data")
-    #: Any SQLAlchemy URL. The default puts SQLite in the data directory, which
-    #: is what makes `fluksio serve` need no infrastructure at all.
+    #: Where the database file goes, as a SQLite URL. The default puts it in
+    #: the data directory, which is what makes `fluksio serve` need no
+    #: infrastructure at all. **SQLite only**: the upserts the metrics
+    #: collector and the run recorder are built on are written against that
+    #: dialect, so another backend would parse, migrate, serve — and then
+    #: silently drop every rollup and fail every run.
     DATABASE_URL: str | None = None
 
     # Flows live on disk as a git repository; secrets stay outside it.
@@ -126,6 +130,12 @@ class Settings(BaseSettings):
     # node at a time.
     FLOW_GPUS: int = 0
     # How long the engine's own metrics, events and run records are kept.
+    #: The largest body `PUT /artifacts` will take, in bytes. A checkpoint or a
+    #: video segment is a legitimate artifact, so this is generous rather than
+    #: small; 0 removes the limit. Nothing capped it at all before, so any
+    #: account or worker credential could fill the data volume.
+    MAX_ARTIFACT_BYTES: int = 2 * 1024 * 1024 * 1024
+
     OBS_RETENTION_DAYS: int = 30
     # How often artifact bytes nothing refers to any more are swept away; 0
     # never sweeps. A flow streaming media writes one artifact per frame, so
@@ -190,7 +200,7 @@ class Settings(BaseSettings):
     @computed_field  # type: ignore[prop-decorator]
     @property
     def SQLALCHEMY_DATABASE_URI(self) -> str:
-        """SQLite in the data directory, unless a URL says otherwise.
+        """SQLite in the data directory, unless a URL names another file.
 
         One engine process owns this database — the same reason the image runs
         a single uvicorn worker — so a file beside the flows is the honest
diff --git a/backend/fluksio/core/db.py b/backend/fluksio/core/db.py
index 0caa694..c24ea76 100644
--- a/backend/fluksio/core/db.py
+++ b/backend/fluksio/core/db.py
@@ -10,12 +10,14 @@ carry on while that single writer works.
 from __future__ import annotations
 
 import logging
+from collections.abc import Iterator
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
 
 from alembic import command
 from alembic.config import Config
-from sqlalchemy import Engine, event, inspect, make_url
+from sqlalchemy import Engine, event, inspect, make_url, text
 from sqlmodel import Session, SQLModel, create_engine, select
 
 from fluksio import crud
@@ -26,12 +28,21 @@ logger = logging.getLogger(__name__)
 
 
 def make_engine(url: str) -> Engine:
-    """One engine for the process, configured for whichever dialect it names."""
+    """One engine for the process. SQLite, and only SQLite.
+
+    Not a limitation that snuck in: `metric_minute` and every run table are
+    written with `sqlalchemy.dialects.sqlite.insert(...).on_conflict_do_update`
+    and with `max(a, b)`, neither of which another dialect has. Pointing this
+    at Postgres used to migrate cleanly, serve, log in — and then lose every
+    observability flush into the collector's hold buffer and fail every run,
+    which is a far worse answer than refusing at startup.
+    """
     if make_url(url).get_backend_name() != "sqlite":
-        # A connection idle across a server restart is dead but still pooled;
-        # the pre-ping spends a round trip to find out instead of failing the
-        # request.
-        return create_engine(url, pool_pre_ping=True)
+        raise RuntimeError(
+            f"DATABASE_URL must be a SQLite URL, not {make_url(url).get_backend_name()!r}. "
+            "The engine's own database is a file beside the flows; the "
+            "analytics Postgres in the compose stack is Umami's, not this."
+        )
 
     database = make_url(url).database
     if database and database != ":memory:":
@@ -48,8 +59,6 @@ engine = make_engine(settings.SQLALCHEMY_DATABASE_URI)
 @event.listens_for(engine, "connect")
 def _sqlite_pragmas(dbapi_connection: Any, _record: Any) -> None:
     """What makes concurrent readers, cascades and waiting-out a writer work."""
-    if engine.dialect.name != "sqlite":
-        return
     cursor = dbapi_connection.cursor()
     # Readers do not block on the writer, which is the whole reason this is
     # usable while a run is streaming metrics into it.
@@ -65,6 +74,29 @@ def _sqlite_pragmas(dbapi_connection: Any, _record: Any) -> None:
     cursor.close()
 
 
+@contextmanager
+def writing(session: Session) -> Iterator[None]:
+    """Take SQLite's write lock before reading, for a read-then-write handler.
+
+    SQLAlchemy begins deferred, so a transaction that reads and then writes
+    has to upgrade — and if another writer committed in between, SQLite
+    refuses with `SQLITE_BUSY_SNAPSHOT` immediately rather than waiting out
+    `busy_timeout`. The engine writes constantly (the metrics flush, a run
+    recording nodes), so any handler that decides *whether* to write by
+    reading first can lose that race and 500.
+
+    Only worth it where the handler really does both: this serialises against
+    other writers for the length of the transaction, which is what WAL's
+    concurrent readers otherwise avoid.
+    """
+    session.execute(text("BEGIN IMMEDIATE"))
+    try:
+        yield
+    except Exception:
+        session.rollback()
+        raise
+
+
 def _alembic_config(connection: Any) -> Config:
     """Alembic driven from here, so an installed wheel needs no alembic.ini."""
     config = Config()
diff --git a/backend/fluksio/flow/modules.py b/backend/fluksio/flow/modules.py
index b14aea4..c3af6b2 100644
--- a/backend/fluksio/flow/modules.py
+++ b/backend/fluksio/flow/modules.py
@@ -29,6 +29,7 @@ import shutil
 import subprocess
 import sys
 import tempfile
+import threading
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -143,6 +144,12 @@ def ensure_venv() -> None:
     )
 
 
+#: Held for the length of a `uv pip sync`. Two applies at once — two browser
+#: tabs, or an apply landing while the boot-time reconcile is still running —
+#: mutate the same venv simultaneously.
+_sync_lock = threading.Lock()
+
+
 def sync(requirements: str) -> tuple[bool, str]:
     """Make the venv hold exactly these packages. Returns success and uv's output.
 
@@ -160,6 +167,19 @@ def sync(requirements: str) -> tuple[bool, str]:
             "into. Install into it with pip or uv, or set NODE_VENV=managed "
             "for a venv the engine owns."
         )
+    if not _sync_lock.acquire(timeout=0.0):
+        return False, (
+            "Another apply is already installing packages. Wait for it to "
+            "finish, then try again."
+        )
+    try:
+        return _sync(requirements)
+    finally:
+        _sync_lock.release()
+
+
+def _sync(requirements: str) -> tuple[bool, str]:
+    """The install itself, under `_sync_lock`."""
     manifest = ""
     try:
         ensure_venv()
diff --git a/backend/fluksio/flow/panels.py b/backend/fluksio/flow/panels.py
index 482f0f8..fdf93c1 100644
--- a/backend/fluksio/flow/panels.py
+++ b/backend/fluksio/flow/panels.py
@@ -12,6 +12,7 @@ which screen hangs where is the deployment's concern, not any one dashboard's.
 
 from __future__ import annotations
 
+import threading
 from pathlib import Path
 
 from pydantic import BaseModel, Field, field_validator
@@ -49,6 +50,17 @@ class PanelsConfig(BaseModel):
     panels: list[PanelDef] = Field(default_factory=list)
 
 
+#: Held across a read-modify-write of the panels file.
+#:
+#: Both `save_panels` and `unpair_panel` are one of those, and interleaving
+#: them silently undid a revocation: a save that read the file before an
+#: unpair wrote it put the old nonce back, and the screen that had just been
+#: unpaired kept working. The nonce carry-forward in the save handler was
+#: written to make that impossible, and the window between its read and its
+#: write is where it happened anyway.
+edit_lock = threading.Lock()
+
+
 def _path() -> Path:
     return settings.PANELS_FILE
 
diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py
index 417d9e5..03d5ec6 100644
--- a/backend/fluksio/flow/queue.py
+++ b/backend/fluksio/flow/queue.py
@@ -25,6 +25,8 @@ from typing import Any, cast
 import orjson
 import redis
 
+from fluksio.flow.state import CONNECT_TIMEOUT_S, SOCKET_TIMEOUT_S
+
 logger = logging.getLogger(__name__)
 
 # The stream is a buffer, not an archive: anything this far behind is long
@@ -407,7 +409,17 @@ class RedisWorkQueue(WorkQueue):
         namespace: str = "queue",
         consumer: str | None = None,
     ) -> None:
-        self._redis = redis.Redis(host=host, port=port, decode_responses=True)
+        self._redis = redis.Redis(
+            host=host,
+            port=port,
+            decode_responses=True,
+            # Longer than any block this asks for, so a claim still waits its
+            # second — but bounded, so a Redis that goes silent without
+            # closing the socket fails rather than hanging the consumer.
+            socket_timeout=SOCKET_TIMEOUT_S,
+            socket_connect_timeout=CONNECT_TIMEOUT_S,
+            health_check_interval=30,
+        )
         self._ns = namespace
         self._consumer = consumer or f"engine-{int(time.time() * 1000) % 1_000_000}"
         self._stream = f"{namespace}:__queue__"
diff --git a/backend/fluksio/flow/state.py b/backend/fluksio/flow/state.py
index 67106fe..23ed33a 100644
--- a/backend/fluksio/flow/state.py
+++ b/backend/fluksio/flow/state.py
@@ -27,6 +27,12 @@ _JSON_OPTS = orjson.OPT_NON_STR_KEYS
 # A sparkline only means something for a value that can be placed on an axis,
 # so the history keeps those and nothing else. 120 points fill a panel-wide
 # chart while leaving Redis a cache rather than a time-series database.
+#: How long a Redis call may take before it is treated as a failure, and
+#: how long connecting may take. A blocking read must ask for less than the
+#: first of these.
+SOCKET_TIMEOUT_S = 5.0
+CONNECT_TIMEOUT_S = 2.0
+
 HISTORY_LIMIT = 120
 
 
@@ -496,6 +502,13 @@ class RedisState(StateBackend):
             db=db,
             password=password,
             decode_responses=False,
+            # Without these a Redis that stops answering without closing the
+            # connection — a partition, a firewall dropping state — hangs the
+            # caller until the kernel gives up, minutes later. That takes the
+            # health endpoint whose job is to notice with it.
+            socket_timeout=SOCKET_TIMEOUT_S,
+            socket_connect_timeout=CONNECT_TIMEOUT_S,
+            health_check_interval=30,
         )
         self._namespace = namespace
         self._ttl = ttl
diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py
index dbd4819..781aafc 100644
--- a/backend/fluksio/main.py
+++ b/backend/fluksio/main.py
@@ -1,7 +1,8 @@
 import asyncio
 import contextlib
+import inspect
 import logging
-from collections.abc import AsyncIterator, Coroutine
+from collections.abc import AsyncIterator, Callable, Coroutine
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from typing import Any
 
@@ -129,109 +130,32 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     # pays a version check for it and nothing else.
     await run_in_threadpool(prepare, db_engine)
     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)
-    # Connectors register their node types before any flow is built with them.
-    load_plugins()
 
-    alerts = AlertManager(event_bus, config=read_alerts_config())
-    execution = ExecutionService(
-        queue=_work_queue(),
-        max_workers=settings.FLOW_MAX_WORKERS,
-        events=event_bus,
-        max_cascades=settings.FLOW_MAX_CASCADES,
-    )
-    store = FlowStore(settings.FLOWS_DIR)
-    # The packages node code imports, before anything tries to import them.
-    await run_in_threadpool(modules.reconcile, store)
-    # Beside the flows rather than in them: an artifact is what a run produced,
-    # not something anyone wrote, so it has no business in the git repository.
-    artifacts = ArtifactStore(settings.FLOWS_DIR.parent / "artifacts")
-    app.state.artifact_store = artifacts
-    accountant = ResourceAccountant(cpus=settings.FLOW_CPUS, gpus=settings.FLOW_GPUS)
-    app.state.resources = accountant
-    # Every machine a node could run on: this one, and whatever attaches.
-    placer = Placer(local=accountant, events=event_bus)
-    # Where more machines can be asked for when nothing attached will do.
-    placer.provisioners = load_provisioners(settings.PROVISIONERS_FILE, event_bus)
-    app.state.placer = placer
-    pool = PythonWorkerPool(
-        python=modules.venv_python(),
-        size=settings.FLOW_MAX_WORKERS,
-        events=event_bus,
-        # A worker in this container writes to the store directly; a remote one
-        # is given a URL instead. Node code calls the same two functions.
-        env={
-            ARTIFACT_DIR_ENV: str(artifacts.root),
-            # Every slot can be busy at once, so a worker left to size its own
-            # thread pool to the machine means as many processes as there are
-            # slots, each believing it has the whole of it. A node that says
-            # what it needs overrides this; one that says nothing gets a share.
-            **fair_share_env(accountant.cpus, settings.FLOW_MAX_WORKERS),
-        },
-    )
-    pool.start()
-    app.state.worker_pool = pool
-    # Assigned rather than passed both ways: the hub tells the placer when a
-    # machine comes or goes, and the placer needs the hub to know what is there.
-    worker_hub = RemoteWorkerHub(on_change=placer.wake)
-    placer.hub = worker_hub
-    app.state.worker_hub = worker_hub
-    controller = FlowController(
-        store=store,
-        state=_state_backend(),
-        events=event_bus,
-        max_workers=settings.FLOW_MAX_WORKERS,
-        fastapi_app=app,
-        execution=execution,
-        alerts=alerts,
-        workers=pool,
-        remote=worker_hub,
-        resources=accountant,
-        placer=placer,
-    )
-    app.state.flow_controller = controller
-    # A "dashboard" alert channel puts its alert into the graph. Bound here
-    # rather than passed in: the manager is built before the controller is.
-    alerts.publish = lambda name, value: controller.publish_message(
-        name, value, ValueSource(kind="api", id="alerts", label="Alerts")
-    )
-    dashboards = DashboardStore(controller.store)
-    app.state.dashboard_store = dashboards
-    controller.dashboards = dashboards
-    # 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,
-        artifacts=artifacts,
-        parallel=settings.FLOW_MAX_RUNS,
-    )
-    app.state.run_service = run_service
-    # Said out loud because it is the only way to tell that a settings file was
-    # read at all — the numbers are what someone raising them is looking for.
-    logger.info(
-        "Engine limits: workers=%s cascades=%s runs=%s",
-        settings.FLOW_MAX_WORKERS,
-        execution.max_cascades,
-        run_service.parallel,
-    )
-    watchdog = LoopWatchdog(event_bus)
-    app.state.watchdog = watchdog
+    # Everything acquired below registers how to release it, and the `finally`
+    # at the end walks the list backwards. It used to be a fixed block of
+    # shutdown steps *after* every acquisition — so a failure part-way through
+    # startup reached none of them and left the worker pool's subprocesses and
+    # every background task behind. Under `--reload` that is once per bad edit.
+    started: list[Callable[[], Any]] = []
 
-    def _background(coro: Coroutine[Any, Any, None], name: str) -> asyncio.Task[None]:
-        """Start a long-lived task that says something if it ever stops.
+    async def _release() -> None:
+        for close in reversed(started):
+            try:
+                result = close()
+                if inspect.isawaitable(result):
+                    await result
+            except Exception:
+                logger.exception("Error while shutting down")
 
-        Each of these loops catches its own exceptions *inside* the loop, so
-        one raised anywhere else simply ended the task — an engine that went
-        on serving with no metrics, no alerts or no artifact sweep and nothing
-        anywhere saying so.
+    def _background(
+        coro: Coroutine[Any, Any, None], name: str, once: bool = False
+    ) -> asyncio.Task[None]:
+        """Start a background task that says something if it ever stops.
+
+        The loops catch their own exceptions *inside* the loop, so one raised
+        anywhere else simply ended the task — an engine that went on serving
+        with no metrics, no alerts or no artifact sweep and nothing anywhere
+        saying so. ``once`` is for a task that is meant to finish.
         """
         task = asyncio.create_task(coro, name=name)
 
@@ -240,7 +164,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
                 return
             exc = done.exception()
             if exc is None:
-                logger.warning("Background task '%s' stopped on its own", name)
+                if not once:
+                    logger.warning("Background task '%s' stopped on its own", name)
                 return
             logger.error("Background task '%s' died: %s", name, exc, exc_info=exc)
             event_bus.publish(
@@ -248,56 +173,173 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
             )
 
         task.add_done_callback(_finished)
+        started.append(task.cancel)
         return task
 
-    watchdog_task = _background(watchdog.run(), "loop-watchdog")
-    alerts_task = _background(alerts.run(), "alert-manager")
-    metrics_task = _background(MetricsCollector(event_bus).run(), "metrics-collector")
-    gc_task = _background(_sweep_artifacts(artifacts, controller), "artifact-gc")
-    await controller.start()
-    run_service.start()
-    # Optional, and off unless someone enrolled this installation: the
-    # connector dials the portal, nothing dials in.
-    cloud_task: asyncio.Task[None] | None = None
-    app.state.cloud_connector = None
-    app.state.cloud_task = None
-    from fluksio.cloud import connector as cloud_connector
-
-    if cloud_config.exists():
-        cloud_connector.start(app)
-        cloud_task = app.state.cloud_task
-    # Watched whether or not one exists now: enrolling from the CLI writes the
-    # config from another process entirely, and an engine already serving
-    # should pick it up rather than need restarting.
-    enrol_task = _background(
-        cloud_connector.watch_enrolment(app), "cloud-enrolment-watch"
-    )
+    # The startup itself is inside the try, so a step that raises part-way
+    # through still reaches the release below.
     try:
+        # Node code is user code, and `print` is how it says things.
+        logs.install()
+        init_secrets(settings.SECRETS_FILE, settings.SECRET_KEY)
+        # Connectors register their node types before any flow is built with them.
+        load_plugins()
+
+        alerts = AlertManager(event_bus, config=read_alerts_config())
+        execution = ExecutionService(
+            queue=_work_queue(),
+            max_workers=settings.FLOW_MAX_WORKERS,
+            events=event_bus,
+            max_cascades=settings.FLOW_MAX_CASCADES,
+        )
+        store = FlowStore(settings.FLOWS_DIR)
+        # The packages node code imports. In the background, because a first
+        # install of anything substantial takes minutes and `uv` is given five
+        # of them twice over — held here, the container never answers its
+        # healthcheck, autoheal restarts it, and it never finishes installing at
+        # all. A node whose import is not there yet fails and is retried, which
+        # is recoverable; a restart loop is not.
+        _background(
+            run_in_threadpool(modules.reconcile, store), "module-reconcile", once=True
+        )
+        # Beside the flows rather than in them: an artifact is what a run produced,
+        # not something anyone wrote, so it has no business in the git repository.
+        artifacts = ArtifactStore(settings.FLOWS_DIR.parent / "artifacts")
+        app.state.artifact_store = artifacts
+        accountant = ResourceAccountant(
+            cpus=settings.FLOW_CPUS, gpus=settings.FLOW_GPUS
+        )
+        app.state.resources = accountant
+        # Every machine a node could run on: this one, and whatever attaches.
+        placer = Placer(local=accountant, events=event_bus)
+        # Where more machines can be asked for when nothing attached will do.
+        placer.provisioners = load_provisioners(settings.PROVISIONERS_FILE, event_bus)
+        app.state.placer = placer
+        pool = PythonWorkerPool(
+            python=modules.venv_python(),
+            size=settings.FLOW_MAX_WORKERS,
+            events=event_bus,
+            # A worker in this container writes to the store directly; a remote one
+            # is given a URL instead. Node code calls the same two functions.
+            env={
+                ARTIFACT_DIR_ENV: str(artifacts.root),
+                # Every slot can be busy at once, so a worker left to size its own
+                # thread pool to the machine means as many processes as there are
+                # slots, each believing it has the whole of it. A node that says
+                # what it needs overrides this; one that says nothing gets a share.
+                **fair_share_env(accountant.cpus, settings.FLOW_MAX_WORKERS),
+            },
+        )
+        pool.start()
+        # On a thread: stopping a worker waits up to five seconds on each child,
+        # and with a pool per declared environment that is a shutdown the event
+        # loop should not be holding.
+        started.append(lambda: run_in_threadpool(pool.stop))
+        app.state.worker_pool = pool
+        # Assigned rather than passed both ways: the hub tells the placer when a
+        # machine comes or goes, and the placer needs the hub to know what is there.
+        worker_hub = RemoteWorkerHub(on_change=placer.wake)
+        placer.hub = worker_hub
+        app.state.worker_hub = worker_hub
+        controller = FlowController(
+            store=store,
+            state=_state_backend(),
+            events=event_bus,
+            max_workers=settings.FLOW_MAX_WORKERS,
+            fastapi_app=app,
+            execution=execution,
+            alerts=alerts,
+            workers=pool,
+            remote=worker_hub,
+            resources=accountant,
+            placer=placer,
+        )
+        app.state.flow_controller = controller
+        # A "dashboard" alert channel puts its alert into the graph. Bound here
+        # rather than passed in: the manager is built before the controller is.
+        alerts.publish = lambda name, value: controller.publish_message(
+            name, value, ValueSource(kind="api", id="alerts", label="Alerts")
+        )
+        dashboards = DashboardStore(controller.store)
+        app.state.dashboard_store = dashboards
+        controller.dashboards = dashboards
+        # 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,
+            artifacts=artifacts,
+            parallel=settings.FLOW_MAX_RUNS,
+        )
+        app.state.run_service = run_service
+        # Said out loud because it is the only way to tell that a settings file was
+        # read at all — the numbers are what someone raising them is looking for.
+        logger.info(
+            "Engine limits: workers=%s cascades=%s runs=%s",
+            settings.FLOW_MAX_WORKERS,
+            execution.max_cascades,
+            run_service.parallel,
+        )
+        watchdog = LoopWatchdog(event_bus)
+        app.state.watchdog = watchdog
+
+        _background(watchdog.run(), "loop-watchdog")
+        _background(alerts.run(), "alert-manager")
+        _background(MetricsCollector(event_bus).run(), "metrics-collector")
+        _background(_sweep_artifacts(artifacts, controller), "artifact-gc")
+        await controller.start()
+        started.append(controller.stop)
+        run_service.start()
+        started.append(lambda: run_in_threadpool(run_service.stop))
+        # A machine asked for and not yet arrived would hold an allocation nobody
+        # is going to use.
+        started.append(
+            lambda: asyncio.gather(
+                *(
+                    run_in_threadpool(provisioner.shutdown)
+                    for provisioner in placer.provisioners
+                )
+            )
+        )
+        started.append(lambda: run_in_threadpool(close_shared_client))
+        # Optional, and off unless someone enrolled this installation: the
+        # connector dials the portal, nothing dials in.
+        cloud_task: asyncio.Task[None] | None = None
+        app.state.cloud_connector = None
+        app.state.cloud_task = None
+        from fluksio.cloud import connector as cloud_connector
+
+        if cloud_config.exists():
+            cloud_connector.start(app)
+            cloud_task = app.state.cloud_task
+
+        def _stop_cloud() -> None:
+            # Re-read from app.state: enrolling at runtime replaces this.
+            running = getattr(app.state, "cloud_task", None) or cloud_task
+            if running is not None:
+                running.cancel()
+
+        started.append(_stop_cloud)
+        # Watched whether or not one exists now: enrolling from the CLI writes the
+        # config from another process entirely, and an engine already serving
+        # should pick it up rather than need restarting.
+        _background(cloud_connector.watch_enrolment(app), "cloud-enrolment-watch")
         # A mounted sub-app gets no lifespan of its own, so the MCP session
         # manager is entered here; without it every /mcp request fails.
         async with _mcp_sessions():
             yield
+    except Exception:
+        # A startup that fails part-way lands here too, because everything
+        # since `event_bus.bind` registered its own release.
+        logger.exception("The engine could not start")
+        raise
     finally:
-        watchdog_task.cancel()
-        alerts_task.cancel()
-        metrics_task.cancel()
-        gc_task.cancel()
-        enrol_task.cancel()
-        # Re-read from app.state: enrolling at runtime replaces this.
-        running_cloud = getattr(app.state, "cloud_task", None) or cloud_task
-        if running_cloud is not None:
-            running_cloud.cancel()
-        await run_in_threadpool(run_service.stop)
-        await controller.stop()
-        # On a thread, like the two above: stopping a worker waits up to five
-        # seconds on each child, and with a pool per declared environment that
-        # is a shutdown the event loop should not be holding.
-        await run_in_threadpool(pool.stop)
-        # A machine asked for and not yet arrived would hold an allocation
-        # nobody is going to use.
-        for provisioner in placer.provisioners:
-            await run_in_threadpool(provisioner.shutdown)
-        await run_in_threadpool(close_shared_client)
+        await _release()
         if settings.MCP_ENABLED:
             from fluksio.mcp.http import aclose
 
diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py
index 74c8de6..bf89a6e 100644
--- a/backend/fluksio/models.py
+++ b/backend/fluksio/models.py
@@ -3,7 +3,7 @@ from datetime import UTC, datetime
 from typing import Any
 
 from pydantic import EmailStr
-from sqlalchemy import JSON, Column
+from sqlalchemy import JSON, Column, Index
 from sqlmodel import Field, SQLModel
 
 from fluksio.core.types import UTCDateTime
@@ -259,6 +259,9 @@ class EngineEvent(SQLModel, table=True):
     """Something worth keeping after the websocket has forgotten it."""
 
     __tablename__ = "engine_event"
+    # The events screen excludes audit rows and reads newest first. A `type`
+    # index alone serves neither half of that.
+    __table_args__ = (Index("ix_engine_event_type_ts", "type", "ts"),)
 
     id: int | None = Field(default=None, primary_key=True)
     ts: datetime = Field(
@@ -286,6 +289,8 @@ class FlowRun(SQLModel, table=True):
     """
 
     __tablename__ = "flow_run"
+    # The history screen filters by flow, newest first.
+    __table_args__ = (Index("ix_flow_run_flow_started", "flow", "started_at"),)
 
     #: The queue entry id, a `manual-` one for a run that never queued, or the
     #: `Run.id` of a batch run.
@@ -327,6 +332,11 @@ class Run(SQLModel, table=True):
     """One finite execution of a flow, with what it was asked and what it made."""
 
     __tablename__ = "run"
+    # The runs list filters by flow and/or status, newest first.
+    __table_args__ = (
+        Index("ix_run_flow_created", "flow", "created_at"),
+        Index("ix_run_status_created", "status", "created_at"),
+    )
 
     id: str = Field(primary_key=True, max_length=64)
     flow: str = Field(index=True, max_length=255)
diff --git a/backend/fluksio/utils.py b/backend/fluksio/utils.py
index 702a744..694cf97 100644
--- a/backend/fluksio/utils.py
+++ b/backend/fluksio/utils.py
@@ -1,3 +1,4 @@
+import hashlib
 import logging
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
@@ -14,6 +15,9 @@ from fluksio.core.config import settings
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
+#: How long to wait on the mail server before giving up on one message.
+SMTP_TIMEOUT_S = 10
+
 
 @dataclass
 class EmailData:
@@ -47,7 +51,14 @@ def send_email(
         html=html_content,
         mail_from=(settings.EMAILS_FROM_NAME, settings.EMAILS_FROM_EMAIL),
     )
-    smtp_options = {"host": settings.SMTP_HOST, "port": settings.SMTP_PORT}
+    # With no timeout smtplib inherits the socket default, which is none at
+    # all — an unreachable mail host then held a threadpool worker until the
+    # kernel gave up, from a route anybody can call.
+    smtp_options: dict[str, Any] = {
+        "host": settings.SMTP_HOST,
+        "port": settings.SMTP_PORT,
+        "timeout": SMTP_TIMEOUT_S,
+    }
     if settings.SMTP_TLS:
         smtp_options["tls"] = True
     elif settings.SMTP_SSL:
@@ -105,24 +116,40 @@ def generate_new_account_email(
     return EmailData(html_content=html_content, subject=subject)
 
 
-def generate_password_reset_token(email: str) -> str:
+def password_fingerprint(hashed_password: str) -> str:
+    """A short digest of the stored hash, for binding a reset link to it.
+
+    What makes a reset token single-use without a table of spent ones: the
+    hash changes the moment the password does, so the link that set it stops
+    verifying. It was replayable for the whole 48 hours otherwise.
+    """
+    return hashlib.sha256(hashed_password.encode()).hexdigest()[:16]
+
+
+def generate_password_reset_token(email: str, hashed_password: str = "") -> str:
     delta = timedelta(hours=settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS)
     now = datetime.now(UTC)
     expires = now + delta
     exp = expires.timestamp()
     encoded_jwt = jwt.encode(
-        {"exp": exp, "nbf": now, "sub": email},
+        {
+            "exp": exp,
+            "nbf": now,
+            "sub": email,
+            "pwd": password_fingerprint(hashed_password),
+        },
         settings.SECRET_KEY,
         algorithm=security.ALGORITHM,
     )
     return encoded_jwt
 
 
-def verify_password_reset_token(token: str) -> str | None:
+def verify_password_reset_token(token: str) -> tuple[str, str] | None:
+    """The email the token names and the password hash it was minted against."""
     try:
         decoded_token = jwt.decode(
             token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
         )
-        return str(decoded_token["sub"])
+        return str(decoded_token["sub"]), str(decoded_token.get("pwd", ""))
     except InvalidTokenError:
         return None
diff --git a/backend/tests/api/routes/test_login.py b/backend/tests/api/routes/test_login.py
index cc6290c..3014782 100644
--- a/backend/tests/api/routes/test_login.py
+++ b/backend/tests/api/routes/test_login.py
@@ -92,7 +92,11 @@ def test_reset_password(client: TestClient, db: Session) -> None:
         is_superuser=False,
     )
     user = create_user(session=db, user_create=user_create)
-    token = generate_password_reset_token(email=email)
+    # Bound to the hash it was minted against, which is what makes the link
+    # stop working once it has set a password.
+    token = generate_password_reset_token(
+        email=email, hashed_password=user.hashed_password
+    )
     headers = user_authentication_headers(client=client, email=email, password=password)
     data = {"new_password": new_password, "token": token}
 
@@ -189,3 +193,30 @@ def test_login_with_argon2_password_keeps_hash(client: TestClient, db: Session)
 
     assert user.hashed_password == original_hash
     assert user.hashed_password.startswith("$argon2")
+
+
+def test_a_reset_link_works_once(client: TestClient, db: Session) -> None:
+    """It used to work over and over for the whole of its 48 hours."""
+    email = random_email()
+    password = random_lower_string()
+    user = create_user(
+        session=db,
+        user_create=UserCreate(email=email, password=password, is_active=True),
+    )
+    token = generate_password_reset_token(
+        email=email, hashed_password=user.hashed_password
+    )
+
+    first = client.post(
+        f"{settings.API_V1_STR}/reset-password/",
+        json={"new_password": random_lower_string(), "token": token},
+    )
+    assert first.status_code == 200
+
+    # The same link again: the password it was minted against has moved.
+    again = client.post(
+        f"{settings.API_V1_STR}/reset-password/",
+        json={"new_password": random_lower_string(), "token": token},
+    )
+    assert again.status_code == 400
+    assert again.json()["detail"] == "Invalid token"
diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py
index 088b7c6..c2834f3 100644
--- a/backend/tests/api/routes/test_observability.py
+++ b/backend/tests/api/routes/test_observability.py
@@ -411,3 +411,39 @@ def test_events_narrow_to_one_run(
     ).json()
 
     assert [event["detail"] for event in events] == ["mine"]
+
+
+def test_a_timeseries_window_can_be_named(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    """`since`/`until` name a span, where `hours` only measures back from now."""
+    base = datetime.now(UTC).replace(second=0, microsecond=0) - timedelta(hours=6)
+    for minute in range(4):
+        db.add(
+            MetricBucket(
+                flow="windowed",
+                node="windowed.n",
+                bucket=base + timedelta(minutes=minute),
+                executions=1,
+                errors=0,
+                messages=1,
+                duration_sum_ms=1.0,
+                duration_max_ms=1.0,
+                lag_sum_ms=0.0,
+                items=1,
+            )
+        )
+    db.commit()
+
+    response = client.get(
+        f"{PREFIX}/timeseries",
+        headers=superuser_token_headers,
+        params={
+            "flow": "windowed",
+            "since": (base + timedelta(minutes=1)).isoformat(),
+            "until": (base + timedelta(minutes=3)).isoformat(),
+        },
+    )
+    assert response.status_code == 200
+    # Two of the four minutes: `since` inclusive, `until` exclusive.
+    assert sum(point["executions"] for point in response.json()) == 2
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index 3dcfce5..3c24418 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -7,6 +7,7 @@ from typing import Any
 import pytest
 from cryptography.hazmat.primitives.asymmetric import rsa
 from fastapi.testclient import TestClient
+from sqlalchemy import delete
 from sqlalchemy.engine import make_url
 from sqlmodel import Session, select
 
@@ -14,7 +15,16 @@ from fluksio.cloud import config as cloud_config
 from fluksio.core.config import settings
 from fluksio.core.db import engine, prepare
 from fluksio.main import app
-from fluksio.models import User
+from fluksio.models import (
+    EngineEvent,
+    FlowRun,
+    MetricBucket,
+    Run,
+    RunArtifact,
+    RunMetric,
+    RunNode,
+    User,
+)
 from tests.utils.portal import INSTALLATION_ID, ISSUER, jwks
 from tests.utils.user import authentication_token_from_email
 from tests.utils.utils import get_superuser_token_headers
@@ -54,6 +64,37 @@ def db() -> Generator[Session, None, None]:
         path.with_name(path.name + suffix).unlink(missing_ok=True)
 
 
+#: Tables the engine writes to on its own, from the collector and the run
+#: service the module-scoped app starts.
+_ENGINE_TABLES = (
+    RunArtifact,
+    RunMetric,
+    RunNode,
+    Run,
+    FlowRun,
+    EngineEvent,
+    MetricBucket,
+)
+
+
+@pytest.fixture(scope="module", autouse=True)
+def _clean_engine_rows() -> Generator[None, None, None]:
+    """Clear what a module wrote before the next one queries it.
+
+    The `db` session is session-scoped and every commit in it is permanent, so
+    a `FlowRun` one module left marked `running` turned up in another module's
+    query. Per-*test* rollback is not available here: the module-scoped
+    `client` runs the real lifespan, and the collector and run service write
+    through sessions of their own that no test transaction wraps. Clearing
+    between modules bounds it to where the writes actually come from.
+    """
+    yield
+    with Session(engine) as session:
+        for model in _ENGINE_TABLES:
+            session.execute(delete(model))
+        session.commit()
+
+
 @pytest.fixture(scope="module")
 def client() -> Generator[TestClient, None, None]:
     with TestClient(app) as c: